Support Delta DeletedRecordCountsHistogramAgg

Ouverte
  • Quelque chose ne va pas ?
#7,571 0 commentaires 0 réactions 0 personnes assignées Voir sur GitHub

Les mainteneurs répondent en général sous 2 jours

Personne n'a encore pris cette issue.

Évaluation

Difficulté
5/5
Temps estimé
Plus d'une semaine
Accessibilité débutants
42/100
Type d'issue
Fonctionnalité
Clarté
Clairement spécifiée
Activité
Active
Stack technique
rust, scala, spark
Domaine
backend, databases

Piste de recherche

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.

Rédigé par le modèle d'indexation à partir du texte de l'issue.

Description

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.

Langage dominant
Scala
Étoiles
1.3k
Forks
414
Merge moyen
3 j 4 h
PR mergées (30 j)
202

Préparer son environnement

Par où commencer

  1. Lisez l'issue en entier, puis le guide de contribution du projet.
  2. Signalez en commentaire que vous la prenez — cela évite que deux personnes fassent le même travail.
  3. Forkez le dépôt et travaillez sur une branche.
  4. Ouvrez une pull request qui référence le numéro de l'issue.

Autres issues de apache/datafusion-comet

Toutes les issues de apache/datafusion-comet

Issues similaires

Plus d'issues Scala

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.