anoopj commented on code in PR #2985:
URL: https://github.com/apache/iceberg-rust/pull/2985#discussion_r3763819571
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -301,59 +349,38 @@ impl FileScanTaskReader {
.with_constant(RESERVED_FIELD_ID_SPEC_ID, spec_id_datum);
}
- if task
- .project_field_ids()
- .contains(&RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER)
- {
- // A data file may physically carry a per-row
`_last_updated_sequence_number`
- // column, e.g. one written by another engine such as Iceberg Java
when carrying
- // rows forward across a rewrite. The spec requires reading such
non-null
- // per-row values unmodified, falling back to the derived value
only where
- // null. That per-row coalesce is not implemented yet, so rather
than silently
- // overwrite genuine per-row values with the derived value, reject
the file
- // loudly. Checks the full pre-projection file schema, since the
column is
- // stripped from the projection mask. Matches on the embedded
field id when
- // present (propagating a malformed id rather than treating it as
absent) and
- // falls back to the column name, so a file read via name mapping
or positional
- // fallback ids (which never equal the reserved id) is still
caught.
- let mut file_has_column = false;
- for field in record_batch_stream_builder.schema().fields() {
- let field_id = match
field.metadata().get(PARQUET_FIELD_ID_META_KEY) {
- Some(id) => Some(id.parse::<i32>().map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("field id not parseable as an i32: {e}"),
- )
- })?),
- None => None,
- };
- if field_id ==
Some(RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER)
- || field.name() ==
RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER
- {
- file_has_column = true;
- break;
- }
- }
- if file_has_column {
- return Err(Error::new(
- ErrorKind::FeatureUnsupported,
- "Reading a physically-stored _last_updated_sequence_number
column is \
- not yet supported; only the derived
(data-sequence-number) value is \
- implemented",
- ));
- }
-
- // Derive the column, gated on the data file's `first_row_id`.
Java gates
- // both lineage columns this way (`ValueReaders.lastUpdated`
returns nulls
- // when the base row id is null); the spec itself only says the
column is
- // assigned the manifest entry's sequence number on read.
+ if project_last_updated_seq {
+ // Materialize the column, gated on the data file's
`first_row_id`. Java gates
+ // it this way (`ValueReaders.lastUpdated` returns nulls when the
base row id is
+ // null); the spec itself only says the column is assigned the
manifest entry's
+ // sequence number on read.
record_batch_transformer_builder = match (task.first_row_id,
task.data_sequence_number)
{
- // Non-null first_row_id: inherit from the data sequence
number.
- (Some(_), Some(seq)) =>
record_batch_transformer_builder.with_constant(
- RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER,
- Datum::long(seq),
- ),
+ (Some(_), Some(seq)) => {
+ let datum = Datum::long(seq);
+ if coalesce_last_updated_seq_leaf.is_some() {
+ // The file physically carries the column: read the
per-row value,
+ // falling back to the data sequence number only where
null.
+
record_batch_transformer_builder.with_coalesced_metadata_column(
+ RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER,
+ datum,
+ )
+ } else if last_updated_seq_present_by_name_only {
Review Comment:
Added the (Some, None) + physical-column test which asserts `DataInvalid`.
We should not hoist because the transformer keys the source column by field id
and can't thread an id-less column. But that only matters when we actually read
the column, which only the `(Some, Some)` arm does. In (None, _) we synthesize
an all-null column without touching the physical data, so an ordinary
v2/null-lineage file that happens to carry a same-named, id-less column is
correctly nulled today.
Hoisting the reject would turn that correct null into a hard
FeatureUnsupported. So the reject is arm-local by design; I added a comment
stating that. If you still want it hoisted I can, but I think arm-local seems
correct to me. wdyt?
--
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]
