anoopj commented on code in PR #2985:
URL: https://github.com/apache/iceberg-rust/pull/2985#discussion_r3763822016
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -1171,32 +1213,175 @@ mod tests {
.await
.unwrap();
- assert_eq!(batches.len(), 2);
- // Identical schema across files -> concat succeeds.
+ assert_eq!(batches.len(), 3);
+ // Identical schema across all three paths -> concat succeeds.
let schema = batches[0].schema();
concat_batches(&schema, &batches)
- .expect("batches from value and null files must share one schema");
+ .expect("constant, null and coalesce files must share one column
type");
+ }
+
+ /// A parquet field carrying the embedded `_last_updated_sequence_number`
field id.
+ fn physical_last_updated_seq_field() -> Field {
+ use crate::metadata_columns::{
+ RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER,
+ RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER,
+ };
+ Field::new(
+ RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER,
+ DataType::Int64,
+ true,
+ )
+ .with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER.to_string(),
+ )]))
}
#[tokio::test]
- async fn test_last_updated_sequence_number_physical_column_unsupported() {
- use
crate::metadata_columns::RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER;
+ async fn test_last_updated_sequence_number_physical_column_coalesced() {
+ let tmp_dir = TempDir::new().unwrap();
+ let dir = tmp_dir.path().to_str().unwrap();
+ // A file that physically carries the column, as Iceberg Java writes
when
+ // carrying rows forward across a rewrite: some rows have a stored
value, some
+ // are null (added/modified rows, inherited on read).
+ let seq_col = Arc::new(Int64Array::from(vec![Some(5), None, Some(8)]))
as ArrayRef;
+ let file_path = write_plain_parquet(
+ dir,
+ "with_seq.parquet",
+ vec![physical_last_updated_seq_field()],
+ vec![seq_col],
+ );
+
+ let task = last_updated_seq_task(file_path, Some(100), Some(9));
+
+ let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(),
Runtime::current()).build();
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as
FileScanTaskStream;
+ let batches: Vec<RecordBatch> = reader
+ .read(tasks)
+ .unwrap()
+ .stream()
+ .try_collect()
+ .await
+ .unwrap();
+
+ // Per-row value where non-null; the data sequence number (9) where
null.
+ assert_last_updated_seq_column(&batches, &[Some(5), Some(9), Some(8)]);
+ }
+
+ #[tokio::test]
+ async fn test_last_updated_sequence_number_coalesced_with_pos_column() {
+ use crate::metadata_columns::{
+ RESERVED_COL_NAME_POS,
RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER,
+ RESERVED_FIELD_ID_POS,
+ };
let tmp_dir = TempDir::new().unwrap();
let dir = tmp_dir.path().to_str().unwrap();
- // A file that physically carries the _last_updated_sequence_number
column, as
- // Iceberg Java writes when carrying rows forward across a rewrite.
- let seq_field = Field::new("_last_updated_sequence_number",
DataType::Int64, true)
- .with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER.to_string(),
- )]));
+ let seq_col = Arc::new(Int64Array::from(vec![Some(5), None, Some(8)]))
as ArrayRef;
+ let file_path = write_plain_parquet(
+ dir,
+ "with_seq_and_pos.parquet",
+ vec![physical_last_updated_seq_field()],
+ vec![seq_col],
+ );
+
+ // Co-project `_pos` (a virtual column appended to the Arrow output
schema) with the
+ // physical coalesce column. This guards that the physical column's
index is
+ // resolved in the Parquet schema, not the Arrow schema (whose indices
shift once
+ // virtual columns are appended).
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
Review Comment:
Double-checked. `write_plain_parquet` always writes id=[1,2,3] as its first
column, so the required id field does resolve against the file. The test is
correctly passing.
--
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]
