Gabriel39 commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3702744673
##########
fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java:
##########
@@ -225,6 +229,114 @@ private static TIcebergTableSink planSink(Table table,
RecordingConnectorContext
return plan.getDataSink().getIcebergTableSink();
}
+ @Test
+ public void getWriteColumnsCarriesPinnedTypedDefaultsAndNullability() {
+ Schema writeSchema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional("value")
+
.withId(2).ofType(Types.IntegerType.get()).withWriteDefault(42).build(),
+ Types.NestedField.optional("text")
+
.withId(3).ofType(Types.StringType.get()).withWriteDefault("O'Reilly").build(),
+ Types.NestedField.optional("payload")
+ .withId(4).ofType(Types.BinaryType.get())
+ .withWriteDefault(ByteBuffer.wrap(new byte[] {0x00,
0x0f, (byte) 0xff})).build(),
+ Types.NestedField.optional(5, "nullable_value",
Types.IntegerType.get()),
+ Types.NestedField.required(6, "required_value",
Types.IntegerType.get()));
+ InMemoryCatalog catalog = freshCatalog();
+ Table table = catalog.createTable(TableIdentifier.of("db1",
"defaults"), writeSchema,
+ PartitionSpec.unpartitioned());
+ RecordingConnectorContext context = contextWithStorage();
+ IcebergWritePlanProvider provider = providerFor(table, context);
+
+ List<ConnectorColumn> writeColumns = provider.getWriteColumns(
+ sessionFor(table, context), new IcebergTableHandle("db1",
"defaults"),
+ Optional.empty()).orElseThrow(AssertionError::new);
+ Map<String, ConnectorColumn> columns = new HashMap<>();
+ for (ConnectorColumn column : writeColumns) {
+ columns.put(column.getName(), column);
+ }
+
+ Assertions.assertFalse(columns.get("id").isNullable());
+ Assertions.assertEquals("42",
columns.get("value").getDefaultValueSql());
+ Assertions.assertEquals("'O''Reilly'",
columns.get("text").getDefaultValueSql());
+ Assertions.assertEquals("UNHEX('000FFF')",
columns.get("payload").getDefaultValueSql());
+ Assertions.assertEquals("NULL",
columns.get("nullable_value").getDefaultValueSql());
+
Assertions.assertNull(columns.get("required_value").getDefaultValueSql());
+ for (ConnectorColumn column : writeColumns) {
+ Assertions.assertNull(column.getDefaultValue(),
+ "request-scoped Iceberg defaults must not leak into cached
metadata");
+ }
+ }
+
+ @Test
+ public void branchWritePinsBranchHeadSchema() {
+ InMemoryCatalog catalog = freshCatalog();
+ Table table = unpartitionedUnsortedTable(catalog);
+ table.newAppend().commit();
+ table.manageSnapshots().createBranch("old_schema",
table.currentSnapshot().snapshotId()).commit();
+ table.updateSchema().renameColumn("name", "renamed_name").commit();
+
+ IcebergWriteSchemaContext writeContext =
IcebergWriteSchemaContext.create(
+ table, "db1.t2", Optional.of("old_schema"), false, false);
+
+ Assertions.assertNotEquals(table.schema().schemaId(),
writeContext.getSchema().schemaId());
+ Assertions.assertNotNull(writeContext.getSchema().findField("name"));
+
Assertions.assertNull(writeContext.getSchema().findField("renamed_name"));
+ Assertions.assertDoesNotThrow(() ->
writeContext.validateCurrentSchema(table, false));
+ }
+
+ @Test
+ public void branchWriteRejectsCurrentRequiredFieldAbsentWithoutDefault() {
+ InMemoryCatalog catalog = freshCatalog();
+ Table table = unpartitionedUnsortedTable(catalog);
+ table.newAppend().commit();
+ table.manageSnapshots().createBranch("old_schema",
table.currentSnapshot().snapshotId()).commit();
+ table.updateSchema().allowIncompatibleChanges()
+ .addRequiredColumn("required_value",
Types.IntegerType.get()).commit();
+
+ DorisConnectorException exception =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> IcebergWriteSchemaContext.create(
+ table, "db1.t2", Optional.of("old_schema"), false,
false));
+
+ Assertions.assertTrue(exception.getMessage().contains("required field
required_value"));
+ Assertions.assertTrue(exception.getMessage().contains(
+ "is absent from the pinned branch schema and has no initial
default"));
+ }
+
+ @Test
+ public void branchWriteRevalidatesFieldMadeRequiredAfterPlanning() {
+ InMemoryCatalog catalog = freshCatalog();
+ Table table = unpartitionedUnsortedTable(catalog);
+ table.newAppend().commit();
+ table.manageSnapshots().createBranch("old_schema",
table.currentSnapshot().snapshotId()).commit();
+ IcebergWriteSchemaContext writeContext =
IcebergWriteSchemaContext.create(
+ table, "db1.t2", Optional.of("old_schema"), false, false);
+
table.updateSchema().allowIncompatibleChanges().requireColumn("name").commit();
+
+ DorisConnectorException exception =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> writeContext.validateCurrentSchema(table, false));
+
+ Assertions.assertTrue(exception.getMessage().contains("required field
name"));
+ Assertions.assertTrue(exception.getMessage().contains(
+ "is optional in the pinned branch schema and can contain
explicit nulls"));
+ }
+
+ @Test
+ public void branchWriteAllowsCurrentRequiredFieldWithInitialDefault() {
+ InMemoryCatalog catalog = freshCatalog();
+ Table table = unpartitionedUnsortedTable(catalog);
+ table.newAppend().commit();
+ table.manageSnapshots().createBranch("old_schema",
table.currentSnapshot().snapshotId()).commit();
+ table.updateSchema().addRequiredColumn(
+ "required_value", Types.IntegerType.get(),
Literal.of(7)).commit();
Review Comment:
@-
--
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]
