ErikBPF opened a new issue, #7573: URL: https://github.com/apache/datafusion-comet/issues/7573
### What is the problem the feature request solves? Liquid-clustering optimization retains Spark projections for Delta's range-partition/interleaving key construction. An isolated Delta `InterleaveBits` query on Parquet input, Spark 4.2.0 / Delta 4.4.0 / released Comet 1.2.0, logs: ```text Comet cannot accelerate InterleaveBits because: interleavebits is not supported ``` Results remain correct through Spark. This is a feature investigation with no measured workload speedup, not a request to move Delta OPTIMIZE/commit logic wholesale into Comet. ### Describe the potential solution Support only the pure InterleaveBits transform for one or two INT children, using a separately agreed optional integration seam. This covers the observed one-key clustering route and the isolated two-input reproducer. Preserve null-as-zero, signed 32-bit boundaries, output width and exact byte order against Delta. Compare both relevant Delta reference implementation paths; unsupported arities or types continue to fall back. Acceptance is an isolated expression suite with native-execution assertions, exact bytes for column and literal inputs with constant folding controlled, nulls, signed extrema and deterministic generated integer pairs. Closing this issue does not imply native clustering support or a measured workload speedup. **Out of scope:** RangePartitionId/partitionerexpr, random tie keys, casts beyond the expression boundary, repartitioning, Hilbert curves, more than two inputs, OPTIMIZE/OPTIMIZE FULL, table metadata and file rewrites. Those need separate scoped issues or integration tasks after their own evidence exists. ### Additional context - [Delta InterleaveBits](https://github.com/delta-io/delta/blob/v4.4.0/spark/src/main/scala/org/apache/spark/sql/delta/expressions/InterleaveBits.scala) takes integer children, treats null as zero and returns binary. Its class uses `CodegenFallback`, which Comet's JVM dispatcher can admit, but the Delta class has no registered handler. Trait support alone is not expression admission. - [MultiDimClusteringFunctions](https://github.com/delta-io/delta/blob/v4.4.0/spark/src/main/scala/org/apache/spark/sql/delta/skipping/MultiDimClusteringFunctions.scala) constructs `RangePartitionId` and `InterleaveBits`. - [MultiDimClustering](https://github.com/delta-io/delta/blob/v4.4.0/spark/src/main/scala/org/apache/spark/sql/delta/skipping/MultiDimClustering.scala) composes those expressions into a range-repartitioned key and secondary random byte. - [Delta interleaving tests](https://github.com/delta-io/delta/blob/v4.4.0/spark/src/test/scala/org/apache/spark/sql/delta/expressions/InterleaveBitsSuite.scala) and [clustering documentation](https://docs.delta.io/delta-clustering/). Minimal query body in a compiled `org.apache.spark.sql.comet.repro` helper: ```scala import org.apache.spark.sql.Column import org.apache.spark.sql.classic.ClassicConversions._ import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute import org.apache.spark.sql.delta.expressions.InterleaveBits import org.apache.spark.sql.functions.hex val input = spark.sql("SELECT CAST(x AS INT) AS x FROM VALUES (0),(9),(10),(99),(100),(NULL) AS t(x)") val key = Column(InterleaveBits(Seq(UnresolvedAttribute("x"), UnresolvedAttribute("x")))) val query = input.select(hex(key).as("key")) query.show(false) query.explain(true) ``` Observed runtime: Spark 4.2.0 / Scala 2.13.18 / Java 17, Delta 4.4.0 and released Comet 1.2.0 (`ab46419fcc99bcd34aa4d89f7b545d5d51694a88`). Current main inspected at `4de6704391c9a7832a2079fc4e8d80fc09e9b209` was not executed. This is missing support, not a current-main regression or a wrong-results report. No measured native-kernel speedup is claimed. Optional integration design still needs maintainer agreement. <details> <summary>Self-contained InterleaveBitsRepro reproducer and launch commands</summary> Save as `InterleaveBitsRepro.scala`. Each mode creates its own fresh synthetic Parquet fixture. The Comet control must contain Comet operators before this expression is tested; passing expected rows alone does not establish native execution of the requested expression. ```scala package org.apache.spark.sql.comet.repro import java.nio.file.{Files, Paths} import org.apache.spark.sql.{Column, SparkSession} import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute import org.apache.spark.sql.classic.ClassicConversions._ import org.apache.spark.sql.functions._ import org.apache.spark.sql.delta.expressions.InterleaveBits object InterleaveBitsRepro { def main(args: Array[String]): Unit = { require(args.length == 2, "usage: stock|comet <fresh output>") 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.comet.explain.fallback.log.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.shuffle.partitions", "2") .getOrCreate() try { val control = spark.range(20).selectExpr("sum(id) AS total") require(control.collect().head.getLong(0) == 190L) if (mode == "comet") require(control.queryExecution.executedPlan.toString.contains("Comet")) val path = out.resolve("input").toString spark.sql("SELECT CAST(v AS BIGINT) v, CAST(v AS INT) x, payload FROM VALUES " + "(0,'a'),(9,'b'),(10,'c'),(99,'d'),(100,'e'),(NULL,'f') AS t(v,payload)") .repartition(2).write.parquet(path) val input = spark.read.parquet(path) val query = input.select(col("payload"), hex(Column(InterleaveBits(Seq( UnresolvedAttribute("x"), UnresolvedAttribute("x"))))).as("key")).orderBy("payload") require(query.collect().map(_.getString(1)).toSeq == Seq("0000000000000000", "00000000000000C3","00000000000000CC","0000000000003C0F", "0000000000003C30","0000000000000000")) println(query.queryExecution.executedPlan.toString) println("PASS: InterleaveBitsRepro " + mode) } finally spark.stop() } } ``` Use Spark 4.2.0 with Java 17. Maven coordinates: - `io.delta:delta-spark_4.2_2.13:4.4.0` - `io.delta:delta-storage:4.4.0` - `org.apache.datafusion:comet-spark-spark4.2_2.13:1.2.0` Set `SPARK_HOME`, `JAVA_HOME`, `DELTA_JARS` (two Delta JAR paths, comma separated), and `COMET_JAR` to these local artifacts. The Comet release contains its matching native library. No Delta contrib JAR is required for this reproducer. Bash commands: ```bash set -euo pipefail export SPARK_LOCAL_IP=127.0.0.1 export COMET_WORKER_THREADS=2 export COMET_MAX_BLOCKING_THREADS=4 repro_root=$(mktemp -d /tmp/comet-delta-expression.XXXXXX) mkdir "$repro_root/classes" "$JAVA_HOME/bin/java" -Xmx512m -XX:ActiveProcessorCount=2 \ -cp "$SPARK_HOME/jars/*" scala.tools.nsc.Main \ -classpath "$SPARK_HOME/jars/*:${DELTA_JARS//,/:}" \ -d "$repro_root/classes" InterleaveBitsRepro.scala "$JAVA_HOME/bin/jar" cf "$repro_root/probe.jar" -C "$repro_root/classes" . for mode in stock comet; do jars="$DELTA_JARS" if [ "$mode" = comet ]; then jars="$jars,$COMET_JAR"; fi "$SPARK_HOME/bin/spark-submit" --master 'local[2]' --driver-memory 2g \ --conf spark.driver.extraJavaOptions=-XX:ActiveProcessorCount=2 \ --jars "$jars" --class org.apache.spark.sql.comet.repro.InterleaveBitsRepro \ "$repro_root/probe.jar" "$mode" "$repro_root/$mode" done ``` Expected: both modes exit zero and print `PASS: InterleaveBitsRepro <mode>`. On the measured Comet 1.2.0 runtime, the requested private expression remains on Spark and the fallback reason appears in the log. The source above pins exact expected results; it does not implement the feature. </details> ### Willingness to contribute I would be willing to contribute a fix for this bug with guidance from the DataFusion community. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
