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/04/29 05:34:38 UTC
[PR] Vectorized parquet row lineage [iceberg]
amogh-jahagirdar opened a new pull request, #12928:
URL: https://github.com/apache/iceberg/pull/12928
(no comment)
--
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
Re: [PR] Spark 3.5, Arrow: Support for Row lineage when using the Parquet Vectorized reader [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2069581692
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/TestHelpers.java:
##########
@@ -113,33 +113,6 @@ public static void assertEqualsSafe(Types.StructType struct, Record rec, Row row
}
}
- public static void assertEqualsBatch(
Review Comment:
No longer needed, it's public but it's in tests in Spark so I think we're OK to directly remove this? Don't think we're breaking any compatibilities guarantees the project provides by removing it.
--
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
Re: [PR] Spark 3.5, Arrow: Support for Row lineage when using the Parquet Vectorized reader [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2106437659
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java:
##########
@@ -567,6 +608,164 @@ public void close() {
}
}
+ private static final class RowIdVectorReader extends VectorizedArrowReader {
+ private static final Field ROW_ID_ARROW_FIELD = ArrowSchemaUtil.convert(MetadataColumns.ROW_ID);
+
+ private final long firstRowId;
+ private final VectorizedReader<VectorHolder> idReader;
+ private final VectorizedReader<VectorHolder> posReader;
+ private NullabilityHolder nulls;
+
+ private RowIdVectorReader(long firstRowId, VectorizedArrowReader idReader) {
+ this.firstRowId = firstRowId;
+ this.idReader = idReader != null ? idReader : nulls();
+ this.posReader = new PositionVectorReader(true);
+ }
+
+ @Override
+ public VectorHolder read(VectorHolder reuse, int numValsToRead) {
+ FieldVector positions = null;
+ FieldVector ids = null;
+
+ try {
+ positions = posReader.read(null, numValsToRead).vector();
+ VectorHolder idsHolder = idReader.read(null, numValsToRead);
+ ids = idsHolder.vector();
+ ArrowVectorAccessor<?, String, ?, ?> idsAccessor =
+ ids == null ? null : ArrowVectorAccessors.getVectorAccessor(idsHolder);
+
+ BigIntVector rowIds = allocateBigIntVector(ROW_ID_ARROW_FIELD, numValsToRead);
+ ArrowBuf dataBuffer = rowIds.getDataBuffer();
+ for (int i = 0; i < numValsToRead; i += 1) {
+ long bufferOffset = (long) i * Long.BYTES;
+ if (idsAccessor == null || isNull(idsHolder, i)) {
+ long rowId = firstRowId + (Long) positions.getObject(i);
+ dataBuffer.setLong(bufferOffset, rowId);
+ } else {
+ long materializedRowId = idsAccessor.getLong(i);
+ dataBuffer.setLong(bufferOffset, materializedRowId);
+ }
+ }
+
+ rowIds.setValueCount(numValsToRead);
+ return VectorHolder.vectorHolder(rowIds, MetadataColumns.ROW_ID, nulls);
+ } finally {
+ if (positions != null) {
+ positions.close();
+ }
+
+ if (ids != null) {
+ ids.close();
+ }
+ }
+ }
+
+ @Override
+ public void setRowGroupInfo(
+ PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {
+ idReader.setRowGroupInfo(source, metadata);
+ posReader.setRowGroupInfo(source, metadata);
+ }
+
+ @Override
+ public void setBatchSize(int batchSize) {
+ if (nulls == null || nulls.size() < batchSize) {
+ this.nulls = newNullabilityHolder(batchSize);
+ }
+
+ idReader.setBatchSize(batchSize);
+ posReader.setBatchSize(batchSize);
+ }
+
+ @Override
+ public void close() {
+ // don't close vectors as they are not owned by readers
Review Comment:
I made this comment a bit more specific to say the "result vectors" since I previously copy/pasted from the other reader but I think the intent is just to be for the result vectors, lmk what you think.
I believe we can and should safely close the intermediate vectors used for calculations on the inheritane path. For instance I think for the position vectors used for calculating row IDs in case it's null or the underlying materialized row id reader, those can safely be closed after reading a batch since those vectors are scoped to the `read`. After that, they don't need to be used externally and I think it makes sense to close them as soon as we know it's not needed anymore.
I think the part that cannot be closed are the vectors containing the contents of the results of the reader (e.g. the `BigIntVector rowIds = allocateBigIntVector(...)` since then we'd be freeing contents before external readers could use them.
--
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
Re: [PR] Spark, Arrow: Support for Row lineage when doing Vectorized Parquet reads [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#issuecomment-2839023768
I've separated out a large chunk of the test refactoring to use Iceberg Records in `writeAndValidate` instead of Avro Records into https://github.com/apache/iceberg/pull/12925 . I think we should get that in first, then I can rebase and this becomes a more focused change.
--
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
Re: [PR] Spark 3.5, Arrow: Support for Row lineage when using the Parquet Vectorized reader [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#issuecomment-2946762052
Thanks @nastra I will go ahead and merge
--
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
Re: [PR] Spark, Arrow: Support for Row lineage when doing Vectorized Parquet reads [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2066607495
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/GenericsHelpers.java:
##########
@@ -70,6 +72,19 @@ public static void assertEqualsSafe(Types.StructType struct, Record expected, Ro
}
}
+ public static void assertEqualsBatch(
+ Types.StructType struct,
+ Iterator<Record> expectedRecords,
+ ColumnarBatch batch,
+ Map<Integer, Object> idToConstant,
+ Integer numRowsAlreadyRead) {
Review Comment:
I'm looking into cleaning this up, I needed to pass the number of rows already read by the vectorized reader since that's needed to determine the overall row position
--
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
Re: [PR] Spark 3.5, Arrow: Support for Row lineage when using the Parquet Vectorized reader [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2106433365
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java:
##########
@@ -567,6 +608,164 @@ public void close() {
}
}
+ private static final class RowIdVectorReader extends VectorizedArrowReader {
+ private static final Field ROW_ID_ARROW_FIELD = ArrowSchemaUtil.convert(MetadataColumns.ROW_ID);
+
+ private final long firstRowId;
+ private final VectorizedReader<VectorHolder> idReader;
+ private final VectorizedReader<VectorHolder> posReader;
+ private NullabilityHolder nulls;
+
+ private RowIdVectorReader(long firstRowId, VectorizedArrowReader idReader) {
+ this.firstRowId = firstRowId;
+ this.idReader = idReader != null ? idReader : nulls();
+ this.posReader = new PositionVectorReader(true);
+ }
+
+ @Override
+ public VectorHolder read(VectorHolder reuse, int numValsToRead) {
+ FieldVector positions = null;
+ FieldVector ids = null;
+
+ try {
+ positions = posReader.read(null, numValsToRead).vector();
+ VectorHolder idsHolder = idReader.read(null, numValsToRead);
+ ids = idsHolder.vector();
+ ArrowVectorAccessor<?, String, ?, ?> idsAccessor =
+ ids == null ? null : ArrowVectorAccessors.getVectorAccessor(idsHolder);
+
+ BigIntVector rowIds = allocateBigIntVector(ROW_ID_ARROW_FIELD, numValsToRead);
+ ArrowBuf dataBuffer = rowIds.getDataBuffer();
+ for (int i = 0; i < numValsToRead; i += 1) {
Review Comment:
Happy to change it if you feel strongly about it, but I mostly just followed the increment pattern of i += 1 already in this class (and this package it looks like). If we do change it, I'd change it for the other instances in this class just to keep things consistent.
--
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
Re: [PR] Spark 3.5, Arrow: Support for Row lineage when using the Parquet Vectorized reader [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2069578123
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedReaderBuilder.java:
##########
@@ -101,33 +99,26 @@ public VectorizedReader<?> message(
Lists.newArrayListWithExpectedSize(icebergFields.size());
for (Types.NestedField field : icebergFields) {
- int id = field.fieldId();
- VectorizedReader<?> reader = readersById.get(id);
- if (idToConstant.containsKey(id)) {
- reorderedFields.add(constantReader(field, idToConstant.get(id)));
- } else if (id == MetadataColumns.ROW_POSITION.fieldId()) {
- if (setArrowValidityVector) {
- reorderedFields.add(VectorizedArrowReader.positionsWithSetArrowValidityVector());
- } else {
- reorderedFields.add(VectorizedArrowReader.positions());
- }
- } else if (id == MetadataColumns.IS_DELETED.fieldId()) {
- reorderedFields.add(new DeletedVectorReader());
- } else if (reader != null) {
- reorderedFields.add(reader);
- } else if (field.initialDefault() != null) {
- reorderedFields.add(
- constantReader(field, convert.apply(field.type(), field.initialDefault())));
- } else if (field.isOptional()) {
- reorderedFields.add(VectorizedArrowReader.nulls());
- } else {
- throw new IllegalArgumentException(
- String.format("Missing required field: %s", field.name()));
- }
+ VectorizedReader<?> reader =
+ VectorizedArrowReader.replaceWithMetadataReader(
+ field, readersById.get(field.fieldId()), idToConstant, setArrowValidityVector);
+ reorderedFields.add(defaultReader(field, reader));
Review Comment:
Same as the refactoring done in https://github.com/apache/iceberg/pull/12836/files
--
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
Re: [PR] Spark 3.5, Arrow: Support for Row lineage when using the Parquet Vectorized reader [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2069768082
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRowLevelOperationsWithLineage.java:
##########
@@ -91,7 +91,6 @@ public void beforeEach() {
assumeThat(formatVersion).isGreaterThanOrEqualTo(3);
// ToDo: Remove these as row lineage inheritance gets implemented in the other readers
assumeThat(fileFormat).isEqualTo(FileFormat.PARQUET);
- assumeThat(vectorized).isFalse();
Review Comment:
I should probably have an additional test which creates a lot of records (more than a single batch in a vectorized read), performs a modification on records with even IDs and asserts the row lineage state. Right now we're effectively just testing a single batch within this test. Theoretically everything should just work but it'll be better to have an explicit test.
--
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
Re: [PR] Spark 3.5, Arrow: Support for Row lineage when using the Parquet Vectorized reader [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2126140039
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java:
##########
@@ -567,6 +608,164 @@ public void close() {
}
}
+ private static final class RowIdVectorReader extends VectorizedArrowReader {
+ private static final Field ROW_ID_ARROW_FIELD = ArrowSchemaUtil.convert(MetadataColumns.ROW_ID);
+
+ private final long firstRowId;
+ private final VectorizedReader<VectorHolder> idReader;
+ private final VectorizedReader<VectorHolder> posReader;
+ private NullabilityHolder nulls;
+
+ private RowIdVectorReader(long firstRowId, VectorizedArrowReader idReader) {
+ this.firstRowId = firstRowId;
+ this.idReader = idReader != null ? idReader : nulls();
+ this.posReader = new PositionVectorReader(true);
+ }
+
+ @Override
+ public VectorHolder read(VectorHolder reuse, int numValsToRead) {
+ FieldVector positions = null;
+ FieldVector ids = null;
+
+ try {
+ positions = posReader.read(null, numValsToRead).vector();
+ VectorHolder idsHolder = idReader.read(null, numValsToRead);
+ ids = idsHolder.vector();
+ ArrowVectorAccessor<?, String, ?, ?> idsAccessor =
+ ids == null ? null : ArrowVectorAccessors.getVectorAccessor(idsHolder);
+
+ BigIntVector rowIds = allocateBigIntVector(ROW_ID_ARROW_FIELD, numValsToRead);
+ ArrowBuf dataBuffer = rowIds.getDataBuffer();
+ for (int i = 0; i < numValsToRead; i += 1) {
Review Comment:
I actually didn't realize that we have so many places that do `i += 1` in for loops. It's not a big deal and I don't feel strong about it but it would be great to fix this throughout the codebase in a separate PR
--
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
Re: [PR] Spark 3.5, Arrow: Support for Row lineage when using the Parquet Vectorized reader [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar merged PR #12928:
URL: https://github.com/apache/iceberg/pull/12928
--
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
Re: [PR] Spark 3.5, Arrow: Support for Row lineage when using the Parquet Vectorized reader [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2106432458
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java:
##########
@@ -461,6 +461,53 @@ public static VectorizedArrowReader positionsWithSetArrowValidityVector() {
return new PositionVectorReader(true);
}
+ public static VectorizedArrowReader rowIds(Long baseRowId, VectorizedArrowReader idReader) {
+ if (baseRowId != null) {
+ return new RowIdVectorReader(baseRowId, idReader);
+ } else {
+ return nulls();
+ }
+ }
+
+ public static VectorizedArrowReader lastUpdated(
+ Long baseRowId, Long fileLastUpdated, VectorizedArrowReader seqReader) {
+ if (fileLastUpdated != null && baseRowId != null) {
+ return new LastUpdatedSeqVectorReader(fileLastUpdated, seqReader);
+ } else {
+ return nulls();
+ }
+ }
+
+ public static VectorizedReader<?> replaceWithMetadataReader(
+ Types.NestedField icebergField,
+ VectorizedReader<?> reader,
+ Map<Integer, ?> idToConstant,
+ boolean setArrowValidityVector) {
+ int id = icebergField.fieldId();
+ if (id == MetadataColumns.ROW_ID.fieldId()) {
+ Long baseRowId = (Long) idToConstant.get(id);
+ return rowIds(baseRowId, (VectorizedArrowReader) reader);
+ } else if (id == MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.fieldId()) {
+ Long baseRowId = (Long) idToConstant.get(id);
Review Comment:
Good catch, I fixed it here.
--
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
Re: [PR] Spark 3.5, Arrow: Support for Row lineage when using the Parquet Vectorized reader [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2106432131
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRowLevelOperationsWithLineage.java:
##########
@@ -91,7 +91,6 @@ public void beforeEach() {
assumeThat(formatVersion).isGreaterThanOrEqualTo(3);
// ToDo: Remove these as row lineage inheritance gets implemented in the other readers
assumeThat(fileFormat).isEqualTo(FileFormat.PARQUET);
- assumeThat(vectorized).isFalse();
Review Comment:
I added a test which tests a lot of records (25000, it takes 6 seconds to run locally). I didn't do the "only update even" records approach since it made the merge a bit opaque to understand. I followed what I already did in the previous test which updates and inserts a single record. The benefit now is we can essentially make sure that reading multiple batches with all the inheritance logic works and there's no weird reuse bugs.
--
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
Re: [PR] Spark, Arrow: Support for Row lineage when doing Vectorized Parquet reads [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2066616375
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/TestHelpers.java:
##########
@@ -115,22 +116,38 @@ public static void assertEqualsSafe(Types.StructType struct, Record rec, Row row
public static void assertEqualsBatch(
Types.StructType struct, Iterator<Record> expected, ColumnarBatch batch) {
- for (int rowId = 0; rowId < batch.numRows(); rowId++) {
- InternalRow row = batch.getRow(rowId);
- Record rec = expected.next();
+ assertEqualsBatch(struct, expected, batch, null, null);
+ }
+
+ public static void assertEqualsBatch(
+ Types.StructType struct,
+ Iterator<Record> expected,
+ ColumnarBatch batch,
+ Integer numRowsRead,
+ Map<Integer, Object> idToConstant) {
+ for (int rowPos = 0; rowPos < batch.numRows(); rowPos++) {
+ InternalRow row = batch.getRow(rowPos);
+ Record expectedRecord = expected.next();
Review Comment:
These changes should no longer be needed?
--
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
Re: [PR] Spark, Arrow: Support for Row lineage when doing Vectorized Parquet reads [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2066591870
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java:
##########
@@ -567,6 +606,146 @@ public void close() {
}
}
+ private static final class RowIdVectorReader extends VectorizedArrowReader {
+ private static final Field ROW_ID_ARROW_FIELD = ArrowSchemaUtil.convert(MetadataColumns.ROW_ID);
+
+ private final long firstRowId;
+ private final VectorizedReader<VectorHolder> idReader;
+ private final VectorizedReader<VectorHolder> posReader;
+ private NullabilityHolder nulls;
+
+ private RowIdVectorReader(long firstRowId, VectorizedArrowReader idReader) {
+ this.firstRowId = firstRowId;
+ this.idReader = idReader;
+ this.posReader = new PositionVectorReader(true);
+ }
+
+ @Override
+ public VectorHolder read(VectorHolder reuse, int numValsToRead) {
+ FieldVector positions = posReader.read(null, numValsToRead).vector();
+ VectorHolder ids = idReader.read(null, numValsToRead);
+ BigIntVector vec = newVector(numValsToRead);
+ ArrowBuf dataBuffer = vec.getDataBuffer();
+ for (int i = 0; i < numValsToRead; i += 1) {
+ long bufferOffset = (long) i * Long.BYTES;
+ if (ids.nullabilityHolder().isNullAt(i) == 1) {
+ long rowId = firstRowId + (Long) positions.getObject(i);
+ dataBuffer.setLong(bufferOffset, rowId);
+ } else {
+ long materializedRowId = ids.vector().getDataBuffer().getLong(bufferOffset);
+ dataBuffer.setLong(bufferOffset, materializedRowId);
+ }
+ }
+
+ vec.setValueCount(numValsToRead);
+
+ return new VectorHolder.RowIdVectorHolder(vec, MetadataColumns.ROW_ID, nulls);
+ }
+
+ @Override
+ public void setRowGroupInfo(
+ PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {
+ idReader.setRowGroupInfo(source, metadata);
+ posReader.setRowGroupInfo(source, metadata);
+ }
+
+ @Override
+ public void setBatchSize(int batchSize) {
+ if (nulls == null || nulls.size() < batchSize) {
+ this.nulls = newNullabilityHolder(batchSize);
+ }
+
+ idReader.setBatchSize(batchSize);
+ posReader.setBatchSize(batchSize);
+ }
+
+ @Override
+ public void close() {
+ // don't close vectors as they are not owned by readers
+ }
+
+ private static BigIntVector newVector(int valueCount) {
+ BigIntVector vector =
+ (BigIntVector) ROW_ID_ARROW_FIELD.createVector(ArrowAllocation.rootAllocator());
+ vector.allocateNew(valueCount);
+ return vector;
+ }
+
+ private static NullabilityHolder newNullabilityHolder(int size) {
+ NullabilityHolder nullabilityHolder = new NullabilityHolder(size);
+ nullabilityHolder.setNotNulls(0, size);
+ return nullabilityHolder;
+ }
+ }
+
+ private static final class LastUpdatedSeqVectorReader extends VectorizedArrowReader {
+ private static final Field LAST_UPDATED_SEQ =
+ ArrowSchemaUtil.convert(MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER);
+
+ private final long lastUpdatedSeq;
+ private final VectorizedReader<VectorHolder> seqReader;
+ private NullabilityHolder nulls;
+
+ private LastUpdatedSeqVectorReader(
+ long lastUpdatedSeq, VectorizedReader<VectorHolder> seqReader) {
+ this.lastUpdatedSeq = lastUpdatedSeq;
+ this.seqReader = seqReader == null ? nulls() : seqReader;
+ }
+
+ @Override
+ public VectorHolder read(VectorHolder reuse, int numValsToRead) {
+ VectorHolder seqNumbers = seqReader.read(null, numValsToRead);
+ BigIntVector vec = newVector(numValsToRead);
+ ArrowBuf dataBuffer = vec.getDataBuffer();
+ for (int i = 0; i < numValsToRead; i += 1) {
+ long bufferOffset = (long) i * Long.BYTES;
+ if (seqNumbers.nullabilityHolder().isNullAt(i) == 1) {
+ dataBuffer.setLong(bufferOffset, lastUpdatedSeq);
+ } else {
+ long materializedRowId = seqNumbers.vector().getDataBuffer().getLong(bufferOffset);
+ dataBuffer.setLong(bufferOffset, materializedRowId);
+ }
+ }
+
+ vec.setValueCount(numValsToRead);
+
+ return new VectorHolder.RowIdVectorHolder(vec, MetadataColumns.ROW_ID, nulls);
+ }
+
+ @Override
+ public void setRowGroupInfo(
+ PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {
+ seqReader.setRowGroupInfo(source, metadata);
+ }
+
+ @Override
+ public void setBatchSize(int batchSize) {
+ if (nulls == null || nulls.size() < batchSize) {
+ this.nulls = newNullabilityHolder(batchSize);
+ }
+
+ seqReader.setBatchSize(batchSize);
+ }
+
+ @Override
+ public void close() {
+ // don't close vectors as they are not owned by readers
+ }
+
+ private static BigIntVector newVector(int valueCount) {
+ BigIntVector vector =
+ (BigIntVector) LAST_UPDATED_SEQ.createVector(ArrowAllocation.rootAllocator());
+ vector.allocateNew(valueCount);
+ return vector;
+ }
+
+ private static NullabilityHolder newNullabilityHolder(int size) {
+ NullabilityHolder nullabilityHolder = new NullabilityHolder(size);
+ nullabilityHolder.setNotNulls(0, size);
+ return nullabilityHolder;
+ }
Review Comment:
This doesn't need to be duplicated, move it out
--
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
Re: [PR] Spark 3.5, Arrow: Support for Row lineage when using the Parquet Vectorized reader [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2106437961
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/parquet/vectorized/TestParquetVectorizedReads.java:
##########
@@ -104,7 +119,30 @@ private void writeAndValidate(
int batchSize,
Function<Record, Record> transform)
throws IOException {
- // Write test data
+ writeAndValidate(
+ writeSchema,
+ expectedSchema,
+ numRecords,
+ seed,
+ nullPercentage,
+ reuseContainers,
+ batchSize,
+ transform,
+ ImmutableMap.of());
+ }
+
+ private void writeAndValidate(
+ Schema writeSchema,
+ Schema expectedSchema,
+ int numRecords,
+ long seed,
+ float nullPercentage,
+ boolean reuseContainers,
+ int batchSize,
+ Function<Record, Record> transform,
+ Map<Integer, Object> idToConstant)
+ throws IOException {
+ // // Write test data
Review Comment:
Fixed!
--
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
Re: [PR] Spark 3.5, Arrow: Support for Row lineage when using the Parquet Vectorized reader [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2070346566
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java:
##########
@@ -567,6 +608,138 @@ public void close() {
}
}
+ private static final class RowIdVectorReader extends VectorizedArrowReader {
+ private static final Field ROW_ID_ARROW_FIELD = ArrowSchemaUtil.convert(MetadataColumns.ROW_ID);
+
+ private final long firstRowId;
+ private final VectorizedReader<VectorHolder> idReader;
+ private final VectorizedReader<VectorHolder> posReader;
+ private NullabilityHolder nulls;
+
+ private RowIdVectorReader(long firstRowId, VectorizedArrowReader idReader) {
+ this.firstRowId = firstRowId;
+ this.idReader = idReader != null ? idReader : nulls();
+ this.posReader = new PositionVectorReader(true);
+ }
+
+ @Override
+ public VectorHolder read(VectorHolder reuse, int numValsToRead) {
+ FieldVector positions = posReader.read(null, numValsToRead).vector();
+ VectorHolder ids = idReader.read(null, numValsToRead);
Review Comment:
On second look, we should be closing these after we we go through the batch. Same with the last udpate seq case, I'll fix this
--
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
Re: [PR] Spark 3.5, Arrow: Support for Row lineage when using the Parquet Vectorized reader [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #12928:
URL: https://github.com/apache/iceberg/pull/12928#discussion_r2069572607
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java:
##########
@@ -461,6 +461,53 @@ public static VectorizedArrowReader positionsWithSetArrowValidityVector() {
return new PositionVectorReader(true);
}
+ public static VectorizedArrowReader rowIds(Long baseRowId, VectorizedArrowReader idReader) {
+ if (baseRowId != null) {
+ return new RowIdVectorReader(baseRowId, idReader);
+ } else {
+ return nulls();
+ }
+ }
+
+ public static VectorizedArrowReader lastUpdated(
+ Long baseRowId, Long fileLastUpdated, VectorizedArrowReader seqReader) {
+ if (fileLastUpdated != null && baseRowId != null) {
+ return new LastUpdatedSeqVectorReader(fileLastUpdated, seqReader);
+ } else {
+ return nulls();
+ }
+ }
+
+ public static VectorizedReader<?> replaceWithMetadataReader(
+ Types.NestedField icebergField,
+ VectorizedReader<?> reader,
+ Map<Integer, ?> idToConstant,
+ boolean setArrowValidityVector) {
+ int id = icebergField.fieldId();
+ if (id == MetadataColumns.ROW_ID.fieldId()) {
+ Long baseRowId = (Long) idToConstant.get(id);
+ return rowIds(baseRowId, (VectorizedArrowReader) reader);
+ } else if (id == MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.fieldId()) {
+ Long baseRowId = (Long) idToConstant.get(id);
+ Long fileSeqNumber = (Long) idToConstant.get(id);
+ return VectorizedArrowReader.lastUpdated(
+ baseRowId, fileSeqNumber, (VectorizedArrowReader) reader);
+ } else if (idToConstant.containsKey(id)) {
+ // containsKey is used because the constant may be null
+ return new ConstantVectorReader<>(icebergField, idToConstant.get(id));
+ } else if (id == MetadataColumns.ROW_POSITION.fieldId()) {
+ if (setArrowValidityVector) {
+ return positionsWithSetArrowValidityVector();
+ } else {
+ return VectorizedArrowReader.positions();
+ }
+ } else if (id == MetadataColumns.IS_DELETED.fieldId()) {
+ return new DeletedVectorReader();
+ }
Review Comment:
This is all pretty similar to what was done with the ParquetValueReader
--
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