ErikBPF opened a new issue, #7572: URL: https://github.com/apache/datafusion-comet/issues/7572
### What is the problem the feature request solves? Delta's DV-writing `BitmapAggregator` falls back to Spark's `ObjectHashAggregate` on Spark 4.2.0 / Delta 4.4.0 / released Comet 1.2.0: ```text unsupported Spark aggregate function: bitmapaggregator ``` The query produces correct results through Spark. The request is to support this Delta-specific writer aggregate, not to report corruption or to implement Spark's public `bitmap_*` SQL family. ### Describe the potential solution Implement this aggregate's Portable encoding path only, using a separately agreed optional Delta integration seam. Preserve the non-null result struct's exact schema and Delta behavior for cardinality, nullable last value and serialized bitmap bytes. Delta's JVM decoder must consume the output. Acceptance: native aggregate execution with exact empty/null/duplicate, sparse/dense and above-2^32 sets; update/merge/spill and bounded memory; Portable state roundtrip through the Delta reference. Admit cross-engine partial-state directions only when proven; otherwise retain fallback. Unsupported formats and input shapes remain on Spark. Validate rejection of negative indexes and `Long.MaxValue` before allocation; use bounded above-2^32 values such as `4294967299`. Do not materialize the maximum representable valid index: the reference allocates a high-word-indexed array, which can be enormous. **Out of scope:** original file identity, absolute row-index generation, native DV-writing MERGE/DELETE, reader admission changes, other bitmap formats, generic Spark bitmap SQL and Delta transaction handling. No native file-writing result is required to close this aggregate issue. Integration with real DV writes is a separate dependent task after file identity and row-index support are proven. ### Additional context - [Delta 4.4.0 BitmapAggregator](https://github.com/delta-io/delta/blob/v4.4.0/spark/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregation/BitmapAggregator.scala): nulls are ignored; update/merge operate on `RoaringBitmapArray`; evaluation returns cardinality/last/bytes. - [Delta bitmap E2E tests](https://github.com/delta-io/delta/blob/v4.4.0/spark/src/test/scala/org/apache/spark/sql/delta/util/BitmapAggregatorE2ESuite.scala). - [Spark 4.2 bitmap_construct_agg](https://spark.apache.org/docs/4.2.0/api/python/reference/pyspark.sql/api/pyspark.sql.functions.bitmap_construct_agg.html) and [its implementation](https://github.com/apache/spark/blob/v4.2.0/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/bitmapExpressions.scala): a different public aggregate with binary result/fixed binary state. It is not a replacement for Delta's type. - [Current Delta scan safety gates](https://github.com/apache/datafusion-comet/blob/4de6704391c9a7832a2079fc4e8d80fc09e9b209/contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala). 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.catalyst.expressions.aggregation.BitmapAggregator import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArray, RoaringBitmapArrayFormat} val input = spark.sql("SELECT CAST(v AS BIGINT) AS v FROM VALUES " + "(0), (9), (10), (99), (100), (NULL) AS t(v)") val bitmap = new BitmapAggregator(UnresolvedAttribute("v"), RoaringBitmapArrayFormat.Portable) val query = input.agg(Column(bitmap.toAggregateExpression()).as("bitmap")) val result = query.collect().head.getStruct(0) assert(result.getLong(0) == 5L && result.getLong(1) == 100L) val bytes: Array[Byte] = result.getAs(2) val decoded = RoaringBitmapArray.readFrom(bytes) assert(decoded.cardinality == 5L) assert(Seq(0L,9L,10L,99L,100L).forall(decoded.contains)) 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 PortableBitmapRepro reproducer and launch commands</summary> Save as `PortableBitmapRepro.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.catalyst.expressions.aggregation.BitmapAggregator import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArray, RoaringBitmapArrayFormat} object PortableBitmapRepro { 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.agg(Column(new BitmapAggregator(UnresolvedAttribute("v"), RoaringBitmapArrayFormat.Portable).toAggregateExpression()).as("bitmap")) val result = query.collect().head.getStruct(0) require(result.getLong(0) == 5L && result.getLong(1) == 100L) val bytes: Array[Byte] = result.getAs(2) val decoded = RoaringBitmapArray.readFrom(bytes) require(decoded.cardinality == 5L && Seq(0L,9L,10L,99L,100L).forall(decoded.contains)) println(query.queryExecution.executedPlan.toString) println("PASS: PortableBitmapRepro " + 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" PortableBitmapRepro.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.PortableBitmapRepro \ "$repro_root/probe.jar" "$mode" "$repro_root/$mode" done ``` Expected: both modes exit zero and print `PASS: PortableBitmapRepro <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]
