[CALCITE-7592] Merge main branch

Author: tkalkirill

build.gradle.kts

@@ -26,6 +26,7 @@ import com.github.vlsi.gradle.release.RepositoryType
 import de.thetaphi.forbiddenapis.gradle.CheckForbiddenApis
 import de.thetaphi.forbiddenapis.gradle.CheckForbiddenApisExtension
 import net.ltgt.gradle.errorprone.errorprone
+import org.apache.calcite.buildtools.asmchecker.AsmCheckerTask
 import org.apache.calcite.buildtools.buildext.dsl.ParenthesisBalancer
 import org.gradle.api.tasks.testing.logging.TestExceptionFormat
 
@@ -37,6 +38,7 @@ plugins {
     publishing
     // Verification
     checkstyle
+    id("calcite.asmchecker")
     calcite.buildext
     jacoco
     id("jacoco-report-aggregation")
@@ -930,6 +932,15 @@ allprojects {
                 }
                 jvmArgs("-Xmx6g")
             }
+            register<AsmCheckerTask>("bytecodeCheck") {
+                group = LifecycleBasePlugin.VERIFICATION_GROUP
+                description = "Checks the bytecode of every .class file in the build directory using ASM."
+                dependsOn("classes")
+            }
+
+            check {
+                dependsOn("bytecodeCheck")
+            }
             hepLargePlanModeTestIncludes[project.path]?.let { includes ->
                 val hepLargePlanModeTask = register<Test>("testHepLargePlanMode") {
                     group = LifecycleBasePlugin.VERIFICATION_GROUP

buildSrc/gradle.properties

@@ -20,3 +20,6 @@ kotlin.code.style=official
 # Plugins
 com.github.autostyle.version=3.2
 com.github.vlsi.vlsi-release-plugins.version=1.52
+
+# Dependencies (keep in sync with root gradle.properties)
+asm.version=9.9.1

buildSrc/settings.gradle.kts

@@ -24,6 +24,7 @@ pluginManagement {
     }
 }
 
+include("asmchecker")
 include("javacc")
 include("fmpp")
 include("buildext")

buildSrc/subprojects/asmchecker/asmchecker.gradle.kts

@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+
+dependencies {
+    val asmVersion = providers.gradleProperty("asm.version").get()
+    implementation("org.ow2.asm:asm:$asmVersion")
+    implementation("org.ow2.asm:asm-analysis:$asmVersion")
+    implementation("org.ow2.asm:asm-commons:$asmVersion")
+    implementation("org.ow2.asm:asm-tree:$asmVersion")
+    implementation("org.ow2.asm:asm-util:$asmVersion")
+}
+
+gradlePlugin {
+    plugins {
+        register("asmchecker") {
+            id = "calcite.asmchecker"
+            implementationClass = "org.apache.calcite.buildtools.asmchecker.AsmCheckerPlugin"
+        }
+    }
+}

buildSrc/subprojects/asmchecker/src/main/kotlin/org/apache/calcite/buildtools/asmchecker/AsmCheckerPlugin.kt

@@ -0,0 +1,26 @@
+/*
+ * 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.calcite.buildtools.asmchecker
+
+import org.gradle.api.Plugin
+import org.gradle.api.Project
+
+open class AsmCheckerPlugin : Plugin<Project> {
+    override fun apply(target: Project) {
+    }
+}

buildSrc/subprojects/asmchecker/src/main/kotlin/org/apache/calcite/buildtools/asmchecker/AsmCheckerTask.kt

@@ -0,0 +1,53 @@
+/*
+ * 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.calcite.buildtools.asmchecker
+
+import java.nio.file.Files
+import java.nio.file.Paths
+import org.gradle.api.DefaultTask
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.TaskAction
+import org.objectweb.asm.ClassReader
+import org.objectweb.asm.ClassWriter
+import org.objectweb.asm.commons.ClassRemapper
+import org.objectweb.asm.commons.Remapper
+import org.objectweb.asm.util.CheckClassAdapter
+
+@CacheableTask
+open class AsmCheckerTask : DefaultTask() {
+
+    @TaskAction
+    fun run() {
+        project.layout.buildDirectory.get().asFile.walk()
+            .onEnter { dir ->
+              // the classes in spark/build/sparkServer/classes, generated by SparkHandlerImpl,
+              // have invalid bytecode, so exclude them
+              "sparkServer${java.io.File.separator}classes" !in dir.path }
+            .filter { file -> file.getName().lowercase().endsWith(".class") }
+            .forEach {
+                val classReader = ClassReader(Files.readAllBytes(Paths.get(it.getPath())))
+                val classVisitor = CheckClassAdapter(ClassWriter(ClassWriter.COMPUTE_MAXS))
+                val classRemapper = ClassRemapper(classVisitor, object : Remapper() {})
+                try {
+                    classReader.accept(classRemapper, ClassReader.EXPAND_FRAMES)
+                } catch (e: java.lang.RuntimeException) {
+                    throw java.lang.RuntimeException("Invalid bytecode file:" + it, e)
+                }
+            }
+    }
+}

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

@@ -30,6 +30,7 @@
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;
 import org.apache.calcite.rel.RelCollations;
+import org.apache.calcite.rel.RelFieldCollation;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.Aggregate;
 import org.apache.calcite.rel.core.AggregateCall;
@@ -264,6 +265,20 @@ protected void createAccumulatorAdders(
               for (int index : agg.call.getArgList()) {
                 args.add(RexInputRef.of(index, inputTypes));
               }
+              // Percentile functions such as PERCENTILE_CONT and
+              // PERCENTILE_DISC take the fraction as their only argument, but
+              // aggregate over the WITHIN GROUP (ORDER BY ...) column. Expose
+              // that collation column as an extra argument so that the
+              // accumulator can collect its values (already sorted by
+              // SourceSorter).
+              if (agg.call.getAggregation().isPercentile()) {
+                for (RelFieldCollation fieldCollation
+                    : agg.call.collation.getFieldCollations()) {
+                  args.add(
+                      RexInputRef.of(fieldCollation.getFieldIndex(),
+                          inputTypes));
+                }
+              }
               return args;
             }
 

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

@@ -499,6 +499,8 @@
 import static org.apache.calcite.sql.fun.SqlStdOperatorTable.OCTET_LENGTH;
 import static org.apache.calcite.sql.fun.SqlStdOperatorTable.OR;
 import static org.apache.calcite.sql.fun.SqlStdOperatorTable.OVERLAY;
+import static org.apache.calcite.sql.fun.SqlStdOperatorTable.PERCENTILE_CONT;
+import static org.apache.calcite.sql.fun.SqlStdOperatorTable.PERCENTILE_DISC;
 import static org.apache.calcite.sql.fun.SqlStdOperatorTable.PI;
 import static org.apache.calcite.sql.fun.SqlStdOperatorTable.PLUS;
 import static org.apache.calcite.sql.fun.SqlStdOperatorTable.POSITION;
@@ -1326,6 +1328,8 @@ void populate3() {
       defineAgg(SINGLE_VALUE, SingleValueImplementor.class);
       defineAgg(COLLECT, CollectImplementor.class);
       defineAgg(ARRAY_AGG, CollectImplementor.class);
+      defineAgg(PERCENTILE_CONT, PercentileImplementor.class);
+      defineAgg(PERCENTILE_DISC, PercentileImplementor.class);
       defineAgg(LISTAGG, ListaggImplementor.class);
       defineAgg(FUSION, FusionImplementor.class);
       defineAgg(MODE, ModeImplementor.class);
@@ -1967,6 +1971,60 @@ static class CollectImplementor extends StrictAggImplementor {
     }
   }
 
+  /** Implementor for the {@code PERCENTILE_CONT} and {@code PERCENTILE_DISC}
+   * aggregate functions.
+   *
+   * <p>The fraction is the sole argument of the aggregate call, while the
+   * values whose percentile is computed come from the
+   * {@code WITHIN GROUP (ORDER BY ...)} column, which
+   * {@link EnumerableAggregateBase#createAccumulatorAdders} exposes as an extra
+   * argument. The input rows are sorted by {@code SourceSorter} before being
+   * accumulated, so the collected values are already in order. */
+  static class PercentileImplementor extends StrictAggImplementor {
+    @Override public List<Type> getNotNullState(AggContext info) {
+      final List<Type> types = new ArrayList<>();
+      types.add(List.class);
+      types.add(double.class);
+      return types;
+    }
+
+    @Override protected void implementNotNullReset(AggContext info,
+        AggResetContext reset) {
+      reset.currentBlock().add(
+          Expressions.statement(
+              Expressions.assign(reset.accumulator().get(0),
+                  Expressions.new_(ArrayList.class))));
+      reset.currentBlock().add(
+          Expressions.statement(
+              Expressions.assign(reset.accumulator().get(1),
+                  Expressions.constant(0d))));
+    }
+
+    @Override protected void implementNotNullAdd(AggContext info,
+        AggAddContext add) {
+      add.currentBlock().add(
+          Expressions.statement(
+              Expressions.assign(add.accumulator().get(1),
+                  EnumUtils.convert(add.arguments().get(0), double.class))));
+
+      add.currentBlock().add(
+          Expressions.statement(
+              Expressions.call(add.accumulator().get(0),
+                  BuiltInMethod.COLLECTION_ADD.method,
+                  Expressions.box(add.arguments().get(1)))));
+    }
+
+    @Override protected Expression implementNotNullResult(AggContext info,
+        AggResultContext result) {
+      final BuiltInMethod method =
+          info.aggregation().kind == SqlKind.PERCENTILE_DISC
+              ? BuiltInMethod.PERCENTILE_DISC
+              : BuiltInMethod.PERCENTILE_CONT;
+      return Expressions.call(method.method, result.accumulator().get(0),
+          result.accumulator().get(1));
+    }
+  }
+
   /** Implementor for the {@code LISTAGG} aggregate function. */
   static class ListaggImplementor extends StrictAggImplementor {
     @Override protected void implementNotNullReset(AggContext info,

core/src/main/java/org/apache/calcite/rel/core/TableModify.java

@@ -32,7 +32,9 @@
 import org.apache.calcite.rel.metadata.RelMetadataQuery;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeField;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.type.SqlTypeUtil;
 
@@ -261,9 +263,37 @@ public boolean isMerge() {
               null);
     }
 
+    inputRowType = expectedInputRowTypeForAssignment(inputRowType);
     return inputRowType;
   }
 
+  private RelDataType expectedInputRowTypeForAssignment(
+      RelDataType expectedRowType) {
+    final RelDataType actualRowType = getInput().getRowType();
+    if (actualRowType.getFieldCount() != expectedRowType.getFieldCount()) {
+      return expectedRowType;
+    }
+    final RelDataTypeFactory typeFactory = getCluster().getTypeFactory();
+    final RelDataTypeFactory.Builder builder = typeFactory.builder();
+    boolean changed = false;
+    final List<RelDataTypeField> expectedFields = expectedRowType.getFieldList();
+    final List<RelDataTypeField> actualFields = actualRowType.getFieldList();
+    for (int i = 0; i < expectedFields.size(); i++) {
+      final RelDataTypeField expectedField = expectedFields.get(i);
+      final RelDataType actualType = actualFields.get(i).getType();
+      final RelDataType expectedType = expectedField.getType();
+      if (!SqlTypeUtil.equalSansNullability(typeFactory, actualType, expectedType)
+          && SqlTypeUtil.canAssignFrom(expectedType, actualType)
+          && !RexUtil.isLosslessCast(actualType, expectedType)) {
+        builder.add(expectedField.getName(), actualType);
+        changed = true;
+      } else {
+        builder.add(expectedField);
+      }
+    }
+    return changed ? builder.build() : expectedRowType;
+  }
+
   @Override public RelWriter explainTerms(RelWriter pw) {
     return super.explainTerms(pw)
         .item("table", table.getQualifiedName())

core/src/main/java/org/apache/calcite/runtime/SqlFunctions.java

@@ -133,6 +133,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.NoSuchElementException;
 import java.util.Objects;
 import java.util.Set;
 import java.util.TimeZone;
@@ -4069,6 +4070,55 @@ public static BigDecimal mod(BigDecimal b0, BigDecimal b1) {
     return bigDecimals[1];
   }
 
+  // PERCENTILE_CONT / PERCENTILE_DISC
+
+  /** Support the PERCENTILE_CONT aggregate function.
+   *
+   * <p>The {@code values} list must already be sorted according to the
+   * {@code WITHIN GROUP (ORDER BY ...)} clause. The fraction must be in the
+   * range 0 to 1 inclusive. The result is a linear interpolation between the
+   * two values that surround the desired position. */
+  public static BigDecimal percentileCont(List<? extends Number> values,
+      double fraction) {
+    final int n = values.size();
+    if (n == 0) {
+      throw new NoSuchElementException(
+          "PERCENTILE_CONT is not defined on an empty group");
+    }
+    final double rank = fraction * (n - 1);
+    final int lo = (int) Math.floor(rank);
+    final int hi = (int) Math.ceil(rank);
+    final BigDecimal loValue = toBigDecimal(values.get(lo));
+    if (lo == hi) {
+      return loValue;
+    }
+    final BigDecimal hiValue = toBigDecimal(values.get(hi));
+    final BigDecimal frac = BigDecimal.valueOf(rank - lo);
+    return loValue.add(hiValue.subtract(loValue).multiply(frac));
+  }
+
+  /** Support the PERCENTILE_DISC aggregate function.
+   *
+   * <p>The {@code values} list must already be sorted according to the
+   * {@code WITHIN GROUP (ORDER BY ...)} clause. The fraction must be in the
+   * range 0 to 1 inclusive. The result is an actual value from the group: the
+   * first whose cumulative distribution is greater than or equal to the
+   * fraction. */
+  public static Object percentileDisc(List<?> values, double fraction) {
+    final int n = values.size();
+    if (n == 0) {
+      throw new NoSuchElementException(
+          "PERCENTILE_DISC is not defined on an empty group");
+    }
+    int index = (int) Math.ceil(fraction * n) - 1;
+    if (index < 0) {
+      index = 0;
+    } else if (index >= n) {
+      index = n - 1;
+    }
+    return requireNonNull(values.get(index));
+  }
+
   // FLOOR
 
   public static double floor(double b0) {