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

オープン
  • 何か問題がありますか？
#7,497 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る

メンテナーはふだん 2 日以内に返信

まだ誰も着手していません。

評価

難易度
4/5
見積もり時間
3〜5日
初心者へのやさしさ
55/100
issue の種類
バグ
明瞭さ
明確に書かれている
活発さ
活発
技術スタック
rust, scala, spark
領域
backend, data-engineering

調査の方向性

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.

索引モデルが issue の本文から書いたものです。

説明

area:shuffle bug priority:high

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.

主要言語
Scala
スター
1.3k
フォーク
414
平均マージ
3日 5時間
マージ済み PR（30日）
210

環境構築

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

apache/datafusion-comet のほかの issue

apache/datafusion-comet の issue をすべて見る

似ている issue

Scala の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。