This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 5c2715767 fix(expr): widen promoted partition values and bounds to
prevent false pruning (#3483)
5c2715767 is described below
commit 5c27157675a7babb611638786195dfe24ad41dab
Author: Anoop Johnson <[email protected]>
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 <[email protected]>
* 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 <[email protected]>
---
.../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 {
