Add fluss-codegen module for runtime code generation

fluss-codegen/pom.xml

@@ -0,0 +1,124 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements.  See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership.  The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.fluss</groupId>
+        <artifactId>fluss</artifactId>
+        <version>0.9-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>fluss-codegen</artifactId>
+
+    <name>Fluss : Code Gen</name>
+    <packaging>jar</packaging>
+    <description>
+        Code generation module for Fluss, providing runtime code generation
+        for high-performance record comparison and projection operations.
+    </description>
+
+    <properties>
+        <janino.version>3.1.9</janino.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.fluss</groupId>
+            <artifactId>fluss-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <!-- Janino for runtime Java compilation -->
+        <dependency>
+            <groupId>org.codehaus.janino</groupId>
+            <artifactId>janino</artifactId>
+            <version>${janino.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.codehaus.janino</groupId>
+            <artifactId>commons-compiler</artifactId>
+            <version>${janino.version}</version>
+        </dependency>
+
+        <!-- Test dependencies -->
+        <dependency>
+            <groupId>org.apache.fluss</groupId>
+            <artifactId>fluss-test-utils</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.fluss</groupId>
+            <artifactId>fluss-common</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <!-- Shade Janino to avoid classpath conflicts -->
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>shade-janino</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                        <configuration>
+                            <artifactSet>
+                                <includes>
+                                    <include>org.codehaus.janino:janino</include>
+                                    <include>org.codehaus.janino:commons-compiler</include>
+                                </includes>
+                            </artifactSet>
+                            <relocations>
+                                <relocation>
+                                    <pattern>org.codehaus.janino</pattern>
+                                    <shadedPattern>org.apache.fluss.shaded.org.codehaus.janino</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>org.codehaus.commons</pattern>
+                                    <shadedPattern>org.apache.fluss.shaded.org.codehaus.commons</shadedPattern>
+                                </relocation>
+                            </relocations>
+                            <filters>
+                                <filter>
+                                    <artifact>*:*</artifact>
+                                    <excludes>
+                                        <exclude>META-INF/*.SF</exclude>
+                                        <exclude>META-INF/*.DSA</exclude>
+                                        <exclude>META-INF/*.RSA</exclude>
+                                    </excludes>
+                                </filter>
+                            </filters>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

fluss-codegen/src/main/java/org/apache/fluss/codegen/CodeGenException.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.codegen;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.exception.FlussRuntimeException;
+
+/**
+ * Exception for all errors occurring during code generation.
+ *
+ * <p>This exception is thrown when:
+ *
+ * <ul>
+ *   <li>Generated code fails to compile (syntax errors, missing imports, etc.)
+ *   <li>Unsupported data types are encountered during code generation
+ *   <li>Generated class fails to instantiate
+ * </ul>
+ */
+@Internal
+public class CodeGenException extends FlussRuntimeException {
+
+    private static final long serialVersionUID = 1L;
+
+    public CodeGenException(String message) {
+        super(message);
+    }
+
+    public CodeGenException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

fluss-codegen/src/main/java/org/apache/fluss/codegen/CodeGeneratorContext.java

@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.codegen;
+
+import org.apache.fluss.utils.InstantiationUtils;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * The context for code generator, maintaining various reusable statements that could be inserted
+ * into different code sections in the final generated class.
+ */
+public class CodeGeneratorContext {
+
+    private static final AtomicLong NAME_COUNTER = new AtomicLong(0);
+
+    /** Holding a list of objects that could be passed into generated class. */
+    private final List<Object> references = new ArrayList<>();
+
+    /** Set of member statements that will be added only once. */
+    private final LinkedHashSet<String> reusableMemberStatements = new LinkedHashSet<>();
+
+    /** Set of constructor statements that will be added only once. */
+    private final LinkedHashSet<String> reusableInitStatements = new LinkedHashSet<>();
+
+    public CodeGeneratorContext() {}
+
+    /**
+     * Adds a reusable member field statement to the member area.
+     *
+     * @param memberStatement the member field declare statement
+     */
+    public void addReusableMember(String memberStatement) {
+        reusableMemberStatements.add(memberStatement);
+    }
+
+    /**
+     * Adds a reusable Object to the member area of the generated class. The object must be
+     * Serializable.
+     *
+     * @param obj the object to be added to the generated class (must be Serializable)
+     * @param fieldNamePrefix prefix field name of the generated member field term
+     * @param fieldTypeTerm field type class name
+     * @return the generated unique field term
+     */
+    public <T extends Serializable> String addReusableObject(
+            T obj, String fieldNamePrefix, String fieldTypeTerm) {
+        String fieldTerm = newName(fieldNamePrefix);
+        addReusableObjectInternal(obj, fieldTerm, fieldTypeTerm);
+        return fieldTerm;
+    }
+
+    private <T extends Serializable> void addReusableObjectInternal(
+            T obj, String fieldTerm, String fieldTypeTerm) {
+        int idx = references.size();
+        // make a deep copy of the object
+        try {
+            Object objCopy = InstantiationUtils.clone(obj);
+            references.add(objCopy);
+        } catch (Exception e) {
+            throw new CodeGenException("Failed to clone object: " + obj, e);
+        }
+
+        reusableMemberStatements.add("private transient " + fieldTypeTerm + " " + fieldTerm + ";");
+        reusableInitStatements.add(
+                fieldTerm + " = ((" + fieldTypeTerm + ") references[" + idx + "]);");
+    }
+
+    /** Adds a reusable init statement which will be placed in constructor. */
+    public void addReusableInitStatement(String statement) {
+        reusableInitStatements.add(statement);
+    }
+
+    /**
+     * @return code block of statements that need to be placed in the member area of the class
+     */
+    public String reuseMemberCode() {
+        return String.join("\n", reusableMemberStatements);
+    }
+
+    /**
+     * @return code block of statements that need to be placed in the constructor
+     */
+    public String reuseInitCode() {
+        return String.join("\n", reusableInitStatements);
+    }
+
+    /**
+     * @return the list of reference objects
+     */
+    public Object[] getReferences() {
+        return references.toArray();
+    }
+
+    /**
+     * Generates a new unique name with the given prefix.
+     *
+     * @param name the name prefix
+     * @return a unique name
+     */
+    public static String newName(String name) {
+        return name + "$" + NAME_COUNTER.getAndIncrement();
+    }
+}