feat: SAP changes

Author: kramerul

build.sh

@@ -0,0 +1,52 @@
+#!/bin/bash
+#
+# 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.
+#
+set -e
+VERSION=1.41.0
+
+# Java 21 doesn't suppport Java 8
+#if [ -d /Library/Java/JavaVirtualMachines/sapmachine-jdk-17.0.11.jdk/Contents/Home ]; then
+#  export JAVA_HOME=/Library/Java/JavaVirtualMachines/sapmachine-jdk-17.0.11.jdk/Contents/Home
+#  export PATH=$JAVA_HOME/bin:$PATH
+#fi
+
+#if [ -d /Library/Java/JavaVirtualMachines/sapmachine-17.jdk/Contents/Home ]; then
+#  export JAVA_HOME=/Library/Java/JavaVirtualMachines/sapmachine-17.jdk/Contents/Home
+#  export PATH=$JAVA_HOME/bin:$PATH
+#fi
+
+for module in linq4j core;
+do
+  ./gradlew :${module}:clean -x :core:javadoc :${module}:publishToMavenLocal
+   mvn install:install-file \
+      -Dfile=${module}/build/libs/calcite-${module}-$VERSION-SNAPSHOT.jar \
+      -DgroupId=org.apache.calcite \
+      -DartifactId=calcite-${module}-sap \
+      -Dversion=$VERSION \
+      -Dpackaging=jar \
+      -DlocalRepositoryPath=../nucleus-resources-app/patches/
+   mvn install:install-file \
+      -Dfile=${module}/build/libs/calcite-${module}-$VERSION-SNAPSHOT.jar \
+      -DgroupId=org.apache.calcite \
+      -DartifactId=calcite-${module}-sap \
+      -Dversion=$VERSION \
+      -Dpackaging=jar \
+      -DlocalRepositoryPath=$HOME/.m2/repository
+   cat ${module}/build/publications/${module}/pom-default.xml | \
+   sed -e "s/calcite-${module}/calcite-${module}-sap/g" -e "s/${VERSION}-SNAPSHOT/${VERSION}/g" > \
+   ../nucleus-resources-app/patches/org/apache/calcite/calcite-${module}-sap/$VERSION/calcite-${module}-sap-$VERSION.pom
+done

core/src/main/java/org/apache/calcite/DataContext.java

