You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by gi...@apache.org on 2026/10/22 09:07:37 UTC
(iceberg-rust) branch gh-readonly-queue/main/pr-3483-000e791d5229d9d136cda2b41fa0d7563fa58e1e created (now 5c2715767)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-3483-000e791d5229d9d136cda2b41fa0d7563fa58e1e
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
at 5c2715767 fix(expr): widen promoted partition values and bounds to prevent false pruning (#3483)
This branch includes the following new commits:
new 5c2715767 fix(expr): widen promoted partition values and bounds to prevent false pruning (#3483)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iceberg-rust) 01/01: fix(expr): widen promoted partition values and bounds to prevent false pruning (#3483)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3483-000e791d5229d9d136cda2b41fa0d7563fa58e1e
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 5c27157675a7babb611638786195dfe24ad41dab
Author: Anoop Johnson <an...@apache.org>
AuthorDate: Thu Oct 22 09:07:11 2026 +0000
fix(expr): widen promoted partition values and bounds to prevent false pruning (#3483)
* fix(expr): widen promoted partition values before comparison
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...@databricks.com>
* fix(expr): widen promoted column bounds in metrics pruning
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.
* fix(scan): widen promoted partition tuples at manifest read
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.
* fix(scan): tighten promotion gate, borrow unpromoted bounds, guard tuple 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.
* test(scan): add end-to-end promotion scan regressions for deletes
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.
* fix(scan): copy dropped partition sources through unchanged
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.
* fix(scan): normalize dropped numeric partition sources by written type
* Feedback
---------
Co-authored-by: Isaac <no...@databricks.com>
---
.../expr/visitors/inclusive_metrics_evaluator.rs | 106 ++++++
crates/iceberg/src/scan/context.rs | 54 ++-
crates/iceberg/src/scan/mod.rs | 217 +++++++++++
crates/iceberg/src/spec/manifest/data_file.rs | 232 ++++++++++-
crates/iceberg/src/spec/values/datum.rs | 11 +
crates/iceberg/src/spec/values/primitive.rs | 26 ++
crates/iceberg/src/test_utils/scan.rs | 422 ++++++++++++++++++++-
7 files changed, 1062 insertions(+), 6 deletions(-)
diff --git a/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs b/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs
index 8ccd1364d..f7abe7ab5 100644
--- a/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs
+++ b/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+use std::borrow::Cow;
+
use fnv::FnvHashSet;
use crate::expr::visitors::bound_predicate_visitor::{BoundPredicateVisitor, visit};
@@ -120,6 +122,7 @@ impl<'a> InclusiveMetricsEvaluator<'a> {
};
if let Some(bound) = bound {
+ let bound = &*Self::promote_bound(bound, reference);
if cmp_fn(bound, datum) {
return ROWS_MIGHT_MATCH;
}
@@ -129,6 +132,19 @@ impl<'a> InclusiveMetricsEvaluator<'a> {
ROWS_MIGHT_MATCH
}
+
+ /// Widens a stored bound to the reference's type so a bound written under a
+ /// narrower type (e.g. an `int` column later promoted to `long`) compares
+ /// against a predicate bound to the wider type. Borrows the bound when nothing
+ /// is promoted, so unpromoted bounds (e.g. strings/binaries) are not copied.
+ 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)
+ }
}
impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
@@ -259,6 +275,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
}
if let Some(lower_bound) = self.lower_bound(field_id) {
+ let lower_bound = &*Self::promote_bound(lower_bound, reference);
if lower_bound.is_nan() {
// NaN indicates unreliable bounds.
// See the InclusiveMetricsEvaluator docs for more.
@@ -269,6 +286,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
}
if let Some(upper_bound) = self.upper_bound(field_id) {
+ let upper_bound = &*Self::promote_bound(upper_bound, reference);
if upper_bound.is_nan() {
// NaN indicates unreliable bounds.
// See the InclusiveMetricsEvaluator docs for more.
@@ -438,6 +456,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
}
if let Some(lower_bound) = self.lower_bound(field_id) {
+ let lower_bound = &*Self::promote_bound(lower_bound, reference);
if lower_bound.is_nan() {
// NaN indicates unreliable bounds. See the InclusiveMetricsEvaluator docs for more.
return ROWS_MIGHT_MATCH;
@@ -450,6 +469,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
}
if let Some(upper_bound) = self.upper_bound(field_id) {
+ let upper_bound = &*Self::promote_bound(upper_bound, reference);
if upper_bound.is_nan() {
// NaN indicates unreliable bounds. See the InclusiveMetricsEvaluator docs for more.
return ROWS_MIGHT_MATCH;
@@ -521,6 +541,92 @@ mod test {
assert!(result);
}
+ fn promotion_schema(primitive: PrimitiveType) -> SchemaRef {
+ Arc::new(
+ Schema::builder()
+ .with_fields(vec![Arc::new(NestedField::optional(
+ 1,
+ "a",
+ Type::Primitive(primitive),
+ ))])
+ .build()
+ .unwrap(),
+ )
+ }
+
+ fn data_file_with_bounds(lower: Datum, upper: Datum) -> DataFile {
+ DataFile {
+ lower_bounds: HashMap::from([(1, lower)]),
+ upper_bounds: HashMap::from([(1, upper)]),
+ ..create_test_data_file()
+ }
+ }
+
+ #[test]
+ fn eval_promotes_int_bounds_for_inequality() {
+ // A column promoted int->long keeps int-typed column bounds in
+ // pre-promotion manifests; a predicate bound to the promoted long type
+ // must still match a file whose range covers it, not silently prune it.
+ let schema = promotion_schema(PrimitiveType::Long);
+ let data_file = data_file_with_bounds(Datum::int(5), Datum::int(10));
+ let predicate = Predicate::Binary(BinaryExpression::new(
+ GreaterThan,
+ Reference::new("a"),
+ Datum::long(4),
+ ))
+ .bind(schema, true)
+ .unwrap();
+
+ assert!(InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap());
+ }
+
+ #[test]
+ fn eval_promotes_int_bounds_for_in() {
+ let schema = promotion_schema(PrimitiveType::Long);
+ let data_file = data_file_with_bounds(Datum::int(5), Datum::int(10));
+ let predicate = Predicate::Set(SetExpression::new(
+ In,
+ Reference::new("a"),
+ FnvHashSet::from_iter([Datum::long(5), Datum::long(6)]),
+ ))
+ .bind(schema, true)
+ .unwrap();
+
+ assert!(InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap());
+ }
+
+ #[test]
+ fn eval_promotes_int_bounds_for_eq() {
+ // With the bounds widened, `== 5` correctly prunes a file whose range is
+ // entirely above 5 (int bounds [6, 10]) instead of failing open.
+ let schema = promotion_schema(PrimitiveType::Long);
+ let data_file = data_file_with_bounds(Datum::int(6), Datum::int(10));
+ let predicate = Predicate::Binary(BinaryExpression::new(
+ Eq,
+ Reference::new("a"),
+ Datum::long(5),
+ ))
+ .bind(schema, true)
+ .unwrap();
+
+ assert!(!InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap());
+ }
+
+ #[test]
+ fn eval_promotes_float_bounds_for_inequality() {
+ let schema = promotion_schema(PrimitiveType::Double);
+ let data_file = data_file_with_bounds(Datum::float(1.5_f32), Datum::float(2.0_f32));
+ let predicate = Predicate::Binary(BinaryExpression::new(
+ GreaterThan,
+ Reference::new("a"),
+ Datum::double(1.0),
+ ))
+ .bind(schema, true)
+ .unwrap();
+
+ assert!(InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap());
+ }
+
#[test]
fn test_all_nulls() {
let result =
diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs
index 211ebf8f0..0e8b0aa50 100644
--- a/crates/iceberg/src/scan/context.rs
+++ b/crates/iceberg/src/scan/context.rs
@@ -30,7 +30,8 @@ use crate::scan::{
};
use crate::spec::{
ManifestContentType, ManifestEntryRef, ManifestFile, ManifestList, NameMapping,
- PartitionSpecRef, SchemaRef, SnapshotRef, SortOrderRef, StructType, TableMetadataRef,
+ PartitionSpecRef, PrimitiveType, SchemaRef, SnapshotRef, SortOrderRef, StructType,
+ TableMetadataRef,
};
use crate::{Error, ErrorKind, Result};
@@ -73,6 +74,28 @@ pub(crate) struct ManifestEntryContext {
pub sort_order: Option<SortOrderRef>,
}
+/// Returns whether any `int` or `float` field of `written` is widened in `current`, to
+/// `long`/`double` or to `unknown` once its source column is dropped, i.e. partition
+/// values written with `written` need normalizing before they compare against predicates
+/// bound to the current schema or against delete-file partitions.
+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)
+ )
+ )
+ })
+}
+
impl ManifestFileContext {
/// Consumes this [`ManifestFileContext`], fetching its Manifest from FileIO and then
/// streaming its constituent [`ManifestEntries`] to the channel provided in the context
@@ -95,7 +118,33 @@ impl ManifestFileContext {
let manifest = object_cache.get_manifest(&manifest_file).await?;
+ // A manifest written under an older schema stores partition values in that
+ // schema's (narrower) types. When the current partition type widens the one the
+ // manifest was written with (e.g. an int source column promoted to long), widen
+ // each entry's partition tuple here so the pruning evaluator and the delete-file
+ // index compare like-typed values. The gate compares the partition types
+ // themselves, not schema ids, since v1 may omit them (defaulting to 0). Column
+ // bounds are widened lazily where the metrics evaluator compares them.
+ let promotion_target: Option<(StructType, StructType)> = match partition_spec.as_ref() {
+ Some(spec) => {
+ let current = spec.partition_type(&snapshot_schema)?;
+ let written = spec.partition_type(manifest.metadata().schema())?;
+ partition_types_differ(¤t, &written).then_some((current, written))
+ }
+ None => None,
+ };
+
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();
+ entry.data_file.partition =
+ entry.data_file.promoted_partition(current, written)?;
+ Arc::new(entry)
+ }
+ None => manifest_entry.clone(),
+ };
+
// Carry the raw id through unresolved, then resolve it to an order for the
// reader. The resolved order is `None` for the unsorted order (empty fields) so
// that `Some` always means genuinely sorted, matching Java's `isSorted()` gate;
@@ -108,8 +157,7 @@ impl ManifestFileContext {
.cloned();
let manifest_entry_context = ManifestEntryContext {
- // TODO: refactor to avoid the expensive ManifestEntry clone
- manifest_entry: manifest_entry.clone(),
+ manifest_entry,
expression_evaluator_cache: expression_evaluator_cache.clone(),
field_ids: field_ids.clone(),
partition_spec_id: manifest_file.partition_spec_id,
diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs
index 59a353c5e..cb77abeb7 100644
--- a/crates/iceberg/src/scan/mod.rs
+++ b/crates/iceberg/src/scan/mod.rs
@@ -1619,6 +1619,223 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_scan_widens_old_partition_so_pruning_and_deletes_match() {
+ // Guards the reader-side normalization through a real scan: a data manifest
+ // written when the partition source `x` was `int` must have its `int(7)`
+ // partition widened to `long(7)` during planning, so (a) a `x = 7` (long)
+ // filter keeps the file and (b) the equality delete keyed by `long(7)`
+ // attaches. Without the normalization the int/long mismatch drops the delete
+ // (and prunes the file under the filter).
+ let mut fixture = TableTestFixture::new();
+ fixture.setup_promoted_partition_with_eq_delete(false).await;
+
+ let tasks: Vec<FileScanTask> = fixture
+ .table
+ .scan()
+ .build()
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(tasks.len(), 1);
+ assert_eq!(
+ tasks[0].deletes().len(),
+ 1,
+ "equality delete must attach after widening the old int partition to long"
+ );
+ assert_eq!(
+ tasks[0].partition(),
+ Some(&Struct::from_iter([Some(Literal::long(7))])),
+ "the widened long(7) tuple must reach FileScanTask::partition()"
+ );
+
+ let tasks: Vec<FileScanTask> = fixture
+ .table
+ .scan()
+ .with_filter(Reference::new("x").equal_to(Datum::long(7)))
+ .build()
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ tasks.len(),
+ 1,
+ "filter on the promoted type must not prune the file"
+ );
+ assert_eq!(tasks[0].deletes().len(), 1);
+
+ let tasks: Vec<FileScanTask> = fixture
+ .table
+ .scan()
+ .with_filter(Reference::new("x").equal_to(Datum::long(8)))
+ .build()
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert!(
+ tasks.is_empty(),
+ "a file in partition x=7 must be pruned by an x=8 filter"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_scan_widens_old_partition_with_pre_promotion_delete() {
+ // Data file and equality delete both written before the promotion (int
+ // partition). They matched on main; after both sides widen to long they must
+ // still match, so the delete stays attached.
+ let mut fixture = TableTestFixture::new();
+ fixture.setup_promoted_partition_with_eq_delete(true).await;
+
+ let tasks: Vec<FileScanTask> = fixture
+ .table
+ .scan()
+ .build()
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(tasks.len(), 1);
+ assert_eq!(tasks[0].deletes().len(), 1);
+ }
+
+ #[tokio::test]
+ async fn test_scan_widens_old_partition_so_deletion_vector_matches() {
+ // A v3 deletion vector written after the promotion (long partition) references
+ // an older data file (int partition). The widening must make the partitions
+ // agree so the DV attaches, instead of failing the DV partition check.
+ let mut fixture = TableTestFixture::new();
+ fixture.setup_promoted_partition_with_dv().await;
+
+ let tasks: Vec<FileScanTask> = fixture
+ .table
+ .scan()
+ .build()
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(tasks.len(), 1);
+ assert_eq!(tasks[0].deletes().len(), 1);
+ }
+
+ /// Evolves `table` so its historical `identity(x)` spec outlives the source column:
+ /// makes the default spec unpartitioned and drops `x` (field id 1) from the schema,
+ /// so the manifests' spec resolves `x` to `unknown`.
+ fn drop_promoted_partition_source(table: Table) -> Table {
+ let evolved_schema = Schema::builder()
+ .with_fields(
+ table
+ .metadata()
+ .current_schema()
+ .as_struct()
+ .fields()
+ .iter()
+ .filter(|field| field.id != 1)
+ .cloned(),
+ )
+ .with_identifier_field_ids(vec![2])
+ .build()
+ .unwrap();
+ let metadata = TableMetadataBuilder::new_from_metadata(table.metadata().clone(), None)
+ .add_default_partition_spec(UnboundPartitionSpec::builder().build())
+ .unwrap()
+ .add_current_schema(evolved_schema)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ table.with_metadata(Arc::new(metadata))
+ }
+
+ #[tokio::test]
+ async fn test_scan_keeps_eq_delete_after_dropping_promoted_source() {
+ // Partition source `x` promoted int -> long, then dropped. The data file (int)
+ // and its equality delete (long) must still normalize to the same long partition
+ // so the delete stays attached; otherwise it silently detaches.
+ let mut fixture = TableTestFixture::new();
+ fixture.setup_promoted_partition_with_eq_delete(false).await;
+ let table = drop_promoted_partition_source(fixture.table);
+
+ let tasks: Vec<FileScanTask> = table
+ .scan()
+ .build()
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(tasks.len(), 1);
+ assert_eq!(tasks[0].deletes().len(), 1);
+ }
+
+ #[tokio::test]
+ async fn test_scan_keeps_deletion_vector_after_dropping_promoted_source() {
+ // Same as above for a deletion vector: without widening the dropped numeric
+ // source the DV partition check errors (long(7) vs int(7)).
+ let mut fixture = TableTestFixture::new();
+ fixture.setup_promoted_partition_with_dv().await;
+ let table = drop_promoted_partition_source(fixture.table);
+
+ let tasks: Vec<FileScanTask> = table
+ .scan()
+ .build()
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(tasks.len(), 1);
+ assert_eq!(tasks[0].deletes().len(), 1);
+ }
+
+ #[tokio::test]
+ async fn test_scan_keeps_eq_delete_when_dropped_source_is_date() {
+ // A promoted numeric source `x` sits beside a dropped `date` source `d`. The gate
+ // fires on `x`'s int -> long widening, but `d` (stored as an int) must be copied
+ // through, not widened to long, or the data file's partition diverges from the
+ // equality delete's and detaches it.
+ let mut fixture = TableTestFixture::new();
+ fixture
+ .setup_promoted_partition_with_eq_delete_and_dropped_date()
+ .await;
+
+ let tasks: Vec<FileScanTask> = fixture
+ .table
+ .scan()
+ .build()
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(tasks.len(), 1);
+ assert_eq!(tasks[0].deletes().len(), 1);
+ }
+
#[tokio::test]
async fn test_plan_files_carries_row_lineage_into_file_scan_task() {
let mut fixture = TableTestFixture::new();
diff --git a/crates/iceberg/src/spec/manifest/data_file.rs b/crates/iceberg/src/spec/manifest/data_file.rs
index 94568ab44..6138e5220 100644
--- a/crates/iceberg/src/spec/manifest/data_file.rs
+++ b/crates/iceberg/src/spec/manifest/data_file.rs
@@ -29,7 +29,9 @@ use super::{
};
use crate::Error;
use crate::error::{Result, invalid_data};
-use crate::spec::{DEFAULT_PARTITION_SPEC_ID, Struct, StructType, Type};
+use crate::spec::{
+ DEFAULT_PARTITION_SPEC_ID, Literal, NestedFieldRef, PrimitiveType, Struct, StructType, Type,
+};
/// Data file carries data file path, partition tuple, metrics, …
#[derive(Debug, PartialEq, Clone, Eq, Builder)]
@@ -184,7 +186,71 @@ pub struct DataFile {
pub(crate) content_size_in_bytes: Option<i64>,
}
+fn promote_partition_value(
+ value: Option<&Literal>,
+ current_field: &NestedFieldRef,
+ written_field: &NestedFieldRef,
+) -> Option<Literal> {
+ match value {
+ Some(Literal::Primitive(primitive)) => {
+ let promoted = match current_field.field_type.as_primitive_type() {
+ // The source column was dropped, so there is no current type to widen to.
+ // Widen a numeric to its promotion ceiling so this value and its deletes
+ // normalize the same way; keep every other type (e.g. a `date`, stored as
+ // an `int`) at its written type so deletes that skip normalization match.
+ Some(PrimitiveType::Unknown) => {
+ match written_field.field_type.as_primitive_type() {
+ Some(PrimitiveType::Int | PrimitiveType::Float) => primitive.widen(),
+ _ => None,
+ }
+ }
+ Some(target) => primitive.promote_to(target),
+ None => None,
+ };
+ promoted.map(Literal::Primitive).or_else(|| value.cloned())
+ }
+ other => other.cloned(),
+ }
+}
+
impl DataFile {
+ /// Returns this file's partition tuple normalized to the current schema: a field
+ /// promoted (`int` -> `long`, `float` -> `double`) since the manifest was written is
+ /// widened to its `current` type; a dropped numeric source, which has no `current`
+ /// type, is widened to its promotion ceiling by its `written` type; all other values
+ /// are copied through. Column bounds are widened separately, lazily, where they are
+ /// compared.
+ ///
+ /// `current` and `written` are the file's spec resolved against the current schema and
+ /// against the manifest's schema, so they match the stored tuple in length.
+ ///
+ /// # Errors
+ /// Returns `DataInvalid` if the stored tuple and `current` have different lengths; the
+ /// lengths should always match, since both come from the same spec.
+ pub(crate) fn promoted_partition(
+ &self,
+ current: &StructType,
+ written: &StructType,
+ ) -> Result<Struct> {
+ if self.partition.fields().len() != current.fields().len() {
+ return Err(invalid_data!(
+ "Partition tuple has {} values but the spec partition type has {} fields",
+ self.partition.fields().len(),
+ current.fields().len()
+ ));
+ }
+
+ Ok(self
+ .partition
+ .iter()
+ .zip(current.fields())
+ .zip(written.fields())
+ .map(|((value, current_field), written_field)| {
+ promote_partition_value(value, current_field, written_field)
+ })
+ .collect())
+ }
+
/// Get the content type of the data file (data, equality deletes, or position deletes)
pub fn content_type(&self) -> DataContentType {
self.content
@@ -417,7 +483,12 @@ impl std::fmt::Display for DataFileFormat {
#[cfg(test)]
mod test {
- use crate::spec::DataContentType;
+ use crate::ErrorKind;
+ use crate::spec::{
+ DataContentType, DataFileBuilder, DataFileFormat, Literal, NestedField, PrimitiveType,
+ Struct, StructType, Type,
+ };
+
#[test]
fn test_data_content_type_default() {
assert_eq!(DataContentType::default(), DataContentType::Data);
@@ -427,4 +498,161 @@ mod test {
fn test_data_content_type_default_value() {
assert_eq!(DataContentType::default() as i32, 0);
}
+
+ #[test]
+ fn promoted_partition_widens_int_and_float() {
+ // Partition columns promoted int -> long and float -> double; a file
+ // written before the promotion still carries the narrower types.
+ let current = StructType::new(vec![
+ NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(),
+ NestedField::required(1001, "c", Type::Primitive(PrimitiveType::Double)).into(),
+ ]);
+ let written = StructType::new(vec![
+ NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Int)).into(),
+ NestedField::required(1001, "c", Type::Primitive(PrimitiveType::Float)).into(),
+ ]);
+
+ let data_file = DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path("x.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .partition(Struct::from_iter([
+ Some(Literal::int(7)),
+ Some(Literal::float(1.5_f32)),
+ ]))
+ .record_count(1)
+ .file_size_in_bytes(1)
+ .partition_spec_id(0)
+ .build()
+ .unwrap();
+
+ assert_eq!(
+ data_file.promoted_partition(¤t, &written).unwrap(),
+ Struct::from_iter([Some(Literal::long(7)), Some(Literal::double(1.5))])
+ );
+ }
+
+ #[test]
+ fn promoted_partition_widens_dropped_numeric_source() {
+ // A dropped partition source resolves to `Unknown`, so there is no current
+ // type; a numeric still widens to its ceiling by its written type, keeping it
+ // aligned with deletes written while the source still existed.
+ let current = StructType::new(vec![
+ NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Unknown)).into(),
+ ]);
+ let written = StructType::new(vec![
+ NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Int)).into(),
+ ]);
+
+ let data_file = DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path("x.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .partition(Struct::from_iter([Some(Literal::int(7))]))
+ .record_count(1)
+ .file_size_in_bytes(1)
+ .partition_spec_id(0)
+ .build()
+ .unwrap();
+
+ assert_eq!(
+ data_file.promoted_partition(¤t, &written).unwrap(),
+ Struct::from_iter([Some(Literal::long(7))])
+ );
+ }
+
+ #[test]
+ fn promoted_partition_preserves_dropped_date_source() {
+ // A `date` is stored as an `int`, so widening a dropped date the way a dropped
+ // int widens would turn it into a `long` and diverge from delete manifests that
+ // skip normalization. A dropped non-numeric source is copied through unchanged.
+ let current = StructType::new(vec![
+ NestedField::required(1000, "d", Type::Primitive(PrimitiveType::Unknown)).into(),
+ ]);
+ let written = StructType::new(vec![
+ NestedField::required(1000, "d", Type::Primitive(PrimitiveType::Date)).into(),
+ ]);
+
+ let data_file = DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path("x.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .partition(Struct::from_iter([Some(Literal::date(20_000))]))
+ .record_count(1)
+ .file_size_in_bytes(1)
+ .partition_spec_id(0)
+ .build()
+ .unwrap();
+
+ assert_eq!(
+ data_file.promoted_partition(¤t, &written).unwrap(),
+ data_file.partition
+ );
+ }
+
+ #[test]
+ fn promoted_partition_leaves_matching_types_unchanged() {
+ let partition_type = StructType::new(vec![
+ NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(),
+ ]);
+
+ let data_file = DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path("x.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .partition(Struct::from_iter([Some(Literal::long(7))]))
+ .record_count(1)
+ .file_size_in_bytes(1)
+ .partition_spec_id(0)
+ .build()
+ .unwrap();
+
+ assert_eq!(
+ data_file
+ .promoted_partition(&partition_type, &partition_type)
+ .unwrap(),
+ data_file.partition
+ );
+ }
+
+ #[test]
+ fn promoted_partition_rejects_tuple_length_mismatch() {
+ let data_file = |values: Vec<Option<Literal>>| {
+ DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path("x.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .partition(Struct::from_iter(values))
+ .record_count(1)
+ .file_size_in_bytes(1)
+ .partition_spec_id(0)
+ .build()
+ .unwrap()
+ };
+ let one_field = StructType::new(vec![
+ NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(),
+ ]);
+ let two_fields = StructType::new(vec![
+ NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(),
+ NestedField::required(1001, "c", Type::Primitive(PrimitiveType::Long)).into(),
+ ]);
+
+ let longer = data_file(vec![Some(Literal::int(7)), Some(Literal::int(8))]);
+ assert_eq!(
+ longer
+ .promoted_partition(&one_field, &one_field)
+ .unwrap_err()
+ .kind(),
+ ErrorKind::DataInvalid
+ );
+
+ let shorter = data_file(vec![Some(Literal::int(7))]);
+ assert_eq!(
+ shorter
+ .promoted_partition(&two_fields, &two_fields)
+ .unwrap_err()
+ .kind(),
+ ErrorKind::DataInvalid
+ );
+ }
}
diff --git a/crates/iceberg/src/spec/values/datum.rs b/crates/iceberg/src/spec/values/datum.rs
index 09c1e8831..5c4339d8b 100644
--- a/crates/iceberg/src/spec/values/datum.rs
+++ b/crates/iceberg/src/spec/values/datum.rs
@@ -1229,6 +1229,17 @@ impl Datum {
&self.r#type
}
+ /// 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. Returns `None` when `target` does not widen this value (decimals
+ /// already compare across precisions), so callers can avoid a copy.
+ pub(crate) fn promote_to(&self, target: &PrimitiveType) -> Option<Datum> {
+ self.literal
+ .promote_to(target)
+ .map(|literal| Datum::new(target.clone(), literal))
+ }
+
/// Returns true if the Literal represents a primitive type
/// that can be a NaN, and that it's value is NaN
pub fn is_nan(&self) -> bool {
diff --git a/crates/iceberg/src/spec/values/primitive.rs b/crates/iceberg/src/spec/values/primitive.rs
index b76978bd5..2204e2242 100644
--- a/crates/iceberg/src/spec/values/primitive.rs
+++ b/crates/iceberg/src/spec/values/primitive.rs
@@ -21,6 +21,8 @@ use std::cmp::Ordering;
use ordered_float::{FloatCore, OrderedFloat};
+use crate::spec::PrimitiveType;
+
/// Values present in iceberg type
///
/// `Float` and `Double` compare as Java's `Float.compare` and `Double.compare` do: `-0.0` is
@@ -105,6 +107,30 @@ fn float_cmp<T: FloatCore>(a: &OrderedFloat<T>, b: &OrderedFloat<T>) -> Ordering
}
impl PrimitiveLiteral {
+ /// Widens this literal to its promotion ceiling (`int` -> `long`, `float` ->
+ /// `double`), returning `None` for any other literal. Used when the target type
+ /// is unknown (the partition source column was dropped) so a value and its
+ /// deletes still normalize to the same type.
+ pub(crate) fn widen(&self) -> Option<PrimitiveLiteral> {
+ match self {
+ PrimitiveLiteral::Int(value) => Some(PrimitiveLiteral::Long(i64::from(*value))),
+ PrimitiveLiteral::Float(value) => {
+ Some(PrimitiveLiteral::Double(OrderedFloat(f64::from(value.0))))
+ }
+ _ => None,
+ }
+ }
+
+ /// Widens this literal to `target` for the value promotions `int` -> `long`
+ /// and `float` -> `double`, returning `None` when `target` does not promote it.
+ pub(crate) fn promote_to(&self, target: &PrimitiveType) -> Option<PrimitiveLiteral> {
+ match (self, target) {
+ (PrimitiveLiteral::Int(_), PrimitiveType::Long)
+ | (PrimitiveLiteral::Float(_), PrimitiveType::Double) => self.widen(),
+ _ => None,
+ }
+ }
+
/// Must follow the declaration order of the variants: `partial_cmp` uses it to order
/// different variants the way the derived `PartialOrd` did.
fn variant_index(&self) -> u8 {
diff --git a/crates/iceberg/src/test_utils/scan.rs b/crates/iceberg/src/test_utils/scan.rs
index 2f96e61ad..039fb6421 100644
--- a/crates/iceberg/src/test_utils/scan.rs
+++ b/crates/iceberg/src/test_utils/scan.rs
@@ -44,7 +44,8 @@ use crate::metadata_columns::{
use crate::spec::{
DataContentType, DataFileBuilder, DataFileFormat, FormatVersion, Literal, ManifestEntry,
ManifestListWriter, ManifestStatus, ManifestWriterBuilder, NestedField, PartitionSpec,
- PrimitiveType, Schema, Struct, StructType, TableMetadata, TableMetadataBuilder, Type,
+ PrimitiveType, Schema, Struct, StructType, TableMetadata, TableMetadataBuilder, Transform,
+ Type, UnboundPartitionSpec,
};
use crate::table::Table;
use crate::test_utils::test_runtime;
@@ -606,6 +607,425 @@ impl TableTestFixture {
manifest_list_write.close().await.unwrap();
}
+ /// Writes a data manifest under an older schema where the identity partition
+ /// source `x` (id 1) is still `int` (so the partition value is stored as `int`),
+ /// plus an equality-delete manifest. Exercises the scan's reader-side widening.
+ /// When `delete_before_promotion` is false the delete is written under the current
+ /// `long` schema (`long` partition), and the old `int` partition must widen to
+ /// match it. When true the delete is also written under the old `int` schema (`int`
+ /// partition), which matched on main and must keep matching once both sides widen.
+ /// The data/delete parquet files themselves are not written, which is enough for
+ /// `plan_files` (it reads only manifests).
+ pub(crate) async fn setup_promoted_partition_with_eq_delete(
+ &mut self,
+ delete_before_promotion: bool,
+ ) {
+ let current_snapshot = self.table.metadata().current_snapshot().unwrap();
+ let parent_snapshot = current_snapshot
+ .parent_snapshot(self.table.metadata())
+ .unwrap();
+ let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
+ let current_partition_spec = self.table.metadata().default_partition_spec();
+
+ // The schema the data manifest was written under: the current schema with the
+ // partition source `x` still `int`.
+ let mut fields = current_schema.as_struct().fields().to_vec();
+ fields[0] = NestedField::required(1, "x", PrimitiveType::Int.into()).into();
+ let written_schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(0)
+ .with_fields(fields)
+ .build()
+ .unwrap(),
+ );
+
+ // Data manifest under the old (int) schema; an Existing entry from the parent
+ // snapshot so its data sequence number is below the equality delete's.
+ let mut data_writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ written_schema.clone(),
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v2_data();
+ data_writer
+ .add_existing_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Existing)
+ .snapshot_id(parent_snapshot.snapshot_id())
+ .sequence_number(parent_snapshot.sequence_number())
+ .file_sequence_number(parent_snapshot.sequence_number())
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/x7.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(100)
+ .record_count(1)
+ .partition(Struct::from_iter([Some(Literal::int(7))]))
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ let data_manifest = data_writer.write_manifest_file().await.unwrap();
+
+ // Equality-delete manifest, Added in the current snapshot so its sequence
+ // number is above the data file's. Written under the old `int` schema or the
+ // current `long` schema depending on `delete_before_promotion`.
+ let (delete_schema, delete_partition) = if delete_before_promotion {
+ (written_schema, Struct::from_iter([Some(Literal::int(7))]))
+ } else {
+ (
+ current_schema.clone(),
+ Struct::from_iter([Some(Literal::long(7))]),
+ )
+ };
+ let mut delete_writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ delete_schema,
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v2_deletes();
+ delete_writer
+ .add_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::EqualityDeletes)
+ .equality_ids(Some(vec![2]))
+ .file_path(format!("{}/eq-delete.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(100)
+ .record_count(1)
+ .partition(delete_partition)
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ let delete_manifest = delete_writer.write_manifest_file().await.unwrap();
+
+ let manifest_list_writer = self
+ .table
+ .file_io()
+ .new_output(current_snapshot.manifest_list())
+ .unwrap()
+ .writer()
+ .await
+ .unwrap();
+ let mut manifest_list_write = ManifestListWriter::v2(
+ manifest_list_writer,
+ current_snapshot.snapshot_id(),
+ current_snapshot.parent_snapshot_id(),
+ current_snapshot.sequence_number(),
+ );
+ manifest_list_write
+ .add_manifests(vec![data_manifest, delete_manifest].into_iter())
+ .unwrap();
+ manifest_list_write.close().await.unwrap();
+ }
+
+ /// Writes a v3 data manifest under an older schema where the identity partition
+ /// source `x` (id 1) is still `int`, plus a v3 deletion-vector manifest under the
+ /// current `long` schema whose DV references the data file with a `long` partition.
+ /// The reader-side widening must make the data file's `int` partition agree with
+ /// the DV's `long` partition so the DV partition check passes and the DV attaches;
+ /// on main the check errors. Parquet/Puffin files are not written (`plan_files`
+ /// reads only manifests).
+ pub(crate) async fn setup_promoted_partition_with_dv(&mut self) {
+ let metadata = TableMetadataBuilder::new_from_metadata(
+ self.table.metadata().clone(),
+ self.table.metadata_location().map(str::to_string),
+ )
+ .upgrade_format_version(FormatVersion::V3)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ self.table = Table::builder()
+ .metadata(metadata)
+ .identifier(self.table.identifier().clone())
+ .file_io(self.table.file_io().clone())
+ .metadata_location(self.table.metadata_location().unwrap().to_string())
+ .runtime(test_runtime())
+ .build()
+ .unwrap();
+
+ let current_snapshot = self.table.metadata().current_snapshot().unwrap();
+ let parent_snapshot = current_snapshot
+ .parent_snapshot(self.table.metadata())
+ .unwrap();
+ 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();
+ fields[0] = NestedField::required(1, "x", PrimitiveType::Int.into()).into();
+ let written_schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(0)
+ .with_fields(fields)
+ .build()
+ .unwrap(),
+ );
+
+ let data_file_path = format!("{}/x7.parquet", &self.table_location);
+
+ let mut data_writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ written_schema,
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v3_data();
+ data_writer
+ .add_existing_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Existing)
+ .snapshot_id(parent_snapshot.snapshot_id())
+ .sequence_number(parent_snapshot.sequence_number())
+ .file_sequence_number(parent_snapshot.sequence_number())
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::Data)
+ .file_path(data_file_path.clone())
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(100)
+ .record_count(1)
+ .partition(Struct::from_iter([Some(Literal::int(7))]))
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ let data_manifest = data_writer.write_manifest_file().await.unwrap();
+
+ let mut dv_writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ current_schema.clone(),
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v3_deletes();
+ dv_writer
+ .add_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::PositionDeletes)
+ .file_path(format!("{}/dv.puffin", &self.table_location))
+ .file_format(DataFileFormat::Puffin)
+ .file_size_in_bytes(100)
+ .record_count(1)
+ .referenced_data_file(Some(data_file_path))
+ .content_offset(Some(0))
+ .content_size_in_bytes(Some(40))
+ .partition(Struct::from_iter([Some(Literal::long(7))]))
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ let dv_manifest = dv_writer.write_manifest_file().await.unwrap();
+
+ let manifest_list_writer = self
+ .table
+ .file_io()
+ .new_output(current_snapshot.manifest_list())
+ .unwrap()
+ .writer()
+ .await
+ .unwrap();
+ let mut manifest_list_write = ManifestListWriter::v3(
+ manifest_list_writer,
+ current_snapshot.snapshot_id(),
+ current_snapshot.parent_snapshot_id(),
+ current_snapshot.sequence_number(),
+ Some(42),
+ );
+ manifest_list_write
+ .add_manifests(vec![data_manifest, dv_manifest].into_iter())
+ .unwrap();
+ manifest_list_write.close().await.unwrap();
+ }
+
+ /// Writes a two-field spec `identity(x)`, `identity(d date)` where the data manifest
+ /// was written with `x` still `int` (so the current `long` schema widens it), then
+ /// drops the date source `d`. A dropped `date` is stored as an `int`, so if it were
+ /// widened like a dropped numeric the data file's partition would become `long` and
+ /// diverge from the equality delete's, detaching it. Parquet files are not written
+ /// (`plan_files` reads only manifests).
+ pub(crate) async fn setup_promoted_partition_with_eq_delete_and_dropped_date(&mut self) {
+ // Evolve to the two-field spec with the date source still present.
+ let mut fields = self
+ .table
+ .metadata()
+ .current_schema()
+ .as_struct()
+ .fields()
+ .to_vec();
+ fields.push(NestedField::optional(9, "d", PrimitiveType::Date.into()).into());
+ let date_schema = Schema::builder().with_fields(fields).build().unwrap();
+ let date_spec = PartitionSpec::builder(date_schema.clone())
+ .add_partition_field("x", "x", Transform::Identity)
+ .unwrap()
+ .add_partition_field("d", "d", Transform::Identity)
+ .unwrap()
+ .build()
+ .unwrap();
+ let metadata = TableMetadataBuilder::new_from_metadata(self.table.metadata().clone(), None)
+ .add_current_schema(date_schema)
+ .unwrap()
+ .add_default_partition_spec(date_spec.into_unbound())
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ self.table = self.table.clone().with_metadata(Arc::new(metadata));
+
+ let current_snapshot = self.table.metadata().current_snapshot().unwrap();
+ let parent_snapshot = current_snapshot
+ .parent_snapshot(self.table.metadata())
+ .unwrap();
+ let current_schema = self.table.metadata().current_schema().clone();
+ let current_partition_spec = self.table.metadata().default_partition_spec();
+
+ // The schema the data manifest was written under: the partition source `x` still
+ // `int`.
+ let mut fields = current_schema.as_struct().fields().to_vec();
+ fields[0] = NestedField::required(1, "x", PrimitiveType::Int.into()).into();
+ let written_schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(0)
+ .with_fields(fields)
+ .build()
+ .unwrap(),
+ );
+
+ // Data manifest under the old (int) schema; an Existing entry from the parent
+ // snapshot so its sequence number is below the equality delete's.
+ let mut data_writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ written_schema,
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v2_data();
+ data_writer
+ .add_existing_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Existing)
+ .snapshot_id(parent_snapshot.snapshot_id())
+ .sequence_number(parent_snapshot.sequence_number())
+ .file_sequence_number(parent_snapshot.sequence_number())
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(current_partition_spec.spec_id())
+ .content(DataContentType::Data)
+ .file_path(format!("{}/x7.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(100)
+ .record_count(1)
+ .partition(Struct::from_iter([
+ Some(Literal::int(7)),
+ Some(Literal::date(20_000)),
+ ]))
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ let data_manifest = data_writer.write_manifest_file().await.unwrap();
+
+ // Equality-delete manifest, Added in the current snapshot (sequence number above
+ // the data file's) under the current `long` schema.
+ let mut delete_writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ current_schema.clone(),
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v2_deletes();
+ delete_writer
+ .add_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(current_partition_spec.spec_id())
+ .content(DataContentType::EqualityDeletes)
+ .equality_ids(Some(vec![2]))
+ .file_path(format!("{}/eq-delete.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(100)
+ .record_count(1)
+ .partition(Struct::from_iter([
+ Some(Literal::long(7)),
+ Some(Literal::date(20_000)),
+ ]))
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ let delete_manifest = delete_writer.write_manifest_file().await.unwrap();
+
+ let manifest_list_writer = self
+ .table
+ .file_io()
+ .new_output(current_snapshot.manifest_list())
+ .unwrap()
+ .writer()
+ .await
+ .unwrap();
+ let mut manifest_list_write = ManifestListWriter::v2(
+ manifest_list_writer,
+ current_snapshot.snapshot_id(),
+ current_snapshot.parent_snapshot_id(),
+ current_snapshot.sequence_number(),
+ );
+ manifest_list_write
+ .add_manifests(vec![data_manifest, delete_manifest].into_iter())
+ .unwrap();
+ manifest_list_write.close().await.unwrap();
+
+ // Drop the date source `d`: the manifests keep the two-field spec while the
+ // current schema loses `d`, so the spec resolves `d` to `unknown`.
+ let evolved_schema = Schema::builder()
+ .with_fields(
+ current_schema
+ .as_struct()
+ .fields()
+ .iter()
+ .filter(|field| field.id != 9)
+ .cloned(),
+ )
+ .build()
+ .unwrap();
+ let metadata = TableMetadataBuilder::new_from_metadata(self.table.metadata().clone(), None)
+ .add_default_partition_spec(UnboundPartitionSpec::builder().build())
+ .unwrap()
+ .add_current_schema(evolved_schema)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ self.table = self.table.clone().with_metadata(Arc::new(metadata));
+ }
+
/// Writes identical Parquet data files (1.parquet, 2.parquet, 3.parquet)
/// and returns the file size in bytes.
fn write_parquet_data_files(&self) -> u64 {