Gabriel39 commented on code in PR #66627:
URL: https://github.com/apache/doris/pull/66627#discussion_r3758740537
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolution.java:
##########
@@ -87,10 +96,75 @@ public static void addColumn(Table table,
ConnectorColumnPath path, IcebergColum
/** Drops the nested field at {@code path}; its parent must resolve to a
struct that contains the leaf. */
public static void dropColumn(Table table, ConnectorColumnPath path) {
- ResolvedColumnPath resolvedPath =
validateNestedStructFieldPath(table.schema(), path, "drop");
- UpdateSchema updateSchema = table.updateSchema();
- updateSchema.deleteColumn(resolvedPath.getFullPath());
- updateSchema.commit();
+ dropColumnWithPartitionSpecFence(table, path, true);
+ }
+
+ static void dropTopLevelColumn(Table table, String columnName) {
+ dropColumnWithPartitionSpecFence(table,
ConnectorColumnPath.of(columnName), false);
+ }
+
+ private static void dropColumnWithPartitionSpecFence(
+ Table table, ConnectorColumnPath path, boolean nested) {
+ TableOperations operations = ((HasTableOperations) table).operations();
+ TableMetadata initial = operations.refresh();
+ // A commit conflict must rerun resolution and retained-spec
validation against refreshed metadata.
+ Tasks.foreach(operations)
+ .retry(initial.propertyAsInt(
+ TableProperties.COMMIT_NUM_RETRIES,
TableProperties.COMMIT_NUM_RETRIES_DEFAULT))
+ .exponentialBackoff(
+
initial.propertyAsInt(TableProperties.COMMIT_MIN_RETRY_WAIT_MS,
+
TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT),
+
initial.propertyAsInt(TableProperties.COMMIT_MAX_RETRY_WAIT_MS,
+
TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT),
+
initial.propertyAsInt(TableProperties.COMMIT_TOTAL_RETRY_TIME_MS,
+
TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
+ 2.0)
+ .onlyRetryOn(CommitFailedException.class)
+ .run(ops -> commitDropAttempt(ops, table.name(), path,
nested));
+ }
+
+ private static void commitDropAttempt(
+ TableOperations operations, String tableName, ConnectorColumnPath
path, boolean nested) {
+ TableMetadata base = operations.refresh();
+ ResolvedColumnPath resolvedPath = nested
+ ? validateNestedStructFieldPath(base.schema(), path, "drop")
+ : resolveColumnPath(base.schema(), path, "drop");
+ validateNotUsedByRetainedPartitionSpec(base, resolvedPath);
+
+ Table attemptTable = new BaseTable(operations.temp(base), tableName);
+ Schema updatedSchema = attemptTable.updateSchema()
+ .deleteColumn(resolvedPath.getFullPath()).apply();
+ int fenceSpecId =
base.specs().stream().mapToInt(PartitionSpec::specId).max().orElse(-1) + 1;
+ String fenceFieldName = "_doris_schema_drop_fence_" +
(base.lastAssignedPartitionId() + 1);
+ PartitionSpec fenceSpec = PartitionSpec.builderFor(base.schema())
+ .withSpecId(fenceSpecId)
+ .alwaysNull(resolvedPath.getFullPath(), fenceFieldName)
+ .build();
+ TableMetadata.Builder builder =
TableMetadata.buildFrom(base).addPartitionSpec(fenceSpec);
+ new
MetadataUpdate.RemovePartitionSpecs(Set.of(fenceSpecId)).applyTo(builder);
Review Comment:
Fixed in 6848a1685f5 by removing the client-predicted add/remove spec
transaction. Iceberg REST cannot assert the spec-ID namespace, so schema DROP
now fails closed for REST operations before commit; direct catalogs use their
atomic metadata CAS. This prevents both concurrent-spec deletion and a
surviving fence.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolution.java:
##########
@@ -87,10 +96,75 @@ public static void addColumn(Table table,
ConnectorColumnPath path, IcebergColum
/** Drops the nested field at {@code path}; its parent must resolve to a
struct that contains the leaf. */
public static void dropColumn(Table table, ConnectorColumnPath path) {
- ResolvedColumnPath resolvedPath =
validateNestedStructFieldPath(table.schema(), path, "drop");
- UpdateSchema updateSchema = table.updateSchema();
- updateSchema.deleteColumn(resolvedPath.getFullPath());
- updateSchema.commit();
+ dropColumnWithPartitionSpecFence(table, path, true);
+ }
+
+ static void dropTopLevelColumn(Table table, String columnName) {
+ dropColumnWithPartitionSpecFence(table,
ConnectorColumnPath.of(columnName), false);
+ }
+
+ private static void dropColumnWithPartitionSpecFence(
+ Table table, ConnectorColumnPath path, boolean nested) {
+ TableOperations operations = ((HasTableOperations) table).operations();
+ TableMetadata initial = operations.refresh();
+ // A commit conflict must rerun resolution and retained-spec
validation against refreshed metadata.
+ Tasks.foreach(operations)
+ .retry(initial.propertyAsInt(
+ TableProperties.COMMIT_NUM_RETRIES,
TableProperties.COMMIT_NUM_RETRIES_DEFAULT))
+ .exponentialBackoff(
+
initial.propertyAsInt(TableProperties.COMMIT_MIN_RETRY_WAIT_MS,
+
TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT),
+
initial.propertyAsInt(TableProperties.COMMIT_MAX_RETRY_WAIT_MS,
+
TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT),
+
initial.propertyAsInt(TableProperties.COMMIT_TOTAL_RETRY_TIME_MS,
+
TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
+ 2.0)
+ .onlyRetryOn(CommitFailedException.class)
+ .run(ops -> commitDropAttempt(ops, table.name(), path,
nested));
+ }
+
+ private static void commitDropAttempt(
+ TableOperations operations, String tableName, ConnectorColumnPath
path, boolean nested) {
+ TableMetadata base = operations.refresh();
+ ResolvedColumnPath resolvedPath = nested
+ ? validateNestedStructFieldPath(base.schema(), path, "drop")
+ : resolveColumnPath(base.schema(), path, "drop");
+ validateNotUsedByRetainedPartitionSpec(base, resolvedPath);
+
+ Table attemptTable = new BaseTable(operations.temp(base), tableName);
+ Schema updatedSchema = attemptTable.updateSchema()
+ .deleteColumn(resolvedPath.getFullPath()).apply();
+ int fenceSpecId =
base.specs().stream().mapToInt(PartitionSpec::specId).max().orElse(-1) + 1;
+ String fenceFieldName = "_doris_schema_drop_fence_" +
(base.lastAssignedPartitionId() + 1);
+ PartitionSpec fenceSpec = PartitionSpec.builderFor(base.schema())
+ .withSpecId(fenceSpecId)
+ .alwaysNull(resolvedPath.getFullPath(), fenceFieldName)
+ .build();
+ TableMetadata.Builder builder =
TableMetadata.buildFrom(base).addPartitionSpec(fenceSpec);
+ new
MetadataUpdate.RemovePartitionSpecs(Set.of(fenceSpecId)).applyTo(builder);
+ builder.setCurrentSchema(updatedSchema, base.lastColumnId());
Review Comment:
Fixed in 6848a1685f5. Non-REST DROP now goes through Iceberg's standard
SchemaUpdate.commit(), preserving its schema-linked property and name-mapping
transformations. Tests cover metrics, Parquet bloom-filter, and Parquet
column-stats property cleanup.
--
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]
