laskoviymishka commented on code in PR #3483:
URL: https://github.com/apache/iceberg-rust/pull/3483#discussion_r4345819208
##########
crates/iceberg/src/scan/context.rs:
##########
@@ -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();
Review Comment:
Follow-up, not for this PR: could deleted entries skip this deep clone? In a
gated manifest every entry is copied here, before `process_data_manifest_entry`
drops non-alive ones and prunes the rest
([scan/mod.rs#L579-L630](https://github.com/apache/iceberg-rust/blob/db4e54454ab9372bf53f8a9c7657deec5f5a3af4/crates/iceberg/src/scan/mod.rs#L579-L630)).
`main` only clones the `Arc`
([context.rs#L111-L112](https://github.com/apache/iceberg-rust/blob/1be091a2135a5908537a5b45786ae7207a723aab/crates/iceberg/src/scan/context.rs#L111-L112)).
As I read it, partition pruning needs the widened tuple, so the alive check is
the part that could move ahead of the clone.
##########
crates/iceberg/src/test_utils/scan.rs:
##########
@@ -604,6 +605,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();
Review Comment:
Non-blocking, fine as a follow-up: could the three promotion fixtures share
one helper for the int-schema override and the old-schema data manifest? This
block repeats [the one in
`setup_promoted_partition_with_eq_delete`](https://github.com/apache/iceberg-rust/blob/db4e54454ab9372bf53f8a9c7657deec5f5a3af4/crates/iceberg/src/test_utils/scan.rs#L630-L631).
The date fixture's source drop
([L1003-L1020](https://github.com/apache/iceberg-rust/blob/db4e54454ab9372bf53f8a9c7657deec5f5a3af4/crates/iceberg/src/test_utils/scan.rs#L1003-L1020))
has the same shape as `drop_promoted_partition_source`
([scan/mod.rs](https://github.com/apache/iceberg-rust/blob/db4e54454ab9372bf53f8a9c7657deec5f5a3af4/crates/iceberg/src/scan/mod.rs#L1318-L1342)),
which could move into `test_utils` and be reused.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
