Repository navigation
Conversation
The eager Arrow writer repeatedly derives the same schema-only field-to-leaf mapping when creating and validating row groups, even though that metadata is invariant for the lifetime of the writer factory. Build an immutable write-schema plan once per factory and share it across row groups and column writers. Reuse the cached target fields, leaf ranges, and Parquet descriptors for both ordinary and CDC paths, while continuing to derive leaves and levels from each incoming batch. This removes repeated schema traversal without retaining input arrays or changing dense, dictionary, REE, view, required-null, or leaf-ownership semantics. Signed-off-by: Hippolyte Barraud <hippolyte.barraud@datadoghq.com>
|
run benchmark arrow_writer |
|
🤖 Benchmark starting (GKE) | trigger Target: arrow_writer Sharding factor: 4 (4 workers). Comparing share_write_schemas (ba9d4b8) to 7dcc160 (merge-base) diff Run configurationrun benchmark arrow_writer
shards: 4
Results will be posted when all workers finish. File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Target: arrow_writer Comparing share_write_schemas (ba9d4b8) to 7dcc160 (merge-base) diff Run configurationrun benchmark arrow_writer
shards: 4
Comparison results: View the complete native critcmp output. The comparison is published in an unlisted Gist because the original inline report exceeded GitHub's comment limit. Per-runner informationarrow_writer — shard 1/4Node: gk3-benchmark-cluster-nap-1kxn11uo-aeaf8131-9gtv Instance: c4a-highmem-16 (12 vCPU / 65 GiB) uname: BENCH_COMMAND: cargo bench --features=arrow,async,test_common,experimental,object_store --bench arrow_writerCPU Details (lscpu)Resource Usagebase (merge-base)
branch
arrow_writer — shard 2/4Node: gk3-benchmark-cluster-nap-1kxn11uo-efb27e7d-nplg Instance: c4a-highmem-16 (12 vCPU / 65 GiB) uname: BENCH_COMMAND: cargo bench --features=arrow,async,test_common,experimental,object_store --bench arrow_writerCPU Details (lscpu)Resource Usagebase (merge-base)
branch
arrow_writer — shard 3/4Node: gk3-benchmark-cluster-nap-1kxn11uo-efb27e7d-mmq9 Instance: c4a-highmem-16 (12 vCPU / 65 GiB) uname: BENCH_COMMAND: cargo bench --features=arrow,async,test_common,experimental,object_store --bench arrow_writerCPU Details (lscpu)Resource Usagebase (merge-base)
branch
arrow_writer — shard 4/4Node: gk3-benchmark-cluster-nap-1kxn11uo-efb27e7d-wpl2 Instance: c4a-highmem-16 (12 vCPU / 65 GiB) uname: BENCH_COMMAND: cargo bench --features=arrow,async,test_common,experimental,object_store --bench arrow_writerCPU Details (lscpu)Resource Usagebase (merge-base)
branch
File an issue against this benchmark runner |
|
🤖 Benchmark starting (GKE) | trigger Target: writer_overhead Sharding factor: 1 (1 workers). Comparing share_write_schemas (ba9d4b8) to 7dcc160 (merge-base) diff Run configurationrun benchmark writer_overhead
shards: 1
Results will be posted when all workers finish. File an issue against this benchmark runner |
|
run benchmark writer_overhead |
|
🤖 Benchmark starting (GKE) | trigger Target: writer_overhead Sharding factor: 1 (1 workers). Comparing share_write_schemas (ba9d4b8) to 7dcc160 (merge-base) diff Run configurationrun benchmark writer_overhead
shards: 1
Results will be posted when all workers finish. File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Target: writer_overhead Comparing share_write_schemas (ba9d4b8) to 7dcc160 (merge-base) diff Run configurationrun benchmark writer_overhead
shards: 1
DetailsPer-runner informationwriter_overhead — shard 1/1Node: gk3-benchmark-cluster-nap-1qdxlfmx-b0cf750b-8frp Instance: c4a-highmem-16 (12 vCPU / 65 GiB) uname: BENCH_COMMAND: cargo bench --features=arrow,async,test_common,experimental,object_store --bench writer_overheadCPU Details (lscpu)Resource Usagebase (merge-base)
branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Target: writer_overhead Comparing share_write_schemas (ba9d4b8) to 7dcc160 (merge-base) diff Run configurationrun benchmark writer_overhead
shards: 1
DetailsPer-runner informationwriter_overhead — shard 1/1Node: gk3-benchmark-cluster-nap-1qdxlfmx-da3d3c4e-7q6k Instance: c4a-highmem-16 (12 vCPU / 65 GiB) uname: BENCH_COMMAND: cargo bench --features=arrow,async,test_common,experimental,object_store --bench writer_overheadCPU Details (lscpu)Resource Usagebase (merge-base)
branch
File an issue against this benchmark runner |
| let schema_plan = | ||
| ArrowWriteSchemaPlan::try_new(file_writer.schema_descr_ptr().as_ref(), &arrow_schema) | ||
| .map(Arc::new); |
There was a problem hiding this comment.
It seems like the ceremony of storing errors for later use is to avoid changing the signature on this function (ArrowRowGroupWriterFactory::new, which is public). But it feels to me that the right long-term shape is to add ArrowRowGroupWriterFactory::try_new(...) -> Result<...>. I would suggest we add that method and either:
(1) Remove new() as a hard breaking change (but an easy one to fix, I'd guess 1 LOC in datafusion or any downstream). It should be fine to ship in next major release.
(2) Keep new() and have it use this machinery internally, but deprecate it and suggest using try_new instead. We could at least remove this complexity in the future.
If we go with (2) we may have some bugs to fix. My agent claims that ParquetError::ArrowError would end up in General. I did not verify this.
There was a problem hiding this comment.
There was a problem hiding this comment.
I think it's worth doing a quick (AI assisted) survey of call-sites at least in DataFusion. But I agree it's worth just biting the bullet and making the breaking change (1).
There was a problem hiding this comment.
@alamb, I know you generally prefer avoiding breaking changes. Do you have a preference here? It feels like (2) mostly delays the inevitable, so I’m inclined to bite the bullet now.
We can make the breaking change, but then we would need to wait for the next major release
Retain the ArrowError variant and original message when caching and replaying schema-plan failures instead of converting them to `General`. Add regression coverage for error variant and message preservation, including repeated factory calls with nullable map-key schemas. Signed-off-by: Hippolyte Barraud <hippolyte.barraud@datadoghq.com>
b7844da to
97447bd
Compare
alamb
added
the
api-change
Add `ArrowRowGroupWriterFactory::try_new` to return schema-planning errors immediately while preserving the original ParquetError. Deprecate `new()`` for removal in 61.0.0, retaining its deferred-error behavior for compatibility. Document the cached-error state and cleanup needed when the deprecated constructor is removed. Migrate internal callers, examples, benchmarks, and integration tests to try_new(). Extend tests to verify immediate and deferred failures. Signed-off-by: Hippolyte Barraud <hippolyte.barraud@datadoghq.com>
|
@etseidl @adriangb 33f6549 adds the new I propose we merge this as-is since it doesn’t break existing users. I’ll open a separate PR for the actual API removal so we can apply the Thanks! |
|
Sounds good. I'm worried about the complexity of manually passing around and re-surfacing errors, but it seems like it might be the right tradeoff to avoid a breaking change. If it causes issues we can always remove it as a bugfix, and we'll remove it before next version anyway. |
etseidl
left a comment
etseidl
left a comment
There was a problem hiding this comment.
I'm feeling deja vu...haven't we been here before 😉
Looks good to me (and my agent). One minor nit I found on my own (it's probably wrong :P)
Co-authored-by: Ed Seidl <etseidl@users.noreply.github.com>
Which issue does this PR close?
Rationale for this change
The eager Arrow writer repeatedly derives the same schema-only field-to-leaf mapping when creating and validating row groups, even though that metadata is invariant for the lifetime of the writer factory.
What changes are included in this PR?
Build an immutable write-schema plan once per factory and share it across row groups and column writers. Reuse the cached target fields, leaf ranges, and Parquet descriptors for both ordinary and CDC paths, while continuing to derive leaves and levels from each incoming batch. This removes repeated schema traversal without retaining input arrays or changing dense, dictionary, REE, view, required-null, or leaf-ownership semantics.
Are these changes tested?
All tests passing.
Are there any significant user-facing changes?
None.