ErikBPF opened a new issue, #7570:
URL: https://github.com/apache/datafusion-comet/issues/7570
### What is the problem the feature request solves?
Delta's `FileSizeHistogramAgg` 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: filesizehistogramagg
```
Correct results are returned through Spark. This is one aggregate-support
request, with no measured speedup claim.
### Describe the potential solution
Add optional support for this aggregate alone, preserving its bin boundaries,
file counts, byte totals, null/empty behavior, update/merge and
serialization.
Keep Delta out of mandatory core dependencies. Use a separately agreed
optional
integration seam; do not design a general plugin system in this issue.
Acceptance for the initial supported path:
- Assert native aggregate execution and exact scalar/grouped results against
Delta.
- Cover sorted custom bins, boundaries, nulls, empty partitions and reference
overflow. Ignore values below the first bin; do not ignore all negative
values.
With bins `[-10,0,10,100]` and values `-5,NULL,0,9,10,99,100`, return
`[-10,0,10,100,1,2,2,1,-5,9,109,100]`. Negative values inside a bin count.
- Test partial/merge/spill state on the admitted native pipeline. Classify
each
cross-engine state direction independently and retain safe fallback where
interoperability is unproved; implementing both directions is not
mandatory.
Intermediate state contains counts and totals only, serialized with
`Platform.putLong`; boundaries are prepended only to the final result.
Preserve
exact bytes on the target architecture, not an invented portable-endian
format.
- With bins `[0,10,100]` and values `0,9,10,99,100,NULL`, return
`[0,10,100,2,2,1,9,109,100]`.
**Out of scope:** deleted-record histograms, bitmap aggregation,
snapshot-wide
native execution, unrelated colocated aggregates, Delta reads/writes and any
production configuration change. This issue can close while those remain
Spark.
### Additional context
[Delta
FileSizeHistogramAgg](https://github.com/delta-io/delta/blob/v4.4.0/spark/src/main/scala/org/apache/spark/sql/delta/stats/FileSizeHistogramUtils.scala)
extends
[FileStatsHistogramAggBase](https://github.com/delta-io/delta/blob/v4.4.0/spark/src/main/scala/org/apache/spark/sql/delta/stats/FileStatsHistogram.scala).
Use those implementations as the state/result oracle, alongside
[Spark 4.2 null/aggregate
semantics](https://spark.apache.org/docs/4.2.0/sql-ref-null-semantics.html).
Related aggregate-state audit: #5975. The separately proposed
`DeletedRecordCountsHistogramAgg` support is not a prerequisite.
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 FileSizeHistogramRepro reproducer and launch
commands</summary>
Save as `FileSizeHistogramRepro.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.FileSizeHistogramUtils.FileSizeHistogramAgg
object FileSizeHistogramRepro {
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(FileSizeHistogramAgg(UnresolvedAttribute("v"),
IndexedSeq(0L,10L,100L)).toAggregateExpression()).as("sizes"))
require(query.collect().head.getSeq[Long](0) ==
Seq(0L,10L,100L,2L,2L,1L,9L,109L,100L))
println(query.queryExecution.executedPlan.toString)
println("PASS: FileSizeHistogramRepro " + 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" FileSizeHistogramRepro.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.FileSizeHistogramRepro \
"$repro_root/probe.jar" "$mode" "$repro_root/$mode"
done
```
Expected: both modes exit zero and print `PASS: FileSizeHistogramRepro
<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]
