ErikBPF opened a new issue, #7571:
URL: https://github.com/apache/datafusion-comet/issues/7571
### What is the problem the feature request solves?
An isolated `DeletedRecordCountsHistogramAgg` query falls back to Spark on
the
measured Spark 4.2.0 / Delta 4.4.0 / Comet 1.2.0 runtime:
```text
unsupported Spark aggregate function: deletedrecordcountshistogramagg
```
Results are correct through Spark. This is a request for one private
aggregate,
not native deletion-vector generation or a wrong-results report.
### Describe the potential solution
Add optional support for this aggregate alone. Preserve its fixed bucket
assignment, null/empty handling, update/merge and exact serialized state.
Use a separately agreed optional integration seam without making Delta a
mandatory core
dependency or coupling this change to FileSizeHistogramAgg.
Acceptance for the initial supported path:
- Assert native aggregate execution and exact scalar/grouped buckets against
Delta.
- Cover null/empty partitions, zero, bucket boundaries and counter overflow.
Negative inputs fail the reference nonnegative requirement; `Long.MaxValue`
belongs to the final bucket.
- Test native partial/merge/spill state. Independently classify cross-engine
directions; unproved directions retain safe fallback rather than becoming
required interoperability work. The normal state has one element-count byte
containing `10`, followed by ten `Platform.putLong` counters: 81 bytes
total.
The header is not a version. Preserve reference count/length validation and
target-architecture byte order.
- Values `0,9,10,99,100,NULL` produce buckets
`[1,1,2,1,0,0,0,0,0,0]`.
**Out of scope:** file-size histograms, DV bitmap generation, DV
reads/writes,
REORG/VACUUM execution, arbitrary colocated aggregates and production
settings.
This issue can close without completing any of those operations.
### Additional context
[Delta's
implementation](https://github.com/delta-io/delta/blob/v4.4.0/spark/src/main/scala/org/apache/spark/sql/delta/stats/DeletedRecordCountsHistogramUtils.scala)
defines both bucket selection and its own serialization. Its output is not
FileSizeHistogramAgg's boundaries/counts/bytes layout.
[Spark 4.2 null/aggregate
semantics](https://spark.apache.org/docs/4.2.0/sql-ref-null-semantics.html)
remain the surrounding query oracle. Coordinate state-admission rules with
#5975.
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 DeletedRecordHistogramRepro reproducer and launch
commands</summary>
Save as `DeletedRecordHistogramRepro.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.stats.DeletedRecordCountsHistogramUtils.DeletedRecordCountsHistogramAgg
object DeletedRecordHistogramRepro {
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(DeletedRecordCountsHistogramAgg(
UnresolvedAttribute("v")).toAggregateExpression()).as("deleted"))
require(query.collect().head.getSeq[Long](0) ==
Seq(1L,1L,2L,1L,0L,0L,0L,0L,0L,0L))
println(query.queryExecution.executedPlan.toString)
println("PASS: DeletedRecordHistogramRepro " + 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" DeletedRecordHistogramRepro.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.DeletedRecordHistogramRepro \
"$repro_root/probe.jar" "$mode" "$repro_root/$mode"
done
```
Expected: both modes exit zero and print `PASS: DeletedRecordHistogramRepro
<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]
