You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "sdd (via GitHub)" <gi...@apache.org> on 2025/03/01 12:14:26 UTC
[PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
sdd opened a new pull request, #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017
Continues the series of scan delete file support PRs.
* Adds parsing of equality delete files into `DeleteFileManager`
* Tests WIP, to follow
Builds on top of https://github.com/apache/iceberg-rust/pull/1011, https://github.com/apache/iceberg-rust/pull/982, https://github.com/apache/iceberg-rust/pull/951 and https://github.com/apache/iceberg-rust/pull/950 and should be merged after them.
Issue: https://github.com/apache/iceberg-rust/issues/630
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2196737201
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
+ .map(|(column, field)| {
+ let col_as_datum_vec = arrow_array_to_datum_iterator(column, field);
+ col_as_datum_vec.map(|c| (c, field.name.to_string()))
+ })
+ .try_collect()?;
+
+ // consume all the iterators in lockstep, creating per-row predicates that get combined
+ // into a single final predicate
+
+ // (2025-06-12) can't use `is_empty` as it depends on unstable library feature `exact_size_is_empty`
+ #[allow(clippy::len_zero)]
+ while datum_columns_with_names[0].0.len() > 0 {
+ let mut row_predicate = AlwaysTrue;
+ for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names {
+ if let Some(item) = column.next() {
+ if let Some(datum) = item? {
+ row_predicate = row_predicate
+ .and(Reference::new(field_name.clone()).equal_to(datum.clone()));
+ }
+ }
+ }
+ result_predicate = result_predicate.and(row_predicate.not());
Review Comment:
In fact it seems the logic in Java Iceberg is constructed the same way as I have done so. See https://github.com/apache/iceberg/blob/3f2267735b80c49f3cfc3e22c29bb610d1183582/data/src/main/java/org/apache/iceberg/data/DeleteFilter.java#L237
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#issuecomment-3234328495
Hey - sure, would love to get this merged
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2179703171
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,233 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
- Err(Error::new(
- ErrorKind::FeatureUnsupported,
- "parsing of equality deletes is not yet supported",
- ))
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
Review Comment:
I think this is incorrect, eq id allows fileds to be nested under struct, so comparing top level field may be incorrect. We should use a visitor to do this.
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
+ .map(|(column, field)| {
+ let col_as_datum_vec = arrow_array_to_datum_iterator(column, field);
+ col_as_datum_vec.map(|c| (c, field.name.to_string()))
+ })
+ .try_collect()?;
+
+ // consume all the iterators in lockstep, creating per-row predicates that get combined
+ // into a single final predicate
+
+ // (2025-06-12) can't use `is_empty` as it depends on unstable library feature `exact_size_is_empty`
+ #[allow(clippy::len_zero)]
+ while datum_columns_with_names[0].0.len() > 0 {
+ let mut row_predicate = AlwaysTrue;
+ for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names {
+ if let Some(item) = column.next() {
+ if let Some(datum) = item? {
+ row_predicate = row_predicate
+ .and(Reference::new(field_name.clone()).equal_to(datum.clone()));
+ }
+ }
+ }
+ result_predicate = result_predicate.and(row_predicate.not());
+ }
+ }
+ Ok(result_predicate.rewrite_not())
+ }
+}
+
+macro_rules! prim_to_datum {
Review Comment:
I would suggest to enhance existing `arrow_struct_to_literal` to return `Datum`, I think most codes could 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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2179413795
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
+ .map(|(column, field)| {
+ let col_as_datum_vec = arrow_array_to_datum_iterator(column, field);
+ col_as_datum_vec.map(|c| (c, field.name.to_string()))
+ })
+ .try_collect()?;
+
+ // consume all the iterators in lockstep, creating per-row predicates that get combined
+ // into a single final predicate
+
+ // (2025-06-12) can't use `is_empty` as it depends on unstable library feature `exact_size_is_empty`
+ #[allow(clippy::len_zero)]
+ while datum_columns_with_names[0].0.len() > 0 {
+ let mut row_predicate = AlwaysTrue;
+ for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names {
+ if let Some(item) = column.next() {
+ if let Some(datum) = item? {
+ row_predicate = row_predicate
+ .and(Reference::new(field_name.clone()).equal_to(datum.clone()));
+ }
+ }
+ }
+ result_predicate = result_predicate.and(row_predicate.not());
+ }
+ }
+ Ok(result_predicate.rewrite_not())
+ }
+}
+
+macro_rules! prim_to_datum {
Review Comment:
There is a `get_arrow_datum` in `arrow/schema.rs` [here](https://github.com/apache/iceberg-rust/blob/36cc12087bc118f1fe10efa6b30db98bd9655ba7/crates/iceberg/src/arrow/schema.rs#L652) which does the inverse, or at least close to it.
`arrow_struct_to_literal` in `arrow/value.rs` does something similar, especially in the `primitive` function of `ArrowArrayToIcebergStructConverter`'s visitor implementation [here](https://github.com/apache/iceberg-rust/blob/36cc12087bc118f1fe10efa6b30db98bd9655ba7/crates/iceberg/src/arrow/value.rs#L203). But, this returns Iceberg `Literal`s rather than `Datum`s. We need `Datum` because we're building an unbound predicate.
Nothing that already exists is an exact match for what we need to do here, as far as I can find.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2179239103
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
+ .map(|(column, field)| {
+ let col_as_datum_vec = arrow_array_to_datum_iterator(column, field);
+ col_as_datum_vec.map(|c| (c, field.name.to_string()))
+ })
+ .try_collect()?;
+
+ // consume all the iterators in lockstep, creating per-row predicates that get combined
+ // into a single final predicate
+
+ // (2025-06-12) can't use `is_empty` as it depends on unstable library feature `exact_size_is_empty`
+ #[allow(clippy::len_zero)]
+ while datum_columns_with_names[0].0.len() > 0 {
+ let mut row_predicate = AlwaysTrue;
+ for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names {
+ if let Some(item) = column.next() {
+ if let Some(datum) = item? {
+ row_predicate = row_predicate
+ .and(Reference::new(field_name.clone()).equal_to(datum.clone()));
+ }
+ }
+ }
+ result_predicate = result_predicate.and(row_predicate.not());
Review Comment:
I've just checked and both approaches give exactly the same resulting predicate from the unit test.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2180719880
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,233 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
- Err(Error::new(
- ErrorKind::FeatureUnsupported,
- "parsing of equality deletes is not yet supported",
- ))
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
Review Comment:
As far as I can tell, `arrow_struct_to_literal` is not used anywhere, but it is `pub`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2179403667
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
Review Comment:
I've fixed this, thanks!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2201925656
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,233 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
- Err(Error::new(
- ErrorKind::FeatureUnsupported,
- "parsing of equality deletes is not yet supported",
- ))
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
Review Comment:
Ready for re-review @liurenjie1024 - I switched to a visitor-based approach and extended the test to cover having a nested struct field as an equality delete column.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2179197908
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
+ .map(|(column, field)| {
+ let col_as_datum_vec = arrow_array_to_datum_iterator(column, field);
+ col_as_datum_vec.map(|c| (c, field.name.to_string()))
+ })
+ .try_collect()?;
+
+ // consume all the iterators in lockstep, creating per-row predicates that get combined
+ // into a single final predicate
+
+ // (2025-06-12) can't use `is_empty` as it depends on unstable library feature `exact_size_is_empty`
+ #[allow(clippy::len_zero)]
+ while datum_columns_with_names[0].0.len() > 0 {
+ let mut row_predicate = AlwaysTrue;
+ for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names {
+ if let Some(item) = column.next() {
+ if let Some(datum) = item? {
+ row_predicate = row_predicate
+ .and(Reference::new(field_name.clone()).equal_to(datum.clone()));
+ }
+ }
+ }
+ result_predicate = result_predicate.and(row_predicate.not());
Review Comment:
No, I'm trying to make it `!row_predicate1 and !row_predicate2`. Hence the last line being `rewrite_not` rather than `negative`. Whether you negate each predicate along the way and `and` them together, or `or` them together and negate them all at the end, the result is logically the same, no?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2179197908
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
+ .map(|(column, field)| {
+ let col_as_datum_vec = arrow_array_to_datum_iterator(column, field);
+ col_as_datum_vec.map(|c| (c, field.name.to_string()))
+ })
+ .try_collect()?;
+
+ // consume all the iterators in lockstep, creating per-row predicates that get combined
+ // into a single final predicate
+
+ // (2025-06-12) can't use `is_empty` as it depends on unstable library feature `exact_size_is_empty`
+ #[allow(clippy::len_zero)]
+ while datum_columns_with_names[0].0.len() > 0 {
+ let mut row_predicate = AlwaysTrue;
+ for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names {
+ if let Some(item) = column.next() {
+ if let Some(datum) = item? {
+ row_predicate = row_predicate
+ .and(Reference::new(field_name.clone()).equal_to(datum.clone()));
+ }
+ }
+ }
+ result_predicate = result_predicate.and(row_predicate.not());
Review Comment:
No, I'm trying to make it `!row_predicate1 and !row_predicate2`. Hence the last line being `rewrite_not` rather than `negative`. Whether you negate each predicate along the way and `and` them together, or `or` them together and negate the entire combined predicate at the end, the result is logically the same, no?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2179211848
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
+ .map(|(column, field)| {
+ let col_as_datum_vec = arrow_array_to_datum_iterator(column, field);
+ col_as_datum_vec.map(|c| (c, field.name.to_string()))
+ })
+ .try_collect()?;
+
+ // consume all the iterators in lockstep, creating per-row predicates that get combined
+ // into a single final predicate
+
+ // (2025-06-12) can't use `is_empty` as it depends on unstable library feature `exact_size_is_empty`
+ #[allow(clippy::len_zero)]
+ while datum_columns_with_names[0].0.len() > 0 {
+ let mut row_predicate = AlwaysTrue;
+ for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names {
+ if let Some(item) = column.next() {
+ if let Some(datum) = item? {
+ row_predicate = row_predicate
+ .and(Reference::new(field_name.clone()).equal_to(datum.clone()));
+ }
+ }
+ }
+ result_predicate = result_predicate.and(row_predicate.not());
+ }
+ }
+ Ok(result_predicate.rewrite_not())
+ }
+}
+
+macro_rules! prim_to_datum {
Review Comment:
I looked at the time I wrote this and could not find anything suitable but I'll take a look again in case I missed something.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2179246166
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
+ .map(|(column, field)| {
+ let col_as_datum_vec = arrow_array_to_datum_iterator(column, field);
+ col_as_datum_vec.map(|c| (c, field.name.to_string()))
+ })
+ .try_collect()?;
+
+ // consume all the iterators in lockstep, creating per-row predicates that get combined
+ // into a single final predicate
+
+ // (2025-06-12) can't use `is_empty` as it depends on unstable library feature `exact_size_is_empty`
+ #[allow(clippy::len_zero)]
+ while datum_columns_with_names[0].0.len() > 0 {
+ let mut row_predicate = AlwaysTrue;
+ for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names {
+ if let Some(item) = column.next() {
+ if let Some(datum) = item? {
+ row_predicate = row_predicate
+ .and(Reference::new(field_name.clone()).equal_to(datum.clone()));
+ }
+ }
+ }
+ result_predicate = result_predicate.and(row_predicate.not());
Review Comment:
It seemed to me to be more understandable to write the predicate as "not matching row 1, and not matching row 2, ..." rather than "not(matching row 1, or matching row 2...)" but they're essentially saying the same thing and result in the exact same query. I'm happy to change it around if the consensus is that one approach is easier to understand than the other.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2179192924
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
Review Comment:
Got it, will fix.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2177176342
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
+ .map(|(column, field)| {
+ let col_as_datum_vec = arrow_array_to_datum_iterator(column, field);
+ col_as_datum_vec.map(|c| (c, field.name.to_string()))
+ })
+ .try_collect()?;
+
+ // consume all the iterators in lockstep, creating per-row predicates that get combined
+ // into a single final predicate
+
+ // (2025-06-12) can't use `is_empty` as it depends on unstable library feature `exact_size_is_empty`
+ #[allow(clippy::len_zero)]
+ while datum_columns_with_names[0].0.len() > 0 {
+ let mut row_predicate = AlwaysTrue;
+ for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names {
+ if let Some(item) = column.next() {
+ if let Some(datum) = item? {
+ row_predicate = row_predicate
+ .and(Reference::new(field_name.clone()).equal_to(datum.clone()));
+ }
+ }
+ }
+ result_predicate = result_predicate.and(row_predicate.not());
Review Comment:
Are you trying to make this as `not(!row_predicate1 and !row_predicate2)`? This is a little odd to read, why not just `row_predicate1 or row_predicate2`? Also the last line should be `negative` rather `rewrite_not`
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
Review Comment:
This is incorrect, see https://iceberg.apache.org/spec/#equality-delete-files:~:text=Each%20row%20of%20the%20delete%20file%20produces%20one%20equality%20predicate%20that%20matches%20any%20row%20where%20the%20delete%20columns%20are%20equal.%20Multiple%20columns%20can%20be%20thought%20of%20as%20an%20AND%20of%20equality%20predicates.%20A%20null%20value%20in%20a%20delete%20column%20matches%20a%20row%20if%20the%20row%27s%20value%20is%20null%2C%20equivalent%20to%20col%20IS%20NULL. If a column is missing, we should treat it as null, and it still helps to prune null value.
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
+ .map(|(column, field)| {
+ let col_as_datum_vec = arrow_array_to_datum_iterator(column, field);
+ col_as_datum_vec.map(|c| (c, field.name.to_string()))
+ })
+ .try_collect()?;
+
+ // consume all the iterators in lockstep, creating per-row predicates that get combined
+ // into a single final predicate
+
+ // (2025-06-12) can't use `is_empty` as it depends on unstable library feature `exact_size_is_empty`
+ #[allow(clippy::len_zero)]
+ while datum_columns_with_names[0].0.len() > 0 {
+ let mut row_predicate = AlwaysTrue;
+ for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names {
+ if let Some(item) = column.next() {
+ if let Some(datum) = item? {
+ row_predicate = row_predicate
+ .and(Reference::new(field_name.clone()).equal_to(datum.clone()));
+ }
+ }
+ }
+ result_predicate = result_predicate.and(row_predicate.not());
+ }
+ }
+ Ok(result_predicate.rewrite_not())
+ }
+}
+
+macro_rules! prim_to_datum {
Review Comment:
I think there exists similar things in other parts? We should reuse them
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2179246166
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
+ .map(|(column, field)| {
+ let col_as_datum_vec = arrow_array_to_datum_iterator(column, field);
+ col_as_datum_vec.map(|c| (c, field.name.to_string()))
+ })
+ .try_collect()?;
+
+ // consume all the iterators in lockstep, creating per-row predicates that get combined
+ // into a single final predicate
+
+ // (2025-06-12) can't use `is_empty` as it depends on unstable library feature `exact_size_is_empty`
+ #[allow(clippy::len_zero)]
+ while datum_columns_with_names[0].0.len() > 0 {
+ let mut row_predicate = AlwaysTrue;
+ for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names {
+ if let Some(item) = column.next() {
+ if let Some(datum) = item? {
+ row_predicate = row_predicate
+ .and(Reference::new(field_name.clone()).equal_to(datum.clone()));
+ }
+ }
+ }
+ result_predicate = result_predicate.and(row_predicate.not());
Review Comment:
It seemed to me to be more understandable to write the predicate as "not matching row 1, and not matching row 2, ..." rather than "not(matching row 1, or matching row 2...)" but they're essentially saying the same thing and result in the exact same predicate. I'm happy to change it around if the consensus is that one approach is easier to understand than the other.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 merged PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#issuecomment-3232863247
Sorry for delay @sdd Would you mind rebase it and fix ci?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#issuecomment-2965671829
@liurenjie1024 @Xuanwo @Fokko this is ready for review now that 982 has been merged! 😁
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2180205465
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
+ .map(|(column, field)| {
+ let col_as_datum_vec = arrow_array_to_datum_iterator(column, field);
+ col_as_datum_vec.map(|c| (c, field.name.to_string()))
+ })
+ .try_collect()?;
+
+ // consume all the iterators in lockstep, creating per-row predicates that get combined
+ // into a single final predicate
+
+ // (2025-06-12) can't use `is_empty` as it depends on unstable library feature `exact_size_is_empty`
+ #[allow(clippy::len_zero)]
+ while datum_columns_with_names[0].0.len() > 0 {
+ let mut row_predicate = AlwaysTrue;
+ for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names {
+ if let Some(item) = column.next() {
+ if let Some(datum) = item? {
+ row_predicate = row_predicate
+ .and(Reference::new(field_name.clone()).equal_to(datum.clone()));
+ }
+ }
+ }
+ result_predicate = result_predicate.and(row_predicate.not());
+ }
+ }
+ Ok(result_predicate.rewrite_not())
+ }
+}
+
+macro_rules! prim_to_datum {
Review Comment:
OK, I'm happy to do that, will make a start on it later
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#discussion_r2180856664
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -308,28 +319,231 @@ impl CachingDeleteFileLoader {
Ok(result)
}
- /// Parses record batch streams from individual equality delete files
- ///
- /// Returns an unbound Predicate for each batch stream
async fn parse_equality_deletes_record_batch_stream(
- streams: ArrowRecordBatchStream,
+ mut stream: ArrowRecordBatchStream,
+ equality_ids: HashSet<i32>,
) -> Result<Predicate> {
- // TODO
+ let mut result_predicate = AlwaysTrue;
+
+ while let Some(record_batch) = stream.next().await {
+ let record_batch = record_batch?;
+
+ if record_batch.num_columns() == 0 {
+ return Ok(AlwaysTrue);
+ }
+
+ let batch_schema_arrow = record_batch.schema();
+ let batch_schema_iceberg = arrow_schema_to_schema(batch_schema_arrow.as_ref())?;
+
+ let mut datum_columns_with_names: Vec<_> = record_batch
+ .columns()
+ .iter()
+ .zip(batch_schema_iceberg.as_struct().fields())
+ // only use columns that are in the set of equality_ids for this delete file
+ .filter(|(field, value)| equality_ids.contains(&value.id))
+ .map(|(column, field)| {
+ let col_as_datum_vec = arrow_array_to_datum_iterator(column, field);
+ col_as_datum_vec.map(|c| (c, field.name.to_string()))
+ })
+ .try_collect()?;
+
+ // consume all the iterators in lockstep, creating per-row predicates that get combined
+ // into a single final predicate
+
+ // (2025-06-12) can't use `is_empty` as it depends on unstable library feature `exact_size_is_empty`
+ #[allow(clippy::len_zero)]
+ while datum_columns_with_names[0].0.len() > 0 {
+ let mut row_predicate = AlwaysTrue;
+ for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names {
+ if let Some(item) = column.next() {
+ if let Some(datum) = item? {
+ row_predicate = row_predicate
+ .and(Reference::new(field_name.clone()).equal_to(datum.clone()));
+ }
+ }
+ }
+ result_predicate = result_predicate.and(row_predicate.not());
+ }
+ }
+ Ok(result_predicate.rewrite_not())
+ }
+}
+
+macro_rules! prim_to_datum {
Review Comment:
It's not really possible to refactor `arrow_struct_to_literal` to return `Datum` since `Datum` is only for `PrimitiveLiteral` rather than the more general `Literal`, and `arrow_struct_to_literal` handles composite types as well as primitives.
Instead, I've managed to re-use the same visitor that `arrow_struct_to_literal` uses, but wrapped in a small `arrow_primitive_to_literal` function instead, allowing me to remove the similar code from caching_delete_file_loader.rs, at the expense of some slightly uglier glue code in `parse_equality_deletes_record_batch_stream`.
Does this address your concern?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Scan Delete Support Part 6: Equality Delete Parsing [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on PR #1017:
URL: https://github.com/apache/iceberg-rust/pull/1017#issuecomment-3241202161
> Sorry for delay @sdd Would you mind rebase it and fix ci?
That's ready again for you now @liurenjie1024 👍🏼
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org