Guosmilesmile commented on code in PR #16097:
URL: https://github.com/apache/iceberg/pull/16097#discussion_r3136550344
##########
core/src/test/java/org/apache/iceberg/data/DataTestBase.java:
##########
@@ -101,7 +101,8 @@ protected boolean
allowsWritingNullValuesForRequiredFields() {
required(114, "dec_9_0", Types.DecimalType.of(9, 0)),
required(115, "dec_11_2", Types.DecimalType.of(11, 2)),
required(116, "dec_38_10", Types.DecimalType.of(38, 10)), // maximum
precision
- required(117, "time", Types.TimeType.get()));
+ required(117, "time", Types.TimeType.get()),
+ required(118, "uuid", Types.UUIDType.get()));
Review Comment:
E2E UT may be something like this. This is just for my own testing purposes,
and I hope it can provide some inspiration.
```java
class TestFlinkUuidType extends CatalogTestBase {
private static final String TABLE_NAME = "test_table";
private Table icebergTable;
@TempDir private Path warehouseDir;
@Parameters(name = "catalogName={0}, baseNamespace={1}")
protected static List<Object[]> parameters() {
return Arrays.asList(
// For now hive metadata is not supported variant, so we only test
hadoop catalog
new Object[] {"testhadoop", Namespace.empty()},
new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1")});
}
@Override
@BeforeEach
public void before() {
super.before();
sql("CREATE DATABASE %s", flinkDatabase);
sql("USE CATALOG %s", catalogName);
sql("USE %s", DATABASE);
Schema schema =
new Schema(
Types.NestedField.required(1, "id",
Types.IntegerType.get()),
Types.NestedField.optional(2, "data",
Types.UUIDType.get()));
icebergTable =
validationCatalog.createTable(
TableIdentifier.of(icebergNamespace, TABLE_NAME),
schema,
PartitionSpec.unpartitioned(),
ImmutableMap.of(
"format-version",
"3",
"write.format.default",
FileFormat.AVRO.name()));
}
@TestTemplate
public void testInsertUuidFromFlink() throws Exception {
UUID expectedUuid =
UUID.fromString("123e4567-e89b-12d3-a456-426614174000");
sql(
"INSERT INTO %s VALUES (1, CAST(X'%s' AS BINARY(16)))",
TABLE_NAME,
"123e4567e89b12d3a456426614174000");
List<Record> records = SimpleDataUtil.tableRecords(icebergTable);
assertThat(records).hasSize(1);
assertThat(records.get(0).getField("data")).isInstanceOf(UUID.class).isEqualTo(expectedUuid);
}
@TestTemplate
public void testReadUuidFromFlink() throws Exception {
UUID expectedUuid =
UUID.fromString("0f8fad5b-d9cb-469f-a165-70867728950e");
ImmutableList.Builder<Record> builder = ImmutableList.builder();
builder.add(GenericRecord.create(icebergTable.schema()).copy("id", 1,
"data", expectedUuid));
new GenericAppenderHelper(icebergTable, FileFormat.PARQUET, warehouseDir)
.appendToTable(builder.build());
icebergTable.refresh();
List<GenericRowData> genericRowData = Lists.newArrayList();
try (CloseableIterable<CombinedScanTask> combinedScanTasks =
icebergTable.newScan().planTasks()) {
for (CombinedScanTask combinedScanTask : combinedScanTasks) {
try (DataIterator<RowData> dataIterator =
ReaderUtil.createDataIterator(
combinedScanTask, icebergTable.schema(),
icebergTable.schema())) {
while (dataIterator.hasNext()) {
GenericRowData rowData = (GenericRowData) dataIterator.next();
genericRowData.add(rowData);
}
}
}
}
assertThat(genericRowData).hasSize(1);
assertThat(genericRowData.get(0).getField(1)).isInstanceOf(byte[].class);
byte[] uuidBytes = (byte[]) genericRowData.get(0).getField(1);
assertThat(uuidBytes).hasSize(16);
ByteBuffer bb = ByteBuffer.wrap(uuidBytes);
UUID actualUuid = new UUID(bb.getLong(), bb.getLong());
assertThat(actualUuid).isEqualTo(expectedUuid);
}
}
```
```shell
Cannot write incompatible dataset to table with schema:
table {
1: id: required int
2: data: optional uuid
}
Provided schema:
table {
1: id: required int
2: data: optional fixed[16]
}
Problems:
* data: fixed[16] cannot be promoted to uuid
java.lang.IllegalArgumentException: Cannot write incompatible dataset to
table with schema:
table {
1: id: required int
2: data: optional uuid
}
Provided schema:
table {
1: id: required int
2: data: optional fixed[16]
}
Problems:
* data: fixed[16] cannot be promoted to uuid
at
org.apache.iceberg.types.TypeUtil.checkSchemaCompatibility(TypeUtil.java:531)
at
org.apache.iceberg.types.TypeUtil.validateWriteSchema(TypeUtil.java:480)
at
org.apache.iceberg.flink.sink.FlinkSink.toFlinkRowType(FlinkSink.java:755)
at
org.apache.iceberg.flink.sink.FlinkSink$Builder.chainIcebergOperators(FlinkSink.java:457)
at
org.apache.iceberg.flink.sink.FlinkSink$Builder.append(FlinkSink.java:488)
at
org.apache.iceberg.flink.IcebergTableSink.createLegacySink(IcebergTableSink.java:217)
at
org.apache.iceberg.flink.IcebergTableSink.lambda$getSinkRuntimeProvider$0(IcebergTableSink.java:164)
at
org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecSink.applySinkProvider(CommonExecSink.java:409)
```
--
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]
