Skip to content

feat(read)!: take physical row positions from the parquet reader - #286

Merged
anoop-narang merged 8 commits into
mainfrom
feat/reader-row-numbers
Aug 31, 2026
Merged

anoop-narang merged 8 commits into
mainfrom
feat/reader-row-numbers

Conversation

@anoop-narang

@anoop-narang anoop-narang commented Aug 31, 2026 •

Copy link
Copy Markdown
Collaborator

Closes #130.

DuckLake needs each row's physical position in its parquet file, for positional deletes (a delete file names rows by pos) and for row lineage (rowid = row_id_start + physical_row_position).

Until DataFusion 55 there was no reader-level row-number column, so this crate synthesized positions with a FileRowNumberExec above a deliberately constrained scan: row-group-aligned partitions with plan-time seeds, wrapped in a PositionalFileSource that refused repartitioning, filter pushdown and sort pushdown, because any of those would have shifted the positions it was counting.

DataFusion 55 ships apache/datafusion#22026. positional_table_schema now builds the scan's TableSchema with a field carrying arrow-rs's RowNumber extension type, and the reader fills it from row-group offsets in the footer. Values stay true physical positions under pruning, row selection, byte-range splitting and reverse-order reads alike.

This converges with official DuckLake, which derives rowid the same way from DuckDB's reader-level COLUMN_IDENTIFIER_FILE_ROW_NUMBER (ducklake_multi_file_reader.cpp::GetVirtualColumnExpression). FileRowNumberExec was the divergence, not the fix.

What it buys

Absolute positions make pruning safe — filter(delete(R)) == delete(filter(R)), because deletion is keyed by position and a predicate is row-local — so the guards come off. DeleteFilterExec, RowIdExec and ColumnRenameExec forward filter pushdown, and resolve_positions pushes its predicate too.

Measured A/B against main, on a 5M-row catalog written by official DuckDB (41 row groups), best-of-5, noise floor ~10%:

query                          main     branch    delta
count(*)                       .0106     .0095    -13%
region + amount predicate      .0132     .0133      0%   (control: already pushed)
rowid = id over all 5M rows    .0147     .0149      0%   (nothing to prune)
selective, rowid projected     .0141     .0049    -68%
deleted ids, rowid projected   .0144     .0060    -60%

Every result matches what official DuckDB returns for the same queries on the same catalog.

The flat full-scan row is the honest part: build_row_group_partitions was already splitting positional scans by hand, so byte-range splitting is a simplification rather than a new capability. The win is pruning. On write paths (build_update_scan_with_snapshot, compaction sources) the plans are executed directly rather than through the physical optimizer, so a single-file UPDATE or rewrite_files now reads on one thread; noted in the docs.

Net ~400 lines of position-synthesis machinery and plan-time footer bookkeeping deleted.

Also fixed

Two NaN pruning bugs found while auditing the paths this touches. Both are pre-existing and both are silent data loss.

  • build_exec_for_file_with_deletes had no NanPruningBarrierExec on either branch. Its plain-scan branch (a file whose delete file resolves to no positions) has always accepted pushed predicates, so a WHERE x > C could drop NaN rows. Verified: removing the barrier makes the added test return 0 rows where 1 is correct.
  • A float min_value was trusted as a lower bound whenever contains_nan was not false, on the reasoning that NaN can only sit above a max. That holds only for positive NaN. DataFusion orders floats with total_cmp, and SELECT (-CAST('NaN' AS DOUBLE)) < CAST('-Infinity' AS DOUBLE) is true. A -NaN row sits beneath a recorded min, so plan-time file pruning could discard the file holding it: SELECT ... WHERE x < C short, and DELETE ... WHERE x < C reporting zero rows affected with the row still present. Both bounds are gated now. Catalogs written by this crate were never exposed (its writer suppresses both bounds when NaN is present); foreign catalogs, including those official DuckLake writes, were.

Breaking

  • FileRowNumberExec is removed. Nothing to shim — its whole contract was "seed positions from plan-time partition starts", which no longer exists.
  • DeleteFilterExec::try_new and RowIdExec::try_new take a trailing pos_index: usize. Pass the index of the scan's physical-position column, which is its last. The index is passed rather than resolved by name because DuckLake reserves no column names, so a table may have its own __ducklake_row_pos; that column's internal name is now suffixed when it would collide.

