anoopj commented on code in PR #3483:
URL: https://github.com/apache/iceberg-rust/pull/3483#discussion_r4361162757
##########
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:
That makes sense. Will send a followup to clean this up.
--
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]
