laskoviymishka commented on code in PR #3453:
URL: https://github.com/apache/iceberg-rust/pull/3453#discussion_r4537060565
##########
crates/iceberg/src/arrow/delete_file_loader.rs:
##########
@@ -115,6 +119,195 @@ impl BasicDeleteFileLoader {
}
}
+/// Checks that every equality id is a field of the scan schema. Java's
`DeleteFilter` fails the
+/// same way rather than matching rows on fewer columns.
+pub(crate) fn validate_equality_ids(schema: &Schema, equality_ids: &[i32]) ->
Result<()> {
+ match equality_ids
+ .iter()
+ .find(|&&id| schema.field_by_id(id).is_none())
+ {
+ Some(id) => Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!("Equality delete field id {id} is not in the scan schema"),
+ )),
+ None => Ok(()),
+ }
+}
+
+/// Reads an equality delete file as `schema` pruned to its equality fields,
the way Java's
+/// `DeleteFilter` selects them, so a field nested in a struct keeps its
parent struct.
+///
+/// Java writes a nested equality field under its parent structs, while
`EqualityDeleteWriter`
+/// writes it as a top-level column. Such a column is put back under its
parents before schema
+/// evolution, so both layouts read the same way.
+pub(crate) fn evolve_equality_delete_stream(
+ record_batch_stream: ArrowRecordBatchStream,
+ schema: &Schema,
+ equality_ids: &[i32],
+) -> Result<ArrowRecordBatchStream> {
+ validate_equality_ids(schema, equality_ids)?;
+ let crate::spec::Type::Struct(pruned) =
+ prune_columns(schema, equality_ids.iter().copied(), false)?
+ else {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "pruning a schema did not return a struct",
+ ));
+ };
+ let projection: Vec<i32> = pruned.fields().iter().map(|field|
field.id).collect();
+ let pruned = Arc::new(
+ Schema::builder()
+ .with_schema_id(schema.schema_id())
+ .with_fields(pruned.fields().to_vec())
+ .build()?,
+ );
+ let mut transformer = RecordBatchTransformerBuilder::new(pruned.clone(),
&projection).build();
+ // The batches of a file share its schema, so the nesting is worked out
once and only
+ // rebuilt if a batch comes with a different schema.
+ let mut nesting: Option<EqualityColumnNesting> = None;
+ Ok(Box::pin(record_batch_stream.map(move |batch| {
+ let batch = batch?;
+ let nesting = match &mut nesting {
+ Some(nesting) if nesting.input_schema == batch.schema() => nesting,
+ slot => slot.insert(EqualityColumnNesting::try_new(batch.schema(),
&pruned)?),
Review Comment:
Could the test also feed a batch with a different schema?
[`test_equality_delete_stream_nests_every_batch`](https://github.com/apache/iceberg-rust/blob/4050928545273ca16cc7e3af832da52d1fd4e576/crates/iceberg/src/arrow/caching_delete_file_loader.rs#L2480-L2515)
sends two batches with the same schema and checks only the values, so it
passes with or without the cache, and this arm never runs for a changed
`batch.schema()`.
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -619,9 +624,10 @@ impl CachingDeleteFileLoader {
}
};
- let root_array: ArrayRef =
Arc::new(StructArray::from(record_batch));
+ let root_array: ArrayRef =
+
push_struct_nulls_down(&(Arc::new(StructArray::from(record_batch)) as
ArrayRef))?;
Review Comment:
Should a null `s` in the delete file follow the spec text or Java? A null
`s` and `s = {f: null}` both load as `s.f IS NOT NULL`. The
[spec](https://github.com/apache/iceberg/blob/6976e020b894f6a6777704df2b8c4458cb291ae9/format/spec.md#L1179-L1183)
reads that way ("if a parent struct column is null it implies the leaf column
is null"), but Java's
[`DeleteFilter`](https://github.com/apache/iceberg/blob/6976e020b894f6a6777704df2b8c4458cb291ae9/data/src/main/java/org/apache/iceberg/data/DeleteFilter.java#L213-L220)
compares the projected struct with
[`nullsFirst`](https://github.com/apache/iceberg/blob/6976e020b894f6a6777704df2b8c4458cb291ae9/api/src/main/java/org/apache/iceberg/types/Comparators.java#L95-L107)
and tells them apart. This isn't from `push_struct_nulls_down` (1fc76289
builds the same predicate) and is unreachable until #2906, so not blocking.
##########
crates/iceberg/src/arrow/delete_file_loader.rs:
##########
@@ -115,6 +119,195 @@ impl BasicDeleteFileLoader {
}
}
+/// Checks that every equality id is a field of the scan schema. Java's
`DeleteFilter` fails the
+/// same way rather than matching rows on fewer columns.
+pub(crate) fn validate_equality_ids(schema: &Schema, equality_ids: &[i32]) ->
Result<()> {
+ match equality_ids
+ .iter()
+ .find(|&&id| schema.field_by_id(id).is_none())
+ {
+ Some(id) => Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!("Equality delete field id {id} is not in the scan schema"),
+ )),
+ None => Ok(()),
+ }
+}
+
+/// Reads an equality delete file as `schema` pruned to its equality fields,
the way Java's
+/// `DeleteFilter` selects them, so a field nested in a struct keeps its
parent struct.
+///
+/// Java writes a nested equality field under its parent structs, while
`EqualityDeleteWriter`
+/// writes it as a top-level column. Such a column is put back under its
parents before schema
+/// evolution, so both layouts read the same way.
+pub(crate) fn evolve_equality_delete_stream(
+ record_batch_stream: ArrowRecordBatchStream,
+ schema: &Schema,
+ equality_ids: &[i32],
+) -> Result<ArrowRecordBatchStream> {
+ validate_equality_ids(schema, equality_ids)?;
+ let crate::spec::Type::Struct(pruned) =
+ prune_columns(schema, equality_ids.iter().copied(), false)?
+ else {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "pruning a schema did not return a struct",
+ ));
+ };
+ let projection: Vec<i32> = pruned.fields().iter().map(|field|
field.id).collect();
+ let pruned = Arc::new(
+ Schema::builder()
+ .with_schema_id(schema.schema_id())
+ .with_fields(pruned.fields().to_vec())
+ .build()?,
+ );
+ let mut transformer = RecordBatchTransformerBuilder::new(pruned.clone(),
&projection).build();
+ // The batches of a file share its schema, so the nesting is worked out
once and only
+ // rebuilt if a batch comes with a different schema.
+ let mut nesting: Option<EqualityColumnNesting> = None;
+ Ok(Box::pin(record_batch_stream.map(move |batch| {
+ let batch = batch?;
+ let nesting = match &mut nesting {
+ Some(nesting) if nesting.input_schema == batch.schema() => nesting,
+ slot => slot.insert(EqualityColumnNesting::try_new(batch.schema(),
&pruned)?),
+ };
+ transformer.process_record_batch(nesting.apply(batch)?)
+ })) as ArrowRecordBatchStream)
+}
+
+/// How to build the columns of an equality delete batch once the top-level
columns of fields
+/// nested in the pruned schema are moved back under their parent structs.
+struct EqualityColumnNesting {
+ input_schema: ArrowSchemaRef,
+ /// The output schema and columns, or `None` when no column has to move.
+ output: Option<(ArrowSchemaRef, Vec<NestedColumn>)>,
+}
+
+/// An output column: a column of the input batch, or a struct built from such
columns.
+enum NestedColumn {
+ Input(usize),
+ Struct(Fields, Vec<NestedColumn>),
+}
+
+impl NestedColumn {
+ fn build(&self, batch: &RecordBatch) -> ArrayRef {
+ match self {
+ NestedColumn::Input(index) => batch.column(*index).clone(),
+ NestedColumn::Struct(fields, children) =>
Arc::new(StructArray::new(
+ fields.clone(),
+ children.iter().map(|child| child.build(batch)).collect(),
+ None,
+ )),
+ }
+ }
+}
+
+impl EqualityColumnNesting {
+ fn try_new(input_schema: ArrowSchemaRef, schema: &Schema) -> Result<Self> {
+ let top_level_ids: HashSet<i32> = schema
+ .as_struct()
+ .fields()
+ .iter()
+ .map(|field| field.id)
+ .collect();
+ let mut flat = HashMap::new();
+ for (index, field) in input_schema.fields().iter().enumerate() {
+ if let Some(id) = try_get_field_id_from_metadata(field)?
+ && !top_level_ids.contains(&id)
+ && schema.field_by_id(id).is_some()
+ {
+ flat.insert(id, index);
+ }
+ }
+ if flat.is_empty() {
+ return Ok(Self {
+ input_schema,
+ output: None,
+ });
+ }
+
+ fn nest(
Review Comment:
Could we add `load_deletes` tests for two equality ids under one parent
(`[3,4]` in the writer layout) and for a two-level `a.b.c` field? The tests
only use the one-level
[`s{f,g}`](https://github.com/apache/iceberg-rust/blob/4050928545273ca16cc7e3af832da52d1fd4e576/crates/iceberg/src/arrow/caching_delete_file_loader.rs#L2233-L2253)
schema with one nested id, so neither a shared parent nor the recursion here
and in `push_struct_nulls_down` is covered.
The parent built here is always
[nullable](https://github.com/apache/iceberg-rust/blob/4050928545273ca16cc7e3af832da52d1fd4e576/crates/iceberg/src/arrow/delete_file_loader.rs#L252-L256).
With a required `s` the load still returns `(s.f IS NULL) OR (s.f != 7)`, so
could we either take it from `field.required` or add a required-parent case?
--
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]