Testing

  • tests/it/positional_pushdown_tests.rs (new): NaN safety per path, pruning and splitting asserted on the reader's own metrics (11 of 12 row groups pruned; skew < 100%) rather than on plan strings, deletes under pruning, LIMIT after deletes, predicates on renamed and synthetic columns, and a table with a real column named __ducklake_row_pos.
  • A differential test reading an official-DuckDB-written catalog with a defaulted column through the new pushdown path.
  • Encryption: rowid on a PME file, pinning that with_virtual_columns and with_file_decryption_properties coexist — untested upstream.
  • The old plan-shape test asserted DataSourceExec was the direct child of FileRowNumberExec; replaced rather than loosened.

Each NaN and pushdown guard was verified load-bearing by disabling it and observing the failure, not just by observing green.

docs/physical-row-positions.md replaces the superseded rowid-lineage-physical-position-plan.md.

Note on coverage

CI's Linux Docker job runs the Postgres and MySQL provider suites, but its feature set omits
write-postgres, so postgres_single_catalog_write_tests and multicatalog_postgres_tests are
compiled out of every run. I ran them locally: 588 passed, 3 failed — and the same 3 fail
identically on unmodified main. Filed as #285, along with the coverage gap. #284 covers a related
by-name binding hazard on the embedded rowid and snapshot-id columns, which is pre-existing and out
of scope here.

DuckLake needs each row's physical position in its parquet file for
positional deletes (a delete file names rows by `pos`) and for row
lineage (`rowid = row_id_start + physical_row_position`).

Until DataFusion 55 there was no reader-level row-number column, so the
crate synthesized positions with `FileRowNumberExec` above a deliberately
constrained scan: row-group-aligned partitions with plan-time seeds,
wrapped in a `PositionalFileSource` that refused repartitioning, filter
pushdown and sort pushdown. Any of those would have shifted the positions
it was counting.

DataFusion 55 ships apache/datafusion#22026, so ask the reader instead:
`positional_table_schema` builds the scan's `TableSchema` with a field
carrying arrow-rs's `RowNumber` extension type. Values come from row-group
offsets in the footer, so they stay true physical positions under pruning,
row selection, byte-range splitting and reverse-order reads alike.

This converges with official DuckLake, which derives `rowid` the same way
from DuckDB's reader-level `COLUMN_IDENTIFIER_FILE_ROW_NUMBER`
(`ducklake_multi_file_reader.cpp::GetVirtualColumnExpression`).
`FileRowNumberExec` was the divergence, not the fix.

Absolute positions make pruning safe — `filter(delete(R)) ==
delete(filter(R))`, because deletion is keyed by position and a predicate
is row-local — so the guards come off:

- `DeleteFilterExec`, `RowIdExec` and `ColumnRenameExec` forward filter
  pushdown, letting a predicate reach row-group / page / bloom pruning on
  a positional path.
- One file splits across scan partitions again.
- `resolve_positions` pushes its predicate too, so DELETE/UPDATE prunes
  instead of reading every row of each candidate file.

Deleted: `positional_source.rs`, `FileRowNumberExec`,
`build_row_group_partitions`, the `ParquetAccessPlan` construction, the
output-partitioning pins, and the per-row-group footer bookkeeping
(`row_group_starts` / `row_group_count`) that existed only to feed them.

BREAKING CHANGE: `FileRowNumberExec` is removed from the public `row_id`
module; it has no meaning now that the reader supplies positions.
`DeleteFilterExec::try_new` and `RowIdExec::try_new` each take an extra
`pos_index: usize` instead of resolving the column by name — DuckLake
reserves no column names, so a table may have its own `__ducklake_row_pos`
column that a name lookup would bind. The internal column is given a
suffixed name when it would collide.

Also fixes a NaN pruning gap this work widened but did not create.
Parquet footer float bounds exclude NaN while `NaN > C` evaluates true, so
pruning on them can silently drop NaN rows.
`build_exec_for_file_with_deletes` had no `NanPruningBarrierExec` on
either branch, and its plain-scan branch (a file whose delete file
resolves to no positions) has always accepted pushed predicates. The
barrier now covers every path that can let a predicate reach the reader,
and the rowid path's `!needs_position` guard — which assumed the
positional branch refused all pushdown — is gone.

`resolve_positions` guards its own pushdown more strictly than the read
path: no float predicates at all (it has no per-file statistics to run the
precise `contains_nan` test), and none on a file carrying a physical
rename (the reader resolves a pushed predicate by name, and there is no
`ColumnRenameExec` on that path). Both were verified load-bearing — with
either disabled, the DELETE reports zero rows affected and removes
nothing.
Follow-up review of the previous commit, focused on divergence from
official DuckLake.

