Repository navigation
fix(expr): widen promoted partition values and bounds to prevent false pruning - #3483
Conversation
| /// Returns a copy whose literal is widened to `target` for the value | ||
| /// promotions `int` -> `long` and `float` -> `double`, so a value written | ||
| /// under the narrower type compares against a predicate bound to the wider | ||
| /// one. Any other `target` returns an unchanged copy; decimals already | ||
| /// compare across precisions. | ||
| pub(crate) fn promote_to(&self, target: &PrimitiveType) -> Datum { | ||
| match (&self.literal, target) { | ||
| (PrimitiveLiteral::Int(value), PrimitiveType::Long) => Datum::long(*value), | ||
| (PrimitiveLiteral::Float(value), PrimitiveType::Double) => Datum::double(value.0), | ||
| _ => self.clone(), | ||
| } | ||
| } |
There was a problem hiding this comment.
I'm lesser known with the PyIceberg codebase, but we could also push this down into the projected schema when reading the manifest. This way we always get the current spec.
This avoids having to promote all the values afterwards.
There was a problem hiding this comment.
Thanks for the suggestion @Fokko . This is actually how the Java implementation works (i.e. bind against the current schema).
- The blocker for doing it at manifest parse time is the
ObjectCache: it caches the parsed Manifest keyed by location and survives in-place metadata upgrades. - If we decoded entries under the current/projected schema at parse, we'd bake one scan's schema into that shared, long-lived entry. A later scan could then read stale or wrong-typed partition/bounds from the cache and hit the same false-pruning bug. Keeping the parse a pure function of the manifest's own embedded schema is what keeps the cache sound.
- Comparison-time promotion keeps the cache pure and is consistent with the existing read path: the manifest-list
ManifestEvaluatorcurrently decodesFieldSummarybounds on demand with the predicate's type.
f we want to centralize it, I can project per-scan after the cache (promote each entry's partition tuple + bounds to the current schema in the scan's entry-read loop, so the evaluators drop their promotion). It's cache-safe but is a bit heavy because it rebuilds each entry's partition/bounds in the planning loop, so I prefer the surgical version that the PR currently does. Happy to switch if you think the single-place projection is worth it.
There was a problem hiding this comment.
Got it. I would suggest getting this fix in for now, since it fixes an important issue. Maybe we can create a separate PR to assess the impact of pushing the promotion down to the reader. I think that would simplify it a lot and make it less error prone at the same time.
There was a problem hiding this comment.
@Fokko I ended up doing the reader-side widening in this PR rather than a follow-up, because it turned out to be necessary rather than just a cleanup: @JeonDaehong found that the comparison-time version didn't fix equality-delete / deletion-vector matching.
There was a problem hiding this comment.
I ended up doing the reader-side widening in this PR rather than a follow-up
Thanks, that makes sense given the delete matching. Column bounds are still widened inside InclusiveMetricsEvaluator (promote_bound), while partition tuples are now widened once per entry in fetch_manifest_and_stream_manifest_entries. Should bounds move there too? If not in this PR, could you open an issue for it so the rest of @Fokko's suggestion stays tracked?
There was a problem hiding this comment.
Yes, I'd prefer to keep bounds out of this PR. Opened #3501 to track moving column bound promo to reader.
JeonDaehong
left a comment
There was a problem hiding this comment.
Ran this against tables written by Java (iceberg-core 1.12.0-SNAPSHOT). b int is identity-partitioned, with one data file in b=7 (ids 1, 2, 3). Then b is promoted to long, and Java writes a delete for id 2 in b=7. Java reads [1, 3] for every row below.
| main | this PR | |
|---|---|---|
| v2 eq delete, full scan | [1, 2, 3] |
[1, 2, 3] |
v2 eq delete, b = 7 |
[] |
[1, 2, 3] |
| v3 DV, full scan | error | error |
v3 DV, b = 7 |
[] |
error |
The pruning fix works: b = 7 no longer drops the file. But DeleteFileIndex has the same int vs long mismatch. The data file's partition comes from the old manifest as Int(7), and the delete's partition comes from the new one as Long(7). So eq_deletes_by_partition never matches the eq delete, and the DV fails the partition check in get_deletes_for_data_file ("does not match the data file's partition"). With this PR, the filtered scans now reach those paths too: one returns the deleted row, the other errors.
Java returns Long for the old data file's partition too (its specs are bound to the current schema), so it doesn't hit this. Could the partition tuple be widened once, when manifest entries are loaded, instead of in each evaluator? That would also cover the delete index. I'm fine with that being a follow-up if you'd rather keep this PR to pruning.
|
@JeonDaehong Thanks for running the test against Java. This is really good data. The delete matching issue is existing, it is now surfacing because this PR fixed the false pruning. But I am leaning towards fixing that also in this PR rather than getting the pruning fix alone in. |
A partition column promoted int->long (or float->double) keeps its originally-written values in manifests produced before the promotion. The entry partition tuple is decoded under the manifest's own narrower type, while the scan predicate is bound to the current wider type. `Datum::partial_cmp`/`PartialEq` require matching types, so the `ExpressionEvaluator` compared e.g. an int partition value against a long predicate literal, got no ordering, and silently pruned the data file, dropping matching rows from filtered scans. The manifest-list `ManifestEvaluator` is unaffected: it decodes `FieldSummary` bounds with `Datum::try_from_bytes`, which already widens a 4-byte int bound to long. Mirror that on the entry path by widening the partition value to the reference's type before comparing. Decimals already compare across precisions, so only int->long and float->double need widening. Co-authored-by: Isaac <no-reply@databricks.com>
Companion to the partition-tuple widening: InclusiveMetricsEvaluator compares a data file's column bounds against the scan predicate, but those bounds are decoded under the manifest's embedded (narrower) schema while the predicate is bound to the current (wider) type. For an int column promoted to long (or float to double), the inequality and IN checks compared mismatched types, got no ordering, and discarded files that could match, dropping rows; the eq check failed open and merely skipped valid pruning. Widen each stored bound to the reference's type before comparing, via the same `Datum::promote_to` helper, in the inequality, eq, and IN paths.
Replaces the comparison-time partition promotion with a reader-side projection, per review. The earlier comparison-time promotion only reached the pruning evaluators, so a promoted partition column still mismatched in the delete index: a data file's int partition never matched an eq delete or DV keyed by long, dropping or mis-applying deletes. Projecting the partition tuple fixes pruning and delete matching in one place.
10c40b0 to
642d8dd
Compare
|
@Fokko @JeonDaehong updated the PR to handle delete file matching as well. |
mbutrovich
left a comment
There was a problem hiding this comment.
Thanks @anoopj for tracking this down.
Should the description say "Part of #3482" instead of "Closes #3482", or should we open a separate issue for the Parquet page index? The Followups section stops mid-sentence. PageIndexEvaluator builds page bounds with the current field type and the physical literal (int32_bound_to_datum). For a file written before the promotion that gives a long datum holding an Int literal, which compares as None against the long predicate, and visit_inequality then skips the page. So that path drops rows instead of only losing pruning. With four 1024-row int32 pages holding 1, 2, 3, 4 and the field typed long, both col > 2 and col < 3 return RowSelector::skip(4096) on the head commit. Row selection is off by default in ArrowReaderBuilder, but Comet turns it on.
| }; | ||
|
|
||
| #[test] | ||
| fn promoted_data_file_partition_matches_delete_after_column_promotion() { |
There was a problem hiding this comment.
Should these two tests move to the scan level? delete_file_index.rs itself doesn't change in this PR, so they check promoted_partition plus Struct equality, which the data_file.rs tests and test_scan_widens_old_partition_so_pruning_and_deletes_match already cover.
Two cases from @JeonDaehong's Java run aren't covered end to end yet. One is a v3 deletion vector written after the promotion against the older data file, which is the case that hit the partition check. The other is a data file and a delete that were both written before the promotion. Those match on main, and with this change both sides get widened, so a test would show they still match. TableTestFixture::setup_v3_manifest_files shows how to upgrade the fixture to v3 for the DV case.
There was a problem hiding this comment.
Done. Moved the both e2e cases at the scan level. I used setup_v3_manifest_files to upgrade the fixture for the DV case as you suggested. Thanks for the pointer!
| fn promote_bound(bound: &Datum, reference: &BoundReference) -> Datum { | ||
| match reference.field().field_type.as_primitive_type() { | ||
| Some(primitive_type) => bound.promote_to(primitive_type), | ||
| None => bound.clone(), | ||
| } | ||
| } |
There was a problem hiding this comment.
Could this borrow the bound when there's nothing to promote? bound.clone(), and the self.clone() fallback in Datum::promote_to, copy every bound the evaluator reads. A predicate on a string or binary column now allocates a String or Vec<u8> per bound per data file during planning, where main compares by reference. A Cow keeps the promoted case and borrows otherwise:
| fn promote_bound(bound: &Datum, reference: &BoundReference) -> Datum { | |
| match reference.field().field_type.as_primitive_type() { | |
| Some(primitive_type) => bound.promote_to(primitive_type), | |
| None => bound.clone(), | |
| } | |
| } | |
| fn promote_bound<'b>(bound: &'b Datum, reference: &BoundReference) -> Cow<'b, Datum> { | |
| reference | |
| .field() | |
| .field_type | |
| .as_primitive_type() | |
| .and_then(|primitive_type| bound.promote_to(primitive_type)) | |
| .map_or(Cow::Borrowed(bound), Cow::Owned) | |
| } |
That needs use std::borrow::Cow; and Datum::promote_to returning Option<Datum> (with its doc comment updated):
pub(crate) fn promote_to(&self, target: &PrimitiveType) -> Option<Datum> {
self.literal
.promote_to(target)
.map(|literal| Datum::new(target.clone(), literal))
}The call sites then bind let bound = &*Self::promote_bound(bound, reference);, and the comparisons go back to cmp_fn(bound, datum), datum.ge(lower_bound) and datum.le(upper_bound) as on main.
| /// Returns whether `current` widens any field of `written` at the same position | ||
| /// (differing primitive types), i.e. partition values written with `written` need | ||
| /// widening to `current` before they compare against predicates bound to the current | ||
| /// schema. | ||
| fn partition_types_differ(current: &StructType, written: &StructType) -> bool { | ||
| current.fields().len() != written.fields().len() | ||
| || current | ||
| .fields() | ||
| .iter() | ||
| .zip(written.fields()) | ||
| .any(|(c, w)| c.field_type.as_primitive_type() != w.field_type.as_primitive_type()) | ||
| } |
There was a problem hiding this comment.
Should this gate fire only for the type pairs that promoted_partition changes? As written, any primitive difference deep-clones every ManifestEntry in the manifest, bounds maps included, on every scan. A partition source widened from decimal(9, 2) to decimal(18, 2) takes that path, but promote_partition_value leaves every value as it is. So does a dropped string source that now resolves to unknown.
The length check can't fire either. Both types come from the same spec, and partition_type pushes one field per spec field. Something like this limits the clone to manifests that need it (it also needs PrimitiveType in the crate::spec import):
| /// Returns whether `current` widens any field of `written` at the same position | |
| /// (differing primitive types), i.e. partition values written with `written` need | |
| /// widening to `current` before they compare against predicates bound to the current | |
| /// schema. | |
| fn partition_types_differ(current: &StructType, written: &StructType) -> bool { | |
| current.fields().len() != written.fields().len() | |
| || current | |
| .fields() | |
| .iter() | |
| .zip(written.fields()) | |
| .any(|(c, w)| c.field_type.as_primitive_type() != w.field_type.as_primitive_type()) | |
| } | |
| /// Returns whether any `int` or `float` field of `written` is widened in `current` (to | |
| /// `long` or `double`, or to `unknown` after its source column is dropped), i.e. partition | |
| /// values written with `written` need widening before they compare against predicates | |
| /// bound to the current schema. | |
| fn partition_types_differ(current: &StructType, written: &StructType) -> bool { | |
| current.fields().iter().zip(written.fields()).any(|(c, w)| { | |
| matches!( | |
| ( | |
| w.field_type.as_primitive_type(), | |
| c.field_type.as_primitive_type() | |
| ), | |
| ( | |
| Some(PrimitiveType::Int), | |
| Some(PrimitiveType::Long | PrimitiveType::Unknown) | |
| ) | ( | |
| Some(PrimitiveType::Float), | |
| Some(PrimitiveType::Double | PrimitiveType::Unknown) | |
| ) | |
| ) | |
| }) | |
| } |
There was a problem hiding this comment.
Applied your version. Added PrimitiveType to the crate::spec import as noted.
| pub(crate) fn promoted_partition(&self, partition_type: &StructType) -> Struct { | ||
| self.partition | ||
| .iter() | ||
| .zip(partition_type.fields()) | ||
| .map(|(value, field)| promote_partition_value(value, field)) | ||
| .collect() | ||
| } |
There was a problem hiding this comment.
What should happen when self.partition and partition_type have different lengths? zip stops at the shorter one, so trailing partition values would be dropped without an error, and the entry would then miss its partition-scoped deletes. The two should always line up because they come from the same spec. Should this return a DataInvalid error on a mismatch, the way get_deletes_for_data_file does when a DV's partition disagrees with its data file?
There was a problem hiding this comment.
Good catch. Fixed such that it now returns DataInvalid.
|
Thanks for the feedback @mbutrovich. I have fixed the PR description and the context you provided on the followup. The code feedback makes sense, and I will be fixing them. |
…ple length Addresses review feedback: - Gate `partition_types_differ` on the type pairs `promoted_partition` actually widens (int->long/unknown, float->double/unknown) instead of any primitive difference, so a decimal-precision widening or a dropped non-numeric source no longer deep-clones every manifest entry on every scan. - `InclusiveMetricsEvaluator::promote_bound` returns `Cow` and `Datum::promote_to` returns `Option`, so a bound that does not widen (e.g. a string or binary) is borrowed instead of cloned per data file during planning. - `DataFile::promoted_partition` returns `DataInvalid` when the stored tuple and the spec partition type have different lengths, instead of silently dropping trailing values (and the partition-scoped deletes they would match). - Drop the two delete-index unit tests that only re-exercised `promoted_partition` plus `Struct` equality, already covered by the data_file and scan-level tests.
Adds plan_files scan regressions over a promoted partition column, routing an old-schema manifest through the reader-side widening: - A v3 deletion vector written after the promotion against the older data file. Without the widening the DV partition check errors (long(7) vs int(7)); the test guards that the data file's partition widens so the DV attaches. - An equality delete written before the promotion against the older data file, so both sides widen to long. Confirms the pre-existing match is preserved. Parameterizes the equality-delete fixture on whether the delete predates the promotion, and adds a v3 deletion-vector fixture.
A dropped partition source resolves to the Unknown primitive type. Widening its stored value to the promotion ceiling keyed on the backing literal, so a date (stored as an int) was silently promoted to long. The promotion gate skips a manifest whose only type difference is a dropped source, so a data manifest that widened a dropped date diverged from the delete manifest that left it alone, detaching the delete. Copy dropped-source values through unchanged and restrict the gate to genuine int to long and float to double promotions on live columns.
laskoviymishka
left a comment
There was a problem hiding this comment.
LGTM.
Widening the tuple once per entry in fetch_manifest_and_stream_manifest_entries, before the data/delete split, is the right place for this, and it fixes the delete-index case @JeonDaehong hit without touching each evaluator. Agree with @mbutrovich's gate/Cow/zip points sorted this looks good to me; the inlines are small test and doc gaps around the scan test and the Unknown path.
One thing for the Followups list: date -> timestamp/timestamp_ns is the other v3 promotion, and promote_bound would leave a Date bound facing a Timestamp predicate there, the same mismatch this fixes. update_schema can't produce it today, so it doesn't need to be in this PR, but I'd name it so the follow-up issue covers it.
| .try_collect() | ||
| .await | ||
| .unwrap(); | ||
| assert_eq!(tasks.len(), 1); |
There was a problem hiding this comment.
The widened tuple now reaches FileScanTask::partition(), and the reader builds identity constants and _partition from it. I'd pin it here so the reader-visible change is covered:
assert_eq!(
tasks[0].partition(),
Some(&Struct::from_iter([Some(Literal::long(7))]))
);| let tasks: Vec<FileScanTask> = fixture | ||
| .table | ||
| .scan() | ||
| .with_filter(Reference::new("x").equal_to(Datum::long(7))) |
There was a problem hiding this comment.
I'd add the negative case too: x = 8 should still prune this file. As it stands, a widening that made every file match would pass this test.
| // The partition source column was dropped, so there is no target type to | ||
| // widen to; widen to the promotion ceiling instead so this value and its | ||
| // deletes still normalize to the same type and the delete index matches. | ||
| Some(PrimitiveType::Unknown) => primitive.widen(), |
There was a problem hiding this comment.
Under Unknown this widens every Int/Float value, not just promoted ones. A dropped date identity source comes out as Long(days), and so does a plain int that was never promoted. That works as a matching key since both sides go through it whenever the gate fires, but this comment and the doc on promoted_partition only describe int -> long / float -> double. I'd say in both places that this branch produces a canonical key rather than a typed value, since it also ends up in FileScanTask::partition().
There was a problem hiding this comment.
You're right that the ceiling-widen-everything version corrupted a dropped date into Long(days). @JeonDaehong / @mbutrovich surfaced the same on the delete-matching side, so I changed the Unknown branch to decide by the manifest's written type: a dropped int/float still widens to its ceiling, but a dropped date (or any non-numeric) is copied through unchanged.
| // type; the value still widens to its ceiling so it keeps matching deletes | ||
| // written while the source still existed. | ||
| let partition_type = StructType::new(vec![ | ||
| NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Unknown)).into(), |
There was a problem hiding this comment.
PartitionSpec::partition_type already maps a dropped identity source to Unknown, so this path is real, but the test builds the Unknown field by hand and wouldn't notice if that mapping changed. I'd build the type from spec.partition_type(&schema) with the source dropped from the schema, so the test goes through the same call context.rs makes.
There was a problem hiding this comment.
The real spec.partition_type(...) drop-to-Unknown mapping is now exercised end to end: the new scan tests drop the source via a helper that empties the spec and removes the field, then plan through the same partition_type call, so a change to that mapping would break them.
|
Re-ran the same Java-written tables on 642d8dd. All four cases now return
Thanks for pulling the delete matching into this PR. The v3 DV case @mbutrovich mentioned passes here. It's only checked against these Java tables, not by a test in the PR. |
|
@laskoviymishka Thanks for the review. Filed #3503 for the timestamp issue. |
laskoviymishka
left a comment
There was a problem hiding this comment.
Re-checked the five commits since my approval at 642d8dd.
Good to land as is.
Two nits inline, both fine as follow-ups, so this is good to merge as is from my side.
| for manifest_entry in manifest.entries() { | ||
| let manifest_entry = match promotion_target.as_ref() { | ||
| Some((current, written)) => { | ||
| let mut entry = manifest_entry.as_ref().clone(); |
There was a problem hiding this comment.
Follow-up, not for this PR: could deleted entries skip this deep clone? In a gated manifest every entry is copied here, before process_data_manifest_entry drops non-alive ones and prunes the rest (scan/mod.rs#L579-L630). main only clones the Arc (context.rs#L111-L112). As I read it, partition pruning needs the widened tuple, so the alive check is the part that could move ahead of the clone.
There was a problem hiding this comment.
yes, that is a good idea.
| let current_schema = current_snapshot.schema(self.table.metadata()).unwrap(); | ||
| let current_partition_spec = self.table.metadata().default_partition_spec(); | ||
|
|
||
| let mut fields = current_schema.as_struct().fields().to_vec(); |
There was a problem hiding this comment.
Non-blocking, fine as a follow-up: could the three promotion fixtures share one helper for the int-schema override and the old-schema data manifest? This block repeats the one in setup_promoted_partition_with_eq_delete. The date fixture's source drop (L1003-L1020) has the same shape as drop_promoted_partition_source (scan/mod.rs), which could move into test_utils and be reused.
There was a problem hiding this comment.
That makes sense. Will send a followup to clean this up.
|
I'll wait for either @JeonDaehong or @mbutrovich to stamp before merge. |
|
@laskoviymishka sent #3516 to fix the followups |
- Widen INT32 and FLOAT page bounds through PrimitiveLiteral::promote_to, the rule apache#3483 added, instead of hand-written Long and Double arms. - Fold project_column into project_promoted_column, so the call sites don't nest a `?` inside an argument. - Give read_once a row_selection_enabled switch and reuse it in the promoted-column test.
) Discard deleted manifest entries before copying their metadata or dispatching scan work. Live entries still widen before partition pruning and delete matching, consistent with the spec and Java's live-entry scan. Share the promotion fixtures' old-schema data manifest and source-drop setup while preserving their schema, format, and sequence-number cases. Follow-up to apache#3483.
Which issue does this PR close?
What changes are included in this PR?
A column promoted int->long (or float->double) keeps its originally-written values in manifests produced before the promotion. Partition pruning and stats based pruning can silently drop them, and cause data loss.
This PR fixes both problems. I can split the partition pruning and stats-based pruning into separate PRs if needed (currently two commits in this PR)
Followups
Parquet-level pruning does not widen promoted columns yet (pre-existing, not
changed here):
RowGroupMetricsEvaluatorfails open: it produces no bound when the current(
long) field type does not match the physical (int32) stat type, so the rowgroup is kept. Missed pruning only.
PageIndexEvaluatordoes not fail open for inequality predicates: for anINT32/INT64/FLOATpage index on a promoted field it builds a bound with thewide field type and the narrow literal, which compares as
None, so the page isskipped and rows are dropped. Reachable only with row selection enabled (off by
default in
ArrowReaderBuilder, on in Comet). Same class as fix(reader): page-index pruning drops rows for INT32/INT64 decimals #3314, which fixed itfor decimals.
Are these changes tested?
Added tests