Support Delta DeletedRecordCountsHistogramAgg

Aperta
  • Qualcosa non va?
#7,571 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 2 giorni

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
5/5
Tempo stimato
Più di una settimana
Idoneità per principianti
42/100
Tipo di issue
Funzionalità
Chiarezza
Specificata chiaramente
Stato di attività
Attiva
Stack tecnologico
rust, scala, spark
Ambito
backend, databases

Direzione di ricerca

Start with the linked Delta 4.4.0 implementation, DeletedRecordCountsHistogramUtils.scala, and the issue’s Spark 4.2.0 reproducer to understand bucket and serialization behavior. Coordinate the optional integration seam and state admission rules with maintainers, including #5975. Done means the aggregate executes natively and passes the specified bucket, null, boundary, overflow, merge, spill, and exact-state checks, with safe fallback for unproved cross-engine directions.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

requires-triage
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:

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
defines both bucket selection and its own serialization. Its output is not
FileSizeHistogramAgg's boundaries/counts/bytes layout.
Spark 4.2 null/aggregate semantics
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.

Self-contained DeletedRecordHistogramRepro reproducer and launch commands

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.

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:

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.

Willingness to contribute

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

Lingua principale
Scala
Stelle
1.3k
Fork
414
Merge medio
3g 5h
PR unite (30g)
210

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di apache/datafusion-comet

Tutte le issue di apache/datafusion-comet

Issue simili

Altre issue su Scala

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.