github-actions[bot] commented on code in PR #69408:
URL: https://github.com/apache/doris/pull/69408#discussion_r4349932512
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java:
##########
@@ -967,17 +966,198 @@ public void dropColumn(ExternalTable dorisTable,
ColumnPath columnPath, long upd
dropColumn(dorisTable, columnPath.getTopLevelName(), updateTime);
return;
}
+ SessionContext context = SessionContext.current();
try (IcebergExternalMetaCache.WritableTableLease lease =
acquireWritableTable(dorisTable)) {
- Table icebergTable = lease.getTable();
- ResolvedColumnPath resolvedPath =
validateNestedStructFieldPath(icebergTable.schema(), columnPath, "drop");
+ dropColumnSafely(lease, dorisTable, columnPath, context);
+ }
+ refreshAfterColumnDrop(dorisTable, updateTime);
+ }
- UpdateSchema updateSchema = icebergTable.updateSchema();
- updateSchema.deleteColumn(resolvedPath.getFullPath());
- commitTableUpdate(lease, updateSchema::commit,
- "Failed to drop nested column: " +
columnPath.getFullPath() + " from table: "
- + icebergTable.name());
+ private void dropColumnSafely(IcebergExternalMetaCache.WritableTableLease
lease,
+ ExternalTable dorisTable, ColumnPath path, SessionContext context)
+ throws UserException {
+ try {
+ // A refresh can perform authenticated catalog IO just like the
eventual commit. Keep target
+ // validation, refresh and commit inside the one catalog
generation retained by this lease.
+ lease.getAuthenticator().execute(() -> {
+ commitColumnDrop(lease, dorisTable, path, context);
+ return null;
+ });
+ } catch (UserException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new UserException("Failed to drop column: " +
path.getFullPath() + ", error message is: "
+ + ExceptionUtils.getRootCauseMessage(e), e);
+ }
+ }
+
+ private void refreshAfterColumnDrop(ExternalTable dorisTable, long
updateTime) throws UserException {
+ try {
+ refreshTable(dorisTable, updateTime);
+ } catch (RuntimeException e) {
+ throw new UserException("Iceberg column DROP committed, but Doris
metadata refresh failed for "
+ + dorisTable.getRemoteDbName() + "." +
dorisTable.getRemoteName()
+ + "; refresh the table before another operation: " +
e.getMessage(), e);
+ }
+ }
+
+ private void commitColumnDrop(IcebergExternalMetaCache.WritableTableLease
lease,
+ ExternalTable dorisTable, ColumnPath path, SessionContext context)
throws UserException {
+ Table table = lease.getTable();
+ if (!path.isNested()) {
+ validateRowLineageColumnMutation(table, path.getTopLevelName(),
"drop");
+ }
+ TableOperations operations = ((HasTableOperations) table).operations();
+ TableMetadata loaded = operations.current();
+ if (loaded == null) {
+ throw new UserException("Cannot drop column from an unloaded
Iceberg table: " + table.name());
+ }
+ ResolvedColumnPath loadedPath = resolveDropPath(loaded.schema(), path);
+ Set<Integer> loadedSubtreeIds = fieldSubtreeIds(loadedPath);
+
+ TableMetadata base = operations.refresh();
+ ResolvedColumnPath resolvedPath = validatePinnedDropIdentity(
+ table.name(), path, loaded, loadedSubtreeIds, base);
+ validateNotUsedByRetainedPartitionSpec(base, resolvedPath);
+
+ // Stage the standard SDK update against exactly the validated base.
Recapturing operations.current()
+ // here could instead delete a same-name replacement published on a
shared handle after refresh.
+ CapturingTableOperations schemaCapture = new
CapturingTableOperations(operations, base);
+ Table stagedTable = new BaseTable(schemaCapture, table.name());
+ if (!path.isNested()) {
+ validateRowLineageColumnMutation(stagedTable,
path.getTopLevelName(), "drop");
+ }
+
stagedTable.updateSchema().deleteColumn(resolvedPath.getFullPath()).commit();
+ TableMetadata schemaUpdate = schemaCapture.committed();
+ if (isRestTableOperations(operations)) {
+ // No retained spec references the target, so a new SDK partition
reference must allocate a field ID.
Review Comment:
[P1] Fence partition references restored by a stale SDK writer
The partition counter need not advance when a source reference is introduced
after validation. For a v2/v3 table with historical `x:1000` and
current/default `y:1001` specs, another client can prepare
`updateSpec().addNonDefaultSpec().addField("x")`, recycling 1000 from its
captured base into a new nondefault `{y,x}` spec. If
`expireSnapshots().cleanExpiredMetadata(true)` then removes the unused x-only
spec, this DROP refresh sees only y and permits deleting x. The paused client
can now commit its new `{y,x}` spec: its REST requirements still pass, and
`lastAssignedPartitionId` remains 1001. This DROP's
UUID/schema/column/partition requirements also all pass. Iceberg validates only
the default spec against the new schema, so the retained nondefault spec is
left referencing the deleted source, recreating the invalid partition-type
state this PR is meant to prevent. Protect the retained-spec set atomically at
commit (or use an equivalent full-metadata CAS), and add this stale-writer
/metadata-expiry case; the current race tests only cover newly allocated
partition IDs.
--
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]
