From acf2cde65a23123eb3e08b3e52461fd958116ee9 Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Tue, 10 Nov 2026 18:02:15 -0300 Subject: [PATCH 1/2] test(delta): add InterleaveBits handoff baseline --- .../delta/DeltaInterleaveBitsChecks.scala | 148 ++++++++++++++++++ dev/repros/delta/README-7573.md | 68 ++++++++ 2 files changed, 216 insertions(+) create mode 100644 dev/repros/delta/DeltaInterleaveBitsChecks.scala create mode 100644 dev/repros/delta/README-7573.md diff --git a/dev/repros/delta/DeltaInterleaveBitsChecks.scala b/dev/repros/delta/DeltaInterleaveBitsChecks.scala new file mode 100644 index 00000000000..eb830223b8b --- /dev/null +++ b/dev/repros/delta/DeltaInterleaveBitsChecks.scala @@ -0,0 +1,148 @@ +/* + * 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.spark.sql.comet.repro + + +import java.nio.charset.StandardCharsets.UTF_8 +import java.nio.file.{Files, Paths} +import org.apache.spark.sql.{Column, DataFrame, Row, SparkSession} +import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute +import org.apache.spark.sql.catalyst.expressions.{Literal} +import org.apache.spark.sql.classic.ClassicConversions._ + +import org.apache.spark.sql.delta.expressions.InterleaveBits +import org.apache.spark.sql.delta.sources.DeltaSQLConf +import org.apache.spark.sql.functions._ +import org.apache.spark.sql.types._ + +/** Selected reference checks for #7573; no expression implementation is added. */ +object DeltaInterleaveBitsChecks { + def main(args: Array[String]): Unit = { + require(args.length == 2, "usage: stock|comet ") + val mode = args(0); require(Set("stock", "comet")(mode)) + val out = Paths.get(args(1)); Files.createDirectory(out) + val spark = SparkSession.builder + .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension" + + (if (mode == "comet") ",org.apache.comet.CometSparkSessionExtensions" else "")) + .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") + .config("spark.comet.enabled", (mode == "comet").toString) + .config("spark.comet.exec.enabled", (mode == "comet").toString) + .config("spark.comet.explain.fallback.enabled", "true") + .config("spark.shuffle.manager", if (mode == "comet") + "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager" + else "org.apache.spark.shuffle.sort.SortShuffleManager") + .config("spark.memory.offHeap.enabled", "true").config("spark.memory.offHeap.size", "512m") + .config("spark.sql.adaptive.enabled", "false").config("spark.sql.shuffle.partitions", "2") + .getOrCreate() + import spark.implicits._ + var passed = 0 + def check(name: String)(body: => Unit): Unit = { + try { body; Files.write(out.resolve(name + ".passed"), Array.emptyByteArray); passed += 1 } + catch { case e: Throwable => Files.write(out.resolve(name + ".error.txt"), e.toString.getBytes(UTF_8)); throw e } + } + def query(name: String, df: DataFrame)(assertRows: Array[Row] => Unit): Unit = check(name) { + val rows = df.collect(); assertRows(rows) + Files.write(out.resolve(name + ".rows.json"), rows.map(_.json).sorted.mkString("[", ",", "]").getBytes(UTF_8)) + Files.write(out.resolve(name + ".schema.json"), df.schema.json.getBytes(UTF_8)) + Files.write(out.resolve(name + ".plan.txt"), df.queryExecution.executedPlan.toString.getBytes(UTF_8)) + } + try { + val control = spark.range(20).selectExpr("sum(id) total") + query("control", control) { rows => require(rows.length == 1 && rows.head.getLong(0) == 190L) + if (mode == "comet") require(control.queryExecution.executedPlan.toString.contains("Comet")) } + def oracle(values: Seq[java.lang.Integer]): Array[Byte] = { + val bytes = new Array[Byte](4 * values.size); var at = 0 + for (bit <- 31 to 0 by -1; v <- values) { val n = if (v == null) 0 else v.intValue() + if (((n >>> bit) & 1) != 0) bytes(at / 8) = (bytes(at / 8) | (1 << (7-at%8))).toByte + at += 1 } + bytes + } + def hexBytes(b: Array[Byte]): String = b.map(x => f"${x & 255}%02X").mkString + val x = 0x000ff0ff; val y = 0xfff00f00 + check("interleave_asymmetric_oracle") { + require(hexBytes(oracle(Seq(x,y))) == "555555AAAA55AAAA") + require(!oracle(Seq(x,y)).sameElements(oracle(Seq(y,x)))) + require(hexBytes(oracle(Seq(-1,null))) == "AAAAAAAAAAAAAAAA") + require(hexBytes(oracle(Seq(null,-1))) == "5555555555555555") + require(!InterleaveBits(Seq(Literal("x"))).checkInputDataTypes().isSuccess) + } + val random = new scala.util.Random(7573L) + val pairs = Seq[(java.lang.Integer,java.lang.Integer)]((x,y),(y,x),(null,-1),(-1,null),(Int.MinValue,Int.MaxValue),(0,0)) ++ + (0 until 256).map(i => (if (i % 17 == 0) null else Integer.valueOf(random.nextInt()), if (i % 19 == 0) null else Integer.valueOf(random.nextInt()))) + val pairSchema = StructType(Seq(StructField("x",IntegerType),StructField("y",IntegerType))) + val pairFrame = spark.createDataFrame(spark.sparkContext.parallelize(pairs.map {case (a,b) => Row(a,b)},2),pairSchema) + val pairPath = out.resolve("pairs").toString; pairFrame.write.parquet(pairPath) + val inputs = spark.read.parquet(pairPath) + val expectedPairCounts = pairs.groupBy(p => p).map { case (pair, values) => pair -> values.size } + def assertInputPairs(rows: Array[Row]): Unit = { + val actual = rows.toSeq.map(r => ( + if (r.isNullAt(0)) null else Integer.valueOf(r.getInt(0)), + if (r.isNullAt(1)) null else Integer.valueOf(r.getInt(1)))) + require(actual.groupBy(p => p).map { case (pair, values) => pair -> values.size } == expectedPairCounts, + "input pairs changed, duplicated or disappeared") + } + def checkMixedRows(rows: Array[Row], literalFirst: Boolean): Unit = { + require(rows.length == pairs.size, "mixed expression output cardinality changed") + assertInputPairs(rows) + rows.foreach { r => + val value = if (literalFirst) {if (r.isNullAt(1)) null else Integer.valueOf(r.getInt(1))} else {if (r.isNullAt(0)) null else Integer.valueOf(r.getInt(0))} + require(r.getAs[Array[Byte]](2).sameElements(oracle(if (literalFirst) Seq(Integer.valueOf(-1),value) else Seq(value,Integer.valueOf(-1))))) } + } + check("interleave_mixed_validator_rejects_incomplete") { + for (bad <- Seq(Array.empty[Row],Array(Row(x,y,oracle(Seq(Integer.valueOf(x),Integer.valueOf(-1))))), Array.fill(pairs.size)(Row(x,y,oracle(Seq(Integer.valueOf(x),Integer.valueOf(-1))))))) { + var rejected = false + try checkMixedRows(bad,false) catch { case _: IllegalArgumentException => rejected = true } + assert(rejected,"mixed validator accepted empty, truncated or duplicated output") + } + } + for (fast <- Seq(false,true)) { + spark.conf.set(DeltaSQLConf.FAST_INTERLEAVE_BITS_ENABLED.key,fast.toString) + for (dims <- Seq(1,2)) { + val key = Column(InterleaveBits(Seq("x","y").take(dims).map(UnresolvedAttribute(_)))).as("key") + val df = inputs.select(col("x"),col("y"),key) + require(df.schema("key").dataType == BinaryType && !df.schema("key").nullable) + query(s"interleave_columns_${dims}_$fast",df)(rows => {require(rows.length == pairs.size);assertInputPairs(rows);rows.foreach { r => + val values = (0 until dims).map(i => if (r.isNullAt(i)) null else Integer.valueOf(r.getInt(i))) + require(r.getAs[Array[Byte]](2).sameElements(oracle(values))) }}) + } + for (literalFirst <- Seq(false,true)) { + val children = if (literalFirst) Seq(Literal(-1),UnresolvedAttribute("y")) else Seq(UnresolvedAttribute("x"),Literal(-1)) + val df = inputs.select(col("x"),col("y"),Column(InterleaveBits(children)).as("key")) + require(df.schema("key").dataType == BinaryType && !df.schema("key").nullable) + query(s"interleave_mixed_${literalFirst}_$fast",df)(rows => checkMixedRows(rows,literalFirst)) + } + spark.conf.set("spark.sql.optimizer.excludedRules","org.apache.spark.sql.catalyst.optimizer.ConstantFolding") + for (dims <- Seq(1,2)) { + val literal = inputs.limit(1).select(Column(InterleaveBits(Seq(x,y).take(dims).map(Literal(_)))).as("key")) + require(literal.queryExecution.optimizedPlan.exists(p => p.expressions.exists(_.exists(_.isInstanceOf[InterleaveBits]))),"literal expression folded despite exclusion") + require(literal.schema("key").dataType == BinaryType && !literal.schema("key").nullable) + query(s"interleave_literals_unfolded_${dims}_$fast",literal)(rows => { + require(rows.length == 1) + require(rows.head.getAs[Array[Byte]](0).sameElements(oracle(Seq(x,y).take(dims).map(Integer.valueOf)))) + }) + } + spark.conf.unset("spark.sql.optimizer.excludedRules") + } + require(passed == 15, s"expected 15 named cases, observed $passed") + Files.write(out.resolve("summary.json"),s"{\"mode\":\"$mode\",\"passed\":$passed,\"native_acceptance\":false}".getBytes(UTF_8)) + println(s"PASS DeltaInterleaveBitsChecks $mode cases=$passed (reference/fallback only)") + } finally spark.stop() + } +} diff --git a/dev/repros/delta/README-7573.md b/dev/repros/delta/README-7573.md new file mode 100644 index 00000000000..0bcacb7904d --- /dev/null +++ b/dev/repros/delta/README-7573.md @@ -0,0 +1,68 @@ + + +# Reference checks for #7573 + +Run commands from `dev/repros/delta`. + +Standalone, test-only probes related to [apache/datafusion-comet#7573](https://github.com/apache/datafusion-comet/issues/7573). They do not implement the requested Comet expression or change admission rules. The issue remains open. + +Run with Spark 4.2.0, Scala 2.13.18 and Java 17. Supply matching Delta Spark and storage JARs, version 4.4.0 or 4.4.1. Use a fresh output directory for every invocation. + +The setup requires `curl`. Set `SPARK_HOME` and `JAVA_HOME` to your local installations before running it. The Comet artifact is `org.apache.datafusion:comet-spark-spark4.2_2.13:1.2.0`; match the Spark 4.2 and Scala 2.13 lines. This assembly contains the platform library used by the validated Linux x86_64 run; no separate Comet companion JAR was used. Other platforms remain unverified. + +```bash +set -eu +: "${SPARK_HOME:?Set SPARK_HOME to a Spark 4.2.0 installation}" +: "${JAVA_HOME:?Set JAVA_HOME to a Java 17 installation}" +export PATH="$JAVA_HOME/bin:$PATH" +DELTA_VERSION=4.4.0 # Use 4.4.1 to repeat the other validated pairing. +ARTIFACT_DIR=$(mktemp -d) +export DELTA_SPARK_JAR="$ARTIFACT_DIR/delta-spark_2.13-$DELTA_VERSION.jar" +export DELTA_STORAGE_JAR="$ARTIFACT_DIR/delta-storage-$DELTA_VERSION.jar" +export COMET_JAR="$ARTIFACT_DIR/comet-spark-spark4.2_2.13-1.2.0.jar" +curl --fail --location --retry 3 -o "$DELTA_SPARK_JAR" \ + "https://repo.maven.apache.org/maven2/io/delta/delta-spark_2.13/$DELTA_VERSION/delta-spark_2.13-$DELTA_VERSION.jar" +curl --fail --location --retry 3 -o "$DELTA_STORAGE_JAR" \ + "https://repo.maven.apache.org/maven2/io/delta/delta-storage/$DELTA_VERSION/delta-storage-$DELTA_VERSION.jar" +curl --fail --location --retry 3 -o "$COMET_JAR" \ + https://repo.maven.apache.org/maven2/org/apache/datafusion/comet-spark-spark4.2_2.13/1.2.0/comet-spark-spark4.2_2.13-1.2.0.jar +mkdir classes +java -Xmx1g -cp "$SPARK_HOME/jars/*" scala.tools.nsc.Main \ + -classpath "$SPARK_HOME/jars/*:$DELTA_SPARK_JAR:$DELTA_STORAGE_JAR" \ + -d classes DeltaInterleaveBitsChecks.scala +jar cf checks.jar -C classes . +"$SPARK_HOME/bin/spark-submit" --master 'local[2]' --driver-memory 2g \ + --jars "$DELTA_SPARK_JAR,$DELTA_STORAGE_JAR" \ + --class org.apache.spark.sql.comet.repro.DeltaInterleaveBitsChecks \ + checks.jar stock /tmp/DeltaInterleaveBitsChecks-stock +"$SPARK_HOME/bin/spark-submit" --master 'local[2]' --driver-memory 2g \ + --jars "$DELTA_SPARK_JAR,$DELTA_STORAGE_JAR,$COMET_JAR" \ + --class org.apache.spark.sql.comet.repro.DeltaInterleaveBitsChecks \ + checks.jar comet /tmp/DeltaInterleaveBitsChecks-comet +``` + +Expected: 15 named cases, matching row/schema JSON across stock and Comet runs, and `native_acceptance: false` in `summary.json`. The Comet lane requires an accelerated control query; passing Delta probes can still execute in Spark. Execution plans are saved beside rows. These are manually run probes, not tests wired into the Comet CI reactor. + +Grounding: selected equivalents of the following versioned upstream tests, not a replacement for their full suites: + +- [InterleaveBitsSuite (v4.4.0)](https://github.com/delta-io/delta/blob/v4.4.0/spark/src/test/scala/org/apache/spark/sql/delta/expressions/InterleaveBitsSuite.scala) +- [InterleaveBitsSuite (v4.4.1)](https://github.com/delta-io/delta/blob/v4.4.1/spark/src/test/scala/org/apache/spark/sql/delta/expressions/InterleaveBitsSuite.scala) + +Validation uses Comet 1.2.0, not a build of this branch. Current-main expression support and the production image remain unverified. From cf533e38b4e9eb93701f59bf82e282e7239d6147 Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Tue, 10 Nov 2026 18:34:25 -0300 Subject: [PATCH 2/2] docs(delta): isolate each InterleaveBits run --- dev/repros/delta/README-7573.md | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/dev/repros/delta/README-7573.md b/dev/repros/delta/README-7573.md index 0bcacb7904d..7d0ae70ab07 100644 --- a/dev/repros/delta/README-7573.md +++ b/dev/repros/delta/README-7573.md @@ -33,7 +33,9 @@ set -eu : "${JAVA_HOME:?Set JAVA_HOME to a Java 17 installation}" export PATH="$JAVA_HOME/bin:$PATH" DELTA_VERSION=4.4.0 # Use 4.4.1 to repeat the other validated pairing. -ARTIFACT_DIR=$(mktemp -d) +RUN_DIR=$(mktemp -d /tmp/comet-interleave.XXXXXX) +ARTIFACT_DIR="$RUN_DIR/artifacts" +mkdir "$ARTIFACT_DIR" export DELTA_SPARK_JAR="$ARTIFACT_DIR/delta-spark_2.13-$DELTA_VERSION.jar" export DELTA_STORAGE_JAR="$ARTIFACT_DIR/delta-storage-$DELTA_VERSION.jar" export COMET_JAR="$ARTIFACT_DIR/comet-spark-spark4.2_2.13-1.2.0.jar" @@ -43,19 +45,19 @@ curl --fail --location --retry 3 -o "$DELTA_STORAGE_JAR" \ "https://repo.maven.apache.org/maven2/io/delta/delta-storage/$DELTA_VERSION/delta-storage-$DELTA_VERSION.jar" curl --fail --location --retry 3 -o "$COMET_JAR" \ https://repo.maven.apache.org/maven2/org/apache/datafusion/comet-spark-spark4.2_2.13/1.2.0/comet-spark-spark4.2_2.13-1.2.0.jar -mkdir classes +mkdir "$RUN_DIR/classes" java -Xmx1g -cp "$SPARK_HOME/jars/*" scala.tools.nsc.Main \ -classpath "$SPARK_HOME/jars/*:$DELTA_SPARK_JAR:$DELTA_STORAGE_JAR" \ - -d classes DeltaInterleaveBitsChecks.scala -jar cf checks.jar -C classes . + -d "$RUN_DIR/classes" DeltaInterleaveBitsChecks.scala +jar cf "$RUN_DIR/checks.jar" -C "$RUN_DIR/classes" . "$SPARK_HOME/bin/spark-submit" --master 'local[2]' --driver-memory 2g \ --jars "$DELTA_SPARK_JAR,$DELTA_STORAGE_JAR" \ --class org.apache.spark.sql.comet.repro.DeltaInterleaveBitsChecks \ - checks.jar stock /tmp/DeltaInterleaveBitsChecks-stock + "$RUN_DIR/checks.jar" stock "$RUN_DIR/stock" "$SPARK_HOME/bin/spark-submit" --master 'local[2]' --driver-memory 2g \ --jars "$DELTA_SPARK_JAR,$DELTA_STORAGE_JAR,$COMET_JAR" \ --class org.apache.spark.sql.comet.repro.DeltaInterleaveBitsChecks \ - checks.jar comet /tmp/DeltaInterleaveBitsChecks-comet + "$RUN_DIR/checks.jar" comet "$RUN_DIR/comet" ``` Expected: 15 named cases, matching row/schema JSON across stock and Comet runs, and `native_acceptance: false` in `summary.json`. The Comet lane requires an accelerated control query; passing Delta probes can still execute in Spark. Execution plans are saved beside rows. These are manually run probes, not tests wired into the Comet CI reactor.