`IcebergTableScan` ignores the session's `batch_size`

Aperta Adatta ai principianti
  • Qualcosa non va?
#100 2 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
2/5
Tempo stimato
1-3 ore
Idoneità per principianti
82/100
Tipo di issue
Bug
Chiarezza
Specificata chiaramente
Stato di attività
Attiva
Stack tecnologico
rust
Ambito
backend

Direzione di ricerca

Inizia da crates/datafusion/src/read/scan.rs: segui IcebergDataSource::open fino a get_batch_stream e verifica come la dimensione del batch della sessione viene passata a TableScanBuilder. L’issue include una modifica suggerita per passare la dimensione configurata e chiamare with_batch_size; il lavoro è terminato quando le scansioni rispettano datafusion.execution.batch_size invece di usare sempre batch da 1.024 righe.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

IcebergDataSource::open does not pass execution.batch_size from the session config to get_batch_stream:

https://github.com/apache/datafusion-iceberg/blob/256c4b7bde01a5eee9e4c6b417a36dbe3eb761ec/crates/datafusion/src/read/scan.rs#L182-L198

get_batch_stream never calls TableScanBuilder::with_batch_size:

https://github.com/apache/datafusion-iceberg/blob/256c4b7bde01a5eee9e4c6b417a36dbe3eb761ec/crates/datafusion/src/read/scan.rs#L297-L305

So scans always read arrow-rs's default of 1,024-row batches. That's an eighth of DataFusion's default of 8,192, and SET datafusion.execution.batch_size has no effect on them.

Suggested fix:

--- a/crates/datafusion/src/read/scan.rs
+++ b/crates/datafusion/src/read/scan.rs
@@ -195,6 +195,7 @@ impl DataSource for IcebergDataSource {
                 .execution
                 .parquet
                 .enable_page_index,
+            context.session_config().batch_size(),
         );
         let stream = futures::stream::once(fut).try_flatten();
 
@@ -281,6 +282,7 @@ async fn get_batch_stream(
     column_names: Option<Vec<String>>,
     predicate: Option<Predicate>,
     enable_page_index: bool,
+    batch_size: usize,
 ) -> Result<Pin<Box<dyn Stream<Item = Result<RecordBatch>> + Send>>> {
     let scan_builder = match snapshot_id {
         Some(snapshot_id) => table.scan().snapshot_id(snapshot_id),
@@ -296,6 +298,7 @@ async fn get_batch_stream(
     }
     let table_scan = scan_builder
         .with_row_selection_enabled(enable_page_index)
+        .with_batch_size(Some(batch_size))
         .build()
         .map_err(to_datafusion_error)?;
Lingua principale
Rust
Stelle
30
Fork
30
Merge medio
3g 19h
PR unite (30g)
15

Preparare l'ambiente

Questo progetto non fornisce container di sviluppo, Dockerfile né guida per i contributori, quindi l'ambiente è a tuo carico: parti dal suo README e consulta la nostra guida al primo contributo per i passaggi generali.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di apache/datafusion-iceberg

Tutte le issue di apache/datafusion-iceberg

Issue simili

Altre issue su Rust

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.