pvary commented on code in PR #17900:
URL: https://github.com/apache/iceberg/pull/17900#discussion_r4183945341
##########
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:
nit: Import one, and keep only the other fully qualified
--
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]