- The internal physical-position column must never reach a user-visible
  schema. Assert that across all five surfaces that could expose one:
  `SELECT *`, the three CDC table functions, and `information_schema`.
  Official DuckLake has no such column in any output, and ours is an
  implementation detail of a single scan.

- A rowid-projected scan is a positional path, so a predicate on a column
  the file predates now reaches the parquet reader, where the expression
  adapter substitutes the DuckLake `initial_default` before pruning. The
  catalog for this test is written by official DuckDB DuckLake, so it
  doubles as a differential check: our read must agree with what official
  itself returns.

Also harden one assumption the previous commit left implicit. The
deletions feed executes partition 0 of its data-file scan and no others.
That was previously guaranteed by an explicit `UnknownPartitioning(1)`
pin, which the previous commit removed as redundant — the scan has one
file group and `DeletedRowsExec` has no DataFusion children, so nothing
repartitions it. Still true, but no longer stated: were it to stop
holding, rows in the other partitions would be dropped silently, giving a
deletion feed that quietly omits deletions. Refuse instead.
…collision

Review follow-ups. Two correctness fixes and a round of test hardening.

**A recorded float `min_value` is not a lower bound when NaN may be present.**
The max side has been gated on `contains_nan` since #203, on the reasoning that
"NaN can never sit below the min". That holds only for positive NaN. DataFusion
orders floats with `total_cmp`, which is signed: negative NaN sorts below every
value, `-Infinity` included —

    SELECT (-CAST('NaN' AS DOUBLE)) < CAST('-Infinity' AS DOUBLE)  -- true

— and `contains_nan` does not record the sign. So a catalog whose stats carry
finite bounds alongside `contains_nan = true` (official DuckLake's INSERT writes
exactly that) could have a file pruned at plan time while it holds a matching
`-NaN` row: `SELECT ... WHERE x < C` silently short, `DELETE ... WHERE x < C`
reporting zero rows affected with the row still there. Both bounds are now
gated; `float_max_is_bound` becomes `float_bound_is_usable`. Catalogs this crate
writes were never exposed — its writer suppresses both bounds when NaN is
present — so the exposure was foreign catalogs.

**The internal position column could be name-bound to a user column in the CDC
feeds.** `unique_row_pos_name` checked only the file's physical schema, but
`present_catalog_schema` re-presents these scans under the *catalog* names and
`ColumnRenameExec` binds every output field by name. A catalog column called
`__ducklake_row_pos` whose physical name in this file differs — renamed after
the file was written, or absent from it — therefore slipped the check: the
output schema carried the name twice and both fields resolved to the user's
column. Every internal column is Int64, so nothing failed to downcast; the feeds
just reported the wrong rows (the added test shows a deletions feed reporting
none at all). Callers that re-present a scan under other names now pass those
names to `positional_table_schema_reserving`.

Also from review:

- `build_exec_for_partial_file` gains the same "no embedded rowid and no
  `row_id_start`" refusal every other rowid path has, matching the C++
  extension's throw, rather than failing later as a confusing rename error.
- `DeleteFilterExec::try_new` / `RowIdExec::try_new` return an error instead of
  panicking on an out-of-range `pos_index`.
- Comments describing the removed synthesis design are corrected.

Test hardening, after measuring what the fixtures actually produced:

- The `positional_pushdown_tests` fixture wrote a SINGLE row group (the parquet
  default is 1Mi rows), so nothing in that module exercised row-group pruning,
  and byte-range splitting ran at 100% skew — one partition read everything.
  Capped at 5000 rows per group, it now spans 12.
- Added assertions on the reader's own metrics rather than on plan strings: a
  selective predicate prunes 11 of 12 row groups, and `output_rows_skew` proves
  more than one partition reads data. Without these, both optimisations could
  stop working with every test still green.
- Added: a predicate on a renamed column through a positional read (the read
  side of the rename rewriting, previously untested); a predicate on the
  synthetic `rowid` (must not be pushed); the CDC collision above.
- Dropped a statistics-parity test that compared `Absent` to `Absent` and so
  asserted nothing; the parity it claimed is not observable through the public
  API.
Review follow-ups.

`build_exec_for_partial_file` serves two callers: the rowid-projected read and
an ordinary time-travel read, which projects the embedded rowid column away and
needs no lineage at all. The "no embedded rowid and no `row_id_start`" refusal
added in the previous commit fired on both, so a foreign catalog whose merged
partial file records neither would now fail a plain time-travel `SELECT` that
previously read correctly. Gate it on the output actually carrying `rowid`, and
move it below the embedded-snapshot check so a file missing both reports the
column this path actually needs.

