anoopj commented on code in PR #2966:
URL: https://github.com/apache/iceberg-rust/pull/2966#discussion_r3744921482
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -909,6 +959,257 @@ mod tests {
);
}
+ /// Writes a plain (unencrypted) single-column Int32 "id" parquet file
with the
+ /// given extra Arrow fields/columns appended, returning the file path.
+ fn write_plain_parquet(
+ dir: &str,
+ name: &str,
+ extra_fields: Vec<Field>,
+ extra_columns: Vec<ArrayRef>,
+ ) -> String {
+ let mut fields =
+ vec![
+ Field::new("id", DataType::Int32,
false).with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ "1".to_string(),
+ )])),
+ ];
+ fields.extend(extra_fields);
+ let arrow_schema = Arc::new(ArrowSchema::new(fields));
+
+ let mut columns: Vec<ArrayRef> =
vec![Arc::new(Int32Array::from(vec![1, 2, 3]))];
+ columns.extend(extra_columns);
+ let batch = RecordBatch::try_new(arrow_schema.clone(),
columns).unwrap();
+
+ let file_path = format!("{dir}/{name}");
+ let file = File::create(&file_path).unwrap();
+ let props = WriterProperties::builder()
+ .set_compression(Compression::SNAPPY)
+ .build();
+ let mut writer = ArrowWriter::try_new(file, arrow_schema,
Some(props)).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ file_path
+ }
+
+ fn last_updated_seq_task(
+ file_path: String,
+ first_row_id: Option<i64>,
+ data_sequence_number: Option<i64>,
+ ) -> FileScanTask {
+ use
crate::metadata_columns::RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER;
+
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+
+ FileScanTask::builder()
+
.with_file_size_in_bytes(std::fs::metadata(&file_path).unwrap().len())
+ .with_start(0)
+ .with_length(0)
+ .with_data_file_path(file_path)
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_schema(schema)
+ .with_project_field_ids(vec![1,
RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER])
+ .with_first_row_id(first_row_id)
+ .with_data_sequence_number(data_sequence_number)
+ .with_case_sensitive(false)
+ .build()
+ }
+
+ #[tokio::test]
+ async fn test_last_updated_sequence_number_null_when_no_first_row_id() {
+ let tmp_dir = TempDir::new().unwrap();
+ let dir = tmp_dir.path().to_str().unwrap();
+ let file_path = write_plain_parquet(dir, "no_first_row_id.parquet",
vec![], vec![]);
+
+ // A file with a null first_row_id (v1/v2, or a pre-upgrade v3
snapshot) produces
+ // a null _last_updated_sequence_number column, even though it has a
data
+ // sequence number; the spec gates both lineage columns on
first_row_id.
+ let task = last_updated_seq_task(file_path, None, 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();
+
+ use
crate::metadata_columns::RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER;
+ let seq_col = batches[0]
+ .column_by_name(RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER)
+ .expect("column should be present")
+ .as_any()
+
.downcast_ref::<arrow_array::RunArray<arrow_array::types::Int32Type>>()
+ .expect("_last_updated_sequence_number should be a RunArray");
+ assert_eq!(seq_col.len(), 3);
+ let values = seq_col
+ .values()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .expect("REE values should be Int64Array");
+ assert_eq!(values.len(), 1);
+ assert!(values.is_null(0));
+ }
+
+ #[tokio::test]
+ async fn test_last_updated_sequence_number_null_when_no_data_seq() {
+ use
crate::metadata_columns::RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER;
+
+ let tmp_dir = TempDir::new().unwrap();
+ let dir = tmp_dir.path().to_str().unwrap();
+ let file_path = write_plain_parquet(dir, "no_data_seq.parquet",
vec![], vec![]);
+
+ // first_row_id present but data_sequence_number absent (a malformed
manifest)
+ // also yields a null column, matching Java's dual gate. Pins the
(Some, None)
+ // arm so a future match collapse cannot silently unwrap it.
+ let task = last_updated_seq_task(file_path, Some(42), None);
+
+ 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();
+
+ let seq_col = batches[0]
+ .column_by_name(RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER)
+ .expect("column should be present")
+ .as_any()
+
.downcast_ref::<arrow_array::RunArray<arrow_array::types::Int32Type>>()
+ .expect("_last_updated_sequence_number should be a RunArray");
+ let values = seq_col
+ .values()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .expect("REE values should be Int64Array");
+ assert!(values.is_null(0));
+ }
+
+ #[tokio::test]
+ async fn test_last_updated_sequence_number_derived_from_data_seq() {
+ use
crate::metadata_columns::RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER;
+
+ let tmp_dir = TempDir::new().unwrap();
+ let dir = tmp_dir.path().to_str().unwrap();
+ let file_path = write_plain_parquet(dir, "with_first_row_id.parquet",
vec![], vec![]);
+
+ // Non-null first_row_id + data sequence number -> the derived value
(the data
+ // sequence number) for every row. This is the only value-producing
arm.
+ let task = last_updated_seq_task(file_path, Some(42), Some(7));
+
+ 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();
+
+ let seq_col = batches[0]
+ .column_by_name(RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER)
+ .expect("column should be present")
+ .as_any()
+
.downcast_ref::<arrow_array::RunArray<arrow_array::types::Int32Type>>()
+ .expect("_last_updated_sequence_number should be a RunArray");
+ assert_eq!(seq_col.len(), 3);
+ let values = seq_col
+ .values()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .expect("REE values should be Int64Array");
+ assert_eq!(values.len(), 1);
Review Comment:
This is a good pint. All four now materialize the column and assert each
row's logical value instead of relying on the run shape.
--
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]
