INSERT into a partitioned table with fanout disabled always fails: the optimizer removes the partition sort

Aperta
  • Qualcosa non va?
#38 1 commento 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
4/5
Tempo stimato
3-5 giorni
Idoneità per principianti
58/100
Tipo di issue
Bug
Chiarezza
Abbastanza chiara
Stato di attività
Attiva
Stack tecnologico
rust
Ambito
databases

Direzione di ricerca

Read crates/datafusion/src/table/mod.rs#L200-L220 and crates/datafusion/src/physical_plan/write.rs#L143-L151, then run test_insert_plan_fanout_disabled_has_sort and reproduce the partitioned INSERT with fanout disabled. Done means physical optimization preserves or restores the partition sort and hash repartition, and the insert succeeds with the writer receiving input clustered by partition.

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

Descrizione

bug
Describe the bug

With write.datafusion.fanout.enabled=false, insert_into places a SortExec on _partition, and a hash RepartitionExec on _partition, in front of IcebergWriteExec (table/mod.rs#L200-L220), because ClusteredWriter needs its input grouped by partition.

IcebergWriteExec doesn't declare required_input_ordering or required_input_distribution. It reports maintains_input_order() == [true] but advertises no output ordering (write.rs#L143-L151). DataFusion's EnforceSorting therefore treats the sort as unnecessary and removes it, and EnforceDistribution replaces the hash repartition.

To Reproduce
  1. Create t6(id INT NOT NULL, cat INT NOT NULL), partitioned by identity(cat), with write.datafusion.fanout.enabled=false.
  2. Run INSERT INTO t6 SELECT id, cat FROM src, where src has 200 rows with cat = id % 10 and batch_size = 8.

Optimized plan with target_partitions = 4:

IcebergCommitExec: table=ns.t6
  CoalescePartitionsExec
    IcebergWriteExec: table=ns.t6
      ProjectionExec: expr=[id@0 as id, cat@1 as cat, iceberg_partition_values(cat) as _partition]
        RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
          DataSourceExec: partitions=1, partition_sizes=[1]

All 10 runs failed, at both target_partitions = 1 and 4:

The input is not sorted! Cannot write to partition that was previously closed: PartitionKey { ... }
Expected behavior

The insert succeeds, with the writer receiving its input clustered by partition.

Additional context
  • test_insert_plan_fanout_disabled_has_sort passes because it inspects the plan that insert_into returns, which hasn't been through physical optimization.
  • Declaring the requirements on IcebergWriteExec would let the optimizer keep or re-insert them: required_input_ordering on _partition when fanout is disabled, and a hash required_input_distribution on _partition.
  • With fanout enabled (the default), the hash repartition is also replaced by round-robin, so every write task writes every partition. The data is correct, but there are more and smaller files than intended.
  • Even with the sort kept, clustered writes still fail for a second reason: #39. The unpartitioned case of this setting fails separately: #43.
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.