You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "amogh-jahagirdar (via GitHub)" <gi...@apache.org> on 2025/07/01 00:32:26 UTC
Re: [PR] Spark, Avro: Add support for row lineage in Avro reader [iceberg]
amogh-jahagirdar commented on code in PR #13070:
URL: https://github.com/apache/iceberg/pull/13070#discussion_r2176182992
##########
core/src/main/java/org/apache/iceberg/avro/ValueReaders.java:
##########
@@ -1235,4 +1265,64 @@ public void setRowPositionSupplier(Supplier<Long> posSupplier) {
this.currentPosition = posSupplier.get() - 1;
}
}
+
+ static class RowIdReader implements ValueReader<Long>, SupportsRowPosition {
+ private final long firstRowId;
+ private final ValueReader<Long> idReader;
+ private final ValueReader<Long> posReader;
+
+ RowIdReader(Long firstRowId, ValueReader<Long> idReader) {
+ this.firstRowId = firstRowId;
+ this.idReader = idReader != null ? idReader : ValueReaders.constant(null);
+ this.posReader = positions();
+ }
+
+ @Override
+ public Long read(Decoder ignored, Object reuse) throws IOException {
+ Long id = idReader.read(ignored, reuse);
Review Comment:
Done
##########
core/src/main/java/org/apache/iceberg/avro/ValueReaders.java:
##########
@@ -228,50 +228,112 @@ public static List<Pair<Integer, ValueReader<?>>> buildReadPlan(
Map<Integer, ?> idToConstant,
BiFunction<Type, Object, Object> convert) {
Map<Integer, Integer> idToPos = idToPos(expected);
-
List<Pair<Integer, ValueReader<?>>> readPlan = Lists.newArrayList();
- List<Schema.Field> fileFields = record.getFields();
- for (int pos = 0; pos < fileFields.size(); pos += 1) {
+ addFileFieldReadersToPlan(readPlan, record.getFields(), fieldReaders, idToPos, idToConstant);
+ addMissingFileReadersToPlan(readPlan, idToPos, expected, idToConstant, convert);
+ return readPlan;
Review Comment:
I didn't end up refactoring like we do in `ParquetValueReaders`, it ended up being a bit hard to read considering avro positions compared to just splitting into two separate methods, one where we just add the file field readers to the read plan first and then we add the missing readers to the plan. That also eliminates cyclomatic complexity
##########
core/src/main/java/org/apache/iceberg/avro/ValueReaders.java:
##########
@@ -1235,4 +1265,64 @@ public void setRowPositionSupplier(Supplier<Long> posSupplier) {
this.currentPosition = posSupplier.get() - 1;
}
}
+
+ static class RowIdReader implements ValueReader<Long>, SupportsRowPosition {
+ private final long firstRowId;
+ private final ValueReader<Long> idReader;
+ private final ValueReader<Long> posReader;
+
+ RowIdReader(Long firstRowId, ValueReader<Long> idReader) {
+ this.firstRowId = firstRowId;
+ this.idReader = idReader != null ? idReader : ValueReaders.constant(null);
+ this.posReader = positions();
+ }
+
+ @Override
+ public Long read(Decoder ignored, Object reuse) throws IOException {
+ Long id = idReader.read(ignored, reuse);
+ long pos = posReader.read(ignored, reuse);
+
+ if (id != null) {
+ return id;
+ }
+
+ return firstRowId + pos;
+ }
+
+ @Override
+ public void skip(Decoder decoder) throws IOException {
+ idReader.skip(decoder);
+ posReader.skip(decoder);
+ }
+
+ @Override
+ public void setRowPositionSupplier(Supplier<Long> posSupplier) {
+ ((SupportsRowPosition) posReader).setRowPositionSupplier(posSupplier);
+ }
+ }
+
+ static class LastUpdatedSeqReader implements ValueReader<Long> {
+ private final long fileLastUpdated;
+ private final ValueReader<Long> seqReader;
+
+ LastUpdatedSeqReader(long fileLastUpdated, ValueReader<Long> seqReader) {
Review Comment:
Done!
##########
core/src/main/java/org/apache/iceberg/avro/ValueReaders.java:
##########
@@ -1235,4 +1265,64 @@ public void setRowPositionSupplier(Supplier<Long> posSupplier) {
this.currentPosition = posSupplier.get() - 1;
}
}
+
+ static class RowIdReader implements ValueReader<Long>, SupportsRowPosition {
+ private final long firstRowId;
+ private final ValueReader<Long> idReader;
+ private final ValueReader<Long> posReader;
+
+ RowIdReader(Long firstRowId, ValueReader<Long> idReader) {
+ this.firstRowId = firstRowId;
+ this.idReader = idReader != null ? idReader : ValueReaders.constant(null);
+ this.posReader = positions();
+ }
+
+ @Override
+ public Long read(Decoder ignored, Object reuse) throws IOException {
+ Long id = idReader.read(ignored, reuse);
+ long pos = posReader.read(ignored, reuse);
+
+ if (id != null) {
+ return id;
+ }
+
+ return firstRowId + pos;
+ }
+
+ @Override
+ public void skip(Decoder decoder) throws IOException {
+ idReader.skip(decoder);
+ posReader.skip(decoder);
+ }
+
+ @Override
+ public void setRowPositionSupplier(Supplier<Long> posSupplier) {
+ ((SupportsRowPosition) posReader).setRowPositionSupplier(posSupplier);
+ }
+ }
+
+ static class LastUpdatedSeqReader implements ValueReader<Long> {
+ private final long fileLastUpdated;
+ private final ValueReader<Long> seqReader;
+
+ LastUpdatedSeqReader(long fileLastUpdated, ValueReader<Long> seqReader) {
+ this.fileLastUpdated = fileLastUpdated;
+ this.seqReader = seqReader;
+ }
+
+ @Override
+ public Long read(Decoder ignored, Object reuse) throws IOException {
+ Long lastUpdated = seqReader.read(ignored, reuse);
Review Comment:
I renamed to `rowLastUpdatedSeqNumber`. I think that's a better name since it indicates specifically that the materialized seq number is for the row and better distinguishes from the file sequence number that's inherited for records that don't have the value materialized
##########
core/src/main/java/org/apache/iceberg/avro/ValueReaders.java:
##########
@@ -274,6 +287,23 @@ public static List<Pair<Integer, ValueReader<?>>> buildReadPlan(
return readPlan;
}
+ public static ValueReader<Long> rowIds(Long baseRowId, ValueReader<?> idReader) {
+ if (baseRowId != null) {
+ return new RowIdReader(baseRowId, (ValueReader<Long>) idReader);
+ } else {
+ return ValueReaders.constant(null);
+ }
+ }
+
+ public static ValueReader<Long> lastUpdated(
+ Long baseRowId, Long fileLastUpdated, ValueReader<?> seqReader) {
Review Comment:
Sure, I updated!
--
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