Support Delta DeletedRecordCountsHistogramAgg

Abierto
  • ¿Algo no está bien?
#7,571 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 2 días

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
5/5
Tiempo estimado
Más de una semana
Aptitud para principiantes
42/100
Tipo de issue
Nueva funcionalidad
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
rust, scala, spark
Área
backend, databases

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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.

Lenguaje dominante
Scala
Estrellas
1.3k
Forks
414
Merge medio
3 d 5 h
PR fusionados (30 d)
210

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de apache/datafusion-comet

Todos los issues de apache/datafusion-comet

Issues similares

Más issues de Scala

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.