Offset overflow - JVM columnar shuffle cuts row-to-Arrow batches by row count only

Abierto
  • ¿Algo no está bien?
#7,497 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
4/5
Tiempo estimado
3-5 días
Aptitud para principiantes
55/100
Tipo de issue
Error
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
rust, scala, spark
Área
backend, data-engineering

Línea de trabajo

Start with native/shuffle/src/spark_unsafe/row.rs, especially process_sorted_row_partition at lines 1579–1588 and its existing row size handling; compare it with process_sorted_partitions at line 1686. Add byte budget coverage alongside the row limit, then follow the issue’s test plan: use a small budget and verify several blocks are written and the same rows read back.

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

Descripción

area:shuffle bug priority:high requires-triage

Part of #7267.

Problem

The JVM columnar shuffle hands sorted UnsafeRows to native code, which converts them to Arrow in batches of spark.comet.shuffle.jvm.batchSize rows, capped by spark.comet.batchSize. Both entry points cut batches by row count only: process_sorted_partitions splits each partition's rows with run.chunks(batch_size) (native/shuffle/src/spark_unsafe/row.rs:1686), and process_sorted_row_partition does the same at :1579-1588. Strings and binaries are appended with arrow's append_value (:1173-1203), which panics with byte array offset overflow once a column passes 2 GiB. Strings inside arrays, maps and structs go through the same builders.

So a batch fails when one string or binary column holds more than 2 GiB, which is more than about 256 KiB per value at 8192 rows. When the column is dictionary-encoded, the writer succeeds and the reader fails instead (#7496). When the dictionary doesn't pay off, the cast back to plain values returns OffsetOverflowError (row.rs:1777). Spark shuffles rows and has no batch to overflow.

#7269 covers the native shuffle writer, which is separate code.

Line references are for main at 4e6b12290f.

Workaround

Lower spark.comet.shuffle.jvm.batchSize.

Proposed fix

Cut batches by bytes as well as rows. The row sizes are known before conversion (row_sizes), and a row's size bounds the bytes it adds to any column, so ending a batch before the sum of its row sizes passes i32::MAX keeps every column valid without looking at the values.

Tests

Inject a small byte budget and check that process_sorted_row_partition writes several blocks for rows that would otherwise share one, with the same rows read back.

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.