Skip to content

Support Delta InterleaveBits for one or two INT inputs #7573

Open

Description

@ErikBPF

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:

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

Minimal query body in a compiled org.apache.spark.sql.comet.repro helper:

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.

Self-contained InterleaveBitsRepro reproducer and launch commands

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.

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:

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.

Willingness to contribute

I would be willing to contribute a fix for this bug with guidance from the DataFusion community.

Activity

  1. dwsmith1983 commented on Nov 9, 2026

    @dwsmith1983
    Contributor

    @ErikBPF I have a private repo that has z ordering and hilbert ordering in Rust already.

  2. ErikBPF commented on Nov 9, 2026

    @ErikBPF
    ContributorAuthor

    @dwsmith1983 Thanks for sharing that you already have Rust implementations of Z-ordering and Hilbert ordering. I'd be happy to help test them and contribute to the Comet integration.

    Is there a public branch or PR we can follow, or a way you'd prefer us to collaborate while the repository is private? We can help with Spark/Delta reference comparisons, especially exact InterleaveBits output for nulls, negative values, integer boundaries and multiple dimensions, followed by mixed Spark/Comet execution and fallback checks on Apollo.

    Which part would be most useful for us to take on, and what build or test entry point should we use? We can keep #7573 focused on Delta-compatible InterleaveBits and coordinate broader ordering work separately.

  3. dwsmith1983 commented on Nov 10, 2026

    @dwsmith1983
    Contributor

    Is there a public branch or PR we can follow, or a way you'd prefer us to collaborate while the repository is private?

    Could you open a draft PR for this issue? I'll move my Rust Z-order and Hilbert code into it and get a base version working, then hand it to you to extend and clean up. It isn't polished, so expect rough edges. So I can push to your branch, please add me as a collaborator on your fork.

  4. dwsmith1983 commented on Nov 11, 2026

    @dwsmith1983
    Contributor

    Before writing the integration, I'd like agreement on two points.

    1. Comet has no hook for an expression class that only a contrib knows about. I'd add a CometExpressionContrib ServiceLoader interface next to CometScanContrib, which the Delta contrib uses to register a serde for InterleaveBits, so core still names no Delta class. The other option is matching the class name in core, the way icebergFunctions.scala handles Iceberg.
    2. The plan is a native Rust kernel, checked against both of Delta's code paths. Routing the expression through the JVM codegen dispatcher would also stop the fallback with no Rust, but it runs Delta's own eval rather than a native kernel.

    @andygrove @comphead does the interface work for you, and is a native kernel what you'd want here?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions