feat(writer): auto-set referenced_data_file on PositionDeleteFileWriter - #169

* 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]>

* 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]>

* 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]>

* perf(iceberg): bound rewrite-manifests memory + calibrate manifest size estimate (risingwavelabs#185)

* perf(iceberg): bound rewrite-manifests memory + calibrate size estimate

Two related improvements to the size-bounded `RewriteManifestsAction`
cluster path added in risingwavelabs#174.

1. Incremental flush (memory) — IGLOO-283. Seal and write each full
   manifest writer inside the streaming loop instead of parking every
   sealed writer in a `full_writers` buffer until the end. Previously a
   large single-key (e.g. unpartitioned) rewrite buffered every output
   entry in memory at once — O(total entries) — and could OOM the
   process (observed: a ~10k-entry, 1000-column table OOMKilled a 2 GiB
   worker). Peak memory is now bounded by the in-flight input manifests
   plus the still-open (below-target) writers (<=1 per active cluster
   key). The load stream keeps its bounded concurrency; entry routing
   stays sequential because writers are stateful.

2. Calibrate `estimate_manifest_entry_size` — IGLOO-284. The flat per-KV
   constants overcounted the real (uncompressed) Avro footprint by ~1.7x,
   so writers sealed at ~0.6x the target and produced manifests well
   under `commit.manifest.target-size-bytes`. Iceberg encodes the stat
   maps as Avro arrays of {key:int, value:...} records, so per-KV costs
   are much smaller; the bound maps now also count the real serialized
   value bytes (via a non-allocating `datum_serialized_len`) so
   wide/untruncated bounds are no longer undercounted. Measured within
   ~15% of the on-disk length on a 60-column manifest (vs ~1.85x before).

Also removes the now-unused `try_for_each_manifest` helper: the cluster
path was its only caller, and the loop is inlined so `write_manifest_file`
can be awaited at seal time.

Tests:
- writer: `test_estimated_size_matches_on_disk_manifest` asserts the
  estimate is within +/-30% of a real on-disk manifest length (fails
  under the old constants).
- rewrite_manifests: `test_rewrite_manifests_rolls_over_and_preserves_count`
  asserts a hot key splits into multiple manifests with no entry lost.

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <[email protected]>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <[email protected]>

---------

Co-authored-by: Copilot Autofix powered by AI <[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: Dylan <[email protected]>
Co-authored-by: Alex Graham <[email protected]>
Co-authored-by: Copilot Autofix powered by AI <[email protected]>
Co-authored-by: Cursor <[email protected]>