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:
[P1] Create a format-v3 table before adding this initial default
`unpartitionedUnsortedTable(catalog)` creates the table with the default
format version (v2), but Iceberg permits a non-null initial default only in v3.
As a result, this test fails at this `commit()` before it exercises
`validateCurrentSchema()`:
```
Tests run: 1, Failures: 0, Errors: 1
Invalid schema for v2:
- Invalid initial default for required_value: non-null default (7) is not
supported until v3
```
I reproduced this with
`IcebergWritePlanProviderTest#branchWriteAllowsCurrentRequiredFieldWithInitialDefault`
on the current head. Please create or upgrade this fixture as format v3 (the
existing `formatVersionThreeTable` helper demonstrates the required property),
then rerun the test so it validates the intended branch-compatibility behavior.
--
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]