Also correct documentation that still asserts the invariant the float min gate
just disproved — `DuckLakeTableColumnStatistics::contains_nan` and
`stats_collect`'s suppression comment both said NaN can only hide above a max.
Left standing, they invite reintroducing the bug they precede.

`present_catalog_schema`'s hazard note is updated to say which half is now
handled (the position column, via the reserved catalog names) and which is not:
`__ducklake_absent_field_<id>`, and the embedded rowid / snapshot-id columns
whose names come from the file footer and so cannot be suffixed. Binding those
by index rather than by name is the durable fix.

Narrow the parallelism claim in the docs and CHANGELOG. Byte-range splitting
substitutes for the hand-rolled row-group partitioning only on read paths, which
run through the physical optimizer. `build_update_scan_with_snapshot` and the
compaction sources are executed directly, so a single-file UPDATE or
rewrite_files now reads on one thread.
Review follow-ups.

`predicate_is_prunable` refused any file whose `name_mapping` was non-empty,
meaning "this file gives a data column a different physical name". But
`build_file_read_config` also inserts a synthetic `_ducklake_internal_row_id ->
rowid` entry for every file that embeds row ids — i.e. every file ever written
by `UPDATE` or by compaction. Those files were therefore never pruned, silently,
since falling back to a full scan is still correct. That is exactly the
population most likely to be mutated again, and it made the CHANGELOG's
"`DELETE` / `UPDATE` position resolution prunes too" false for them.

`FileReadConfig` now carries `renames_data_columns`, taken from the layout's
mapping before the synthetic entry is added, and the guard reads that. Testing
the mapping's *values* instead would have been fragile: a catalog column named
`rowid` can produce a real rename whose new name is also `rowid`.

The position column is now identified by its `parquet.virtual.row_number`
extension type rather than by being Int64. The embedded rowid and snapshot-id
columns are Int64 too, so the old check passed on precisely the columns a
misalignment would land on. The extension type survives projection, so this
holds on the scan's output schema — which turns the three `debug_assert!`s in
the CDC paths into a release-build guarantee.

Test additions:

- A `DELETE` against an UPDATE-rewritten file, asserting on the row-group metric
  that it still prunes. A row-count assertion passes either way, which is why
  the regression above went unnoticed.
- `pushdown_filters` was enabled nowhere in the tree, so the row-level
  `RowSelection` path — the sparse case the old synthesis could not survive at
  all — was never exercised. Added a variant that turns it on and checks every
  surviving rowid.

Documentation corrections: `compaction.rs` still described the removed
row-group partitioning ("partitions are the bin's row groups, not its files"),
which now reads as a live argument for a premise that no longer holds;
`resolve_positions` was missing from the list of paths that read single-threaded;
and the BREAKING entry omitted the removal of the public `row_id::row_pos_field`
and the change in `ROW_POS_COLUMN_NAME`'s meaning from a fixed name to a base
name that may be suffixed.

`ColumnRenameExec::is_type_preserving_projection` is `pub(crate)`; it was public
by accident.
Removing `build_row_group_partitions` left three paths reading one file on one
thread: `resolve_positions` (DELETE), `build_update_scan_with_snapshot`
(UPDATE), and the compaction rewrite. The read paths were unaffected — the
physical optimizer's `repartition_file_scans` rule splits their file groups —
but these three build a plan and run it themselves via `physical_plan::collect`
/ `execute`, so no optimizer rule ever sees them. The old hand-rolled
partitioning had baked the split into the scan config, which is why it applied
there at all.

Documenting that as a caveat was the wrong call: the change is meant to make
these paths faster, not slower, and nothing about reader-derived positions
requires giving up parallelism. `DuckLakeTable::split_across_partitions` asks
the source for the byte-range split at construction instead.

Safe for the same reason pruning is: the physical position comes from the
reader, not from arrival order. All three consumers read positions out of the
column — `resolve_positions` collects them into a `HashSet`,
`apply_update_to_batches` extends a cumulative vector from the position column,
and the compaction coalesce already tolerated multi-partition input — so a
byte-range split cannot move a row's position.

Verified the split happens rather than assuming it: with a target of 8
partitions the DELETE source scan reports 8, where before this it reported 1.
Correctness under the split is already covered — every DELETE, UPDATE and
compaction test now runs multi-partition.

