Repository navigation
Conversation
| return false; | ||
| } | ||
|
|
||
| protected Set<FileFormat> supportedFormats() { |
There was a problem hiding this comment.
I don't really like this one.
Could we just re-define FILE_FORMATS, and FORMAT_AND_GENERATOR in TestSparkVectorizedFormatModel?
private static final FileFormat[] FILE_FORMATS = {FileFormat.ORC, FileFormat.PARQUET};
private static final List<Arguments> FORMAT_AND_GENERATOR
This relies on the fact, how junit works internally, but no change is required in all of the tests.
I'm open for other ideas as well, this is just the first one occured to me
There was a problem hiding this comment.
Thank you for the suggestion @pvary I tried it out and here's what I found:
Redefining FILE_FORMATS in the subclass works as you described, those tests drop AVRO with no base class changes.
Redefining FORMAT_AND_GENERATOR doesn't work as is, because it references DataGenerators, which is package-private in org.apache.iceberg.data while TestSparkVectorizedFormatModel is in org.apache.iceberg.spark.data.
I tried making DataGenerators, DataGenerators.ALL and the DataGenerator interface public, and with that both fields can be redefined in the subclass. All AVRO cases are gone and I could remove supportedFormats() and the per test assumptions entirely. TestSparkFormatModel and TestFlinkFormatModel still pass.
Would you like me to go with that, or would you rather avoid widening the visibility of those test classes?
There was a problem hiding this comment.
I'm fine with making the DataGenerator and related stuff public in the test packages
There was a problem hiding this comment.
Thank you @pvary.
- Made
DataGenerator,DataGeneratorsandDataGenerators.ALLpublic in the test package. TestSparkVectorizedFormatModelredefines bothFILE_FORMATSandFORMAT_AND_GENERATOR.- Dropped
supportedFormats()and all the per testassumeSupportedFormat()calls from the base class.
There was a problem hiding this comment.
One thing I hit while doing this. When the subclass redefines FORMAT_AND_GENERATOR, it also has to repeat every filter the base applies. #17617 added supportsGenerator yesterday, my subclass didn't have it, and ORC + Variant came back and failed CI.
To fix it I'd have to either make supportsGenerator public, or copy the check into the subclass. Neither seems clean, and any new filter will have the same problem IIUC.
| return false; | ||
| } | ||
|
|
||
| protected boolean readOnly() { |
There was a problem hiding this comment.
Shall we split the base test class to read-tests, and read+write-tests?
Using this method in all test seem awkward a bit.
If we decide so, we might want to do it in a different PR
There was a problem hiding this comment.
I agree with you, would love your advice.
Two ways we could do it:
- Keep the read tests and shared helpers in
BaseFormatModelTests, move the write tests into aBaseReadWriteFormatModelTeststhat extends it. Existing tests move to the new class. - Insert a
BaseReadFormatModelTestsabove the currentBaseFormatModelTestsand put the read tests there. No existing subclass changes its extends.
And TestSparkVectorizedFormatModel would only extend the read-only class.
Would you like to do it in a follow-up PR or this PR?
There was a problem hiding this comment.
Please separate out the test to:
- ReadFormatModelTests as a base, and move the read tests there
- BaseFormatModelTests inheriting from ReadFormatModelTests
I would do this way
There was a problem hiding this comment.
This should be in a PR before this change
159b391 to
6b66972
Compare
| import org.apache.spark.sql.vectorized.ColumnarBatch; | ||
| import org.junit.jupiter.params.provider.Arguments; | ||
|
|
||
| public class TestSparkVectorizedFormatModel extends BaseFormatModelTests<ColumnarBatch> { |
6b66972 to
50d5209
Compare
50d5209 to
625fe88
Compare
| @ParameterizedTest | ||
| @FieldSource("FILE_FORMATS") | ||
| void testNestedDefaultValue(FileFormat fileFormat) throws IOException { | ||
| assumeFalse(supportsBatchReads(), "Vectorized reads do not support nested types"); |
There was a problem hiding this comment.
could we handle this with unsupportedTypeIds?
There was a problem hiding this comment.
I added assumeSupportedTypes() to skip tests when the schema contains a type listed in unsupportedTypeIds().
| @FieldSource("FILE_FORMATS") | ||
| void testReaderBuilderSplit(FileFormat fileFormat) throws IOException { | ||
|
|
||
| assumeFalse(supportsBatchReads(), "Batch reads return batches, not rows"); |
There was a problem hiding this comment.
Do we need to override this test, or can we test differently?
There was a problem hiding this comment.
I added a countRows() hook that sums the row counts across ColumnarBatch instances and uses .size() for row readers. This allows the shared tests to work for both.
| @FieldSource("FILE_FORMATS") | ||
| void testReaderBuilderReuseContainers(FileFormat fileFormat) throws IOException { | ||
|
|
||
| assumeFalse(supportsBatchReads(), "Batch reads reuse differently"); |
There was a problem hiding this comment.
Neither vectorized reader returns the same object twice. ColumnarBatchReader.read always builds a new ColumnarBatch and uses reuse only to decide whether to close the previous vectors IIUC:
if (reuse == null) {
closeVectors();
}
return new ColumnBatchLoader(numRowsToRead).loadDataToColumnBatch();VectorizedSparkOrcReaders.read doesn't take a reuse argument at all.
I checked with and without reuseContainers() on both formats and each call returns a new object either way. Is that expected for batch reads? Please correct me if I am wrong.
There was a problem hiding this comment.
What about the underlying data structures? I expect that the Arrow Vlvectors should be reused.
There was a problem hiding this comment.
The reuse logic is in VectorizedArrowReader.java.
In ColumnarBatchReader.java, the previous VectorHolder is passed back to the reader, allowing the underlying Arrow vector to be reused.
I am getting each vector in a batch like this,
private ValueVector arrowVector(ColumnarBatch batch, int col) {
IcebergArrowColumnVector vector = (IcebergArrowColumnVector) batch.column(col);
return vector.vectorAccessor().getVector();
}I have updated the shared test to compare the underlying Arrow vectors for every column, both with and without reuseContainers().
| @FieldSource("FILE_FORMATS") | ||
| void testReadMetadataColumnIsDeleted(FileFormat fileFormat) throws IOException { | ||
|
|
||
| assumeFalse(supportsBatchReads(), "DeletedColumnVector requires a delete filter"); |
There was a problem hiding this comment.
Please provide a better message. DeletedColumnVector is Spark specific.
What is the exact issue here? The batch readers doesn't read the DVs either, so checking false seems ok.
There was a problem hiding this comment.
It doesn't return false, it throws:
TestSparkVectorizedFormatModel > testReadMetadataColumnIsDeleted(FileFormat) > [2] PARQUET FAILED
java.lang.NullPointerException: Cannot load from byte/boolean array because "this.isDeleted" is null
at org.apache.iceberg.spark.data.vectorized.DeletedColumnVector.getBoolean(DeletedColumnVector.java:61)
at org.apache.spark.sql.vectorized.ColumnarBatchRow.getBoolean(ColumnarBatchRow.java:108)
at org.apache.spark.sql.vectorized.ColumnarBatchRow.get(ColumnarBatchRow.java:181)
at org.apache.iceberg.spark.data.TestHelpers.assertEquals(TestHelpers.java:772)
at org.apache.iceberg.spark.data.TestHelpers.assertEquals(TestHelpers.java:734)
at org.apache.iceberg.spark.data.TestHelpers.assertEquals(TestHelpers.java:719)
at org.apache.iceberg.spark.data.TestSparkVectorizedFormatModel.assertRecordsEqual(TestSparkVectorizedFormatModel.java:102)
The array is only filled when a delete filter is set IIUC. ORC passes, so this is Parquet only if I am not wrong. I will fix the message. Should I file an issue for this, or send a fix PR? Please correct me if I am wrong.
There was a problem hiding this comment.
Yeah, the array is only created by the delete filter.
Let me think what could we do here.
There was a problem hiding this comment.
Taking a look at supportsBatchReads - it always return false.
I think we should override this method.
Also we should update testReaderBuilderRecordsPerBatchNotSupported to have a batch and a non-bach version. Maybe with override and rename to testReaderBuilderRecordsPerBatch
There was a problem hiding this comment.
I updated the skip message to explain the issue and limited the skip to vectorized Parquet. ORC still runs the test.
There was a problem hiding this comment.
I updated testReaderBuilderRecordsPerBatch to test both row and batch readers. Row readers check that recordsPerBatch() throws an exception, while batch readers check the batch size and total row count.
| @FieldSource("FILE_FORMATS") | ||
| void testReadMetadataColumnRowLineage(FileFormat fileFormat) throws IOException { | ||
|
|
||
| // TODO: ORC's vectorized reader treats _row_id as a plain constant |
There was a problem hiding this comment.
Shorten and leave an issue id in the text
| @ParameterizedTest | ||
| @FieldSource("FILE_FORMATS") | ||
| void testSchemaEvolutionEmptyProjection(FileFormat fileFormat) throws IOException { | ||
| assumeFalse(supportsBatchReads(), "Row count assertion does not apply to batches"); |
There was a problem hiding this comment.
I added a countRows() hook that sums the row counts across ColumnarBatch instances and uses .size() for row readers. This allows the shared tests to work for both.
|
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions. |
|
@joyhaldar: Are you planning to work on this? Shall we take over? Maybe @Guosmilesmile could help, if you want. |
Sorry for the delay, I was still planning to work on this, but I've been consumed by other priorities. I would still require some time to get back to this. I would not mind if you'd like to take it over, but I am planning to make time in the coming weeks. |
Hi @joyhaldar, if you're still short on time I'd be happy to pick this up. I'd keep your commits and address @pvary's comments (splitting testReaderBuilderRecordsPerBatchNotSupported into batch and non-batch versions, and checking reuse of the underlying Arrow vectors). If you'd rather finish it yourself, no problem at all, I'll leave it with you. |
Co-authored-by: Joy Haldar <joy.haldar@target.com>
Hello, sorry for the delay, I am working this over the weekend. |
…o 4.2 Co-authored-by: Joy Haldar <joy.haldar@target.com>
Co-authored-by: Joy Haldar <joy.haldar@target.com>
Co-authored-by: Joy Haldar <joy.haldar@target.com>
There was a problem hiding this comment.
Please check this size checks. Are they still valid for the batches?
There was a problem hiding this comment.
Updated this to use countRows(readRecords) so it counts rows for both row and batch readers. I also checked the other size assertions in the shared test class and could not find anymore to update.
| } | ||
|
|
||
| private ValueVector arrowVector(ColumnarBatch batch, int col) { | ||
| IcebergArrowColumnVector vector = (IcebergArrowColumnVector) batch.column(col); |
There was a problem hiding this comment.
Are we sure about this? At least an instanceof please
There was a problem hiding this comment.
I have added,
assertThat(batch.column(col)).isInstanceOf(IcebergArrowColumnVector.class);before typecasting.
| } | ||
|
|
||
| protected void assumeSupportedTypes(Schema schema) { | ||
| Collection<Type.TypeID> unsupportedTypeIds = unsupportedTypeIds(); |
There was a problem hiding this comment.
MAybe update the javadoc for unsupportedTypeIds?
There was a problem hiding this comment.
I have updated the Javadoc to mention both filtering unsupported columns and skipping tests for unsupported types.
| Type.TypeID.STRUCT, | ||
| Type.TypeID.LIST, | ||
| Type.TypeID.MAP, | ||
| Type.TypeID.UUID, |
There was a problem hiding this comment.
Yes, I removed GEOMETRY and GEOGRAPHY from the unsupported types locally and reran the tests. Four tests failed with UnsupportedOperationException in ArrowSchemaUtil.java:184, as the Arrow reader doesn't support these types yet. Adding them back skips those cases. It is routing to UnsupportedOperationException line in ArrowSchemaUtil.
Stacktrace:
TestSparkVectorizedFormatModel > testDataWriterGenericWriteEngineRead(FileFormat, DataGenerator) > [20] PARQUET, Geometry FAILED
java.lang.UnsupportedOperationException: Unsupported primitive type: geometry(OGC:CRS84)
at org.apache.iceberg.arrow.ArrowSchemaUtil$IcebergToArrowTypeConverter.primitive(ArrowSchemaUtil.java:184)
at org.apache.iceberg.arrow.ArrowSchemaUtil$IcebergToArrowTypeConverter.primitive(ArrowSchemaUtil.java:68)
at org.apache.iceberg.types.TypeUtil.visit(TypeUtil.java:823)
at org.apache.iceberg.arrow.ArrowSchemaUtil.convert(ArrowSchemaUtil.java:65)
at org.apache.iceberg.arrow.vectorized.VectorizedArrowReader.allocateFieldVector(VectorizedArrowReader.java:228)
at org.apache.iceberg.arrow.vectorized.VectorizedArrowReader.read(VectorizedArrowReader.java:151)
at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader$ColumnBatchLoader.readDataToColumnVectors(ColumnarBatchReader.java:81)
at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader$ColumnBatchLoader.loadDataToColumnBatch(ColumnarBatchReader.java:69)
at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader.read(ColumnarBatchReader.java:56)
at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader.read(ColumnarBatchReader.java:38)
at org.apache.iceberg.parquet.VectorizedParquetReader$FileIterator.next(VectorizedParquetReader.java:147)
at org.apache.iceberg.spark.data.TestSparkVectorizedFormatModel.assertRecordsEqual(TestSparkVectorizedFormatModel.java:137)
TestSparkVectorizedFormatModel > testDataWriterGenericWriteEngineRead(FileFormat, DataGenerator) > [21] PARQUET, Geography FAILED
java.lang.UnsupportedOperationException: Unsupported primitive type: geography(OGC:CRS84, spherical)
at org.apache.iceberg.arrow.ArrowSchemaUtil$IcebergToArrowTypeConverter.primitive(ArrowSchemaUtil.java:184)
at org.apache.iceberg.arrow.ArrowSchemaUtil$IcebergToArrowTypeConverter.primitive(ArrowSchemaUtil.java:68)
at org.apache.iceberg.types.TypeUtil.visit(TypeUtil.java:823)
at org.apache.iceberg.arrow.ArrowSchemaUtil.convert(ArrowSchemaUtil.java:65)
at org.apache.iceberg.arrow.vectorized.VectorizedArrowReader.allocateFieldVector(VectorizedArrowReader.java:228)
at org.apache.iceberg.arrow.vectorized.VectorizedArrowReader.read(VectorizedArrowReader.java:151)
at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader$ColumnBatchLoader.readDataToColumnVectors(ColumnarBatchReader.java:81)
at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader$ColumnBatchLoader.loadDataToColumnBatch(ColumnarBatchReader.java:69)
at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader.read(ColumnarBatchReader.java:56)
at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader.read(ColumnarBatchReader.java:38)
at org.apache.iceberg.parquet.VectorizedParquetReader$FileIterator.next(VectorizedParquetReader.java:147)
at org.apache.iceberg.spark.data.TestSparkVectorizedFormatModel.assertRecordsEqual(TestSparkVectorizedFormatModel.java:137)
There was a problem hiding this comment.
ORC supports Struct, Array, Map, UUID
There was a problem hiding this comment.
You’re right. I have made the set per-format.
| } | ||
|
|
||
| @Override | ||
| protected void assertRecordsEqual( |
There was a problem hiding this comment.
Does this work when there are multiple batches?
My AI friend is concerned because the ColumnarBatch is only a thin wrapper.
This works now, because there is only a single batch in the tests
There was a problem hiding this comment.
You are right. I tested with recordsPerBatch(2) and found that collecting all batches before comparing caused failures.
When we were doing this,
readRecords = ImmutableList.copyOf(reader);the earlier batch data was overwritten when moving to the next batch, and assertEquals was failing.
I updated assertRecordsEqual() to iterate over records or batches and compare with expected records when reading, before moving to the next batch.
Co-authored-by: Joy Haldar <joy.haldar@target.com>
| Collection<Type.TypeID> unsupportedTypeIds = unsupportedTypeIds(); | ||
|
|
||
| Type unsupportedType = | ||
| TypeUtil.find(schema, type -> unsupportedTypeIds.contains(type.typeId())); |
There was a problem hiding this comment.
This checks for the main struct as well which will be skipped because vectorized read doesn't support structs ATM.
There was a problem hiding this comment.
Fixed, the root struct is now excluded in containsType().
| ReadBuilder<T, ?> builder = | ||
| FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile).project(schema); | ||
|
|
||
| if (supportsBatchReads()) { | ||
| builder.recordsPerBatch(2); | ||
| } |
There was a problem hiding this comment.
Could we create a shared builder helper, like:
private ReadBuilder<T, ?> readBuilder(FileFormat format, InputFile file) {
ReadBuilder<T, ?> builder = FormatModelRegistry.readBuilder(format, engineType(), file);
return supportsBatchReads() ? builder.recordsPerBatch(3) : builder;
}
Notice: used 3 instead of 2, so we have partial batches
There was a problem hiding this comment.
Added readBuilder() with recordsPerBatch(3) and used it in all four places.
There was a problem hiding this comment.
Changed to assertThat(countRows(readRecords)).isZero().
There was a problem hiding this comment.
Changed to assertThat(countRows(readRecords)).isGreaterThan(0).
| assumeFalse( | ||
| supportsBatchReads() && fileFormat == FileFormat.PARQUET, | ||
| "Reading _deleted without a delete filter is not supported by the vectorized reader"); |
There was a problem hiding this comment.
Could we move this code to the Vectorized tests?
Having these checks hides them when we want to understand what is supported in Vectorized reads, and what is not
There was a problem hiding this comment.
I have moved, TestSparkVectorizedFormatModel overrides the test, skips Parquet and calls super.
| void testReadMetadataColumnRowLineage(FileFormat fileFormat) throws IOException { | ||
|
|
||
| // TODO: remove when https://github.com/apache/iceberg/issues/18543 is fixed | ||
| assumeFalse( |
There was a problem hiding this comment.
I have moved to an override in TestSparkVectorizedFormatModel, skips ORC with a TODO for #18543.
| void testReadMetadataColumnRowLineageExistingValues(FileFormat fileFormat) throws IOException { | ||
|
|
||
| // TODO: remove when https://github.com/apache/iceberg/issues/18543 is fixed | ||
| assumeFalse( |
There was a problem hiding this comment.
I have moved to an override in TestSparkVectorizedFormatModel, skips ORC with a TODO for #18543.
| format -> | ||
| Arrays.stream(DataGenerators.ALL) | ||
| .filter(generator -> supportsGenerator(format, generator)) | ||
| .map(generator -> Arguments.of(format, generator))) |
There was a problem hiding this comment.
StructOfPrimitive, ListOfPrimitive, MapOfPrimitive don't skip here, but skip based on types.... yikes...
There was a problem hiding this comment.
Fixed. FORMAT_AND_GENERATOR is now skipping generators with unsupported types up front, so StructOfPrimitive etc. are no longer generated as Parquet test cases. They are still running for ORC.
Co-authored-by: Joy Haldar <joy.haldar@target.com>
ecb8eea to
ae0b191
Compare
| List<T> readRecords; | ||
|
|
||
| try (CloseableIterable<T> reader = | ||
| FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile) |
There was a problem hiding this comment.
Do we want to switch this to readBuilder() too?
And maybe in other places as well, like: testReadFileWithoutFieldIdsUsingNameMapping, readAndAssertMetadataColumn?
There was a problem hiding this comment.
Done. I have changed all engine reads in the base now to use readBuilder(), except testReaderBuilderRecordsPerBatch, which sets its own batch size.
| private static final FileFormat[] FILE_FORMATS = {FileFormat.ORC, FileFormat.PARQUET}; | ||
|
|
||
| // Spark has no TIME or TIMESTAMP_NANO type. The Arrow-based Parquet reader also cannot read | ||
| // nested or geospatial columns, the vectorized ORC reader can. |
There was a problem hiding this comment.
Is this true? ORC + geospat is working?
There was a problem hiding this comment.
No sorry, ORC doesn’t support geospatial, hence it isn’t tested. Fixed the comment.
|
|
||
| private static boolean supportsVectorized(FileFormat format, DataGenerator generator) { | ||
| return supportsGenerator(format, generator) | ||
| && !containsType(generator.schema(), UNSUPPORTED_TYPE_IDS.get(format)); |
There was a problem hiding this comment.
Boolean, date, timestamp and timestamptz aren't covered by any generator test in the vectorized suite. Could we move time_col into its own Time generator, like TimestampNano, so dropping a whole generator only removes the type it's about? Or only drop whole generators for nested types and let supportedSchema filter TIME.
There was a problem hiding this comment.
Went with your second option. Filtering TIME per column now.
Co-authored-by: Joy Haldar <joy.haldar@target.com>
| for (T batch : readRecords) { | ||
| assertThat(countRows(List.of(batch))).isLessThanOrEqualTo(100); | ||
| } | ||
| assertThat(countRows(readRecords)).isEqualTo(genericRecords.size()); |
| InternalRowConverter.convert(schema, expected.get(rowOffset + i)), | ||
| batch.getRow(i)); | ||
| } | ||
| rowOffset += batch.numRows(); |
| private static final Set<Type.TypeID> PARQUET_UNSUPPORTED_TYPE_IDS = | ||
| Set.of( | ||
| Type.TypeID.TIME, | ||
| Type.TypeID.TIMESTAMP_NANO, | ||
| Type.TypeID.STRUCT, | ||
| Type.TypeID.LIST, | ||
| Type.TypeID.MAP, | ||
| Type.TypeID.GEOMETRY, | ||
| Type.TypeID.GEOGRAPHY); |
There was a problem hiding this comment.
| private static final Set<Type.TypeID> PARQUET_UNSUPPORTED_TYPE_IDS = | |
| Set.of( | |
| Type.TypeID.TIME, | |
| Type.TypeID.TIMESTAMP_NANO, | |
| Type.TypeID.STRUCT, | |
| Type.TypeID.LIST, | |
| Type.TypeID.MAP, | |
| Type.TypeID.GEOMETRY, | |
| Type.TypeID.GEOGRAPHY); | |
| private static final Set<Type.TypeID> PARQUET_UNSUPPORTED_TYPE_IDS = | |
| ImmutableSet.<Type.TypeID>builder() | |
| .addAll(SPARK_UNSUPPORTED_TYPE_IDS) | |
| .addAll(ARROW_UNSUPPORTED_TYPE_IDS) | |
| .build(); |
| @@ -2872,21 +2918,15 @@ private void readAndAssertMetadataColumn( | |||
| InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile(); | |||
| List<T> readRecords; | |||
There was a problem hiding this comment.
No, I have removed it.
| return false; | ||
| } | ||
|
|
||
| protected void assertRecordsEqual( |
There was a problem hiding this comment.
Move this after the @TempDir, or even better, move the TempDir to the top
There was a problem hiding this comment.
I have moved it to the top.
| .recordsPerBatch(100)) | ||
| .hasMessageContaining("Batch reading is not supported") | ||
| .isInstanceOf(UnsupportedOperationException.class); | ||
| if (!supportsBatchReads()) { |
There was a problem hiding this comment.
Could we split this to 2 tests with assumeTrue/assumeFalse to run them only if relevant?
There was a problem hiding this comment.
I have split it into testReaderBuilderRecordsPerBatchNotSupported and testReaderBuilderRecordsPerBatch.
Co-authored-by: Joy Haldar <joy.haldar@target.com>
Adds a TCK subclass for Spark's vectorized
ColumnarBatchread path.TestSparkVectorizedFormatModelextendsReadFormatModelTestsand redefinesFILE_FORMATSandFORMAT_AND_GENERATORto skip Avro, which has noColumnarBatchmodel registered. This required makingDataGenerator,DataGeneratorsandDataGenerators.ALLpublic in the test package.Two base class changes were needed:
assertRecordsEqual()- read tests built expectations viaconvertToEngineRecords, which can't work for a batch type