Support Delta DeletedRecordCountsHistogramAgg

Offen
  • Stimmt etwas nicht?
#7,571 0 Kommentare 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen

Maintainer antworten meist innerhalb von 2 Tagen

Dieses Issue hat noch niemand übernommen.

Bewertung

Schwierigkeit
5/5
Geschätzter Aufwand
Über eine Woche
Anfängerfreundlichkeit
42/100
Issue-Typ
Feature
Klarheit
Klar beschrieben
Aktivitätsstatus
Aktiv
Tech-Stack
rust, scala, spark
Bereich
backend, databases

Rechercherichtung

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.

Vom Indexierungsmodell aus dem Issue-Text verfasst.

Beschreibung

area:aggregation enhancement
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.

Vorherrschende Sprache
Scala
Sterne
1.3k
Forks
414
Ø Merge
3 T. 4 Std.
Gemergte PRs (30 T.)
202

Entwicklungsumgebung

Erste Schritte

  1. Lesen Sie das ganze Issue und danach den Beitragsleitfaden des Projekts.
  2. Schreiben Sie ins Issue, dass Sie es übernehmen — das erspart doppelte Arbeit.
  3. Forken Sie das Repository und arbeiten Sie in einem Branch.
  4. Öffnen Sie einen Pull Request, der die Issue-Nummer nennt.

Mehr aus apache/datafusion-comet

Alle Issues in apache/datafusion-comet

Ähnliche Issues

Weitere Issues zu Scala

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.