stevenzwu commented on code in PR #13070:
URL: https://github.com/apache/iceberg/pull/13070#discussion_r2175492438
##########
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:
nit: `idFromFile` like the Parquet reader is more informative
##########
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:
nit: I know this is to be consistent with the Parquet value readers class.
but I found the `fileSeqNumber` (used by the caller method) is a more intuitive
name.
##########
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:
nit: fileLastUpdated -> fileSeqNumber
##########
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:
nit: lastUpdated -> lastUpdateSeqFromFile
--
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]
