CALCITE-6465: Rework code generator to use Flink code splitter

bom/build.gradle.kts

@@ -111,6 +111,7 @@ dependencies {
         apiv("org.apache.commons:commons-pool2")
         apiv("org.apache.commons:commons-collections4")
         apiv("org.apache.commons:commons-text")
+        apiv("org.apache.flink:flink-table-code-splitter")
         apiv("org.apache.geode:geode-core")
         apiv("org.apache.hadoop:hadoop-client", "hadoop")
         apiv("org.apache.hadoop:hadoop-common", "hadoop")

core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableInterpretable.java

@@ -111,7 +111,10 @@ public static Bindable toBindable(Map<String, Object> parameters,
             parameters);
 
     final ClassDeclaration expr = relImplementor.implementRoot(rel, prefer);
-    String s = Expressions.toString(expr.memberDeclarations, "\n", false);
+    String s =
+        Expressions.toString(expr.memberDeclarations,
+            "\n", false,
+            CalciteSystemProperty.METHOD_SPLITTING_THRESHOLD.value());
 
     if (CalciteSystemProperty.DEBUG.value()) {
       Util.debugCode(System.out, s);

core/src/main/java/org/apache/calcite/config/CalciteSystemProperty.java

@@ -420,6 +420,18 @@ public final class CalciteSystemProperty<T> {
   public static final CalciteSystemProperty<Integer> FUNCTION_LEVEL_CACHE_MAX_SIZE =
       intProperty("calcite.function.cache.maxSize", 0, v -> v >= 0);
 
+  /**
+   * The length of code before automatic method splitting is enabled
+   *
+   * <p>A value of <code>0</code>, the default, disables automatic method splitting.
+   *
+   * <p>Some queries can generate methods exceeding the JVM limit of 4000 characters per method.
+   * Use this setting to automatically detect and split methods larger than the specified
+   * limit.
+   */
+  public static final CalciteSystemProperty<Integer> METHOD_SPLITTING_THRESHOLD =
+      intProperty("calcite.linq.method_splitting_threshold", 0);
+
   private static CalciteSystemProperty<Boolean> booleanProperty(String key,
       boolean defaultValue) {
     // Note that "" -> true (convenient for command-lines flags like '-Dflag')

core/src/main/java/org/apache/calcite/interpreter/JaninoRexCompiler.java

@@ -186,7 +186,9 @@ static Scalar.Producer baz(ParameterExpression context_,
     final ClassDeclaration classDeclaration =
         Expressions.classDecl(Modifier.PUBLIC, "Buzz", null,
             ImmutableList.of(Scalar.Producer.class), declarations);
-    String s = Expressions.toString(declarations, "\n", false);
+    String s =
+        Expressions.toString(declarations, "\n", false,
+            CalciteSystemProperty.METHOD_SPLITTING_THRESHOLD.value());
     if (CalciteSystemProperty.DEBUG.value()) {
       Util.debugCode(System.out, s);
     }

core/src/main/java/org/apache/calcite/rex/RexExecutorImpl.java

@@ -100,7 +100,8 @@ private static String compile(RexBuilder rexBuilder, List<RexNode> constExps,
         Expressions.methodDecl(Modifier.PUBLIC, Object[].class,
             BuiltInMethod.FUNCTION1_APPLY.method.getName(),
             ImmutableList.of(root0_), blockBuilder.toBlock());
-    String code = Expressions.toString(methodDecl);
+    String code =
+        Expressions.toString(methodDecl, CalciteSystemProperty.METHOD_SPLITTING_THRESHOLD.value());
     if (CalciteSystemProperty.DEBUG.value()) {
       Util.debugCode(System.out, code);
     }

gradle.properties

@@ -102,6 +102,7 @@ dropwizard-metrics.version=4.0.5
 # do not upgrade this, new versions are Category X license.
 elasticsearch.version=7.10.2
 embedded-redis.version=0.6
+flink-table-code-splitter.version=1.20.0
 jts-core.version=1.19.0
 jts-io-common.version=1.19.0
 proj4j.version=1.2.2

innodb/src/main/java/org/apache/calcite/adapter/innodb/InnodbToEnumerableConverter.java

@@ -150,7 +150,9 @@ static List<String> innodbFieldNames(final RelDataType rowType) {
                 InnodbMethod.INNODB_QUERYABLE_QUERY.method, fields,
                 selectFields, cond, ascOrder));
     if (CalciteSystemProperty.DEBUG.value()) {
-      System.out.println("Innodb: " + Expressions.toString(enumerable));
+      System.out.println(
+          "Innodb: " + Expressions.toString(enumerable,
+          CalciteSystemProperty.METHOD_SPLITTING_THRESHOLD.value()));
     }
     list.add(Expressions.return_(null, enumerable));
     return implementor.result(physType, list.toBlock());

linq4j/build.gradle.kts

@@ -20,4 +20,5 @@ dependencies {
 
     implementation("com.google.guava:guava")
     implementation("org.apache.calcite.avatica:avatica-core")
+    implementation("org.apache.flink:flink-table-code-splitter")
 }

linq4j/src/main/java/org/apache/calcite/linq4j/tree/AbstractNode.java

@@ -51,7 +51,8 @@ public Type getType() {
   }
 
   @Override public String toString() {
-    ExpressionWriter writer = new ExpressionWriter(true);
+    // Use the JVM limit for method size (4K) as the method splitting threshold.
+    ExpressionWriter writer = new ExpressionWriter(true, 4000);
     accept(writer, 0, 0);
     return writer.toString();
   }

linq4j/src/main/java/org/apache/calcite/linq4j/tree/ClassDeclaration.java

@@ -16,6 +16,9 @@
  */
 package org.apache.calcite.linq4j.tree;
 
+import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.table.codesplit.JavaCodeSplitter;
+
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 import java.lang.reflect.Modifier;
@@ -46,19 +49,35 @@ public ClassDeclaration(int modifier, String name, @Nullable Type extended,
   }
 
   @Override public void accept(ExpressionWriter writer) {
+    // Conditionally serialize the class declaration directly to the supplied writer if method
+    // splitting is disabled or to a temporary writer that will generate code for the method that
+    // can be split before being serialized to the supplied writer.
+    final ExpressionWriter writerForClassWithoutSplitting = writer.usesMethodSplitting()
+        ? writer.duplicateState() : writer;
+
     String modifiers = Modifier.toString(modifier);
-    writer.append(modifiers);
+    writerForClassWithoutSplitting.append(modifiers);
     if (!modifiers.isEmpty()) {
-      writer.append(' ');
+      writerForClassWithoutSplitting.append(' ');
     }
-    writer.append(classClass).append(' ').append(name);
+    writerForClassWithoutSplitting.append(classClass).append(' ').append(name);
     if (extended != null) {
-      writer.append(" extends ").append(extended);
+      writerForClassWithoutSplitting.append(" extends ").append(extended);
     }
     if (!implemented.isEmpty()) {
-      writer.list(" implements ", ", ", "", implemented);
+      writerForClassWithoutSplitting.list(" implements ", ", ", "", implemented);
+    }
+    writerForClassWithoutSplitting.list(" {\n", "", "}", memberDeclarations);
+
+    if (writer.usesMethodSplitting()) {
+      final int defaultMaxMembersGeneratedCode = 10000;
+
+      writer.append(
+          StringUtils.stripStart(
+              JavaCodeSplitter.split(writerForClassWithoutSplitting.toString(),
+                  writer.getMethodSplittingThreshold(), defaultMaxMembersGeneratedCode),
+              " "));
     }
-    writer.list(" {\n", "", "}", memberDeclarations);
     writer.newlineAndIndent();
   }
 

linq4j/src/main/java/org/apache/calcite/linq4j/tree/ExpressionWriter.java

@@ -32,15 +32,25 @@ class ExpressionWriter {
 
   private final Spacer spacer = new Spacer(0);
   private final StringBuilder buf = new StringBuilder();
-  private boolean indentPending;
   private final boolean generics;
+  private final int methodSplittingThreshold;
+  private boolean indentPending;
 
   ExpressionWriter() {
-    this(true);
+    this(true, 0);
   }
 
-  ExpressionWriter(boolean generics) {
+  ExpressionWriter(boolean generics, int methodSplittingThreshold) {
     this.generics = generics;
+    this.methodSplittingThreshold = methodSplittingThreshold;
+  }
+
+  public ExpressionWriter duplicateState() {
+    final ExpressionWriter writer =
+        new ExpressionWriter(this.generics, this.methodSplittingThreshold);
+    writer.indentPending = this.indentPending;
+    writer.spacer.add(this.spacer.get());
+    return writer;
   }
 
   public void write(Node expression) {
@@ -73,6 +83,14 @@ public boolean requireParentheses(Expression expression, int lprec,
     return true;
   }
 
+  public boolean usesMethodSplitting() {
+    return methodSplittingThreshold != 0;
+  }
+
+  public int getMethodSplittingThreshold() {
+    return methodSplittingThreshold;
+  }
+
   /**
    * Increases the indentation level.
    */

linq4j/src/main/java/org/apache/calcite/linq4j/tree/Expressions.java

@@ -59,7 +59,17 @@ private Expressions() {}
    */
   public static String toString(List<? extends Node> expressions, String sep,
       boolean generics) {
-    final ExpressionWriter writer = new ExpressionWriter(generics);
+    return toString(expressions, sep, generics, 0);
+  }
+
+  /**
+   * Converts a list of expressions to Java source code, optionally emitting
+   * extra type information in generics. Optionally splits the generated method if
+   * the method is too large.
+   */
+  public static String toString(List<? extends Node> expressions, String sep,
+      boolean generics, int methodSplittingThreshold) {
+    final ExpressionWriter writer = new ExpressionWriter(generics, methodSplittingThreshold);
     for (Node expression : expressions) {
       writer.write(expression);
       writer.append(sep);
@@ -71,7 +81,15 @@ public static String toString(List<? extends Node> expressions, String sep,
    * Converts an expression to Java source code.
    */
   public static String toString(Node expression) {
-    return toString(Collections.singletonList(expression), "", true);
+    return toString(expression, 0);
+  }
+
+  /**
+   * Converts an expression to Java source code.
+   * Optionally splits the generated method if the method is too large.
+   */
+  public static String toString(Node expression, int methodSplittingTheshold) {
+    return toString(Collections.singletonList(expression), "", true, methodSplittingTheshold);
   }
 
   /**

linq4j/src/test/java/org/apache/calcite/linq4j/test/ExpressionTest.java

@@ -21,24 +21,29 @@
 import org.apache.calcite.linq4j.tree.BlockStatement;
 import org.apache.calcite.linq4j.tree.Blocks;
 import org.apache.calcite.linq4j.tree.ClassDeclaration;
+import org.apache.calcite.linq4j.tree.ConditionalStatement;
 import org.apache.calcite.linq4j.tree.DeclarationStatement;
 import org.apache.calcite.linq4j.tree.Expression;
 import org.apache.calcite.linq4j.tree.Expressions;
 import org.apache.calcite.linq4j.tree.FieldDeclaration;
 import org.apache.calcite.linq4j.tree.FunctionExpression;
 import org.apache.calcite.linq4j.tree.MethodCallExpression;
+import org.apache.calcite.linq4j.tree.MethodDeclaration;
 import org.apache.calcite.linq4j.tree.NewExpression;
 import org.apache.calcite.linq4j.tree.Node;
 import org.apache.calcite.linq4j.tree.ParameterExpression;
 import org.apache.calcite.linq4j.tree.Shuttle;
 import org.apache.calcite.linq4j.tree.Types;
 
+import org.apache.commons.lang3.StringUtils;
+
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets;
 
 import org.checkerframework.checker.nullness.qual.Nullable;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
@@ -1719,6 +1724,60 @@ public void checkBlockBuilder(boolean optimizing, String expected) {
         Expressions.toString(Expressions.constant(set)));
   }
 
+  @Test void testLargeMethod() {
+//     public class MyClass {
+//       public void hugeMethod() {
+//         if (true) {
+//           // Long variable name declaration 1;
+//         } else if (false) {
+//           // Long variable name declaration 2;
+//         } else if (false) {
+//           // Long variable name declaration 3;
+//         } else if (false) {
+//           // Long variable name declaration 4;
+//         } else
+//           // Long variable name declaration 4;
+//         }
+//       }
+
+    // Generate long variable names.
+    final String longName = StringUtils.repeat('a', 5000);
+
+    DeclarationStatement longDecl =
+        Expressions.declare(0, longName, Expressions.constant(10));
+    final ConditionalStatement ifThenElse =
+        Expressions.ifThenElse(
+            Expressions.constant(true),
+            longDecl,
+            Expressions.constant(false),
+            longDecl,
+            Expressions.constant(false),
+            longDecl,
+            longDecl);
+
+    final MethodDeclaration hugeMethod =
+        Expressions.methodDecl(Modifier.PUBLIC,
+        Void.TYPE,
+        "hugeMethod",
+        Collections.emptyList(),
+        Blocks.toFunctionBlock(ifThenElse));
+
+    final ClassDeclaration classDecl =
+        new ClassDeclaration(Modifier.PUBLIC,
+        "MyClass",
+        null,
+        ImmutableList.of(),
+        ImmutableList.of(hugeMethod));
+
+    final String originalClass = Expressions.toString(classDecl);
+    final String classWithMethodSplitting = Expressions.toString(classDecl, 4000);
+
+    // Some local variables in hugeMethod have been extracted to be fields named local$<number> in
+    // the enclosing class definition.
+    Assertions.assertFalse(originalClass.contains("local$"));
+    Assertions.assertTrue(classWithMethodSplitting.contains("local$"));
+  }
+
   /** An enum. */
   enum MyEnum {
     X,

ubenchmark/src/jmh/java/org/apache/calcite/adapter/enumerable/CodeGenerationBenchmark.java

@@ -184,8 +184,11 @@ public void setup() {
         EnumerableRelImplementor relImplementor =
             new EnumerableRelImplementor(plan.getCluster().getRexBuilder(), new HashMap<>());
         info.classExpr = relImplementor.implementRoot(plan, EnumerableRel.Prefer.ARRAY);
+
+        // Set the method splitting threshold to the JVM limit of 4K.
         info.javaCode =
-            Expressions.toString(info.classExpr.memberDeclarations, "\n", false);
+            Expressions.toString(info.classExpr.memberDeclarations,
+                "\n", false, 4000);
         info.plan = plan;
         planInfos[i] = info;
       }