@@ -69,6 +69,19 @@ public interface DataContext {
    */
   @Nullable Object get(String name);
 
+  /**
+   * Returns a context variable.
+   *
+   * <p>Supported variables include: "sparkContext", "currentTimestamp",
+   * "localTimestamp".</p>
+   *
+   * @param name Name of variable
+   * @param clazz Type of variable
+   */
+  @Nullable default <T> T get(String name, Class<T> clazz) {
+    return (T) get(name);
+  }
+
   /** Variable that may be asked for in a call to {@link DataContext#get}. */
   enum Variable {
     UTC_TIMESTAMP("utcTimestamp", Long.class),

core/src/main/java/org/apache/calcite/adapter/clone/CloneSchema.java

@@ -26,6 +26,8 @@
 import org.apache.calcite.rel.RelCollation;
 import org.apache.calcite.rel.RelCollations;
 import org.apache.calcite.rel.type.RelProtoDataType;
+import org.apache.calcite.schema.lookup.LikePattern;
+import org.apache.calcite.schema.lookup.Lookup;
 import org.apache.calcite.schema.QueryableTable;
 import org.apache.calcite.schema.Schema;
 import org.apache.calcite.schema.SchemaFactory;

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

@@ -55,6 +55,7 @@
 public class EnumerableBatchNestedLoopJoin extends Join implements EnumerableRel {
 
   private final ImmutableBitSet requiredColumns;
+  private final @Nullable Double originRowCount;
   protected EnumerableBatchNestedLoopJoin(
       RelOptCluster cluster,
       RelTraitSet traits,
@@ -63,18 +64,32 @@ protected EnumerableBatchNestedLoopJoin(
       RexNode condition,
       Set<CorrelationId> variablesSet,
       ImmutableBitSet requiredColumns,
-      JoinRelType joinType) {
+      JoinRelType joinType,
+      double originRowCount) {
     super(cluster, traits, ImmutableList.of(), left, right, condition, variablesSet, joinType);
     this.requiredColumns = requiredColumns;
+    this.originRowCount = originRowCount;
   }
 
+  @Deprecated
   public static EnumerableBatchNestedLoopJoin create(
       RelNode left,
       RelNode right,
       RexNode condition,
       ImmutableBitSet requiredColumns,
       Set<CorrelationId> variablesSet,
       JoinRelType joinType) {
+    return create(left, right, condition, requiredColumns, variablesSet, joinType, null);
+  }
+
+  public static EnumerableBatchNestedLoopJoin create(
+      RelNode left,
+      RelNode right,
+      RexNode condition,
+      ImmutableBitSet requiredColumns,
+      Set<CorrelationId> variablesSet,
+      JoinRelType joinType,
+      @Nullable Double originRowCount) {
     final RelOptCluster cluster = left.getCluster();
     final RelMetadataQuery mq = cluster.getMetadataQuery();
     final RelTraitSet traitSet =
@@ -89,9 +104,11 @@ public static EnumerableBatchNestedLoopJoin create(
         condition,
         variablesSet,
         requiredColumns,
-        joinType);
+        joinType,
+        originRowCount);
   }
 
+
   @Override public @Nullable Pair<RelTraitSet, List<RelTraitSet>> passThroughTraits(
       final RelTraitSet required) {
     return EnumerableTraitsUtils.passThroughTraitsForJoin(
@@ -115,8 +132,8 @@ public static EnumerableBatchNestedLoopJoin create(
   @Override public EnumerableBatchNestedLoopJoin copy(RelTraitSet traitSet,
       RexNode condition, RelNode left, RelNode right, JoinRelType joinType,
       boolean semiJoinDone) {
-    return new EnumerableBatchNestedLoopJoin(getCluster(), traitSet,
-        left, right, condition, variablesSet, requiredColumns, joinType);
+    return new EnumerableBatchNestedLoopJoin(getCluster(), traitSet, left, right, condition,
+        variablesSet, requiredColumns, joinType, originRowCount);
   }
 
   @Override public @Nullable RelOptCost computeSelfCost(
@@ -149,6 +166,10 @@ public static EnumerableBatchNestedLoopJoin create(
     return pw.item("batchSize", variablesSet.size());
   }
 
+  public Double getOriginRowCount() {
+    return originRowCount;
+  }
+
   @Override public Result implement(EnumerableRelImplementor implementor, Prefer pref) {
     final BlockBuilder builder = new BlockBuilder();
     final Result leftResult =

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

@@ -16,14 +16,21 @@
  */
 package org.apache.calcite.adapter.enumerable;
 
+import java.util.function.Predicate;
+
+import java.util.function.Supplier;
+
+import org.apache.calcite.adapter.jdbc.JdbcConvention;
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelOptRuleCall;
 import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.RelTrait;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.CorrelationId;
 import org.apache.calcite.rel.core.Join;
 import org.apache.calcite.rel.core.JoinRelType;
 import org.apache.calcite.rel.logical.LogicalJoin;
+import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rex.RexBuilder;
 import org.apache.calcite.rex.RexCorrelVariable;
 import org.apache.calcite.rex.RexInputRef;
@@ -32,6 +39,7 @@
 import org.apache.calcite.tools.RelBuilder;
 import org.apache.calcite.tools.RelBuilderFactory;
 import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.Util;
 
 import org.immutables.value.Value;
 
@@ -79,10 +87,45 @@ public EnumerableBatchNestedLoopJoinRule(RelBuilderFactory relBuilderFactory,
   @Override public boolean matches(RelOptRuleCall call) {
     Join join = call.rel(0);
     JoinRelType joinType = join.getJoinType();
-    return joinType == JoinRelType.INNER
+    return (joinType == JoinRelType.INNER
         || joinType == JoinRelType.LEFT
         || joinType == JoinRelType.ANTI
-        || joinType == JoinRelType.SEMI;
+        || joinType == JoinRelType.SEMI )
+           && !allLeafsMatch(join, new HasSingleJdbcSource());
+  }
+
+  private static class HasSingleJdbcSource implements Predicate<RelNode> {
+    private JdbcConvention convention = null;
+    @Override
+    public boolean test(RelNode node) {
+      for (RelTrait trait : node.getTraitSet()) {
+        if (trait instanceof JdbcConvention) {
+          JdbcConvention otherConvention = (JdbcConvention) trait;
+          // The first leaf in the tree sets the convention
+          if (convention == null) {
+            convention = otherConvention;
+            return true;
+          }
+          // All other leafs must match the stored convention
+          return convention == otherConvention;
+        }
+      }
+      return false;
+    }
+  }
+
+  private boolean allLeafsMatch(RelNode node, HasSingleJdbcSource predicate) {
+      List<RelNode> inputs = node.getInputs();
+      if ( inputs.isEmpty()) {
+        return predicate.test(node);
+      } else {
+        for ( RelNode input : inputs) {
+          if ( !allLeafsMatch(input.stripped(), predicate)) {
+            return false;
+          }
+        }
+      }
+      return true;
   }
 
   @Override public void onMatch(RelOptRuleCall call) {
@@ -93,17 +136,15 @@ public EnumerableBatchNestedLoopJoinRule(RelBuilderFactory relBuilderFactory,
     final RelBuilder relBuilder = call.builder();
 
     final Set<CorrelationId> correlationIds = new HashSet<>();
-    final List<RexNode> corrVarList = new ArrayList<>();
 
-    final int batchSize = config.batchSize();
-    for (int i = 0; i < batchSize; i++) {
+    final RelDataType leftRowType = join.getLeft().getRowType();
+    int batchSize = config.batchSize();
+    Supplier<RexNode> corrVarSupplier = () -> {
       CorrelationId correlationId = cluster.createCorrel();
       correlationIds.add(correlationId);
-      corrVarList.add(
-          rexBuilder.makeCorrel(join.getLeft().getRowType(),
-              correlationId));
-    }
-    final RexNode corrVar0 = corrVarList.get(0);
+      return rexBuilder.makeCorrel(leftRowType,correlationId);
+    };
+    final RexNode corrVar0 = corrVarSupplier.get();
 
     final ImmutableBitSet.Builder requiredColumns = ImmutableBitSet.builder();
 
@@ -123,12 +164,14 @@ public EnumerableBatchNestedLoopJoinRule(RelBuilderFactory relBuilderFactory,
     final List<RexNode> conditionList = new ArrayList<>();
     conditionList.add(condition);
 
+    if ( requiredColumns.cardinality() > 2) {
+      batchSize = batchSize * 2 / requiredColumns.cardinality();
+    }
     // Add batchSize-1 other conditions
     for (int i = 1; i < batchSize; i++) {
-      final int corrIndex = i;
       final RexNode condition2 = condition.accept(new RexShuttle() {
         @Override public RexNode visitCorrelVariable(RexCorrelVariable variable) {
-          return variable.equals(corrVar0) ? corrVarList.get(corrIndex) : variable;
+          return variable.equals(corrVar0) ? corrVarSupplier.get() : variable;
         }
       });
       conditionList.add(condition2);
@@ -137,6 +180,8 @@ public EnumerableBatchNestedLoopJoinRule(RelBuilderFactory relBuilderFactory,
     // Push a filter with batchSize disjunctions
     relBuilder.push(join.getRight()).filter(relBuilder.or(conditionList));
     final RelNode right = relBuilder.build();
+    final double originRowCount =
+        Util.first(call.getMetadataQuery().getRowCount(join), Double.MAX_VALUE);
 
     call.transformTo(
         EnumerableBatchNestedLoopJoin.create(
@@ -147,7 +192,8 @@ public EnumerableBatchNestedLoopJoinRule(RelBuilderFactory relBuilderFactory,
             join.getCondition(),
             requiredColumns.build(),
             correlationIds,
-            join.getJoinType()));
+            join.getJoinType(),
+            originRowCount));
   }
 
   /** Rule configuration. */

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

@@ -121,11 +121,10 @@ public static EnumerableLimit create(final RelNode input, @Nullable RexNode offs
   static Expression getExpression(RexNode rexNode) {
     if (rexNode instanceof RexDynamicParam) {
       final RexDynamicParam param = (RexDynamicParam) rexNode;
-      return Expressions.convert_(
-          Expressions.call(DataContext.ROOT,
-              BuiltInMethod.DATA_CONTEXT_GET.method,
-              Expressions.constant("?" + param.getIndex())),
-          Integer.class);
+      return Expressions.call(DataContext.ROOT,
+              BuiltInMethod.DATA_CONTEXT_GET_TYPED.method,
+              Expressions.constant("?" + param.getIndex()),
+          Expressions.constant(Integer.class));
     } else {
       // TODO: Enumerable runtime only supports INT types for FETCH and OFFSET, not BIGINT types.
       //  Currently, using BIGINT types for execution will result in an error message.

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

@@ -438,7 +438,7 @@ public static EnumerableMergeJoin create(RelNode left, RelNode right,
     BlockBuilder builder = new BlockBuilder();
     final Result leftResult =
         implementor.visitChild(this, 0, (EnumerableRel) left, pref);
-    final Expression leftExpression =
+    Expression leftExpression =
         builder.append("left", leftResult.block);
     final ParameterExpression left_ =
         Expressions.parameter(leftResult.physType.getJavaRowType(), "left");
@@ -511,15 +511,21 @@ public static EnumerableMergeJoin create(RelNode left, RelNode right,
         PhysTypeImpl.of(typeFactory, comparatorRowType, JavaRowFormat.LIST);
     final RelCollation collation = RelCollations.of(fieldCollations);
     final Expression comparator = comparatorPhysType.generateMergeJoinComparator(collation);
-
+    if (joinType == JoinRelType.INNER || joinType == JoinRelType.SEMI) {
+     leftExpression = Expressions.call(BuiltInMethod.MERGE_JOIN_NOT_NULL_ENUMERABLE.method, leftExpression,
+          Expressions.lambda(
+              leftKeyPhysType.record(leftExpressions), left_));
+    }
     return implementor.result(
         physType,
         builder.append(
             Expressions.call(
                 BuiltInMethod.MERGE_JOIN.method,
                 Expressions.list(
                     leftExpression,
-                    rightExpression,
+                    Expressions.call(BuiltInMethod.MERGE_JOIN_NOT_NULL_ENUMERABLE.method, rightExpression,
+                        Expressions.lambda(
+                            rightKeyPhysType.record(rightExpressions), right_)),
                     Expressions.lambda(
                         leftKeyPhysType.record(leftExpressions), left_),
                     Expressions.lambda(

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

@@ -22,6 +22,7 @@
 import org.apache.calcite.linq4j.tree.Expressions;
 import org.apache.calcite.linq4j.tree.ParameterExpression;
 import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTrait;
 import org.apache.calcite.plan.RelTraitSet;
 import org.apache.calcite.rel.RelCollation;
 import org.apache.calcite.rel.RelNode;
@@ -47,7 +48,7 @@ protected EnumerableMergeUnion(RelOptCluster cluster, RelTraitSet traitSet,
       throw new IllegalArgumentException("EnumerableMergeUnion with no collation");
     }
     for (RelNode input : inputs) {
-      final RelCollation inputCollation = input.getTraitSet().getCollation();
+      final RelTrait inputCollation = input.getTraitSet().getCollation();
       if (inputCollation == null || !inputCollation.satisfies(collation)) {
         throw new IllegalArgumentException("EnumerableMergeUnion input does "
             + "not satisfy collation. EnumerableMergeUnion collation: "

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

@@ -157,11 +157,10 @@ public ClassDeclaration implementRoot(EnumerableRel rootRel,
     final Collection<Statement> stashed =
         Collections2.transform(stashedParameters.values(),
             input -> Expressions.declare(Modifier.FINAL, input,
-                Expressions.convert_(
-                    Expressions.call(DataContext.ROOT,
-                        BuiltInMethod.DATA_CONTEXT_GET.method,
-                        Expressions.constant(input.name)),
-                    input.type)));
+                Expressions.call(DataContext.ROOT,
+                    BuiltInMethod.DATA_CONTEXT_GET_TYPED.method,
+                    Expressions.constant(input.name),
+                    Expressions.constant(input.type))));
 
     final BlockStatement block =
         Expressions.block(

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

@@ -37,6 +37,7 @@
 import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.stream.Collectors;
 
 import static org.apache.calcite.util.BuiltInMethod.ARRAY_COPY;
 import static org.apache.calcite.util.BuiltInMethod.LIST_TO_ARRAY;
@@ -218,7 +219,9 @@ public enum JavaRowFormat {
                 BuiltInMethod.LIST_N.method,
                 Expressions.newArrayInit(
                     Comparable.class,
-                    expressions)),
+                    expressions.stream()
+                        .map(it -> it.type == Object.class ? Expressions.convert_(it,Comparable.class) : it)
+                        .collect(Collectors.toList()))),
             List.class);
       }
     }