luoyuxia commented on code in PR #3601:
URL: https://github.com/apache/fluss/pull/3601#discussion_r3608116098
##########
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java:
##########
@@ -293,14 +310,189 @@ private TableIdentifier
toIcebergTableIdentifier(TablePath tablePath) {
return TableIdentifier.of(tablePath.getDatabaseName(),
tablePath.getTableName());
}
- private void createTable(TablePath tablePath, Catalog.TableBuilder
tableBuilder) {
+ private void createTable(
+ TablePath tablePath,
+ Catalog.TableBuilder tableBuilder,
+ Schema newIcebergSchema,
+ PartitionSpec expectedSpec,
+ SortOrder expectedSortOrder,
+ Map<String, String> expectedProperties,
+ boolean isCreatingFlussTable) {
try {
tableBuilder.create();
} catch (AlreadyExistsException e) {
- throw new TableAlreadyExistException("Table " + tablePath + "
already exists.");
+ // Table already exists in Iceberg. Check schema compatibility for
idempotency,
+ TableIdentifier icebergId = toIcebergTableIdentifier(tablePath);
+ Table existingTable = icebergCatalog.loadTable(icebergId);
+ Schema existingSchema = existingTable.schema();
+ if (!isIcebergSchemaCompatibleWithSchema(existingSchema,
newIcebergSchema)) {
+ throw new TableAlreadyExistException(
+ String.format(
+ "The table %s already exists in Iceberg
catalog, but the table schema is not compatible. "
+ + "Existing schema: %s, new schema:
%s. "
+ + "Please first drop the table in
Iceberg catalog or use a new table name.",
+ tablePath, existingSchema, newIcebergSchema),
+ e);
+ }
+
+ if (!isIcebergPartitionSpecCompatible(
+ existingTable.spec(), existingSchema, expectedSpec,
newIcebergSchema)) {
+ throw new TableAlreadyExistException(
+ String.format(
+ "The table %s already exists in Iceberg
catalog, but the partition spec is not compatible. "
+ + "Existing spec: %s, new spec: %s. "
+ + "Please first drop the table in
Iceberg catalog or use a new table name.",
+ tablePath, existingTable.spec(), expectedSpec),
+ e);
+ }
+
+ if (!isIcebergSortOrderCompatible(
+ existingTable.sortOrder(),
+ existingSchema,
+ expectedSortOrder,
+ newIcebergSchema)) {
+ throw new TableAlreadyExistException(
+ String.format(
+ "The table %s already exists in Iceberg
catalog, but the sort order is not compatible. "
+ + "Existing sort order: %s, new sort
order: %s. "
+ + "Please first drop the table in
Iceberg catalog, or pre-create the "
+ + "Iceberg table without a custom sort
order, or with ASC(%s) set explicitly, "
+ + "or use a new table name.",
+ tablePath,
+ existingTable.sortOrder(),
+ expectedSortOrder,
+ OFFSET_COLUMN_NAME),
+ e);
+ }
+
+ if (!isIcebergPropertiesCompatible(existingTable.properties(),
expectedProperties)) {
+ throw new TableAlreadyExistException(
+ String.format(
+ "The table %s already exists in Iceberg
catalog, but the table properties are not compatible. "
+ + "Existing properties: %s, new
properties: %s. "
+ + "Please first drop the table in
Iceberg catalog or use a new table name.",
+ tablePath, existingTable.properties(),
expectedProperties),
+ e);
+ }
+
+ // if creating a new fluss table, we should ensure the lake table
is empty
+ if (isCreatingFlussTable && existingTable.currentSnapshot() !=
null) {
+ throw new TableAlreadyExistException(
+ String.format(
+ "The table %s already exists in Iceberg
catalog, and the table is not empty. "
+ + "Please first drop the table in
Iceberg catalog or use a new table name.",
+ tablePath),
+ e);
+ }
}
}
+ @VisibleForTesting
+ boolean isIcebergSchemaCompatibleWithSchema(Schema existingSchema, Schema
newSchema) {
+ Schema normalizedExisting =
TypeUtil.assignIncreasingFreshIds(existingSchema);
+ Schema normalizedNew = TypeUtil.assignIncreasingFreshIds(newSchema);
+
+ List<Types.NestedField> existingFields = normalizedExisting.columns();
+ List<Types.NestedField> newFields = normalizedNew.columns();
+
+ if (existingFields.size() != newFields.size()) {
+ return false;
+ }
+
+ for (int i = 0; i < existingFields.size(); i++) {
+ Types.NestedField existing = existingFields.get(i);
+ Types.NestedField expected = newFields.get(i);
+ if (!existing.name().equals(expected.name())
+ || !existing.type().equals(expected.type())
+ || existing.isOptional() != expected.isOptional()) {
+ return false;
+ }
+ }
+
+ return true;
Review Comment:
Blocking: This compatibility check ignores Schema.identifierFieldIds(). A
pre-created Iceberg table with the same columns, partition spec, sort order,
and properties but no identifier fields therefore passes this check.
IcebergLakeWriter#createRecordWriter then sees an empty identifier set and
selects AppendOnlyTaskWriter, so UPDATE_BEFORE and DELETE records are appended
instead of emitted as equality deletes, breaking primary-key semantics. Please
compare identifier fields by resolved column names (raw IDs can differ between
schemas) and add coverage for missing or incorrect identifiers.
--
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]
