Offset overflow - Comet's grouped accumulators build one array for all emitted groups

Offen
  • Stimmt etwas nicht?
#7,501 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
4/5
Geschätzter Aufwand
3-5 Tage
Anfängerfreundlichkeit
42/100
Issue-Typ
Bug
Klarheit
Größtenteils klar
Aktivitätsstatus
Aktiv
Tech-Stack
rust, scala, spark
Bereich
backend, databases

Rechercherichtung

Start with the Rust accumulator implementations named in the issue: native/spark-expr/src/agg_funcs/collect.rs, max_min_by.rs, list_agg.rs, and approx_percentile.rs; trace their evaluate, state, and EmitTo::First paths. Check how errors are returned and how repeated partial emits handle retained groups. Add focused tests for overflow and repeated emits; done means overflow returns an error instead of panicking and chunked emits avoid quadratic work.

Vom Indexierungsmodell aus dem Issue-Text verfasst.

Beschreibung

area:aggregation bug priority:high

Part of #7267.

Problem

Comet's own GroupsAccumulators build one array for all the groups they emit, in evaluate and in state (which also runs when the aggregate spills). With EmitTo::All that is every group in the task, so the array overflows once the values collected across all groups pass 2 GiB, even though each group's own value is small. Spark keeps one buffer per group.

  • collect_list (native/spark-expr/src/agg_funcs/collect.rs:382-428) uses concat or interleave, which return OffsetOverflowError.
  • collect_set (collect.rs:717-732) and max_by/min_by (max_min_by.rs) decode with arrow-row's convert_rows, which panics with offset overflow (arrow-row 59.3.2 src/variable.rs:294).
  • listagg (list_agg.rs:304-317) and the approx_percentile state (approx_percentile.rs:414-421) build with StringArray::from and BinaryBuilder, which panic with byte array offset overflow.

For example, a final collect_list(json) GROUP BY user whose reduce partition holds more than 2 GiB of strings fails.

This is the accumulator side of #7278. Two things here matter for #7278's fix:

  • Chunked emit would call EmitTo::First(n) repeatedly, and in these accumulators each call costs time proportional to everything still retained. collect_list's compact (collect.rs:259) re-takes every retained batch, and collect_set's (collect.rs:622) re-inserts every remaining entry. Emitting in chunks would be quadratic.
  • The comment at collect.rs:509-518 says the element count is what limits the emitted List<i32>. For string and binary elements, the child's bytes are limited to 2 GiB as well.

Line references are for main at 4e6b12290f.

Workaround

Raise spark.sql.shuffle.partitions, so each task emits fewer groups.

Proposed fix

  • Return errors instead of panicking.
  • Make EmitTo::First(n) cheap, so the aggregate can emit groups in chunks bounded by bytes, together with #7278.

Tests

Check that an emit whose total passes the limit returns an error rather than panicking, using a few groups with large values (or a limit lowered for the test). Check the cost of repeated EmitTo::First with enough groups to show the quadratic time.

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.