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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