Reverts the caveats the previous two commits added to the CHANGELOG and
`docs/physical-row-positions.md`, and corrects `compaction.rs`, which had just
been updated to say partitions are the bin's files one each.
…t is read

Review follow-ups on the two preceding commits.

`split_across_partitions` asked for `target_partitions` outright. DataFusion
splits purely by byte size and has no idea how many row groups a file has, so a
1.5 MiB two-row-group file at 8 partitions produced six that read nothing and
eight that each fetched the footer — these scans install no shared metadata
cache. The partitioning this replaced capped at the row-group count for exactly
that reason; `FileReadConfig` carries `row_group_count` again and the split
honours it. That also bounds compaction, where the fan-out is the product of the
bin size and the per-file split.

The split now also honours `optimizer.repartition_file_scans`. Previously a user
turning it off still got split DELETE/UPDATE scans, because
`DataSourceExec::repartitioned` reads only `repartition_file_min_size` — the
flag is checked by the optimizer rule these paths deliberately never reach.

`delete_prunes_on_a_file_that_embeds_row_ids` did not test what it claimed:
reverting the guard to the buggy form leaves it green. `predicate_is_prunable`
is read only inside `resolve_positions`, whose plan `DuckLakeDeleteExec` builds
and runs itself and which never appears in `EXPLAIN ANALYZE`; the metric the
test asserted came from a plain `SELECT` over a different file, since catalog
statistics eliminate the small UPDATE-written one before planning. Pinned with a
unit test on the guard itself — which does fail when reverted — and the DELETE
half now targets a row inside the rewritten file, where a mis-bound pushed
predicate would prune everything and delete nothing.

`rowids_survive_row_level_filter_pushdown` now asserts `pushdown_rows_pruned`,
so it cannot pass on row-group pruning alone. `RowIdExec::try_new` gains a test
for the case its new check exists to catch: a plain Int64 column, the shape of
the embedded rowid the old Int64-only check accepted.

On divergence, since this changes pruning: gating the float min is an
engine-layer difference from official DuckLake, not a format one. Official gates
only the max, which is right for DuckDB — it normalizes NaN, so `-NaN = NaN` and
either sign sorts above every value. Arrow's `total_cmp` is sign-sensitive, so
`-NaN` sorts below `-Infinity` here, and gating both is what keeps pruning
consistent with the filter semantics this engine actually has. The underlying
ordering difference is inherited from arrow, predates this work, and shows up in
query results rather than only in pruning; documented in COMPATIBILITY.md.

Also: the doc block for `row_pos_virtual_field` had drifted onto
`validate_row_pos_field`; the CHANGELOG now records that the position column
must carry the `parquet.virtual.row_number` extension type and that no public
constructor for one remains; and the pruning claim is narrowed to `DELETE`,
since an UPDATE's source scan pushes no predicate.
Closes out the review findings on the preceding commits.

`predicate_is_prunable` refused a file that renamed ANY data column. Absent
columns take a placeholder physical name, so that flag is true for every file
written before an `ADD COLUMN` — permanently, and for predicates that never
touch the new column. A table that has ever gained a column lost `DELETE`
pruning across its whole history. `FileReadConfig` now carries the catalog
indices whose physical name differs in this file, derived positionally from the
read schema rather than from `name_mapping`, and only a predicate that
references one of them is refused. An index outside the physical schema is also
refused now rather than assumed safe.

The three `debug_assert!`s validating the position column discarded the error
that names the offending field and indexed before bounds-checking, so an
out-of-range index panicked with "index out of bounds" instead of the message.
They are `#[cfg(debug_assertions)]` blocks that surface both. (This was reported
fixed one commit earlier; that edit replaced text with itself and changed
nothing.)

Compaction had no coverage under a split scan — the path where the fan-out is
largest, since `merge_adjacent_files` builds one source scan per file in the bin
and each now splits. Every other compaction fixture is far below
`repartition_file_min_size`, so it stays single-partition however the config is
set. Added a merge over two 60k-row files under a splitting config, asserting no
row is lost or duplicated and every merged row keeps a distinct rowid.

Documented that a single-file `rewrite_data_files` no longer preserves source
row order, for the same reason an unsorted merge never did.
@anoop-narang
anoop-narang merged commit 662bda9 into main Aug 31, 2026
5 checks passed
@anoop-narang
anoop-narang deleted the feat/reader-row-numbers branch August 31, 2026 14:22
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Use DataFusion Parquet virtual row_number for DuckLake row positions

1 participant