ORC: bind struct fields by Iceberg field id in generic and Spark 3.1 readers - #256
ORC: bind struct fields by Iceberg field id in generic and Spark 3.1 readers#256cbb330 wants to merge 1 commit into
Conversation
cbb330
left a comment
cbb330
left a comment
There was a problem hiding this comment.
Provenance annotations for each code block, categorized as: backported exactly / backported with adaptations / net new / inspiration from shipped code. Reference: apache/iceberg 77e7dbb24 (PR #15776).
| // Maps each expected field position to the ORC column index it should read from. Only used by | ||
| // the id-based binding constructor; left null by the deprecated positional constructor so that | ||
| // {@link #readInternal} preserves the exact positional behavior for readers that have not been | ||
| // converted to id-based binding. | ||
| private final int[] orcFieldIndex; |
There was a problem hiding this comment.
Backported with adaptations. The orcFieldIndex field is from upstream's StructReader. The doc-comment explaining the null sentinel is reworded for this fork to make explicit that positional binding is preserved for unconverted engine readers.
| /** | ||
| * @deprecated since 1.2.1, positional binding assumes the projected field order matches the | ||
| * file's physical column order. Use {@link #StructReader(TypeDescription, List, | ||
| * Types.StructType, Map)} for id-based binding instead. Kept for engine readers that have | ||
| * not yet migrated. | ||
| */ | ||
| @Deprecated | ||
| protected StructReader( | ||
| List<OrcValueReader<?>> readers, Types.StructType struct, Map<Integer, ?> idToConstant) { | ||
| List<Types.NestedField> fields = struct.fields(); | ||
| this.readers = new OrcValueReader[fields.size()]; | ||
| this.isConstantOrMetadataField = new boolean[fields.size()]; | ||
| this.orcFieldIndex = null; |
There was a problem hiding this comment.
Backported with adaptations. Upstream keeps the positional constructor @Deprecated too; we mirror that (other engines still call it). Adaptation: this.orcFieldIndex = null; is the sentinel that makes readInternal fall back to exact positional behavior.
| protected StructReader( | ||
| TypeDescription orcType, | ||
| List<OrcValueReader<?>> readers, | ||
| Types.StructType struct, | ||
| Map<Integer, ?> idToConstant) { | ||
| List<Types.NestedField> fields = struct.fields(); | ||
| this.readers = new OrcValueReader[fields.size()]; | ||
| this.isConstantOrMetadataField = new boolean[fields.size()]; | ||
| this.orcFieldIndex = new int[fields.size()]; | ||
|
|
||
| Map<Integer, OrcValueReader<?>> readersById = readersByFieldId(orcType, readers); | ||
| Map<Integer, Integer> fieldIdToOrcIndex = buildFieldIdToOrcIndex(orcType); | ||
|
|
||
| for (int pos = 0; pos < fields.size(); pos += 1) { | ||
| Types.NestedField field = fields.get(pos); | ||
| OrcValueReader<?> fileReader = readersById.get(field.fieldId()); | ||
|
|
||
| if (idToConstant.containsKey(field.fieldId())) { | ||
| this.isConstantOrMetadataField[pos] = true; | ||
| this.readers[pos] = constants(idToConstant.get(field.fieldId())); | ||
| } else if (field.equals(MetadataColumns.ROW_POSITION)) { | ||
| this.isConstantOrMetadataField[pos] = true; | ||
| this.readers[pos] = new RowPositionReader(); | ||
| } else if (field.equals(MetadataColumns.IS_DELETED)) { | ||
| this.isConstantOrMetadataField[pos] = true; | ||
| this.readers[pos] = constants(false); | ||
| } else if (fileReader != null) { | ||
| this.isConstantOrMetadataField[pos] = false; | ||
| this.orcFieldIndex[pos] = fieldIdToOrcIndex.getOrDefault(field.fieldId(), -1); | ||
| this.readers[pos] = fileReader; | ||
| } else if (MetadataColumns.isMetadataColumn(field.name())) { | ||
| // in case of any other metadata field, fill with nulls | ||
| this.isConstantOrMetadataField[pos] = true; | ||
| this.readers[pos] = constants(null); | ||
| } else { | ||
| throw new IllegalArgumentException( | ||
| String.format("Missing ORC reader for field %s (%s)", field.name(), field.fieldId())); | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
Backported exactly (minus deliberately omitted branches). This id-binding constructor matches the upstream StructReader line-for-line, except the row-lineage branches (ROW_ID / LAST_UPDATED_SEQUENCE_NUMBER + their handle* helpers) and the || field.type().typeId() == Type.TypeID.UNKNOWN clause are dropped — neither construct exists on 1.2.x.
| private Map<Integer, Integer> buildFieldIdToOrcIndex(TypeDescription orcType) { | ||
| List<TypeDescription> children = orcType.getChildren(); | ||
| Map<Integer, Integer> mapping = Maps.newHashMap(); | ||
| for (int i = 0; i < children.size(); i++) { | ||
| mapping.put(ORCSchemaUtil.fieldId(children.get(i)), i); | ||
| } | ||
| return mapping; | ||
| } | ||
|
|
||
| private Map<Integer, OrcValueReader<?>> readersByFieldId( | ||
| TypeDescription orcType, List<OrcValueReader<?>> readerList) { | ||
| List<TypeDescription> children = orcType.getChildren(); | ||
| Preconditions.checkState( | ||
| children.size() == readerList.size(), | ||
| "Invalid ORC reader binding: children=%s readers=%s", | ||
| children.size(), | ||
| readerList.size()); | ||
| Map<Integer, OrcValueReader<?>> readersById = Maps.newHashMap(); | ||
| for (int i = 0; i < children.size(); i += 1) { | ||
| readersById.put(ORCSchemaUtil.fieldId(children.get(i)), readerList.get(i)); | ||
| } | ||
| return readersById; | ||
| } |
There was a problem hiding this comment.
Backported exactly. buildFieldIdToOrcIndex and readersByFieldId are verbatim from upstream (including the Preconditions.checkState binding invariant).
| private T readInternal(T struct, ColumnVector[] columnVectors, int row) { | ||
| for (int c = 0, vectorIndex = 0; c < readers.length; ++c) { | ||
| int vectorIndex = 0; | ||
| for (int c = 0; c < readers.length; ++c) { | ||
| ColumnVector vector; | ||
| if (isConstantOrMetadataField[c]) { | ||
| vector = null; | ||
| } else if (orcFieldIndex != null) { | ||
| vector = columnVectors[orcFieldIndex[c]]; | ||
| } else { | ||
| vector = columnVectors[vectorIndex]; | ||
| vectorIndex++; | ||
| vector = columnVectors[vectorIndex++]; |
There was a problem hiding this comment.
Backported exactly. readInternal's three-way index selection (constant/metadata → null, id-binding → orcFieldIndex[c], positional → vectorIndex++) is identical to upstream.
| List<String> names, | ||
| List<OrcValueReader<?>> fields) { | ||
| return GenericOrcReaders.struct(fields, expected, idToConstant); | ||
| return GenericOrcReaders.struct(record, fields, expected, idToConstant); |
There was a problem hiding this comment.
Backported exactly. Visitor call site now passes the record TypeDescription — identical one-line change to upstream (record was already provided by the visitor).
| static OrcValueReader<?> struct( | ||
| TypeDescription orcType, | ||
| List<OrcValueReader<?>> readers, | ||
| Types.StructType struct, | ||
| Map<Integer, ?> idToConstant) { | ||
| return new StructReader(orcType, readers, struct, idToConstant); | ||
| } |
There was a problem hiding this comment.
Backported with adaptations. Mirrors upstream's 4-arg struct. Adaptation: upstream (Spark v3.5+) deletes the positional overload; here it is retained just below as @Deprecated because Spark 2.4/3.2/3.3 and Flink still use it on 1.2.x.
| protected StructReader( | ||
| TypeDescription orcType, | ||
| List<OrcValueReader<?>> readers, | ||
| Types.StructType struct, | ||
| Map<Integer, ?> idToConstant) { | ||
| super(orcType, readers, struct, idToConstant); | ||
| this.numFields = struct.fields().size(); | ||
| } |
There was a problem hiding this comment.
Backported with adaptations. 4-arg subclass constructor mirroring upstream; the positional subclass constructor is retained @Deprecated (see below).
| List<String> names, | ||
| List<OrcValueReader<?>> fields) { | ||
| return SparkOrcValueReaders.struct(fields, expected, idToConstant); | ||
| return SparkOrcValueReaders.struct(record, fields, expected, idToConstant); |
There was a problem hiding this comment.
Backported exactly. Visitor call site passes record — identical to upstream, applied to the 3.1 file.
| * exercise branches of the id-binding {@code StructReader} constructor that the shared projection | ||
| * tests do not, and assert that id-binding produces the same results positional binding would. | ||
| */ | ||
| public class TestGenericOrcReaderIdBinding { |
There was a problem hiding this comment.
Net new, using inspiration from shipped tests. Not part of the upstream commit. The metadata-column approach (project without metadata fields, hand the reader the full schema incl. _pos/_deleted) is modeled on the shipped spark/v3.1/.../TestSparkOrcReadMetadataColumns.java; the write/project/read harness follows TestGenericReadProjection / TestGenericData. Covers type promotion, reordered projection, metadata columns, and idToConstant.
Comparison against
|
Noise-free compare vs
|
The work itself (additions), as a clean compareThe cbb330/iceberg-apache@orc-id-binding-baseline...orc-id-binding-work It is two throwaway branches on a fork of
So the three-dot diff is precisely the backport delta, as additions (identical to this PR's own Files-changed view, just isolated to the 6 files and hosted where it can sit next to
(The equivalent view with zero extra branches is simply this PR's Files changed tab.) All |
|
Updated for upstream fidelity in
|
227648d to
94fd480
Compare
Divergence vs
|
Re: red CI on this PR — pre-existing infra, fixed by #258The failing checks here are not caused by this change. Every test job auto-fails in ~2s at the "Prepare all required actions" step because the Evidence it's unrelated to this PR:
Fix: #258 bumps |
…readers Backport the upstream id-based StructReader binding (apache/iceberg 77e7dbb, PR #15776) to the li-1.2.x ORC reader, adapted for the 1.2.x type system. The ORC StructReader previously bound struct fields positionally, consuming column vectors in projection order. This replaces that with id-based binding: each expected field is matched to its ORC column by Iceberg field id, which is resilient to field reordering and schema evolution. Scope: only GenericOrcReaders and Spark 3.1 SparkOrcValueReaders are converted. The positional constructor/overloads are retained and marked @deprecated because the other engine readers (Spark 2.4/3.2/3.3, Flink) still use them. Row-lineage, UNKNOWN, and VARIANT constructs from the upstream commit are omitted as they do not exist on 1.2.x. Behavior-preserving for well-formed reads: id-based binding produces identical results to positional binding on all current inputs. Co-authored-by: Cursor <cursoragent@cursor.com>
94fd480 to
592778c
Compare
|
Closing: this was based on the wrong branch ( Rebuilt as a clean 2-PR stack on
|
1 participant
Summary
Backports the upstream id-based ORC
StructReaderbinding (apache/iceberg77e7dbb24, PR #15776) to the li-1.2.x line, adapted for the 1.2.x type system.What changes: the ORC reader used to bind struct fields positionally — it consumed ORC column vectors in projection order, assuming the projected field order matches the file's physical column order. This PR switches the converted readers to id-based binding: each expected field is matched to its ORC column by its Iceberg field id, which is robust to field reordering and schema evolution.
Why it's safe: for the files this fork reads today,
buildOrcProjectionalready normalizes column order, so id-based binding produces identical results to positional binding on all current inputs. It is a behavior-preserving robustness change — and the prerequisite for the follow-up work (see Next steps).How id-based binding works (quick primer)
An ORC file stores columns in a physical order. When Iceberg reads it, it has an expected schema (the fields it wants), and every Iceberg field has a permanent numeric field id. Two ways to match expected fields to file columns:
The real logic lives in the shared
OrcValueReaders.StructReaderbase class (iniceberg-orc); the per-engine reader edits are thin plumbing that thread the ORC file schema (TypeDescription) down to it and opt in.Provenance of each change
OrcValueReaders: id-binding constructor +readersByFieldId/buildFieldIdToOrcIndex+readInternalselectionGenericOrcReaders4-argstruct+ subclass ctor;GenericOrcReaderrecordthreadingSparkOrcValueReaders/SparkOrcReader(v3.1)@Deprecatedpositional overload that upstream deletedTestGenericOrcReaderIdBindingTestSparkOrcReadMetadataColumns/TestGenericReadProjectionPer-line classification is in the inline review annotations on this PR.
Adaptations vs
apache/main(why it diverges at all)1.2.x lacks constructs the upstream commit references, so these are intentionally omitted (porting them verbatim would not compile on 1.2.x):
ROW_ID/LAST_UPDATED_SEQUENCE_NUMBERbranches, thehandleRowIdField/handleLastUpdatedSeqFieldhelpers, and theRowIdReader/LastUpdatedSeqReaderclasses.UNKNOWNtype — the|| field.type().typeId() == Type.TypeID.UNKNOWNclause in the metadata fallback.VARIANT— the variant reader/visitor.Everything else is aligned to upstream verbatim (method ordering, Javadoc, whitespace, no defensive null-guards).
Scope
Only
GenericOrcReadersand Spark 3.1SparkOrcValueReadersare converted. The positional constructor/overloads are retained and@Deprecatedbecause the unconverted readers (Spark 2.4/3.2/3.3, Flink 1.14/1.15/1.16) still call them — those readers keep identical positional behavior.Reviewing the divergence from upstream
77e7dbb24apache/main: apache/iceberg@main...cbb330:iceberg-apache:compare/orc-id-binding-vs-apache-main (if GitHub says "taking too long to generate," hit Retry — a large-repo cross-fork rendering quirk)apache/main" comment on this PR.git fetch apache main && git diff apache/main -- orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.javaCompanion PR
The identical change is backported to the 1.5.x line (targeting Spark 3.5) in #257. The ORC core is byte-identical between the two backports — proof (renders "the branches are identical"): cbb330/iceberg-apache@compare/orc-id-binding-shared-core...compare/orc-id-binding-shared-core-1.5.x
Testing Done
Validated on JDK 11:
./gradlew :iceberg-orc:test :iceberg-data:test— pass. NewTestGenericOrcReaderIdBindingcovers type promotion (int→long, float→double, decimal widening), reordered projection,_pos/_deletedmetadata columns, andidToConstant../gradlew -DsparkVersions=3.1 -DscalaVersion=2.12 :iceberg-spark:iceberg-spark-3.1_2.12:test— pass (incl.TestSparkOrcReadMetadataColumns, whose non-vectorized path exercises the new id-binding constructor's metadata branches)../gradlew :iceberg-orc:revapi— pass. The public additions do not trip RevAPI, so no.palantir/revapi.ymlchange is needed.spotlessApplyclean; the untouched engines (Flink 1.14/1.15/1.16, Spark 3.2/3.3) still compile against the retained deprecated positional path.Next steps (this is step 1 of 2)
This PR lands only the binding mechanism and deliberately does not change read output. The motivating follow-up is column default values: once fields are bound by id, a subsequent change can use those bound ids to resolve and apply an Iceberg field's initial-default value for fields that are present in the expected schema but absent from the ORC file (which today read as
null). Applying defaults correctly is only possible once binding is keyed on field id rather than position — so this PR is the prerequisite. The defaults work will land as a separate PR on top of this one.