You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "Guosmilesmile (via GitHub)" <gi...@apache.org> on 2026/04/01 05:12:03 UTC
[PR] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Guosmilesmile opened a new pull request, #15843:
URL: https://github.com/apache/iceberg/pull/15843
Added schema evolution tests to BaseFormatModelTests:
* Adding column (reading with wider schema)
* Projection/Removing column (reading with narrower schema)
* Removing and adding a column with the same name (field ID mismatch)
* Allowed type changes (int→long, float→double)
* Reorder columns
* Rename column (same field ID, different name)
* Required → Optional (nullability widening)
* Read with empty projection (zero columns)
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3110085223
##########
core/src/main/java/org/apache/iceberg/avro/AvroSchemaUtil.java:
##########
@@ -90,7 +90,7 @@ public static org.apache.iceberg.Schema toIceberg(Schema schema) {
return new org.apache.iceberg.Schema(fields);
}
- static boolean hasIds(Schema schema) {
+ public static boolean hasIds(Schema schema) {
Review Comment:
I don't like these changes. We are adding public methods for testing purposes only.
Not an elegant, but possible solution is to create an util class in the `data/srcr/test/java/org/apache/iceberg/avro` directory which is public, but only in test package.
This way we don't expose these methods, but still could use in our tests.
@nastra: Any better ideas?
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3110126092
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -700,4 +1089,164 @@ private static String splitSizeProperty(FileFormat fileFormat) {
"No split size property defined for format: " + fileFormat);
};
}
+
+ private void writeRecordsWithoutFieldIds(
+ FileFormat fileFormat, Schema schema, List<Record> records) throws IOException {
+ switch (fileFormat) {
+ case PARQUET -> writeParquetWithoutFieldIds(schema, records);
+ case AVRO -> writeAvroWithoutFieldIds(schema, records);
+ case ORC -> writeOrcWithoutFieldIds(schema, records);
+ default -> throw new UnsupportedOperationException("Unsupported file format: " + fileFormat);
+ }
+ }
+
+ private void writeAvroWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = RemoveIds.removeIds(schema);
+
+ OutputFile outputFile = encryptedFile.encryptingOutputFile();
+ DatumWriter<GenericData.Record> datumWriter = new GenericDatumWriter<>(avroSchemaWithoutIds);
+ try (OutputStream out = outputFile.create();
+ DataFileWriter<GenericData.Record> writer = new DataFileWriter<>(datumWriter)) {
+ writer.create(avroSchemaWithoutIds, out);
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.append(avroRecord);
+ }
+ }
+
+ try (DataFileStream<GenericData.Record> reader =
+ new DataFileStream<>(outputFile.toInputFile().newStream(), new GenericDatumReader<>())) {
+ assertThat(AvroSchemaUtil.hasIds(reader.getSchema())).isFalse();
+ }
+ }
+
+ private void writeParquetWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = RemoveIds.removeIds(schema);
+
+ Path tempFile = Files.createTempFile("iceberg-test-no-ids-", ".parquet");
+ Files.deleteIfExists(tempFile);
+
+ try {
+ try (ParquetWriter<GenericData.Record> writer =
+ AvroParquetWriter.<GenericData.Record>builder(
+ new org.apache.hadoop.fs.Path(tempFile.toUri()))
+ .withDataModel(GenericData.get())
+ .withSchema(avroSchemaWithoutIds)
+ .withConf(new org.apache.hadoop.conf.Configuration())
+ .build()) {
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.write(avroRecord);
+ }
+ }
+
+ try (ParquetFileReader reader =
+ ParquetFileReader.open(
+ HadoopInputFile.fromPath(
+ new org.apache.hadoop.fs.Path(tempFile.toUri()),
+ new org.apache.hadoop.conf.Configuration()))) {
Review Comment:
import
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3086608868
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(projectedSchema)
+ .engineProjection(engineSchema(projectedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ projectedSchema,
+ convertToEngineRecords(expectedGenericRecords, projectedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column). The NameMapping maps file column name "col_b" to old field ID 2. Since the read schema
+ * expects field ID 6, the new col_b should be null.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionReNameMapping(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", null);
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ readSchema, convertToEngineRecords(expectedGenericRecords, readSchema), readRecords);
+ }
+
+ /**
+ * Schema evolution: Allowed type changes. Write with DefaultSchema (col_b:int, col_d:float), read
+ * with promoted types (col_b:long, col_d:double).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotion(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Read with promoted types: `col_b` int → long, `col_d` float → double
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.LongType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.DoubleType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", ((Integer) record.getField("col_b")).longValue());
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", ((Float) record.getField("col_d")).doubleValue());
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ readSchema, convertToEngineRecords(expectedGenericRecords, readSchema), readRecords);
+ }
+
+ /**
+ * Schema evolution: Reorder columns. Write with DefaultSchema {col_a, col_b, col_c, col_d,
+ * col_e}, read with reordered schema {col_e, col_c, col_a, col_d, col_b}.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionReorderColumns(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema reorderedSchema =
+ new Schema(
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(reorderedSchema)
+ .engineProjection(engineSchema(reorderedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(reorderedSchema);
+ for (Types.NestedField col : reorderedSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ reorderedSchema,
+ convertToEngineRecords(expectedGenericRecords, reorderedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Rename column. Write with DefaultSchema where col_b has field ID 2. Read with
+ * a schema where the same field ID 2 is renamed to "column_b". Since Iceberg binds by field ID,
+ * the renamed column should still read the original data correctly.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionRenameColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // rename col_b(id=2) -> column_b, col_d(id=4) -> column_d
+ Schema renamedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "column_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "column_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(renamedSchema)
+ .engineProjection(engineSchema(renamedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(renamedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("column_b", record.getField("col_b"));
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("column_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ renamedSchema, convertToEngineRecords(expectedGenericRecords, renamedSchema), readRecords);
+ }
+
+ /**
+ * Schema evolution: Required → Optional. Write with DefaultSchema where all columns are required.
+ * Read with a schema where some columns are changed to optional. Iceberg allows widening required
+ * to optional. The data should still be read correctly.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionRequiredToOptional(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // change col_b and col_d to optional
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.optional(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : readSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ return expected;
Review Comment:
Why is this conversion 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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3147308813
##########
spark/v3.4/build.gradle:
##########
@@ -49,6 +49,14 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") {
}
}
+ configurations {
+ parquetTestArtifactsRuntime {
+ canBeConsumed = false
+ canBeResolved = true
+ transitive = false
+ }
+ }
+
Review Comment:
Could you please help me understand why we need 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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3198897572
##########
spark/v3.4/build.gradle:
##########
@@ -49,6 +49,14 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") {
}
}
+ configurations {
+ parquetTestArtifactsRuntime {
+ canBeConsumed = false
+ canBeResolved = true
+ transitive = false
+ }
+ }
+
Review Comment:
Could we move `ParquetWritingTestUtils` from `parquet/src/test/java/` to `parquet/src/testFixtures/java/`? That would let us drop the `parquetTestArtifactsRuntime` workaround in all 4 Spark builds, since the existing `testFixtures(project(':iceberg-parquet'))` dependency would already provide 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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3086489398
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
Review Comment:
Make this based on the `writeSchema`
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3110108792
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -609,6 +649,354 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> evolvedColumns = Lists.newArrayList(writeSchema.columns());
+
+ int maxFieldId =
+ writeSchema.columns().stream().mapToInt(Types.NestedField::fieldId).max().orElse(0);
+ evolvedColumns.add(
+ Types.NestedField.optional("new_string_col")
+ .withId(maxFieldId + 1)
+ .ofType(Types.StringType.get())
+ .build());
+ evolvedColumns.add(
+ Types.NestedField.optional("new_int_col")
+ .withId(maxFieldId + 2)
+ .ofType(Types.IntegerType.get())
+ .build());
+ Schema readSchema = new Schema(evolvedColumns);
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, readSchema, expectedGenericRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> writeColumns = writeSchema.columns();
+ assumeThat(writeColumns).hasSizeGreaterThanOrEqualTo(2);
+ Schema projectedSchema =
+ new Schema(writeColumns.get(0), writeColumns.get(writeColumns.size() - 1));
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ for (Types.NestedField col : projectedSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, projectedSchema, expectedGenericRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionDropAndReAddSameNameColumn(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", null);
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, readSchema, expectedGenericRecords);
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionIntToLong(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.IntegerType.get(),
+ Types.LongType.get(),
+ value -> value == null ? null : ((Integer) value).longValue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionFloatToDouble(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.FloatType.get(),
+ Types.DoubleType.get(),
+ value -> value == null ? null : ((Float) value).doubleValue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionDecimalPrecision(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.DecimalType.of(9, 2),
+ Types.DecimalType.of(18, 2),
+ java.util.function.Function.identity());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionDateToTimestamp(FileFormat fileFormat) throws IOException {
+ assumeThat(TypeUtil.isPromotionAllowed(Types.DateType.get(), Types.TimestampType.withoutZone()))
Review Comment:
Why do we have here?
The Spec says that it should be allowed. So we should fail if we change the spec.
And when we change the spec, then we should change the tests to. Using `assume` could easily hide the issue if the `isPromotionAllowed` implementation is changed
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3086548798
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(projectedSchema)
+ .engineProjection(engineSchema(projectedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ projectedSchema,
+ convertToEngineRecords(expectedGenericRecords, projectedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column). The NameMapping maps file column name "col_b" to old field ID 2. Since the read schema
+ * expects field ID 6, the new col_b should be null.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionReNameMapping(FileFormat fileFormat) throws IOException {
Review Comment:
Rename the test to describe that we test readding the column with a same name.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3086496444
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
Review Comment:
These tests are getting very similar with a few method injection points.
Could we refactor them a bit, so they are more generic and the code duplication is avioded?
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "psvri (via GitHub)" <gi...@apache.org>.
psvri commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3098729948
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(projectedSchema)
+ .engineProjection(engineSchema(projectedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ projectedSchema,
+ convertToEngineRecords(expectedGenericRecords, projectedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column). The NameMapping maps file column name "col_b" to old field ID 2. Since the read schema
+ * expects field ID 6, the new col_b should be null.
Review Comment:
I too feel the same. We should have separate tests which uses name mapping as part of TCK.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3086589796
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(projectedSchema)
+ .engineProjection(engineSchema(projectedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ projectedSchema,
+ convertToEngineRecords(expectedGenericRecords, projectedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column). The NameMapping maps file column name "col_b" to old field ID 2. Since the read schema
+ * expects field ID 6, the new col_b should be null.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionReNameMapping(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", null);
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ readSchema, convertToEngineRecords(expectedGenericRecords, readSchema), readRecords);
+ }
+
+ /**
+ * Schema evolution: Allowed type changes. Write with DefaultSchema (col_b:int, col_d:float), read
+ * with promoted types (col_b:long, col_d:double).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotion(FileFormat fileFormat) throws IOException {
Review Comment:
Again, don't duplicate code, just use a private method
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3129978833
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -700,4 +1041,165 @@ private static String splitSizeProperty(FileFormat fileFormat) {
"No split size property defined for format: " + fileFormat);
};
}
+
+ private void writeRecordsWithoutFieldIds(
+ FileFormat fileFormat, Schema schema, List<Record> records) throws IOException {
+ switch (fileFormat) {
+ case PARQUET -> writeParquetWithoutFieldIds(schema, records);
+ case AVRO -> writeAvroWithoutFieldIds(schema, records);
+ case ORC -> writeOrcWithoutFieldIds(schema, records);
+ default -> throw new UnsupportedOperationException("Unsupported file format: " + fileFormat);
+ }
+ }
+
+ private void writeAvroWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = AvroTestHelpers.removeIds(schema);
+
+ OutputFile outputFile = encryptedFile.encryptingOutputFile();
+ DatumWriter<GenericData.Record> datumWriter = new GenericDatumWriter<>(avroSchemaWithoutIds);
+ try (OutputStream out = outputFile.create();
+ DataFileWriter<GenericData.Record> writer = new DataFileWriter<>(datumWriter)) {
+ writer.create(avroSchemaWithoutIds, out);
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.append(avroRecord);
+ }
+ }
+
+ try (DataFileStream<GenericData.Record> reader =
+ new DataFileStream<>(outputFile.toInputFile().newStream(), new GenericDatumReader<>())) {
+ assertThat(AvroTestHelpers.hasIds(reader.getSchema())).isFalse();
+ }
+ }
+
+ private void writeParquetWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = AvroTestHelpers.removeIds(schema);
+
+ Path tempFile = Files.createTempFile("iceberg-test-no-ids-", ".parquet");
+ Files.deleteIfExists(tempFile);
+
+ try {
+ try (ParquetWriter<GenericData.Record> writer =
+ AvroParquetWriter.<GenericData.Record>builder(
+ new org.apache.hadoop.fs.Path(tempFile.toUri()))
+ .withDataModel(GenericData.get())
+ .withSchema(avroSchemaWithoutIds)
+ .withConf(new Configuration())
+ .build()) {
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.write(avroRecord);
+ }
+ }
+
+ try (ParquetFileReader reader =
+ ParquetFileReader.open(
+ HadoopInputFile.fromPath(
+ new org.apache.hadoop.fs.Path(tempFile.toUri()), new Configuration()))) {
+ assertThat(ParquetSchemaUtil.hasIds(reader.getFooter().getFileMetaData().getSchema()))
+ .isFalse();
+ }
+
+ byte[] bytes = Files.readAllBytes(tempFile);
+ try (OutputStream out = encryptedFile.encryptingOutputFile().create()) {
+ out.write(bytes);
+ }
+ } finally {
+ Files.deleteIfExists(tempFile);
+ }
+ }
+
+ private void writeOrcWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ TypeDescription typeWithIds = ORCSchemaUtil.convert(schema);
+ TypeDescription typeWithoutIds = TestORCSchemaUtil.removeIds(typeWithIds);
+
+ Configuration conf = new Configuration();
+ OrcFile.WriterOptions options =
+ OrcFile.writerOptions(conf).useUTCTimestamp(true).setSchema(typeWithoutIds);
+
+ OrcRowWriter<Record> rowWriter = GenericOrcWriter.buildWriter(schema, typeWithIds);
+
+ Path tempFile = Files.createTempFile("iceberg-test-no-ids-", ".orc");
+ Files.deleteIfExists(tempFile);
Review Comment:
Same question as above
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3129463044
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -609,6 +649,354 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> evolvedColumns = Lists.newArrayList(writeSchema.columns());
+
+ int maxFieldId =
+ writeSchema.columns().stream().mapToInt(Types.NestedField::fieldId).max().orElse(0);
+ evolvedColumns.add(
+ Types.NestedField.optional("new_string_col")
+ .withId(maxFieldId + 1)
+ .ofType(Types.StringType.get())
+ .build());
+ evolvedColumns.add(
+ Types.NestedField.optional("new_int_col")
+ .withId(maxFieldId + 2)
+ .ofType(Types.IntegerType.get())
+ .build());
+ Schema readSchema = new Schema(evolvedColumns);
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, readSchema, expectedGenericRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> writeColumns = writeSchema.columns();
+ assumeThat(writeColumns).hasSizeGreaterThanOrEqualTo(2);
+ Schema projectedSchema =
+ new Schema(writeColumns.get(0), writeColumns.get(writeColumns.size() - 1));
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ for (Types.NestedField col : projectedSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, projectedSchema, expectedGenericRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionDropAndReAddSameNameColumn(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", null);
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, readSchema, expectedGenericRecords);
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionIntToLong(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.IntegerType.get(),
+ Types.LongType.get(),
+ value -> value == null ? null : ((Integer) value).longValue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionFloatToDouble(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.FloatType.get(),
+ Types.DoubleType.get(),
+ value -> value == null ? null : ((Float) value).doubleValue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionDecimalPrecision(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.DecimalType.of(9, 2),
+ Types.DecimalType.of(18, 2),
+ java.util.function.Function.identity());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionDateToTimestamp(FileFormat fileFormat) throws IOException {
+ assumeThat(TypeUtil.isPromotionAllowed(Types.DateType.get(), Types.TimestampType.withoutZone()))
Review Comment:
Raise a issue to mark it. https://github.com/apache/iceberg/issues/16083
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3123564975
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -700,4 +1089,164 @@ private static String splitSizeProperty(FileFormat fileFormat) {
"No split size property defined for format: " + fileFormat);
};
}
+
+ private void writeRecordsWithoutFieldIds(
+ FileFormat fileFormat, Schema schema, List<Record> records) throws IOException {
+ switch (fileFormat) {
+ case PARQUET -> writeParquetWithoutFieldIds(schema, records);
+ case AVRO -> writeAvroWithoutFieldIds(schema, records);
+ case ORC -> writeOrcWithoutFieldIds(schema, records);
+ default -> throw new UnsupportedOperationException("Unsupported file format: " + fileFormat);
+ }
+ }
+
+ private void writeAvroWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = RemoveIds.removeIds(schema);
+
+ OutputFile outputFile = encryptedFile.encryptingOutputFile();
+ DatumWriter<GenericData.Record> datumWriter = new GenericDatumWriter<>(avroSchemaWithoutIds);
+ try (OutputStream out = outputFile.create();
+ DataFileWriter<GenericData.Record> writer = new DataFileWriter<>(datumWriter)) {
+ writer.create(avroSchemaWithoutIds, out);
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.append(avroRecord);
+ }
+ }
+
+ try (DataFileStream<GenericData.Record> reader =
+ new DataFileStream<>(outputFile.toInputFile().newStream(), new GenericDatumReader<>())) {
+ assertThat(AvroSchemaUtil.hasIds(reader.getSchema())).isFalse();
+ }
+ }
+
+ private void writeParquetWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = RemoveIds.removeIds(schema);
+
+ Path tempFile = Files.createTempFile("iceberg-test-no-ids-", ".parquet");
+ Files.deleteIfExists(tempFile);
+
+ try {
+ try (ParquetWriter<GenericData.Record> writer =
+ AvroParquetWriter.<GenericData.Record>builder(
+ new org.apache.hadoop.fs.Path(tempFile.toUri()))
+ .withDataModel(GenericData.get())
+ .withSchema(avroSchemaWithoutIds)
+ .withConf(new org.apache.hadoop.conf.Configuration())
+ .build()) {
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.write(avroRecord);
+ }
+ }
+
+ try (ParquetFileReader reader =
+ ParquetFileReader.open(
+ HadoopInputFile.fromPath(
+ new org.apache.hadoop.fs.Path(tempFile.toUri()),
+ new org.apache.hadoop.conf.Configuration()))) {
+ assertThat(ParquetSchemaUtil.hasIds(reader.getFooter().getFileMetaData().getSchema()))
+ .isFalse();
+ }
+
+ byte[] bytes = Files.readAllBytes(tempFile);
+ try (OutputStream out = encryptedFile.encryptingOutputFile().create()) {
+ out.write(bytes);
+ }
+ } finally {
+ Files.deleteIfExists(tempFile);
+ }
+ }
+
+ private void writeOrcWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ TypeDescription typeWithIds = ORCSchemaUtil.convert(schema);
+ TypeDescription typeWithoutIds = ORCSchemaUtil.removeIds(typeWithIds);
+
+ Configuration conf = new Configuration();
+ OrcFile.WriterOptions options =
+ OrcFile.writerOptions(conf).useUTCTimestamp(true).setSchema(typeWithoutIds);
+
+ OrcRowWriter<Record> rowWriter = GenericOrcWriter.buildWriter(schema, typeWithIds);
+
+ Path tempFile = Files.createTempFile("iceberg-test-no-ids-", ".orc");
+ Files.deleteIfExists(tempFile);
+ org.apache.hadoop.fs.Path hadoopPath = new org.apache.hadoop.fs.Path(tempFile.toUri());
+
+ try {
+ try (Writer orcWriter = OrcFile.createWriter(hadoopPath, options)) {
+ VectorizedRowBatch batch = typeWithoutIds.createRowBatch();
+ for (Record record : records) {
+ rowWriter.write(record, batch);
+ if (batch.size == batch.getMaxSize()) {
+ orcWriter.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+
+ if (batch.size > 0) {
+ orcWriter.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+
+ try (Reader reader = OrcFile.createReader(hadoopPath, OrcFile.readerOptions(conf))) {
+ assertThat(ORCSchemaUtil.hasIds(reader.getSchema())).isFalse();
+ }
+
+ byte[] bytes = Files.readAllBytes(tempFile);
+ try (OutputStream out = encryptedFile.encryptingOutputFile().create()) {
+ out.write(bytes);
+ }
+ } finally {
+ Files.deleteIfExists(tempFile);
+ }
+ }
+
+ private void runTypePromotionCheck(
+ FileFormat fileFormat, Type fromType, Type toType, Function<Object, Object> promoteValue)
+ throws IOException {
+ String columnName = "col";
+ Schema writeSchema = new Schema(Types.NestedField.required(1, columnName, fromType));
+ Schema readSchema = new Schema(Types.NestedField.required(1, columnName, toType));
+
+ List<Record> genericRecords = RandomGenericData.generate(writeSchema, 10, 1L);
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField(columnName, promoteValue.apply(record.getField(columnName)));
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, readSchema, expectedGenericRecords);
Review Comment:
Ok,keepo the samp style.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3105371911
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(projectedSchema)
+ .engineProjection(engineSchema(projectedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ projectedSchema,
+ convertToEngineRecords(expectedGenericRecords, projectedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column). The NameMapping maps file column name "col_b" to old field ID 2. Since the read schema
+ * expects field ID 6, the new col_b should be null.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionReNameMapping(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", null);
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ readSchema, convertToEngineRecords(expectedGenericRecords, readSchema), readRecords);
+ }
+
+ /**
+ * Schema evolution: Allowed type changes. Write with DefaultSchema (col_b:int, col_d:float), read
+ * with promoted types (col_b:long, col_d:double).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotion(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Read with promoted types: `col_b` int → long, `col_d` float → double
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.LongType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.DoubleType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", ((Integer) record.getField("col_b")).longValue());
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", ((Float) record.getField("col_d")).doubleValue());
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ readSchema, convertToEngineRecords(expectedGenericRecords, readSchema), readRecords);
+ }
+
+ /**
+ * Schema evolution: Reorder columns. Write with DefaultSchema {col_a, col_b, col_c, col_d,
+ * col_e}, read with reordered schema {col_e, col_c, col_a, col_d, col_b}.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionReorderColumns(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema reorderedSchema =
+ new Schema(
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(reorderedSchema)
+ .engineProjection(engineSchema(reorderedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(reorderedSchema);
+ for (Types.NestedField col : reorderedSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ reorderedSchema,
+ convertToEngineRecords(expectedGenericRecords, reorderedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Rename column. Write with DefaultSchema where col_b has field ID 2. Read with
+ * a schema where the same field ID 2 is renamed to "column_b". Since Iceberg binds by field ID,
+ * the renamed column should still read the original data correctly.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionRenameColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // rename col_b(id=2) -> column_b, col_d(id=4) -> column_d
+ Schema renamedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "column_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "column_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(renamedSchema)
+ .engineProjection(engineSchema(renamedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(renamedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("column_b", record.getField("col_b"));
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("column_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ renamedSchema, convertToEngineRecords(expectedGenericRecords, renamedSchema), readRecords);
+ }
+
+ /**
+ * Schema evolution: Required → Optional. Write with DefaultSchema where all columns are required.
+ * Read with a schema where some columns are changed to optional. Iceberg allows widening required
+ * to optional. The data should still be read correctly.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionRequiredToOptional(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // change col_b and col_d to optional
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.optional(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : readSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ return expected;
Review Comment:
We need to reorder the original data according to the `reorderSchema` to get the desired data.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3105364001
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(projectedSchema)
+ .engineProjection(engineSchema(projectedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ projectedSchema,
+ convertToEngineRecords(expectedGenericRecords, projectedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column). The NameMapping maps file column name "col_b" to old field ID 2. Since the read schema
+ * expects field ID 6, the new col_b should be null.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionReNameMapping(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
Review Comment:
Yes, remove it now.
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
Review Comment:
Ok, base on writeSchema
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3105364493
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
Review Comment:
Got 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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3110088487
##########
orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java:
##########
@@ -515,11 +515,11 @@ static boolean isOptional(TypeDescription orcType) {
return true;
}
- static TypeDescription removeIds(TypeDescription type) {
+ public static TypeDescription removeIds(TypeDescription type) {
Review Comment:
don't add anything to the public API
##########
orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java:
##########
@@ -515,11 +515,11 @@ static boolean isOptional(TypeDescription orcType) {
return true;
}
- static TypeDescription removeIds(TypeDescription type) {
+ public static TypeDescription removeIds(TypeDescription type) {
return OrcSchemaVisitor.visit(type, new RemoveIds());
}
- static boolean hasIds(TypeDescription orcSchema) {
+ public static boolean hasIds(TypeDescription orcSchema) {
Review Comment:
don't add anything to the public API
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3147687820
##########
spark/v3.4/build.gradle:
##########
@@ -49,6 +49,14 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") {
}
}
+ configurations {
+ parquetTestArtifactsRuntime {
+ canBeConsumed = false
+ canBeResolved = true
+ transitive = false
+ }
+ }
+
Review Comment:
When I add `testImplementation (project(path: ':iceberg-parquet', configuration: 'testFixturesRuntimeElements'))` in spark I meet a error.
```> Module 'org.apache.iceberg:iceberg-parquet' has been rejected:
Cannot select module with conflict on capability 'org.apache.iceberg:iceberg-parquet-test-fixtures:1.10.0-SNAPSHOT' also provided by ['project :iceberg-parquet' (testFixturesApiElements), 'project :iceberg-parquet' (testFixturesRuntimeElements)]
```
Spark tests already compile against the main :iceberg-parquet artifact and its testFixtures, but some test code/resources from parquet are only needed at execution time. Keeping them in a separate runtime-only configuration avoids exposing parquet's internal test classes on the test compile classpath, and transitive = false makes sure we only pull in that test jar itself rather than an extra dependency graph.
parquetTestArtifactsRuntime is an internal, resolvable-only configuration. canBeConsumed = false means it is not published for other projects to consume, canBeResolved = true allows this project to resolve it into files and append them to the test classpath, and transitive = false ensures we only pull the :iceberg-parquet testArtifacts jar itself rather than its full transitive dependency graph.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3142564573
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -1330,6 +1357,288 @@ private void readAndAssertGenericRecords(
readAndAssertGenericRecords(fileFormat, schema, sourceRecords.stream().map(transform).toList());
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> evolvedColumns = Lists.newArrayList(writeSchema.columns());
+
+ int maxFieldId =
+ writeSchema.columns().stream().mapToInt(Types.NestedField::fieldId).max().orElse(0);
+ evolvedColumns.add(
+ Types.NestedField.optional("new_string_col")
+ .withId(maxFieldId + 1)
+ .ofType(Types.StringType.get())
+ .build());
+ evolvedColumns.add(
+ Types.NestedField.optional("new_int_col")
+ .withId(maxFieldId + 2)
+ .ofType(Types.IntegerType.get())
+ .build());
+ Schema readSchema = new Schema(evolvedColumns);
+ readAndAssertEngineRecords(
+ fileFormat,
+ readSchema,
+ genericRecords,
+ record -> {
+ Record expected = copy(record, writeSchema, readSchema);
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ });
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> writeColumns = writeSchema.columns();
+ assumeThat(writeColumns).hasSizeGreaterThanOrEqualTo(2);
+ Schema projectedSchema =
+ new Schema(writeColumns.get(0), writeColumns.get(writeColumns.size() - 1));
+
+ readAndAssertEngineRecords(
+ fileFormat,
+ projectedSchema,
+ genericRecords,
+ record -> copy(record, projectedSchema, projectedSchema));
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionDropAndReAddSameNameColumn(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ readAndAssertEngineRecords(
+ fileFormat,
+ readSchema,
+ genericRecords,
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", null);
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ });
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionIntToLong(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.IntegerType.get(),
+ Types.LongType.get(),
+ value -> value == null ? null : ((Integer) value).longValue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionFloatToDouble(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.FloatType.get(),
+ Types.DoubleType.get(),
+ value -> value == null ? null : ((Float) value).doubleValue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionDecimalPrecision(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.DecimalType.of(9, 2),
+ Types.DecimalType.of(18, 2),
+ java.util.function.Function.identity());
Review Comment:
Import
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3086585989
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(projectedSchema)
+ .engineProjection(engineSchema(projectedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ projectedSchema,
+ convertToEngineRecords(expectedGenericRecords, projectedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column). The NameMapping maps file column name "col_b" to old field ID 2. Since the read schema
+ * expects field ID 6, the new col_b should be null.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionReNameMapping(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", null);
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ readSchema, convertToEngineRecords(expectedGenericRecords, readSchema), readRecords);
+ }
+
+ /**
+ * Schema evolution: Allowed type changes. Write with DefaultSchema (col_b:int, col_d:float), read
+ * with promoted types (col_b:long, col_d:double).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotion(FileFormat fileFormat) throws IOException {
Review Comment:
We should create different tests for the different allowed type promotions.
Integer to Long: int → long
Float to Double: float → double
Decimal Precision: decimal(P, S) → decimal(P', S) where (only increases the total precision, not the scale)
Date to Timestamp (v3+): date → timestamp or timestamp_ns
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3086544285
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(projectedSchema)
+ .engineProjection(engineSchema(projectedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ projectedSchema,
+ convertToEngineRecords(expectedGenericRecords, projectedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column). The NameMapping maps file column name "col_b" to old field ID 2. Since the read schema
+ * expects field ID 6, the new col_b should be null.
Review Comment:
The `NameMapping` is a different feature. It allows the user to read files written outside Iceberg to map the columns to the Iceberg fieldIds, so this last sentence is not needed and it is incorrect.
OTOH we need a test for NameMapping where we read a Parquet/ORC/Avro file without Iceberg metadata (not written by Iceberg writers). And we should be able to map the column names to the Iceberg fields using the NameMapping
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3110087033
##########
core/src/main/java/org/apache/iceberg/avro/RemoveIds.java:
##########
@@ -87,7 +87,7 @@ private static Schema.Field copyField(Schema.Field field, Schema newSchema) {
return copy;
}
- static Schema removeIds(org.apache.iceberg.Schema schema) {
+ public static Schema removeIds(org.apache.iceberg.Schema schema) {
Review Comment:
don't add anything to the public API
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3123557042
##########
core/src/main/java/org/apache/iceberg/avro/AvroSchemaUtil.java:
##########
@@ -90,7 +90,7 @@ public static org.apache.iceberg.Schema toIceberg(Schema schema) {
return new org.apache.iceberg.Schema(fields);
}
- static boolean hasIds(Schema schema) {
+ public static boolean hasIds(Schema schema) {
Review Comment:
I saw that many of the methods in these utility classes are static, which made me think I could just add them directly. I've now revised my approach by adding these methods to the utility classes under the test package of the corresponding module, avoiding any changes to the production code.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#issuecomment-4419835728
Merged to main.
Thanks @Guosmilesmile for the PR and @huaxingao for the review!
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3132080793
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -609,6 +640,315 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> evolvedColumns = Lists.newArrayList(writeSchema.columns());
+
+ int maxFieldId =
+ writeSchema.columns().stream().mapToInt(Types.NestedField::fieldId).max().orElse(0);
+ evolvedColumns.add(
+ Types.NestedField.optional("new_string_col")
+ .withId(maxFieldId + 1)
+ .ofType(Types.StringType.get())
+ .build());
+ evolvedColumns.add(
+ Types.NestedField.optional("new_int_col")
+ .withId(maxFieldId + 2)
+ .ofType(Types.IntegerType.get())
+ .build());
+ Schema readSchema = new Schema(evolvedColumns);
+ readAndAssertEngineRecords(
+ fileFormat,
+ readSchema,
+ genericRecords,
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
Review Comment:
Good suggestion, make it simple.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3110095592
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -73,6 +108,11 @@ public abstract class BaseFormatModelTests<T> {
protected abstract void assertEquals(Schema schema, List<T> expected, List<T> actual);
+ protected Object convertConstantToEngine(Type type, Object value) {
+ return null;
Review Comment:
should this throw an exception as a default implementation?
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3199602893
##########
spark/v3.4/build.gradle:
##########
@@ -49,6 +49,14 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") {
}
}
+ configurations {
+ parquetTestArtifactsRuntime {
+ canBeConsumed = false
+ canBeResolved = true
+ transitive = false
+ }
+ }
+
Review Comment:
@huaxingao @pvary Thanks for the suggestion, I have made some change. If you have time,help to review it again.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3148259789
##########
spark/v3.4/build.gradle:
##########
@@ -105,8 +113,12 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") {
testImplementation project(path: ':iceberg-api', configuration: 'testArtifacts')
testImplementation project(path: ':iceberg-core', configuration: 'testArtifacts')
testImplementation project(path: ':iceberg-data', configuration: 'testArtifacts')
+ testImplementation project(path: ':iceberg-orc', configuration: 'testArtifacts')
testImplementation (project(path: ':iceberg-open-api', configuration: 'testFixturesRuntimeElements'))
testImplementation libs.awaitility
+ testImplementation(testFixtures(project(':iceberg-parquet')))
Review Comment:
Only 3.4 need this. Other version has this code.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3148609705
##########
spark/v3.4/build.gradle:
##########
@@ -49,6 +49,14 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") {
}
}
+ configurations {
+ parquetTestArtifactsRuntime {
+ canBeConsumed = false
+ canBeResolved = true
+ transitive = false
+ }
+ }
+
Review Comment:
Ok. I get it now, and I don't have a better solution.
@huaxingao: Do you think this change could be ok for the Spark code? I don't see any issues, but I would like to have someone from more Spark knowledge to take a look.
Thanks,
Peter
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3123563753
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -700,4 +1089,164 @@ private static String splitSizeProperty(FileFormat fileFormat) {
"No split size property defined for format: " + fileFormat);
};
}
+
+ private void writeRecordsWithoutFieldIds(
+ FileFormat fileFormat, Schema schema, List<Record> records) throws IOException {
+ switch (fileFormat) {
+ case PARQUET -> writeParquetWithoutFieldIds(schema, records);
+ case AVRO -> writeAvroWithoutFieldIds(schema, records);
+ case ORC -> writeOrcWithoutFieldIds(schema, records);
+ default -> throw new UnsupportedOperationException("Unsupported file format: " + fileFormat);
+ }
+ }
+
+ private void writeAvroWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = RemoveIds.removeIds(schema);
+
+ OutputFile outputFile = encryptedFile.encryptingOutputFile();
+ DatumWriter<GenericData.Record> datumWriter = new GenericDatumWriter<>(avroSchemaWithoutIds);
+ try (OutputStream out = outputFile.create();
+ DataFileWriter<GenericData.Record> writer = new DataFileWriter<>(datumWriter)) {
+ writer.create(avroSchemaWithoutIds, out);
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.append(avroRecord);
+ }
+ }
+
+ try (DataFileStream<GenericData.Record> reader =
+ new DataFileStream<>(outputFile.toInputFile().newStream(), new GenericDatumReader<>())) {
+ assertThat(AvroSchemaUtil.hasIds(reader.getSchema())).isFalse();
+ }
+ }
+
+ private void writeParquetWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = RemoveIds.removeIds(schema);
+
+ Path tempFile = Files.createTempFile("iceberg-test-no-ids-", ".parquet");
+ Files.deleteIfExists(tempFile);
+
+ try {
+ try (ParquetWriter<GenericData.Record> writer =
+ AvroParquetWriter.<GenericData.Record>builder(
+ new org.apache.hadoop.fs.Path(tempFile.toUri()))
+ .withDataModel(GenericData.get())
+ .withSchema(avroSchemaWithoutIds)
+ .withConf(new org.apache.hadoop.conf.Configuration())
+ .build()) {
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.write(avroRecord);
+ }
+ }
+
+ try (ParquetFileReader reader =
+ ParquetFileReader.open(
+ HadoopInputFile.fromPath(
+ new org.apache.hadoop.fs.Path(tempFile.toUri()),
+ new org.apache.hadoop.conf.Configuration()))) {
Review Comment:
Path need, but Configuration don't, remove 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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3207336678
##########
parquet/src/testFixtures/java/org/apache/iceberg/parquet/TestParquetWritingUtils.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import org.apache.parquet.io.InputFile;
+import org.apache.parquet.io.OutputFile;
+
+/** Utilities for tests that need to write Parquet files. */
+public class TestParquetWritingUtils {
Review Comment:
Since we have the same name `ParquetWritingTestUtils` in test package, I rename to `ParquetFileTestUtils`
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3110090749
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -73,6 +108,11 @@ public abstract class BaseFormatModelTests<T> {
protected abstract void assertEquals(Schema schema, List<T> expected, List<T> actual);
+ protected Object convertConstantToEngine(Type type, Object value) {
+ return null;
+ }
+ ;
Review Comment:
Remove `;`
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3111151754
##########
core/src/main/java/org/apache/iceberg/avro/AvroSchemaUtil.java:
##########
@@ -90,7 +90,7 @@ public static org.apache.iceberg.Schema toIceberg(Schema schema) {
return new org.apache.iceberg.Schema(fields);
}
- static boolean hasIds(Schema schema) {
+ public static boolean hasIds(Schema schema) {
Review Comment:
I guess the other alternative could be to move `writeAvroWithoutFieldIds` into a separate class under the avro package, so that you could access hasIds() from there. Same for the ORC case
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3107125741
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(projectedSchema)
+ .engineProjection(engineSchema(projectedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ projectedSchema,
+ convertToEngineRecords(expectedGenericRecords, projectedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column). The NameMapping maps file column name "col_b" to old field ID 2. Since the read schema
+ * expects field ID 6, the new col_b should be null.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionReNameMapping(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", null);
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ readSchema, convertToEngineRecords(expectedGenericRecords, readSchema), readRecords);
+ }
+
+ /**
+ * Schema evolution: Allowed type changes. Write with DefaultSchema (col_b:int, col_d:float), read
+ * with promoted types (col_b:long, col_d:double).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotion(FileFormat fileFormat) throws IOException {
Review Comment:
> Date to Timestamp (v3+): date → timestamp or timestamp_ns
Added `testSchemaEvolutionTypePromotionDateToTimestamp` and `testSchemaEvolutionTypePromotionDateToTimestampNano`. Currently, these two test cases are failing, and the behavior of Flink and Spark are completely different. Moreover, this type of conversion is not supported in `TypeUtil.isPromotionAllowed`, but V3 types do support this conversion. So I'm wondering and suspecting that this is implemented at the engine layer rather than the file format layer.
Still investigating, feel free to add any additional information if you have any.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3107127694
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
Review Comment:
Yes, add a private method to refactor 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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3129947686
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -609,6 +640,315 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> evolvedColumns = Lists.newArrayList(writeSchema.columns());
+
+ int maxFieldId =
+ writeSchema.columns().stream().mapToInt(Types.NestedField::fieldId).max().orElse(0);
+ evolvedColumns.add(
+ Types.NestedField.optional("new_string_col")
+ .withId(maxFieldId + 1)
+ .ofType(Types.StringType.get())
+ .build());
+ evolvedColumns.add(
+ Types.NestedField.optional("new_int_col")
+ .withId(maxFieldId + 2)
+ .ofType(Types.IntegerType.get())
+ .build());
+ Schema readSchema = new Schema(evolvedColumns);
+ readAndAssertEngineRecords(
+ fileFormat,
+ readSchema,
+ genericRecords,
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
Review Comment:
We could create a method like this:
```
private Record copy(Record source, Schema sourceSchema, Schema targetSchema) {
Record result = GenericRecord.create(targetSchema);
for (Types.NestedField col : sourceSchema.columns()) {
result.setField(col.name(), record.getField(col.name()));
}
return result;
}
```
And use it multiple places. Here we could use like:
```suggestion
Record expected = copy(record, writeSchema, readSchema);
expected.setField("new_string_col", null);
expected.setField("new_int_col", null);
return expected;
```
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3129977168
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -700,4 +1041,165 @@ private static String splitSizeProperty(FileFormat fileFormat) {
"No split size property defined for format: " + fileFormat);
};
}
+
+ private void writeRecordsWithoutFieldIds(
+ FileFormat fileFormat, Schema schema, List<Record> records) throws IOException {
+ switch (fileFormat) {
+ case PARQUET -> writeParquetWithoutFieldIds(schema, records);
+ case AVRO -> writeAvroWithoutFieldIds(schema, records);
+ case ORC -> writeOrcWithoutFieldIds(schema, records);
+ default -> throw new UnsupportedOperationException("Unsupported file format: " + fileFormat);
+ }
+ }
+
+ private void writeAvroWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = AvroTestHelpers.removeIds(schema);
+
+ OutputFile outputFile = encryptedFile.encryptingOutputFile();
+ DatumWriter<GenericData.Record> datumWriter = new GenericDatumWriter<>(avroSchemaWithoutIds);
+ try (OutputStream out = outputFile.create();
+ DataFileWriter<GenericData.Record> writer = new DataFileWriter<>(datumWriter)) {
+ writer.create(avroSchemaWithoutIds, out);
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.append(avroRecord);
+ }
+ }
+
+ try (DataFileStream<GenericData.Record> reader =
+ new DataFileStream<>(outputFile.toInputFile().newStream(), new GenericDatumReader<>())) {
+ assertThat(AvroTestHelpers.hasIds(reader.getSchema())).isFalse();
+ }
+ }
+
+ private void writeParquetWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = AvroTestHelpers.removeIds(schema);
+
+ Path tempFile = Files.createTempFile("iceberg-test-no-ids-", ".parquet");
+ Files.deleteIfExists(tempFile);
Review Comment:
Why do we do this double writing? Could we just write the result to the `encryptedFile` immediately?
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3148267423
##########
spark/v3.4/build.gradle:
##########
@@ -49,6 +49,14 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") {
}
}
+ configurations {
+ parquetTestArtifactsRuntime {
+ canBeConsumed = false
+ canBeResolved = true
+ transitive = false
+ }
+ }
+
Review Comment:
I replace the code with above line. Get the same error .
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #15843:
URL: https://github.com/apache/iceberg/pull/15843
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3105365012
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(projectedSchema)
+ .engineProjection(engineSchema(projectedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ projectedSchema,
+ convertToEngineRecords(expectedGenericRecords, projectedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column). The NameMapping maps file column name "col_b" to old field ID 2. Since the read schema
+ * expects field ID 6, the new col_b should be null.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionReNameMapping(FileFormat fileFormat) throws IOException {
Review Comment:
Rename to testSchemaEvolutionDropAndReAddSameNameColumn.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3142725293
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -1330,6 +1357,288 @@ private void readAndAssertGenericRecords(
readAndAssertGenericRecords(fileFormat, schema, sourceRecords.stream().map(transform).toList());
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> evolvedColumns = Lists.newArrayList(writeSchema.columns());
+
+ int maxFieldId =
+ writeSchema.columns().stream().mapToInt(Types.NestedField::fieldId).max().orElse(0);
+ evolvedColumns.add(
+ Types.NestedField.optional("new_string_col")
+ .withId(maxFieldId + 1)
+ .ofType(Types.StringType.get())
+ .build());
+ evolvedColumns.add(
+ Types.NestedField.optional("new_int_col")
+ .withId(maxFieldId + 2)
+ .ofType(Types.IntegerType.get())
+ .build());
+ Schema readSchema = new Schema(evolvedColumns);
+ readAndAssertEngineRecords(
+ fileFormat,
+ readSchema,
+ genericRecords,
+ record -> {
+ Record expected = copy(record, writeSchema, readSchema);
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ });
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> writeColumns = writeSchema.columns();
+ assumeThat(writeColumns).hasSizeGreaterThanOrEqualTo(2);
+ Schema projectedSchema =
+ new Schema(writeColumns.get(0), writeColumns.get(writeColumns.size() - 1));
+
+ readAndAssertEngineRecords(
+ fileFormat,
+ projectedSchema,
+ genericRecords,
+ record -> copy(record, projectedSchema, projectedSchema));
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
Review Comment:
My bed, Clearn it now.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3107127035
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(projectedSchema)
+ .engineProjection(engineSchema(projectedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ projectedSchema,
+ convertToEngineRecords(expectedGenericRecords, projectedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column). The NameMapping maps file column name "col_b" to old field ID 2. Since the read schema
+ * expects field ID 6, the new col_b should be null.
Review Comment:
This part is complex, and the validation needs to be implemented by writing externally without the id. I have added a UT, please take a look when you have time.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3111088600
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -609,6 +649,354 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> evolvedColumns = Lists.newArrayList(writeSchema.columns());
+
+ int maxFieldId =
+ writeSchema.columns().stream().mapToInt(Types.NestedField::fieldId).max().orElse(0);
+ evolvedColumns.add(
+ Types.NestedField.optional("new_string_col")
+ .withId(maxFieldId + 1)
+ .ofType(Types.StringType.get())
+ .build());
+ evolvedColumns.add(
+ Types.NestedField.optional("new_int_col")
+ .withId(maxFieldId + 2)
+ .ofType(Types.IntegerType.get())
+ .build());
+ Schema readSchema = new Schema(evolvedColumns);
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, readSchema, expectedGenericRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> writeColumns = writeSchema.columns();
+ assumeThat(writeColumns).hasSizeGreaterThanOrEqualTo(2);
+ Schema projectedSchema =
+ new Schema(writeColumns.get(0), writeColumns.get(writeColumns.size() - 1));
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ for (Types.NestedField col : projectedSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, projectedSchema, expectedGenericRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionDropAndReAddSameNameColumn(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", null);
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, readSchema, expectedGenericRecords);
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionIntToLong(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.IntegerType.get(),
+ Types.LongType.get(),
+ value -> value == null ? null : ((Integer) value).longValue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionFloatToDouble(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.FloatType.get(),
+ Types.DoubleType.get(),
+ value -> value == null ? null : ((Float) value).doubleValue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionDecimalPrecision(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.DecimalType.of(9, 2),
+ Types.DecimalType.of(18, 2),
+ java.util.function.Function.identity());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionDateToTimestamp(FileFormat fileFormat) throws IOException {
+ assumeThat(TypeUtil.isPromotionAllowed(Types.DateType.get(), Types.TimestampType.withoutZone()))
+ .isTrue();
+ runTypePromotionCheck(
+ fileFormat,
+ Types.DateType.get(),
+ Types.TimestampType.withoutZone(),
+ value -> value == null ? null : LocalDateTime.of((LocalDate) value, LocalTime.MIDNIGHT));
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionDateToTimestampNano(FileFormat fileFormat)
+ throws IOException {
+ assumeThat(
+ TypeUtil.isPromotionAllowed(
+ Types.DateType.get(), Types.TimestampNanoType.withoutZone()))
+ .isTrue();
+ runTypePromotionCheck(
+ fileFormat,
+ Types.DateType.get(),
+ Types.TimestampNanoType.withoutZone(),
+ value -> value == null ? null : LocalDateTime.of((LocalDate) value, LocalTime.MIDNIGHT));
+ }
+
+ /**
+ * Schema evolution: Reorder columns. Write with DefaultSchema {col_a, col_b, col_c, col_d,
+ * col_e}, read with reordered schema {col_e, col_c, col_a, col_d, col_b}.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionReorderColumns(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema reorderedSchema =
+ new Schema(
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()));
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(reorderedSchema);
+ for (Types.NestedField col : reorderedSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, reorderedSchema, expectedGenericRecords);
+ }
+
+ /**
+ * Schema evolution: Rename column. Write with DefaultSchema where col_b has field ID 2. Read with
+ * a schema where the same field ID 2 is renamed to "column_b". Since Iceberg binds by field ID,
+ * the renamed column should still read the original data correctly.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionRenameColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // rename col_b(id=2) -> column_b, col_d(id=4) -> column_d
+ Schema renamedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "column_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "column_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(renamedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("column_b", record.getField("col_b"));
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("column_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, renamedSchema, expectedGenericRecords);
+ }
+
+ /**
+ * Schema evolution: Required → Optional. Write with DefaultSchema where all columns are required.
+ * Read with a schema where some columns are changed to optional. Iceberg allows widening required
+ * to optional. The data should still be read correctly.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionRequiredToOptional(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // change col_b and col_d to optional
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.optional(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : readSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, readSchema, expectedGenericRecords);
+ }
+
+ /**
+ * Schema evolution: Read with empty projection. Write with DefaultSchema, read with an empty
+ * schema (no columns). The reader should return the correct number of rows but with no data
+ * columns.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionEmptyProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema emptySchema = new Schema();
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(emptySchema)
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
Review Comment:
hasSameSizeAs(...)
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3086598791
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(projectedSchema)
+ .engineProjection(engineSchema(projectedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ projectedSchema,
+ convertToEngineRecords(expectedGenericRecords, projectedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column). The NameMapping maps file column name "col_b" to old field ID 2. Since the read schema
+ * expects field ID 6, the new col_b should be null.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionReNameMapping(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", null);
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ readSchema, convertToEngineRecords(expectedGenericRecords, readSchema), readRecords);
+ }
+
+ /**
+ * Schema evolution: Allowed type changes. Write with DefaultSchema (col_b:int, col_d:float), read
+ * with promoted types (col_b:long, col_d:double).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotion(FileFormat fileFormat) throws IOException {
Review Comment:
If problematic, maybe use a single column schema and use `RandomGenericData.generate`?
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3086511319
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ List<T> expectedEngineRecords = convertToEngineRecords(expectedGenericRecords, readSchema);
+ assertEquals(readSchema, expectedEngineRecords, readRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema projectedSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(projectedSchema)
+ .engineProjection(engineSchema(projectedSchema))
+ .build()) {
+ readRecords = ImmutableList.copyOf(reader);
+ }
+
+ assertThat(readRecords).hasSize(genericRecords.size());
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ assertEquals(
+ projectedSchema,
+ convertToEngineRecords(expectedGenericRecords, projectedSchema),
+ readRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column). The NameMapping maps file column name "col_b" to old field ID 2. Since the read schema
+ * expects field ID 6, the new col_b should be null.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionReNameMapping(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
+ List<T> readRecords;
+ try (CloseableIterable<T> reader =
+ FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
+ .project(readSchema)
+ .engineProjection(engineSchema(readSchema))
Review Comment:
Do we have a fix already merged that the `.engineProjection(engineSchema(readSchema))` is not required anymore?
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3203894598
##########
parquet/src/testFixtures/java/org/apache/iceberg/parquet/TestParquetWritingUtils.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import org.apache.parquet.io.InputFile;
+import org.apache.parquet.io.OutputFile;
+
+/** Utilities for tests that need to write Parquet files. */
+public class TestParquetWritingUtils {
Review Comment:
nit: maybe `ParquetWritingTestUtils`? `Test*` prefix is more for JUnit test classes.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3086487291
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -628,6 +628,401 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.required(2, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()),
+ Types.NestedField.optional(6, "new_string_col", Types.StringType.get()),
+ Types.NestedField.optional(7, "new_int_col", Types.IntegerType.get()));
Review Comment:
Make this based on the `writeSchema`
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3110123047
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -700,4 +1089,164 @@ private static String splitSizeProperty(FileFormat fileFormat) {
"No split size property defined for format: " + fileFormat);
};
}
+
+ private void writeRecordsWithoutFieldIds(
+ FileFormat fileFormat, Schema schema, List<Record> records) throws IOException {
+ switch (fileFormat) {
+ case PARQUET -> writeParquetWithoutFieldIds(schema, records);
+ case AVRO -> writeAvroWithoutFieldIds(schema, records);
+ case ORC -> writeOrcWithoutFieldIds(schema, records);
+ default -> throw new UnsupportedOperationException("Unsupported file format: " + fileFormat);
+ }
+ }
+
+ private void writeAvroWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = RemoveIds.removeIds(schema);
+
+ OutputFile outputFile = encryptedFile.encryptingOutputFile();
+ DatumWriter<GenericData.Record> datumWriter = new GenericDatumWriter<>(avroSchemaWithoutIds);
+ try (OutputStream out = outputFile.create();
+ DataFileWriter<GenericData.Record> writer = new DataFileWriter<>(datumWriter)) {
+ writer.create(avroSchemaWithoutIds, out);
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.append(avroRecord);
+ }
+ }
+
+ try (DataFileStream<GenericData.Record> reader =
+ new DataFileStream<>(outputFile.toInputFile().newStream(), new GenericDatumReader<>())) {
+ assertThat(AvroSchemaUtil.hasIds(reader.getSchema())).isFalse();
+ }
+ }
+
+ private void writeParquetWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = RemoveIds.removeIds(schema);
+
+ Path tempFile = Files.createTempFile("iceberg-test-no-ids-", ".parquet");
+ Files.deleteIfExists(tempFile);
+
+ try {
+ try (ParquetWriter<GenericData.Record> writer =
+ AvroParquetWriter.<GenericData.Record>builder(
+ new org.apache.hadoop.fs.Path(tempFile.toUri()))
+ .withDataModel(GenericData.get())
+ .withSchema(avroSchemaWithoutIds)
+ .withConf(new org.apache.hadoop.conf.Configuration())
Review Comment:
Do we need this?
If we need this could we import?
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3123495373
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -609,6 +649,354 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> evolvedColumns = Lists.newArrayList(writeSchema.columns());
+
+ int maxFieldId =
+ writeSchema.columns().stream().mapToInt(Types.NestedField::fieldId).max().orElse(0);
+ evolvedColumns.add(
+ Types.NestedField.optional("new_string_col")
+ .withId(maxFieldId + 1)
+ .ofType(Types.StringType.get())
+ .build());
+ evolvedColumns.add(
+ Types.NestedField.optional("new_int_col")
+ .withId(maxFieldId + 2)
+ .ofType(Types.IntegerType.get())
+ .build());
+ Schema readSchema = new Schema(evolvedColumns);
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, readSchema, expectedGenericRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> writeColumns = writeSchema.columns();
+ assumeThat(writeColumns).hasSizeGreaterThanOrEqualTo(2);
+ Schema projectedSchema =
+ new Schema(writeColumns.get(0), writeColumns.get(writeColumns.size() - 1));
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ for (Types.NestedField col : projectedSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, projectedSchema, expectedGenericRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionDropAndReAddSameNameColumn(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", null);
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, readSchema, expectedGenericRecords);
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionIntToLong(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.IntegerType.get(),
+ Types.LongType.get(),
+ value -> value == null ? null : ((Integer) value).longValue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionFloatToDouble(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.FloatType.get(),
+ Types.DoubleType.get(),
+ value -> value == null ? null : ((Float) value).doubleValue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionDecimalPrecision(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.DecimalType.of(9, 2),
+ Types.DecimalType.of(18, 2),
+ java.util.function.Function.identity());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionDateToTimestamp(FileFormat fileFormat) throws IOException {
+ assumeThat(TypeUtil.isPromotionAllowed(Types.DateType.get(), Types.TimestampType.withoutZone()))
Review Comment:
I tried it myself, and currently neither Flink nor Spark supports this feature. Below are my test cases and the results I got.
Flink:
```java
@TestTemplate
public void testDateToTimestamp() throws Exception {
String tableName = "test_insert_into_timestamp";
sql(
"CREATE TABLE %s(id INT, data date) WITH ('write.format.default'='%s','format-version'='3')",
tableName, format.name());
try {
Table table =
validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
sql("INSERT INTO %s SELECT 1, DATE '2024-01-01'", tableName);
sql("INSERT INTO %s SELECT 2, DATE '2024-01-02'", tableName);
sql("INSERT INTO %s SELECT 3, DATE '2024-01-03'", tableName);
table
.updateSchema()
.updateColumn("data", Types.TimestampType.withoutZone())
.commit();
table.refresh();
sql("SELECT * FROM %s ORDER BY id", tableName);
} finally {
sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
}
}
```
```shell
Cannot change column type: data: date -> timestamp
java.lang.IllegalArgumentException: Cannot change column type: data: date -> timestamp
at org.apache.iceberg.relocated.com.google.common.base.Preconditions.checkArgument(Preconditions.java:463)
at org.apache.iceberg.SchemaUpdate.updateColumn(SchemaUpdate.java:285)
at org.apache.iceberg.flink.TestFlinkTableSink.testDateToTimestamp(TestFlinkTableSink.java:341)
at java.base/java.lang.reflect.Method.invoke(Method.java:580)
at java.base/java.util.Optional.ifPresent(Optional.java:178)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
```
Spark:
```java
@TestTemplate
public void testDateToTimestamp() {
String tableName = tableName("date_to_timestamp_table");
sql(
"CREATE TABLE %s (id INT NOT NULL, dt DATE) "
+ "USING iceberg TBLPROPERTIES ('format-version'='3')",
tableName);
sql("INSERT INTO %s VALUES (1, DATE '2023-01-01')", tableName);
sql("ALTER TABLE %s ALTER COLUMN dt TYPE TIMESTAMP", tableName);
List<Object[]> sql = sql("select * from %s", tableName);
System.out.println(sql);
}
```
```shell
Cannot change column type: dt: date -> timestamptz
java.lang.IllegalArgumentException: Cannot change column type: dt: date -> timestamptz
at org.apache.iceberg.relocated.com.google.common.base.Preconditions.checkArgument(Preconditions.java:463)
at org.apache.iceberg.SchemaUpdate.updateColumn(SchemaUpdate.java:285)
at org.apache.iceberg.spark.Spark3Util.applySchemaChanges(Spark3Util.java:179)
at org.apache.iceberg.spark.SparkCatalog.commitChanges(SparkCatalog.java:781)
at org.apache.iceberg.spark.SparkCatalog.alterTable(SparkCatalog.java:294)
at org.apache.spark.sql.execution.datasources.v2.AlterTableExec.run(AlterTableExec.scala:40)
at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result$lzycompute(V2CommandExec.scala:43)
at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result(V2CommandExec.scala:43)
```
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3123560901
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -73,6 +108,11 @@ public abstract class BaseFormatModelTests<T> {
protected abstract void assertEquals(Schema schema, List<T> expected, List<T> actual);
+ protected Object convertConstantToEngine(Type type, Object value) {
+ return null;
Review Comment:
This was code from another PR and has been cleaned up.
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3123495373
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -609,6 +649,354 @@ void testReaderBuilderRecordsPerBatchNotSupported(FileFormat fileFormat) throws
.isInstanceOf(UnsupportedOperationException.class);
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> evolvedColumns = Lists.newArrayList(writeSchema.columns());
+
+ int maxFieldId =
+ writeSchema.columns().stream().mapToInt(Types.NestedField::fieldId).max().orElse(0);
+ evolvedColumns.add(
+ Types.NestedField.optional("new_string_col")
+ .withId(maxFieldId + 1)
+ .ofType(Types.StringType.get())
+ .build());
+ evolvedColumns.add(
+ Types.NestedField.optional("new_int_col")
+ .withId(maxFieldId + 2)
+ .ofType(Types.IntegerType.get())
+ .build());
+ Schema readSchema = new Schema(evolvedColumns);
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ for (Types.NestedField col : writeSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, readSchema, expectedGenericRecords);
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> writeColumns = writeSchema.columns();
+ assumeThat(writeColumns).hasSizeGreaterThanOrEqualTo(2);
+ Schema projectedSchema =
+ new Schema(writeColumns.get(0), writeColumns.get(writeColumns.size() - 1));
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(projectedSchema);
+ for (Types.NestedField col : projectedSchema.columns()) {
+ expected.setField(col.name(), record.getField(col.name()));
+ }
+
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, projectedSchema, expectedGenericRecords);
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
+ * DefaultSchema where col_b has field ID 2. Read with a schema where col_b has field ID 6 (a new
+ * column).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionDropAndReAddSameNameColumn(FileFormat fileFormat) throws IOException {
+
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ // Remove col_b and add a new col_b with a different field ID
+ Schema readSchema =
+ new Schema(
+ Types.NestedField.required(1, "col_a", Types.StringType.get()),
+ Types.NestedField.optional(6, "col_b", Types.IntegerType.get()),
+ Types.NestedField.required(3, "col_c", Types.LongType.get()),
+ Types.NestedField.required(4, "col_d", Types.FloatType.get()),
+ Types.NestedField.required(5, "col_e", Types.DoubleType.get()));
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField("col_a", record.getField("col_a"));
+ expected.setField("col_b", null);
+ expected.setField("col_c", record.getField("col_c"));
+ expected.setField("col_d", record.getField("col_d"));
+ expected.setField("col_e", record.getField("col_e"));
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, readSchema, expectedGenericRecords);
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionIntToLong(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.IntegerType.get(),
+ Types.LongType.get(),
+ value -> value == null ? null : ((Integer) value).longValue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionFloatToDouble(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.FloatType.get(),
+ Types.DoubleType.get(),
+ value -> value == null ? null : ((Float) value).doubleValue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionDecimalPrecision(FileFormat fileFormat) throws IOException {
+ runTypePromotionCheck(
+ fileFormat,
+ Types.DecimalType.of(9, 2),
+ Types.DecimalType.of(18, 2),
+ java.util.function.Function.identity());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionTypePromotionDateToTimestamp(FileFormat fileFormat) throws IOException {
+ assumeThat(TypeUtil.isPromotionAllowed(Types.DateType.get(), Types.TimestampType.withoutZone()))
Review Comment:
I tried it myself, and currently neither Flink nor Spark supports this feature. Below are my test cases and the results I got.
Flink:
```java
@TestTemplate
public void testDateToTimestamp() throws Exception {
String tableName = "test_insert_into_timestamp";
sql(
"CREATE TABLE %s(id INT, data date) WITH ('write.format.default'='%s','format-version'='3')",
tableName, format.name());
try {
Table table =
validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));
sql("INSERT INTO %s SELECT 1, DATE '2024-01-01'", tableName);
sql("INSERT INTO %s SELECT 2, DATE '2024-01-02'", tableName);
sql("INSERT INTO %s SELECT 3, DATE '2024-01-03'", tableName);
table
.updateSchema()
.updateColumn("data", Types.TimestampType.withoutZone())
.commit();
table.refresh();
sql("SELECT * FROM %s ORDER BY id", tableName);
} finally {
sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
}
}
```
```shell
Cannot change column type: data: date -> timestamp
java.lang.IllegalArgumentException: Cannot change column type: data: date -> timestamp
at org.apache.iceberg.relocated.com.google.common.base.Preconditions.checkArgument(Preconditions.java:463)
at org.apache.iceberg.SchemaUpdate.updateColumn(SchemaUpdate.java:285)
at org.apache.iceberg.flink.TestFlinkTableSink.testDateToTimestamp(TestFlinkTableSink.java:341)
at java.base/java.lang.reflect.Method.invoke(Method.java:580)
at java.base/java.util.Optional.ifPresent(Optional.java:178)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
```
Spark:
```java
@TestTemplate
public void testDateToTimestamp() {
String tableName = tableName("date_to_timestamp_table");
sql(
"CREATE TABLE %s (id INT NOT NULL, dt DATE) "
+ "USING iceberg TBLPROPERTIES ('format-version'='3')",
tableName);
sql("INSERT INTO %s VALUES (1, DATE '2023-01-01')", tableName);
sql("ALTER TABLE %s ALTER COLUMN dt TYPE timestamp_ntz", tableName);
List<Object[]> sql = sql("select * from %s", tableName);
System.out.println(sql);
}
```
```shell
Cannot change column type: dt: date -> timestamptz
java.lang.IllegalArgumentException: Cannot change column type: dt: date -> timestamptz
at org.apache.iceberg.relocated.com.google.common.base.Preconditions.checkArgument(Preconditions.java:463)
at org.apache.iceberg.SchemaUpdate.updateColumn(SchemaUpdate.java:285)
at org.apache.iceberg.spark.Spark3Util.applySchemaChanges(Spark3Util.java:179)
at org.apache.iceberg.spark.SparkCatalog.commitChanges(SparkCatalog.java:781)
at org.apache.iceberg.spark.SparkCatalog.alterTable(SparkCatalog.java:294)
at org.apache.spark.sql.execution.datasources.v2.AlterTableExec.run(AlterTableExec.scala:40)
at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result$lzycompute(V2CommandExec.scala:43)
at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result(V2CommandExec.scala:43)
```
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3142563631
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -1330,6 +1357,288 @@ private void readAndAssertGenericRecords(
readAndAssertGenericRecords(fileFormat, schema, sourceRecords.stream().map(transform).toList());
}
+ /**
+ * Schema evolution: Adding column (reading with wider schema). Write with DefaultSchema, read
+ * with additional optional columns. The new columns should be filled with null values.
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionAddColumn(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> evolvedColumns = Lists.newArrayList(writeSchema.columns());
+
+ int maxFieldId =
+ writeSchema.columns().stream().mapToInt(Types.NestedField::fieldId).max().orElse(0);
+ evolvedColumns.add(
+ Types.NestedField.optional("new_string_col")
+ .withId(maxFieldId + 1)
+ .ofType(Types.StringType.get())
+ .build());
+ evolvedColumns.add(
+ Types.NestedField.optional("new_int_col")
+ .withId(maxFieldId + 2)
+ .ofType(Types.IntegerType.get())
+ .build());
+ Schema readSchema = new Schema(evolvedColumns);
+ readAndAssertEngineRecords(
+ fileFormat,
+ readSchema,
+ genericRecords,
+ record -> {
+ Record expected = copy(record, writeSchema, readSchema);
+
+ expected.setField("new_string_col", null);
+ expected.setField("new_int_col", null);
+ return expected;
+ });
+ }
+
+ /**
+ * Schema evolution: Projection / Removing column (reading with narrower schema). Write with
+ * DefaultSchema, read with only a subset of columns (skipping middle columns).
+ */
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testSchemaEvolutionProjection(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema writeSchema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Types.NestedField> writeColumns = writeSchema.columns();
+ assumeThat(writeColumns).hasSizeGreaterThanOrEqualTo(2);
+ Schema projectedSchema =
+ new Schema(writeColumns.get(0), writeColumns.get(writeColumns.size() - 1));
+
+ readAndAssertEngineRecords(
+ fileFormat,
+ projectedSchema,
+ genericRecords,
+ record -> copy(record, projectedSchema, projectedSchema));
+ }
+
+ /**
+ * Schema evolution: Removing and adding a column with the same name (name mapping). Write with
Review Comment:
Pls. don't mention name mapping. It has nothing to do with this 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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3110133129
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -700,4 +1089,164 @@ private static String splitSizeProperty(FileFormat fileFormat) {
"No split size property defined for format: " + fileFormat);
};
}
+
+ private void writeRecordsWithoutFieldIds(
+ FileFormat fileFormat, Schema schema, List<Record> records) throws IOException {
+ switch (fileFormat) {
+ case PARQUET -> writeParquetWithoutFieldIds(schema, records);
+ case AVRO -> writeAvroWithoutFieldIds(schema, records);
+ case ORC -> writeOrcWithoutFieldIds(schema, records);
+ default -> throw new UnsupportedOperationException("Unsupported file format: " + fileFormat);
+ }
+ }
+
+ private void writeAvroWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = RemoveIds.removeIds(schema);
+
+ OutputFile outputFile = encryptedFile.encryptingOutputFile();
+ DatumWriter<GenericData.Record> datumWriter = new GenericDatumWriter<>(avroSchemaWithoutIds);
+ try (OutputStream out = outputFile.create();
+ DataFileWriter<GenericData.Record> writer = new DataFileWriter<>(datumWriter)) {
+ writer.create(avroSchemaWithoutIds, out);
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.append(avroRecord);
+ }
+ }
+
+ try (DataFileStream<GenericData.Record> reader =
+ new DataFileStream<>(outputFile.toInputFile().newStream(), new GenericDatumReader<>())) {
+ assertThat(AvroSchemaUtil.hasIds(reader.getSchema())).isFalse();
+ }
+ }
+
+ private void writeParquetWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = RemoveIds.removeIds(schema);
+
+ Path tempFile = Files.createTempFile("iceberg-test-no-ids-", ".parquet");
+ Files.deleteIfExists(tempFile);
+
+ try {
+ try (ParquetWriter<GenericData.Record> writer =
+ AvroParquetWriter.<GenericData.Record>builder(
+ new org.apache.hadoop.fs.Path(tempFile.toUri()))
+ .withDataModel(GenericData.get())
+ .withSchema(avroSchemaWithoutIds)
+ .withConf(new org.apache.hadoop.conf.Configuration())
+ .build()) {
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.write(avroRecord);
+ }
+ }
+
+ try (ParquetFileReader reader =
+ ParquetFileReader.open(
+ HadoopInputFile.fromPath(
+ new org.apache.hadoop.fs.Path(tempFile.toUri()),
+ new org.apache.hadoop.conf.Configuration()))) {
+ assertThat(ParquetSchemaUtil.hasIds(reader.getFooter().getFileMetaData().getSchema()))
+ .isFalse();
+ }
+
+ byte[] bytes = Files.readAllBytes(tempFile);
+ try (OutputStream out = encryptedFile.encryptingOutputFile().create()) {
+ out.write(bytes);
+ }
+ } finally {
+ Files.deleteIfExists(tempFile);
+ }
+ }
+
+ private void writeOrcWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ TypeDescription typeWithIds = ORCSchemaUtil.convert(schema);
+ TypeDescription typeWithoutIds = ORCSchemaUtil.removeIds(typeWithIds);
+
+ Configuration conf = new Configuration();
+ OrcFile.WriterOptions options =
+ OrcFile.writerOptions(conf).useUTCTimestamp(true).setSchema(typeWithoutIds);
+
+ OrcRowWriter<Record> rowWriter = GenericOrcWriter.buildWriter(schema, typeWithIds);
+
+ Path tempFile = Files.createTempFile("iceberg-test-no-ids-", ".orc");
+ Files.deleteIfExists(tempFile);
+ org.apache.hadoop.fs.Path hadoopPath = new org.apache.hadoop.fs.Path(tempFile.toUri());
+
+ try {
+ try (Writer orcWriter = OrcFile.createWriter(hadoopPath, options)) {
+ VectorizedRowBatch batch = typeWithoutIds.createRowBatch();
+ for (Record record : records) {
+ rowWriter.write(record, batch);
+ if (batch.size == batch.getMaxSize()) {
+ orcWriter.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+
+ if (batch.size > 0) {
+ orcWriter.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+
+ try (Reader reader = OrcFile.createReader(hadoopPath, OrcFile.readerOptions(conf))) {
+ assertThat(ORCSchemaUtil.hasIds(reader.getSchema())).isFalse();
+ }
+
+ byte[] bytes = Files.readAllBytes(tempFile);
+ try (OutputStream out = encryptedFile.encryptingOutputFile().create()) {
+ out.write(bytes);
+ }
+ } finally {
+ Files.deleteIfExists(tempFile);
+ }
+ }
+
+ private void runTypePromotionCheck(
+ FileFormat fileFormat, Type fromType, Type toType, Function<Object, Object> promoteValue)
+ throws IOException {
+ String columnName = "col";
+ Schema writeSchema = new Schema(Types.NestedField.required(1, columnName, fromType));
+ Schema readSchema = new Schema(Types.NestedField.required(1, columnName, toType));
+
+ List<Record> genericRecords = RandomGenericData.generate(writeSchema, 10, 1L);
+ writeGenericRecords(fileFormat, writeSchema, genericRecords);
+
+ List<Record> expectedGenericRecords =
+ genericRecords.stream()
+ .map(
+ record -> {
+ Record expected = GenericRecord.create(readSchema);
+ expected.setField(columnName, promoteValue.apply(record.getField(columnName)));
+ return expected;
+ })
+ .toList();
+
+ readAndAssertEngineRecords(fileFormat, readSchema, expectedGenericRecords);
Review Comment:
I still think that this would be better
```suggestion
readAndAssertEngineRecords(fileFormat, readSchema, genericRecords, record -> {
Record expected = GenericRecord.create(readSchema);
expected.setField(columnName, promoteValue.apply(record.getField(columnName)));
return expected;
});
```
--
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] Data: Add TCK tests for Schema Evolution in BaseFormatModelTests [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #15843:
URL: https://github.com/apache/iceberg/pull/15843#discussion_r3132078209
##########
data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java:
##########
@@ -700,4 +1041,165 @@ private static String splitSizeProperty(FileFormat fileFormat) {
"No split size property defined for format: " + fileFormat);
};
}
+
+ private void writeRecordsWithoutFieldIds(
+ FileFormat fileFormat, Schema schema, List<Record> records) throws IOException {
+ switch (fileFormat) {
+ case PARQUET -> writeParquetWithoutFieldIds(schema, records);
+ case AVRO -> writeAvroWithoutFieldIds(schema, records);
+ case ORC -> writeOrcWithoutFieldIds(schema, records);
+ default -> throw new UnsupportedOperationException("Unsupported file format: " + fileFormat);
+ }
+ }
+
+ private void writeAvroWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = AvroTestHelpers.removeIds(schema);
+
+ OutputFile outputFile = encryptedFile.encryptingOutputFile();
+ DatumWriter<GenericData.Record> datumWriter = new GenericDatumWriter<>(avroSchemaWithoutIds);
+ try (OutputStream out = outputFile.create();
+ DataFileWriter<GenericData.Record> writer = new DataFileWriter<>(datumWriter)) {
+ writer.create(avroSchemaWithoutIds, out);
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.append(avroRecord);
+ }
+ }
+
+ try (DataFileStream<GenericData.Record> reader =
+ new DataFileStream<>(outputFile.toInputFile().newStream(), new GenericDatumReader<>())) {
+ assertThat(AvroTestHelpers.hasIds(reader.getSchema())).isFalse();
+ }
+ }
+
+ private void writeParquetWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = AvroTestHelpers.removeIds(schema);
+
+ Path tempFile = Files.createTempFile("iceberg-test-no-ids-", ".parquet");
+ Files.deleteIfExists(tempFile);
Review Comment:
Good point. I didn't find a suitable API at first, but after searching again, I've refactored this part.
--
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