viirya commented on code in PR #3662:
URL: https://github.com/apache/iceberg-rust/pull/3662#discussion_r4522943282
##########
crates/iceberg/src/transaction/snapshot/simple.rs:
##########
@@ -180,28 +178,25 @@ impl SimpleSnapshotProducer {
self.validate_added_data_files(table)?;
Review Comment:
On a cache hit we don't check that the cached manifests' specs still exist.
If a concurrent commit removes spec 0 (allowed once it's no longer the
default), the retry still fails, but only because `build_summary` can't resolve
the spec ("Failed to create snapshot summary."). That's the right outcome, but
it happens by accident. Could we check this explicitly here, or at least note
it next to the "a spec the table still has" doc on `validate_added_data_files`?
##########
crates/iceberg/src/transaction/snapshot/simple.rs:
##########
@@ -128,24 +126,24 @@ impl SimpleSnapshotProducer {
.await
}
- /// Checks that every file is written under the table's default partition
- /// spec with a compatible partition value.
+ /// Checks that every file uses a known partition spec with a compatible
+ /// partition value.
///
/// Runs against the table the manifests are written for; a later rebase
/// does not invalidate manifests already written under a spec the table
/// still has.
fn validate_added_data_files(&self, table: &Table) -> Result<()> {
let metadata = table.metadata();
for data_file in &self.intent.data_files {
- // TODO(#3500): accept any spec known to the table and group the
- // files by spec when writing manifests, instead of requiring the
- // default spec.
- if metadata.default_partition_spec_id() !=
data_file.partition_spec_id {
- return Err(invalid_data!(
- "Data file partition spec id does not match table default
partition spec id"
- ));
- }
- validate_partition_value(data_file.partition(),
metadata.default_partition_type())?;
+ let spec = metadata
+ .partition_spec_by_id(data_file.partition_spec_id)
+ .ok_or_else(|| {
+ invalid_data!("Unknown partition spec id {}",
data_file.partition_spec_id)
Review Comment:
Now that files from several specs can be mixed, it would help to include the
file path in this error and in the incompatible-value error below, e.g. `Cannot
find partition spec {id} for data file {path}` as in Java.
Also, `partition_type` is now rebuilt for every file. Grouping first and
validating each group once would avoid that, and would let
`write_added_manifest` take the resolved spec instead of looking it up again.
##########
crates/iceberg/src/transaction/snapshot/simple.rs:
##########
@@ -411,14 +406,96 @@ mod tests {
assert_eq!(producer.manifest_counter, 1);
}
+ #[tokio::test]
+ async fn test_append_uses_file_spec_after_default_spec_changes() {
+ let table = make_v2_minimal_table();
+ let file = data_file(&table, "test/1.parquet");
+ let mut state = producer(vec![file.clone()]);
+ let first_snapshot = added_snapshot(&mut
state.apply(&table).await.unwrap());
+ let first_list = table
+ .manifest_list_reader(&first_snapshot)
+ .load()
+ .await
+ .unwrap();
+
+ let metadata = table
+ .metadata()
+ .clone()
+ .into_builder(None)
+
.add_default_partition_spec(UnboundPartitionSpec::builder().build())
Review Comment:
The retry half of this test still passes if `build_summary` is reverted to
`default_partition_spec()`, because the new default has fewer fields. Using a
default spec with more fields (e.g. the bucket spec from append.rs) and
asserting the retried summary would pin the fix to the retry path. The
fresh-producer half is already covered by
`test_fast_append_groups_files_by_partition_spec`, so it could go.
##########
crates/iceberg/src/transaction/append.rs:
##########
@@ -1001,6 +1001,196 @@ mod tests {
assert_eq!(data_file, *manifest.entries()[0].data_file());
}
+ fn with_bucket_partition_spec(table: Table) -> Table {
+ let spec =
PartitionSpec::builder(table.metadata().current_schema().clone())
+ .add_partition_field("x", "x", Transform::Identity)
+ .unwrap()
+ .add_partition_field("y", "y_bucket", Transform::Bucket(16))
+ .unwrap()
+ .build()
+ .unwrap()
+ .into_unbound();
+ let metadata = table
+ .metadata()
+ .clone()
+ .into_builder(None)
+ .add_default_partition_spec(spec)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ table.with_metadata(Arc::new(metadata))
+ }
+
+ #[tokio::test]
+ async fn test_fast_append_groups_files_by_partition_spec() {
+ for table in [
+ make_v1_table(),
+ make_v2_minimal_table(),
+ make_v3_minimal_table(),
+ ] {
+ let metadata = table
+ .metadata()
+ .clone()
+ .into_builder(None)
+ .set_properties(HashMap::from([(
+
TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT.to_string(),
+ "2".to_string(),
+ )]))
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ let table = table.with_metadata(Arc::new(metadata));
+ let table =
with_bucket_partition_spec(table).with_metadata_location(
+
"s3://bucket/test/location/metadata/00000-9c12d441-03fe-4693-9a96-a0705ddf69c1.metadata.json".to_string(),
+ );
+ let old_file = DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path("test/old.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(100)
+ .record_count(1)
+ .partition_spec_id(0)
+ .partition(Struct::from_iter([Some(Literal::long(300))]))
+ .build()
+ .unwrap();
+ let mut other_old_file = old_file.clone();
+ other_old_file.file_path = "test/other-old.parquet".to_string();
+ let mut new_file = old_file.clone();
+ new_file.file_path = "test/new.parquet".to_string();
+ new_file.partition_spec_id =
table.metadata().default_partition_spec_id();
+ new_file.partition =
+ Struct::from_iter([Some(Literal::long(300)),
Some(Literal::int(3))]);
+
+ let action =
Transaction::new(&table).fast_append().add_data_files(vec![
+ new_file.clone(),
+ old_file.clone(),
+ other_old_file.clone(),
+ old_file.clone(),
Review Comment:
The duplicate inputs here and the already-live check at the end are
spec-agnostic and already covered by `test_fast_append_dedupes_*` and
`test_validate_files_not_live_sees_files_added_by_rebase`. Dropping them would
keep this test focused on grouping and the summary.
--
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]
