Repository navigation
Performance improvments - #1
Merged
Merged
Conversation
![baz-reviewer[bot]](https://avatars.githubusercontent.com/in/933528?s=60&v=4){kind=small}
| repository = "https://github.com/apache/iceberg-rust" | ||
| # Check the MSRV policy in README.md before changing this | ||
| rust-version = "1.88" | ||
| rust-version = "1.92" |
There was a problem hiding this comment.
This 1.92 MSRV bump isn’t reflected in README.md/CHANGELOG.md, so could we update the user-facing docs alongside it?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In Cargo.toml around
line 39, where `rust-version = "1.92"` is set (the MSRV bump), update the repo’s
user-facing documentation to match. Specifically, edit README.md to clearly state
Iceberg Rust’s current minimum supported Rust version is 1.92 (and adjust any language
about “rolling MSRV” or crates.io links), and update CHANGELOG.md to add a new entry
noting the MSRV bump from 1.88 to 1.92. Also search for any other
MSRV/Build-from-source/PyO3/Python binding build prerequisites references in the repo
and update them so downstream users see the new minimum before build-time failures.
Comment on lines
+404
to
+406
| pub trait IcebergWriterBuilder<I = DefaultInput, O = DefaultOutput>: Send + Sync + 'static | ||
| where I: Send + 'static | ||
| { |
There was a problem hiding this comment.
Adding I: Send + 'static to the public IcebergWriterBuilder trait breaks downstream impls with non-Send/'static inputs, and the same bound is already on IcebergWriter; should we keep it off the trait and constrain only the async paths that need it?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
`crates/iceberg/src/writer/mod.rs` around lines 404-406, the `pub trait
IcebergWriterBuilder<I = DefaultInput, O = DefaultOutput>` adds a new public
where-clause `I: Send + 'static`, which is a breaking change for downstream impls.
Refactor by removing or relaxing these public bounds on the trait header and instead
apply `Send + 'static` only on the specific implementations/code paths that actually
require it (e.g., the async execution boundary inside writer/builder logic), using
method-level or internal where-clauses rather than constraining every downstream
`IcebergWriterBuilder` impl. Also check `IcebergWriter` around lines 415-417 for
duplicated bounds and ensure the `Send + 'static` requirement is only enforced where
necessary, so external custom writers/builders don’t have to change their type
signatures unless required by the actual async/threading usage.
Comment on lines
+183
to
+186
| pub fn split( | ||
| &self, | ||
| batch: &RecordBatch, | ||
| ) -> Result<Vec<(PartitionKey, RecordBatch, Vec<usize>)>> { |
There was a problem hiding this comment.
RecordBatchPartitionSplitter::split is public and changing its return type to Result<Vec<(PartitionKey, RecordBatch, Vec<usize>)>> breaks downstream destructuring — should we keep the old tuple shape and expose row IDs via a separate API?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
crates/iceberg/src/arrow/record_batch_partition_splitter.rs around lines 179-223, the
public method `RecordBatchPartitionSplitter::split` currently returns
`Result<Vec<(PartitionKey, RecordBatch, Vec<usize>)>>`, which breaks downstream users
expecting the old 2-tuple `(PartitionKey, RecordBatch)`. Refactor by restoring `split`
to the original 2-tuple return type and extracting the existing row-id producing logic
into a new public API (e.g., `split_with_row_ids`) that returns the 3-tuple. Update the
in-repo tests and any internal callers to use the new row-id API, while ensuring the
original `split` behavior remains unchanged for backward compatibility.
Comment on lines
83
to
85
| (PrimitiveType::Date, PrimitiveLiteral::Int(v)) => { | ||
| Date32Type::to_naive_date(*v).year() - UNIX_EPOCH_YEAR | ||
| Date32Type::to_naive_date_opt(*v).unwrap().year() - UNIX_EPOCH_YEAR | ||
| } |
There was a problem hiding this comment.
Could we propagate Err(DataInvalid) from Year::transform_literal and Month::transform_literal instead of unwrapping Date32Type::to_naive_date_opt(*v)? Raw i32 Iceberg date literals can hit None near the bounds and panic here.
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
crates/iceberg/src/transform/temporal.rs around lines 81-86 in `impl TransformFunction
for Year::transform_literal`, replace the `Date32Type::to_naive_date_opt(*v).unwrap()`
with logic that handles the `None` case by returning an `Err` (e.g.,
`ErrorKind::DataInvalid`/invalid date literal) instead of panicking. Do the same in
`impl TransformFunction for Month::transform_literal` around lines 179-183 by removing
the duplicated `unwrap()` calls for `year()` and `month0()` and propagating an error
when the conversion returns None. Optionally factor the `i32 days -> (year, month0)`
conversion into a small private helper to avoid repeating the same error-handling
pattern across Year and Month.
Comment on lines
+52
to
+57
| /// Write data with a partition key and return position information for each record. | ||
| async fn write_with_position( | ||
| &mut self, | ||
| partition_key: PartitionKey, | ||
| input: I, | ||
| ) -> Result<Vec<PositionDeleteInput>>; |
There was a problem hiding this comment.
Adding write_with_position without a default breaks external PartitioningWriter impls; should we make it a default method or split it into an extension trait to keep compatibility?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
crates/iceberg/src/writer/partitioning/mod.rs around lines 52-57, the
`PartitioningWriter` trait adds `async fn write_with_position(...) ->
Result<Vec<PositionDeleteInput>>` as a required method with no default body, causing a
breaking compile error for downstream crates that only implement `write`/`close`.
Refactor to preserve compatibility by providing a default implementation for
`write_with_position` (e.g., default to calling `write` and returning an empty
`Vec<PositionDeleteInput>`), or alternatively move `write_with_position` into a separate
extension trait and implement it for existing writers via a blanket impl. Ensure in-repo
writers keep overriding it where they actually produce position delete info.
Comment on lines
641
to
+644
| } else { | ||
| // inner writer is not initialized yet | ||
| 0 | ||
| // No row group has been flushed yet; return only committed bytes. | ||
| // should_roll won't fire until the first flush gives us a real ratio. | ||
| flushed |
There was a problem hiding this comment.
current_written_size() returning only bytes_written() before the first flush leaves RollingFileWriter::should_roll() at 0, so the first file can grow past target_file_size; should we include in_progress_size() until the first flush like before?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
crates/iceberg/src/writer/file_writer/parquet_writer.rs around lines 641-645 inside
`impl CurrentFileStatus for ParquetWriter { fn current_written_size(&self) -> usize }`,
the else-branch currently returns only `flushed` when `accumulated_uncompressed_proxy ==
0`, which ignores `inner.in_progress_size()` until the first row-group flush. This
causes `RollingFileWriter::should_roll()` to see an unrealistically small size for the
initial buffered batches and allows the first file to grow past `target_file_size`.
Refactor `current_written_size()` so that when no flush-based compression ratio has been
established yet, it includes `in_progress` (matching the previous behavior), and only
switch to the ratio-based `flushed + in_progress * ratio` once
`accumulated_uncompressed_proxy` is > 0. Add or adjust a unit/integration assertion (if
available) to cover rolling behavior before the first flush.
Comment on lines
+36
to
+38
| pyo3 = { version = "0.28", features = ["extension-module", "abi3-py310"] } | ||
| iceberg-datafusion = { path = "../../crates/integrations/datafusion" } | ||
| datafusion-ffi = { version = "52.1" } | ||
| datafusion-ffi = "53.0.0" |
There was a problem hiding this comment.
Bumping pyo3 = 0.28 / datafusion-ffi = 53.0.0 leaves bindings/python/uv.lock on datafusion==52.*, so make install/make test can hit the >=53 test requirement in bindings/python/tests/test_datafusion_table_provider.py — should we update the lockfile/dev dependency together?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
bindings/python/Cargo.toml around lines 36-38, the dependency bumps pyo3 to 0.28 and
datafusion-ffi to 53.0.0 but the Python dev environment remains pinned to
datafusion==52.*. Update bindings/python/uv.lock (and any Python dependency spec used by
bindings/python/Makefile’s `uv sync --group dev`) so the installed `datafusion`
package matches the new minimum (>=53) required by
bindings/python/tests/test_datafusion_table_provider.py. Verify by running the
documented `make install`/`make test` path that local builds no longer install
DataFusion 52 and that the tests pass.
Comment on lines
165
to
+175
| PrimitiveType::Date => "date".to_string(), | ||
| PrimitiveType::Timestamp => "timestamp".to_string(), | ||
| PrimitiveType::TimestampNs => "timestamp_ns".to_string(), | ||
| PrimitiveType::Timestamptz | PrimitiveType::TimestamptzNs => { | ||
| return Err(Error::new( | ||
| ErrorKind::FeatureUnsupported, | ||
| format!("Conversion from {p:?} is not supported"), | ||
| )); | ||
| } | ||
| PrimitiveType::Timestamptz => "timestamp".to_string(), | ||
| PrimitiveType::TimestamptzNs => "timestamp_ns".to_string(), | ||
| // PrimitiveType::Timestamptz | PrimitiveType::TimestamptzNs => { | ||
| // return Err(Error::new( | ||
| // ErrorKind::FeatureUnsupported, | ||
| // format!("Conversion from {p:?} is not supported"), | ||
| // )); | ||
| // } |
There was a problem hiding this comment.
PrimitiveType::Timestamptz/TimestamptzNs are silently mapped to plain Glue timestamp/timestamp_ns, losing timezone semantics — should we gate this behind a flag or return ErrorKind::FeatureUnsupported by default? Also, can we remove the commented-out match arms for these variants?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
In crates/catalog/glue/src/schema.rs around lines 165-175, inside the
`SchemaVisitor::primitive` match: 1. Remove the commented-out block for
`PrimitiveType::Timestamptz | PrimitiveType::TimestamptzNs`. 2. For the live arms,
either restore `ErrorKind::FeatureUnsupported` as the default or introduce an explicit
opt-in flag/config on `convert_to_glue_table()`/the builder to allow lossy timezone
dropping; the default should be non-lossy and fail. 3. Add/adjust tests (including
`test_schema_with_simple_fields` expectations for c14/c15) to verify both strict and
opt-in behavior.
Co-authored-by: xxhx <[email protected]>
Signed-off-by: Mingzhuo Yin <[email protected]>
risingwavelabs#161) * fix: avoid panic in snapshot summary update_totals on malformed values Previously update_totals called .unwrap() when parsing snapshot summary properties (total-*, added-*, removed-*) as u64. A value written by a non-Rust writer that exceeds u64::MAX (or is otherwise malformed) caused ParseIntError { kind: PosOverflow } and aborted the writer. The same function used checked subtraction (`new_total -= value`), which could panic in debug builds on a removed > previous + added skew from corrupt history. Replace both with parse-or-zero + saturating arithmetic so a single corrupt prior summary cannot indefinitely break compaction / commit for a table. Behavior matches Java Iceberg's SnapshotSummary, which treats missing or unparseable values as 0. Add tests for the three failure modes (overflowing previous total, removed > previous + added, non-numeric value). * fix: avoid panic in snapshot summary truncate overwrite on malformed values `update_snapshot_summaries` routes the truncate-overwrite path through `truncate_table_summary`, which parsed previous-summary totals via `get_prop` as `i32` and propagated `ParseIntError` with `?`. The caller then did `.map_err(...).unwrap()`, so a malformed, empty, or u64-overflowing previous total (legal in summaries written by other engines) would panic the writer. The `i32` ceiling was also too narrow for real-world `TOTAL_FILE_SIZE` values (~2.1 GB), making the panic reachable on perfectly valid input. Apply the same parse-or-zero / non-panicking policy already used in `update_totals`: - Replace `get_prop` with `previous_total_or_zero`, which parses as `u64` and returns 0 for missing, malformed, or overflowing values. - Make `truncate_table_summary` infallible and collapse the six near-identical blocks into a single (total, removed) loop. - Drop the `.map_err(...).unwrap()` in `update_snapshot_summaries`. Add a test that drives the full truncate-overwrite path with a previous summary containing garbage, empty, u64-overflowing, i32-overflowing (but valid u64), well-formed, and missing totals. Asserts no panic, all totals reset to 0, malformed/missing previous totals do not emit a bogus removed_* counter, and well-formed totals (including the value that would have broken the old i32 parser) still flow into removed_*.
* fix puffin file reader * Remove REST commit error body logging
* Add myself to code owners Added vovacf201 to code owners so I could approve PRs. * fix: typos (mutli->multi, unparseable->unparsable) --------- Co-authored-by: Alex Graham <[email protected]>
- Add add_entry_preserving_status; use in overwrite/rewrite manifest rewrite - Tolerate missing file_sequence_number on Existing/Deleted when sequence set - Track removed files in summaries; previous snapshot via branch ref - fix(glue): support timestamptz / timestamptz_ns in Glue schema conversion
rewrite_files: replace broken concurrent rewrite implementation with a correct two-phase approach — Phase 1 loads all manifests concurrently via `buffered` (ordered, preserving manifest-list order for V3 first_row_id correctness), Phase 2 applies the original sequential filter-and-rewrite logic over the pre-loaded results. Removes the commented-out dead code block, the incomplete rewrite_tasks pipeline that never populated its task list, and the use-after-move on manifest. snapshot: remove duplicate `removed_data_files` struct field introduced by the PR, and fix double `remove_file()` call in `summary()` that was causing deleted-records and removed-files-size to be reported at 2x their actual values for Replace operations. parquet_writer: fix first-flush ratio miss by sampling in_progress_size() immediately before each write instead of reading the stale last_in_progress_size field (which was always 0 on the first call). Removes the now-unnecessary last_in_progress_size field. Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
amitgilad3
added a commit
that referenced
this pull request
Jul 14, 2026
* fix: speed commits + improve parquet writer est * fix(rest): refresh token after unauthorized response (risingwavelabs#157) * feat: support dropping schema fields (risingwavelabs#155) Co-authored-by: xxhx <[email protected]> * fix: skip opendal TimeoutLayer under madsim (risingwavelabs#160) Signed-off-by: Mingzhuo Yin <[email protected]> * fix: avoid panic in snapshot summary update_totals on malformed values (risingwavelabs#161) * fix: avoid panic in snapshot summary update_totals on malformed values Previously update_totals called .unwrap() when parsing snapshot summary properties (total-*, added-*, removed-*) as u64. A value written by a non-Rust writer that exceeds u64::MAX (or is otherwise malformed) caused ParseIntError { kind: PosOverflow } and aborted the writer. The same function used checked subtraction (`new_total -= value`), which could panic in debug builds on a removed > previous + added skew from corrupt history. Replace both with parse-or-zero + saturating arithmetic so a single corrupt prior summary cannot indefinitely break compaction / commit for a table. Behavior matches Java Iceberg's SnapshotSummary, which treats missing or unparseable values as 0. Add tests for the three failure modes (overflowing previous total, removed > previous + added, non-numeric value). * fix: avoid panic in snapshot summary truncate overwrite on malformed values `update_snapshot_summaries` routes the truncate-overwrite path through `truncate_table_summary`, which parsed previous-summary totals via `get_prop` as `i32` and propagated `ParseIntError` with `?`. The caller then did `.map_err(...).unwrap()`, so a malformed, empty, or u64-overflowing previous total (legal in summaries written by other engines) would panic the writer. The `i32` ceiling was also too narrow for real-world `TOTAL_FILE_SIZE` values (~2.1 GB), making the panic reachable on perfectly valid input. Apply the same parse-or-zero / non-panicking policy already used in `update_totals`: - Replace `get_prop` with `previous_total_or_zero`, which parses as `u64` and returns 0 for missing, malformed, or overflowing values. - Make `truncate_table_summary` infallible and collapse the six near-identical blocks into a single (total, removed) loop. - Drop the `.map_err(...).unwrap()` in `update_snapshot_summaries`. Add a test that drives the full truncate-overwrite path with a previous summary containing garbage, empty, u64-overflowing, i32-overflowing (but valid u64), well-formed, and missing totals. Asserts no panic, all totals reset to 0, malformed/missing previous totals do not emit a bogus removed_* counter, and well-formed totals (including the value that would have broken the old i32 parser) still flow into removed_*. * fix: fix iceberg V3 puffin file reader (risingwavelabs#162) * fix puffin file reader * Remove REST commit error body logging * Add myself to code owners (risingwavelabs#164) * Add myself to code owners Added vovacf201 to code owners so I could approve PRs. * fix: typos (mutli->multi, unparseable->unparsable) --------- Co-authored-by: Alex Graham <[email protected]> * fix(iceberg): preserve manifest metadata and snapshot summary context - Add add_entry_preserving_status; use in overwrite/rewrite manifest rewrite - Tolerate missing file_sequence_number on Existing/Deleted when sequence set - Track removed files in summaries; previous snapshot via branch ref - fix(glue): support timestamptz / timestamptz_ns in Glue schema conversion * fix(iceberg): fix concurrent manifest loading and parquet size tracking rewrite_files: replace broken concurrent rewrite implementation with a correct two-phase approach — Phase 1 loads all manifests concurrently via `buffered` (ordered, preserving manifest-list order for V3 first_row_id correctness), Phase 2 applies the original sequential filter-and-rewrite logic over the pre-loaded results. Removes the commented-out dead code block, the incomplete rewrite_tasks pipeline that never populated its task list, and the use-after-move on manifest. snapshot: remove duplicate `removed_data_files` struct field introduced by the PR, and fix double `remove_file()` call in `summary()` that was causing deleted-records and removed-files-size to be reported at 2x their actual values for Replace operations. parquet_writer: fix first-flush ratio miss by sampling in_progress_size() immediately before each write instead of reading the stale last_in_progress_size field (which was always 0 on the first call). Removes the now-unnecessary last_in_progress_size field. Co-Authored-By: Claude Sonnet 4.6 <[email protected]> --------- Signed-off-by: Mingzhuo Yin <[email protected]> Co-authored-by: Xinhao Xu <[email protected]> Co-authored-by: xxhx <[email protected]> Co-authored-by: Mingzhuo Yin <[email protected]> Co-authored-by: vovacf201 <[email protected]> Co-authored-by: Dylan <[email protected]> Co-authored-by: Alex Graham <[email protected]> Co-authored-by: Claude Sonnet 4.6 <[email protected]>
amitgilad3
added a commit
that referenced
this pull request
Jul 14, 2026
* fix(transaction): account for added delete files in snapshot summary (risingwavelabs#166) SnapshotProducer::summary() fed only added_data_files into the SnapshotSummaryCollector. The added_delete_files vec is written to delete manifests by manifest_file(), but never reached the collector, so added-delete-files / added-position-deletes / added-equality-deletes were never emitted and update_totals rolled total-delete-files / total-position-deletes / total-equality-deletes forward as previous(0) + added(0) on every commit — summaries reported zero delete files on tables whose manifest lists carried hundreds of live ones. Engines that consult the summary to pick a read path (or to reject unsupported delete types up front) then mis-plan against the actual manifests. Add the missing added_delete_files loop in summary(), mirroring the added_data_files one; the collector already branches on data_file.content to update the matching position/equality delete counters. Add a regression test that builds a producer with one data file plus position- and equality-delete files against a parent snapshot with non-zero delete totals, asserting the added-* delete counters are emitted and the total-* counters roll forward instead of staying at zero. * fix(writer): sort position-delete entries globally by `(file_path, pos)` (risingwavelabs#167) * fix(writer): sort position-delete entries globally by (file_path, pos) DeltaWriter previously called write_position_deletes() once per insert/delete pass, producing multiple independently sorted runs concatenated into a single parquet file. Readers that rely on the Iceberg v2 sort guarantee would fail to apply deletions correctly. Buffer all position-delete inputs in a BTreeMap<Arc<str>, Vec<i64>> keyed by file path. At close(), drain the buffer and emit it sorted by (file_path, pos) in fixed-size chunks so the underlying RollingFileWriter retains its file-splitting behavior. Add two regression tests that confirm per-file and cross-write global ordering. Co-Authored-By: Claude Sonnet 4.6 <[email protected]> * style(writer): apply rustfmt formatting to delta_writer Co-Authored-By: Claude Sonnet 4.6 <[email protected]> --------- Co-authored-by: Claude Sonnet 4.6 <[email protected]> * feat(writer): auto-set referenced_data_file on PositionDeleteFileWriter (risingwavelabs#169) * feat(writer): auto-set referenced_data_file on PositionDeleteFileWriter close When a position delete writer closes with exactly one output file and all written entries reference a single distinct data file path, set referenced_data_file on the resulting DataFile. This allows readers to scope delete file application without scanning the file_path column, matching Iceberg Java behavior and eliminating unnecessary I/O. Co-Authored-By: Claude Sonnet 4.6 <[email protected]> * fix(writer): set referenced_data_file on all rolled position delete files The builders.len() == 1 guard prevented referenced_data_file from being set when the rolling writer produced multiple output files due to size rollover. Since rolling is purely size-driven, distinct_paths.len() == 1 already guarantees that every rolled output file contains entries for the same single data file. Removing the builders.len() check ensures the field is set on all output files, so readers can skip them for non-matching data files even in high-volume single-file delete scenarios. Co-Authored-By: Claude Sonnet 4.6 <[email protected]> * refactor(writer): clean up test comment in position_delete_file_writer Co-Authored-By: Claude Sonnet 4.6 <[email protected]> --------- Co-authored-by: Claude Sonnet 4.6 <[email protected]> * fix: stream manifest lists during snapshot expiration to bound memory (risingwavelabs#170) RemoveSnapshotAction::commit, when clear_expired_meta_data is enabled, computes the set of reachable partition-spec ids by calling load_manifest_lists(retained_snapshots).try_collect(), which holds every retained snapshot's manifest list in memory at once. It only needs the small set of reachable spec ids. On tables whose retained manifest lists are large (e.g. high-churn tables without manifest compaction), this is O(N) in memory and can OOM the process during commit. Add a streaming helper `for_each_manifest_list` that loads manifest lists with bounded concurrency (buffer_unordered) and invokes a closure on each, dropping the list immediately afterwards. RemoveSnapshotAction now folds the reachable partition-spec ids through it instead of collecting a Vec. The result is identical; peak memory drops from O(all retained manifest lists) to O(concurrency lists in flight). reachable_schemas already comes from Snapshot::schema_id() and needs no manifest load. * fix: stream manifest loading in snapshot-expiration file cleanup (risingwavelabs#171) * fix: stream manifest lists during snapshot expiration to bound memory RemoveSnapshotAction::commit, when clear_expired_meta_data is enabled, computes the set of reachable partition-spec ids by calling load_manifest_lists(retained_snapshots).try_collect(), which holds every retained snapshot's manifest list in memory at once. It only needs the small set of reachable spec ids. On tables whose retained manifest lists are large (e.g. high-churn tables without manifest compaction), this is O(N) in memory and can OOM the process during commit. Add a streaming helper `for_each_manifest_list` that loads manifest lists with bounded concurrency (buffer_unordered) and invokes a closure on each, dropping the list immediately afterwards. RemoveSnapshotAction now folds the reachable partition-spec ids through it instead of collecting a Vec. The result is identical; peak memory drops from O(all retained manifest lists) to O(concurrency lists in flight). reachable_schemas already comes from Snapshot::schema_id() and needs no manifest load. * fix: stream manifest loading in snapshot-expiration file cleanup ReachableFileCleanupStrategy (used after a remove_snapshots commit when clear_expired_files is enabled) loaded entire collections up front: - clean_files / prune_referenced_manifests collected Vec<(SnapshotRef, ManifestList)> via load_manifest_lists, then folded the entries into HashSets. - find_files_to_delete collected Vec<(ManifestFile, Manifest)> via load_manifests, then folded data-file paths into a HashSet. In each case the fully-loaded Vec was held in memory simultaneously with the derived set, roughly doubling peak memory on tables with many/large manifests. Add a streaming `for_each_manifest` helper (analogous to `for_each_manifest_list`) and convert the three cleanup sites to fold entries into their sets while dropping each loaded list/manifest immediately. Peak memory for the load drops from O(all loaded lists/manifests + derived set) to O(concurrency in flight + derived set). Behavior is unchanged. Note: this bounds the *loading*; the derived sets (manifest files, data-file paths) are still O(N) by nature of the diff-based cleanup algorithm. Fully bounding cleanup needs a batched/partitioned deletion strategy (separate follow-up). * fix: stream manifest loading in rewrite-manifests and append/overwrite validation (risingwavelabs#172) * fix: stream manifest loading in rewrite-manifests and append/overwrite validation Two more spots loaded an entire manifest collection up front and held it for the duration of a fold: - rewrite_manifests routes entries from `manifests_to_rewrite` into clustered manifest writers, but first collected every manifest via load_manifests. - the append/overwrite validation path scans the current snapshot's manifests to detect duplicate adds and validate deletes; it also loaded all of them via load_manifests before scanning, even though it early-exits once both checks are satisfied. Add a fallible streaming helper `try_for_each_manifest` (analogous to `for_each_manifest`) and use it in rewrite_manifests to route entries while dropping each manifest immediately. Convert the validation scan to a streamed `buffer_unordered` loop that drops each manifest after scanning and stops pulling manifests entirely once the early-exit condition is met. Behavior is unchanged (both paths already processed manifests in buffer_unordered completion order). Peak memory drops from O(all loaded manifests) to O(concurrency in flight), and the validation path additionally avoids loading manifests past the early-exit. * docs: fix stale comment in append/overwrite validation scan The leading comment said the block loads all manifests concurrently, but it now streams manifests one at a time with early-exit. Update it to describe the scan's purpose, removing the contradiction with the streaming note below. * Add rewrite manifests target size (risingwavelabs#174) * fix: stream manifest loading in rewrite-manifests and append/overwrite validation Two more spots loaded an entire manifest collection up front and held it for the duration of a fold: - rewrite_manifests routes entries from `manifests_to_rewrite` into clustered manifest writers, but first collected every manifest via load_manifests. - the append/overwrite validation path scans the current snapshot's manifests to detect duplicate adds and validate deletes; it also loaded all of them via load_manifests before scanning, even though it early-exits once both checks are satisfied. Add a fallible streaming helper `try_for_each_manifest` (analogous to `for_each_manifest`) and use it in rewrite_manifests to route entries while dropping each manifest immediately. Convert the validation scan to a streamed `buffer_unordered` loop that drops each manifest after scanning and stops pulling manifests entirely once the early-exit condition is met. Behavior is unchanged (both paths already processed manifests in buffer_unordered completion order). Peak memory drops from O(all loaded manifests) to O(concurrency in flight), and the validation path additionally avoids loading manifests past the early-exit. * docs: fix stale comment in append/overwrite validation scan The leading comment said the block loads all manifests concurrently, but it now streams manifests one at a time with early-exit. Update it to describe the scan's purpose, removing the contradiction with the streaming note below. * feat(rewrite_manifests): bound output manifest size to target (8 MiB) The cluster_by path of RewriteManifestsAction routed every alive entry for a partition cluster key into a single ManifestWriter, producing one unbounded manifest per key. On high-ingest tables this yields huge manifests and an oversized manifest list. Add size-based rollover honoring commit.manifest.target-size-bytes (default 8 MiB), matching Spark and the append/overwrite/rewrite_files paths: - ManifestWriter tracks a running estimated serialized size and exposes estimated_manifest_size(); estimate captures file path + per-column stat maps + fixed overhead. - When an open writer reaches the target it is sealed (deferred finalize, since the streaming routing closure is sync) and the next entry for that key opens a fresh writer. File count is preserved (created==replaced still holds); manifest ordering is not required for correctness. Adds a unit test for the size signal. * test(rewrite_manifests): cover target-size manifest rollover Add integration test requested in PR review: set commit.manifest.target-size-bytes=1 so the cluster_by path rolls each alive entry into its own output manifest, verifying that a hot cluster key no longer produces a single unbounded manifest. Refactors the shared setup helper to accept table properties on creation. * chore: update code owners (add @netgusto & @agaddis02, remove @jonathanc-n) (risingwavelabs#176) * chore: add @netgusto to code owners * chore: remove @jonathanc-n; sync owner list * chore: add @agaddis02 to code owners * fix(rewrite_manifests): honor target size from snapshot properties (risingwavelabs#175) The rewrite-manifests target size (`commit.manifest.target-size-bytes`) was only read from the persistent *table* properties. Callers that set it per-commit via `set_snapshot_properties` (the transient, non-persisted mechanism) had no effect on manifest sizing — the value was recorded in the snapshot summary but ignored when writing manifests. Resolve the target size in order: staged snapshot property (this action's `set_snapshot_properties`) first, then the table property, then the Iceberg default (8 MiB). This lets a caller size the manifests for a single rewrite without mutating persistent table configuration, while preserving the existing table-property and default behavior. Add an integration test proving the snapshot property overrides a conflicting table property and does not mutate it. * feat: Variant Support (risingwavelabs#145) * cherry-pick feat: Variant Support * Preserve Arrow variant extension metadata * Fix Arrow variant projection and schema conversion * Deduplicate variant integration tests by extracting shared helpers Extract `load_variant_table()` and `assert_variant_column()` to remove duplicated catalog setup and variant column assertions across tests. Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix(avro): derive unique variant record names from enclosing field id Avro record names must be unique within a schema, so a fixed "variant" record name breaks schemas with more than one variant column. Use r{field_id} derived from the enclosing field id, mirroring Java's TypeToSchema and the current apache#2188 revision. Co-Authored-By: Claude Fable 5 <[email protected]> * fix(reader): reject shredded variant columns in projection A `typed_value` sub-field means the variant payload is shredded, which we cannot reconstruct yet. Projecting only metadata/value would silently drop data, so fail loudly instead (ported from apache#2188). Co-Authored-By: Claude Fable 5 <[email protected]> * fix(reader): derive variant extension metadata from Parquet VARIANT logical type Files written by Spark/iceberg-java annotate variant groups with the Parquet VARIANT logical type but embed no Arrow schema, so the Arrow schema used for projection is derived from the Parquet schema at read time. Without parquet's variant_experimental feature the derivation produced a plain struct with no extension metadata, and arrow_schema_to_schema recursed into the id-less variant sub-fields, failing with "Field id not found in metadata". Enable variant_experimental so the derived schema carries arrow.parquet.variant, which the schema converter and projection rely on. Co-Authored-By: Claude Fable 5 <[email protected]> * fix: auto-select format version for DataFusion-created tables Port the check_format_compatibility scope changes from the current apache#2188 revision, previously skipped to keep divergence minimal. The cherry-picked check validates table metadata at creation, which broke DataFusion CREATE TABLE with nanosecond timestamps (DataFusion's default TIMESTAMP) on the default v2 format: df_scan_all_types.toml and df_timestamp_predicate_pushdown.toml failed with "Invalid schema for vv2". These failures were masked in CI by fail-fast on the (now fixed) read_variant integration tests. - DataFusion CREATE TABLE picks the format version from Schema::calc_min_compatible_format() (at least v2, v3 when the schema requires it, e.g. timestamp_ns / variant). - check_format_compatibility now also validates initial_default floors, reports problems sorted by field id, and fixes the doubled "vv2" in the message (FormatVersion's Display already includes the "v"). - Table metadata validation checks only the current schema, matching the upstream revision. Co-Authored-By: Claude Fable 5 <[email protected]> * chore: adopt remaining upstream apache#2188 deltas to minimize divergence Port everything still differing from the current apache#2188 revision, except the fork's deliberate fixes (variant extension metadata, projection lock-step traversal, variant_experimental derivation): - Glue/HMS catalogs map variant column types ("variant" / "unknown", matching iceberg-java's HiveSchemaUtil, apache/iceberg#15964) instead of rejecting with FeatureUnsupported. - Port upstream's variant tests for partition/sort source rejection, arrow value conversion, id reassignment, schema indexing, column pruning, parquet path indexing, and type_to_arrow_type. - Align index.rs/prune_columns.rs import style with upstream (files now byte-identical). Not ported: transaction/update_schema.rs hunks (patch code the fork's older base does not have) and public-api.txt (mechanism absent from this base). Co-Authored-By: Claude Fable 5 <[email protected]> --------- Co-authored-by: Christian Thiel <[email protected]> Co-authored-by: Claude Opus 4.6 <[email protected]> * fix: rewrite/overwrite transactions mishandle DELETE manifests (risingwavelabs#179) * fix: overwrite txn corrupts DELETE manifest Signed-off-by: Mingzhuo Yin <[email protected]> * fix rewrite txn * fix: preserve delete-manifest content type and entry status on rewrite `existing_manifest` rewrites a manifest in place when only some of its entries are removed. Two defects in that path, both present in the verbatim-duplicated `overwrite_files` / `rewrite_files` twins: 1. The writer was opened with a hardcoded `ManifestContentType::Data`. For a delete manifest this makes `check_data_file` reject the first surviving entry, so any overwrite/rewrite that removes a subset of the delete files tracked by a delete manifest fails the whole commit with `DataInvalid`. This is the common case for compaction on a table with row-level deletes. Pass the source manifest's own content through. 2. Surviving entries were carried forward with `add_entry`, which restamps them as `Added` under the new snapshot and clears their file sequence number. This inflates `added_files_count` / `added_rows_count` and makes an unchanged file look newly added to incremental readers. Use `add_existing_entry`, skipping tombstones first, matching what `rewrite_manifests` and `manifest_filter` already do. Also fixes an operator-precedence bug in `delete_entries` (`overwrite_files` only): `A || B && C` parses as `A || (B && C)`, so every `PositionDeletes` entry was marked `Deleted` regardless of whether its path was in `removed_delete_file_paths`. `add_existing_entry` now has a real caller, so drop its stale `#[allow(dead_code)]`. Adds four regression tests over a shared fixture, each verified to fail against the corresponding pre-fix behaviour. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]> --------- Signed-off-by: Mingzhuo Yin <[email protected]> Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]> * fix: remove dead code * refactor: unify overwrite & rewrite txn (risingwavelabs#182) * chore: add silver-ymz to code owners (risingwavelabs#183) * Improve RewriteManifestsAction to reuse rewritten manifests on conflict retry (risingwavelabs#181) * perf(iceberg): stream manifest loading in orphan file removal (risingwavelabs#177) Orphan file removal loaded all manifest lists and manifests into memory simultaneously before scanning, holding O(all snapshots × manifests) in peak memory. collect_snapshot_files() used load_manifest_lists and load_manifests which collect every loaded item into a Vec before the caller can fold them. On tables with many snapshots or large manifests, this could OOM the process during orphan cleanup. Replaced load_manifest_lists/load_manifests with the streaming helpers for_each_manifest_list/for_each_manifest (introduced upstream in PRs risingwavelabs#170-risingwavelabs#172). Manifest list paths are extracted from snapshots upfront, then each manifest list and manifest is streamed with bounded concurrency and dropped immediately after its entries are folded into the reachable set. Removed load_manifest_lists from utils.rs as it now has zero callers. * feat: add some trival traits for DeleteVector (risingwavelabs#184) Signed-off-by: Mingzhuo Yin <[email protected]> * fix(iceberg): preserve manifest metadata and snapshot summary context - Add add_entry_preserving_status; use in overwrite/rewrite manifest rewrite - Tolerate missing file_sequence_number on Existing/Deleted when sequence set - Track removed files in summaries; previous snapshot via branch ref - fix(glue): support timestamptz / timestamptz_ns in Glue schema conversion * Performance improvments (#1) * fix: speed commits + improve parquet writer est * fix(rest): refresh token after unauthorized response (risingwavelabs#157) * feat: support dropping schema fields (risingwavelabs#155) Co-authored-by: xxhx <[email protected]> * fix: skip opendal TimeoutLayer under madsim (risingwavelabs#160) Signed-off-by: Mingzhuo Yin <[email protected]> * fix: avoid panic in snapshot summary update_totals on malformed values (risingwavelabs#161) * fix: avoid panic in snapshot summary update_totals on malformed values Previously update_totals called .unwrap() when parsing snapshot summary properties (total-*, added-*, removed-*) as u64. A value written by a non-Rust writer that exceeds u64::MAX (or is otherwise malformed) caused ParseIntError { kind: PosOverflow } and aborted the writer. The same function used checked subtraction (`new_total -= value`), which could panic in debug builds on a removed > previous + added skew from corrupt history. Replace both with parse-or-zero + saturating arithmetic so a single corrupt prior summary cannot indefinitely break compaction / commit for a table. Behavior matches Java Iceberg's SnapshotSummary, which treats missing or unparseable values as 0. Add tests for the three failure modes (overflowing previous total, removed > previous + added, non-numeric value). * fix: avoid panic in snapshot summary truncate overwrite on malformed values `update_snapshot_summaries` routes the truncate-overwrite path through `truncate_table_summary`, which parsed previous-summary totals via `get_prop` as `i32` and propagated `ParseIntError` with `?`. The caller then did `.map_err(...).unwrap()`, so a malformed, empty, or u64-overflowing previous total (legal in summaries written by other engines) would panic the writer. The `i32` ceiling was also too narrow for real-world `TOTAL_FILE_SIZE` values (~2.1 GB), making the panic reachable on perfectly valid input. Apply the same parse-or-zero / non-panicking policy already used in `update_totals`: - Replace `get_prop` with `previous_total_or_zero`, which parses as `u64` and returns 0 for missing, malformed, or overflowing values. - Make `truncate_table_summary` infallible and collapse the six near-identical blocks into a single (total, removed) loop. - Drop the `.map_err(...).unwrap()` in `update_snapshot_summaries`. Add a test that drives the full truncate-overwrite path with a previous summary containing garbage, empty, u64-overflowing, i32-overflowing (but valid u64), well-formed, and missing totals. Asserts no panic, all totals reset to 0, malformed/missing previous totals do not emit a bogus removed_* counter, and well-formed totals (including the value that would have broken the old i32 parser) still flow into removed_*. * fix: fix iceberg V3 puffin file reader (risingwavelabs#162) * fix puffin file reader * Remove REST commit error body logging * Add myself to code owners (risingwavelabs#164) * Add myself to code owners Added vovacf201 to code owners so I could approve PRs. * fix: typos (mutli->multi, unparseable->unparsable) --------- Co-authored-by: Alex Graham <[email protected]> * fix(iceberg): preserve manifest metadata and snapshot summary context - Add add_entry_preserving_status; use in overwrite/rewrite manifest rewrite - Tolerate missing file_sequence_number on Existing/Deleted when sequence set - Track removed files in summaries; previous snapshot via branch ref - fix(glue): support timestamptz / timestamptz_ns in Glue schema conversion * fix(iceberg): fix concurrent manifest loading and parquet size tracking rewrite_files: replace broken concurrent rewrite implementation with a correct two-phase approach — Phase 1 loads all manifests concurrently via `buffered` (ordered, preserving manifest-list order for V3 first_row_id correctness), Phase 2 applies the original sequential filter-and-rewrite logic over the pre-loaded results. Removes the commented-out dead code block, the incomplete rewrite_tasks pipeline that never populated its task list, and the use-after-move on manifest. snapshot: remove duplicate `removed_data_files` struct field introduced by the PR, and fix double `remove_file()` call in `summary()` that was causing deleted-records and removed-files-size to be reported at 2x their actual values for Replace operations. parquet_writer: fix first-flush ratio miss by sampling in_progress_size() immediately before each write instead of reading the stale last_in_progress_size field (which was always 0 on the first call). Removes the now-unnecessary last_in_progress_size field. Co-Authored-By: Claude Sonnet 4.6 <[email protected]> --------- Signed-off-by: Mingzhuo Yin <[email protected]> Co-authored-by: Xinhao Xu <[email protected]> Co-authored-by: xxhx <[email protected]> Co-authored-by: Mingzhuo Yin <[email protected]> Co-authored-by: vovacf201 <[email protected]> Co-authored-by: Dylan <[email protected]> Co-authored-by: Alex Graham <[email protected]> Co-authored-by: Claude Sonnet 4.6 <[email protected]> --------- Signed-off-by: Mingzhuo Yin <[email protected]> Co-authored-by: Tran Ngoc Tuan <[email protected]> Co-authored-by: Claude Sonnet 4.6 <[email protected]> Co-authored-by: vovacf201 <[email protected]> Co-authored-by: Yuhao Su <[email protected]> Co-authored-by: Christian Thiel <[email protected]> Co-authored-by: Mingzhuo Yin <[email protected]> Co-authored-by: Cursor <[email protected]> Co-authored-by: Dylan <[email protected]> Co-authored-by: Alex Graham <[email protected]> Co-authored-by: Xinhao Xu <[email protected]> Co-authored-by: xxhx <[email protected]>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
User description
Which issue does this PR close?
What changes are included in this PR?
Are these changes tested?
Generated description
Below is a concise technical summary of the changes proposed in this PR:
Optimize the manifest rewrite flow in the rewrite_files and utils utilities so concurrent manifest loading with order-preserving buffering feeds per-index rewrite tasks that drop deleted data files while retaining the original manifest ordering needed for row lineage. Improve
ParquetWriterbyte accounting by tracking observed compressed-to-uncompressed ratios from row-group flushes socurrent_written_size(and consequentlyshould_roll) reflects buffered data more accurately before the first flush.ParquetWriternow tracks compressed versus proxy-uncompressed deltas from row-group flushes socurrent_written_sizeextrapolates buffered bytes using the observed compression ratio, improving roll triggering responsiveness.Modified files (1)
Latest Contributors(2)
utils::load_manifests, classify each manifest based on deleted paths, and rewrite only those that still contain entries, reassembling the outputs in the original order and letting fully deleted manifests drop out while documenting the concurrency strategy inlib.rsmetadata.Modified files (3)
Latest Contributors(2)
Modified files (1)
Latest Contributors(2)