fix(expr): widen promoted partition values and bounds to prevent false pruning

[anoopj]

Which issue does this PR close?

• Part of Filtered scans silently drop rows for type-promoted columns (int->long, float->double) #3482 .

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):

• RowGroupMetricsEvaluator fails open: it produces no bound when the current
  (long) field type does not match the physical (int32) stat type, so the row
  group is kept. Missed pruning only.
• PageIndexEvaluator does not fail open for inequality predicates: for an
  INT32/INT64/FLOAT page index on a promoted field it builds a bound with the
  wide field type and the narrow literal, which compares as None, so the page is
  skipped 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 it
  for decimals.

Are these changes tested?

Added tests

[Fokko]

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.

[anoopj]

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 ManifestEvaluator currently decodes FieldSummary bounds 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.

[Fokko]

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.

[anoopj]

@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.

[mbutrovich]

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?

[JeonDaehong]

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.

[anoopj]

@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.

[anoopj]

@Fokko @JeonDaehong updated the PR to handle delete file matching as well.

[mbutrovich]

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.

[mbutrovich]

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.

[mbutrovich]

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:

Suggested change

	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.

[mbutrovich]

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):

Suggested change

	/// 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)
	)
	)
	})
	}

[mbutrovich]

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?

[anoopj]

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.