github-actions[bot] commented on code in PR #66627:
URL: https://github.com/apache/doris/pull/66627#discussion_r3752071581
##########
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();
Review Comment:
[P1] Pin the loaded table identity across refreshes
This refresh discards the identity of the `Table` returned by `loadTable`.
If a REST table is dropped and recreated under the same identifier before this
refresh (or between retry attempts), `RESTTableOperations` adopts the
replacement metadata and this code builds `AssertTableUUID` from the
replacement UUID, so the stale DROP can remove a column from the new table.
Capture the originally loaded identity before refreshing and reject any attempt
whose refreshed table differs; cover both the initial-refresh and retry
replacement cases.
##########
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)
Review Comment:
[P2] Make the synthetic partition name collision-free
For an unpartitioned table this always chooses
`_doris_schema_drop_fence_1000`. That is also a legal user column name; if an
unrelated column already has it, `PartitionSpec.Builder.alwaysNull` rejects the
target name because it belongs to a different source field, so dropping any
other column fails before commit. Choose a fence name after checking the
complete schema (or use a construction that guarantees no unrelated-field
collision), and add a regression for this valid schema.
##########
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:
[P2] Preserve schema-drop property cleanup
This combined commit bypasses `SchemaUpdate.commit()`'s
`applyChangesToMetadata` step. A normal Iceberg drop removes the deleted path's
metrics, Parquet bloom-filter, and Parquet stats overrides, but
`buildFrom(base).setCurrentSchema(...)` retains them. The drop therefore leaves
stale writer settings that can break a later metadata update or silently apply
if the same name is added again. Apply the same property transformation in this
metadata transaction and cover it with a configured-column drop test.
##########
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:
[P1] Do not remove a client-predicted spec ID after REST rebasing
`AssertLastAssignedPartitionId` does not fence the spec-ID namespace. A
concurrent non-default spec can reuse existing partition-field IDs while
consuming `fenceSpecId`, so every generated requirement still passes. When the
REST server applies this request, `AddPartitionSpec` renumbers the fence to the
next free spec ID, but this update still removes the original numeric ID: the
concurrent spec is deleted and the fence remains after its source column is
dropped. Tie removal to the actually added spec or reject this spec-ID rebase,
and add a regression where the concurrent spec does not advance
`lastAssignedPartitionId`.
##########
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();
Review Comment:
[P1] Preserve the original field identity across retries
Each retry resolves `path` against the latest schema. If another writer
drops the original field and adds or renames a different field to the same
path, the first attempt conflicts, but the retry resolves the replacement field
ID and successfully deletes it. The table UUID is unchanged, so pinning only
table identity does not prevent this ABA. Capture the originally targeted field
ID (and relevant subtree identity for parent drops), require every attempt to
resolve that same object, and add a same-path replacement regression.
##########
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);
Review Comment:
[P1] Preserve v1 partition IDs after the fence
On a never-partitioned format-v1 table this transient field is assigned ID
1000. `addPartitionSpec` advances the final `lastAssignedPartitionId` to 1000,
while `RemovePartitionSpecs` removes only the spec. After this drop, the first
real `updateSpec().addField(...)` therefore gets 1001; v1's sequential-ID check
requires the first field to be 1000, so that commit fails. Keep the fence from
leaving v1's counter ahead of its current sequential layout and add a
regression that drops an ordinary column, then adds the table's first partition
field.
--
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]
