fmorillo7694 commented on code in PR #17900:
URL: https://github.com/apache/iceberg/pull/17900#discussion_r4193421648
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java:
##########
@@ -692,6 +693,177 @@ void testSchemaEvolutionNonBackwardsCompatible() throws
Exception {
}
}
+ @Test
+ void testWriteVariant() throws Exception {
+ Schema schema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", Types.VariantType.get()));
+ createV3Table("t1", schema);
+
+ executeVariantSink(
+ Lists.newArrayList(new VariantInput(schema, 1), new
VariantInput(schema, 2)), this.env);
+
+ assertVariantPayloads(ImmutableMap.of(1, 1L, 2, 2L));
+ }
+
+ @Test
+ void testSchemaEvolutionAddVariantField() throws Exception {
+ Schema variantSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()),
+ Types.NestedField.optional(3, "payload", Types.VariantType.get()));
+ createV3Table("t1", SimpleDataUtil.SCHEMA);
+
+ executeVariantSink(
+ Lists.newArrayList(
+ new VariantInput(SimpleDataUtil.SCHEMA, 1), new
VariantInput(variantSchema, 2)),
+ this.env);
+
+ Table table =
CATALOG_EXTENSION.catalog().loadTable(TableIdentifier.of(DATABASE, "t1"));
+
assertThat(table.schema().findField("payload").type()).isEqualTo(Types.VariantType.get());
+ Map<Integer, Long> expected = Maps.newHashMap();
+ expected.put(1, null);
+ expected.put(2, 2L);
+ assertVariantPayloads(expected);
+ }
+
+ @Test
+ void testSchemaEvolutionStringToVariantFails() throws Exception {
+ assertIncompatibleVariantEvolution(
+ Types.StringType.get(),
+ Types.VariantType.get(),
+ "Cannot change column type: payload: string -> variant");
+ }
+
+ @Test
+ void testSchemaEvolutionVariantToStringFails() throws Exception {
+ assertIncompatibleVariantEvolution(
+ Types.VariantType.get(),
+ Types.StringType.get(),
+ "Cannot change column type: payload: variant -> string");
+ }
+
+ private void assertIncompatibleVariantEvolution(
+ Type tableType, Type writeType, String expectedMessage) throws Exception
{
+ Schema tableSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", tableType));
+ Schema writeSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", writeType));
+ createV3Table("t1", tableSchema);
+
+ try {
+ executeVariantSink(
+ Lists.newArrayList(new VariantInput(writeSchema, 1)),
+ StreamExecutionEnvironment.getExecutionEnvironment());
+ fail();
+ } catch (JobExecutionException e) {
+ assertThat(
+ ExceptionUtils.findThrowable(
+ e, t -> t.getMessage() != null &&
t.getMessage().contains(expectedMessage)))
+ .isNotEmpty();
+ }
+ }
+
+ private static void createV3Table(String name, Schema schema) {
+ CATALOG_EXTENSION
+ .catalog()
+ .createTable(
+ TableIdentifier.of(DATABASE, name),
+ schema,
+ PartitionSpec.unpartitioned(),
+ null,
+ ImmutableMap.of(TableProperties.FORMAT_VERSION, "3"));
+ }
+
+ private static void executeVariantSink(List<VariantInput> inputs,
StreamExecutionEnvironment env)
+ throws Exception {
+ DataStream<VariantInput> dataStream =
+ env.fromData(inputs, TypeInformation.of(new TypeHint<>() {}));
+ env.setParallelism(1);
+ DynamicIcebergSink.forInput(dataStream)
+ .generator(new VariantGenerator())
+ .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+ .writeParallelism(1)
+ .immediateTableUpdate(true)
+ .append();
+ env.execute("Test Iceberg Variant DataStream");
+ }
+
+ /** Reads table t1 and checks the "k" field of each row's variant payload,
keyed by row id. */
+ private static void assertVariantPayloads(Map<Integer, Long> expectedById)
throws IOException {
+ Map<Integer, Long> actualById = Maps.newHashMap();
+ try (CloseableIterable<Record> records =
+ IcebergGenerics.read(
+
CATALOG_EXTENSION.catalog().loadTable(TableIdentifier.of(DATABASE, "t1")))
+ .build()) {
+ for (Record record : records) {
+ org.apache.iceberg.variants.Variant payload =
Review Comment:
Done.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java:
##########
@@ -85,6 +86,27 @@ void testPreservesRowKind() {
assertThat(convert(deleteRow, SCHEMA,
SCHEMA2).getRowKind()).isEqualTo(RowKind.DELETE);
}
+ @Test
+ void testVariantWithDataConversion() {
+ Schema sourceSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", Types.VariantType.get()));
+ Schema targetSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "payload", Types.VariantType.get()));
+ assertThat(CompareSchemasVisitor.visit(sourceSchema, targetSchema))
Review Comment:
Done, switched to the four-argument `visit`.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java:
##########
@@ -692,6 +693,177 @@ void testSchemaEvolutionNonBackwardsCompatible() throws
Exception {
}
}
+ @Test
+ void testWriteVariant() throws Exception {
+ Schema schema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", Types.VariantType.get()));
+ createV3Table("t1", schema);
+
+ executeVariantSink(
+ Lists.newArrayList(new VariantInput(schema, 1), new
VariantInput(schema, 2)), this.env);
+
+ assertVariantPayloads(ImmutableMap.of(1, 1L, 2, 2L));
+ }
+
+ @Test
+ void testSchemaEvolutionAddVariantField() throws Exception {
Review Comment:
Done, it now runs with `immediateTableUpdate` set to both `true` and
`false`, and both pass.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -210,6 +215,14 @@ private void addColumn(int parentId, Types.NestedField
field) {
private void updateColumn(Types.NestedField existingField, Types.NestedField
targetField) {
String existingColumnName =
this.existingSchema.findColumnName(existingField.fieldId());
+ // A column cannot be promoted to variant; fail instead of silently
skipping the update
+ Preconditions.checkArgument(
Review Comment:
Agreed, failing in `TableUpdater.findOrCreateSchema` before the schema
update starts would be cleaner. I will do that as a follow-up PR once this one
is in.
--
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]
