github-actions[bot] commented on code in PR #68957:
URL: https://github.com/apache/doris/pull/68957#discussion_r4309971663
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterTableCommand.java:
##########
@@ -543,27 +555,68 @@ public <R, C> R accept(PlanVisitor<R, C> visitor, C
context) {
@Override
public void run(ConnectContext ctx, StmtExecutor executor) throws
Exception {
+ // The Lance mutation budget starts at statement entry and is shared
by admission
+ // (the snapshot read below), the executor's admission queue, the
dispatch and the wait.
+ long statementStartMs = System.currentTimeMillis();
validate(ctx);
Review Comment:
[P2] Bound the metadata reads performed during validation
With a cold schema cache (for example after REFRESH TABLE), validate() calls
LanceIndexMutationValidator.validateCreateIndex -> table.getColumn ->
LanceExternalTable.initSchema -> LanceCatalogClient.readTableSnapshot, which
performs namespace/Dataset IO synchronously on the statement thread. These
reads run before any bounded wait starts, so even a 1-second
lance_index_mutation_budget_cap_seconds can leave CREATE INDEX blocked for the
storage timeout; the cancel flag is only observed later. Admission can
similarly reload the schema via storedColumnName(). Enforce the shared deadline
for these resolution/schema reads as well, with native resources remaining
owned by their worker; recording the start time or checking the deadline after
a synchronous read does not bound the wait.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceIndexAdmission.java:
##########
@@ -129,35 +144,139 @@ static void admitCreate(SnapshotLoader loader,
LanceExternalCatalog catalog,
rejectIfAmbiguousLookupColumn(snapshot, def.getCols().get(0));
// 6. Schema contract v1 from the stored column name (never the raw
user spelling). The
// build itself is the admission-depth type validation against the
pinned snapshot.
- LanceSchemaContractBuilder.build(snapshot.getTopLevelFields(),
storedColumnName(table, def.getCols().get(0)));
- // Every validation above is preserved for the synchronous execution
path; until that
- // path lands, a mutation that would be admitted terminates with the
shared rejection.
- catalogMgr.withLanceIndexAdmission(catalog, target, () -> {
- LanceIndexMutationValidator.rejectUnsupportedOperation(
- def.isOrReplace() ? "CREATE OR REPLACE INDEX" : "CREATE
INDEX", "catalog tables");
- return null;
- });
+ String storedColumn = storedColumnName(table, def.getCols().get(0));
+ LanceIndexSchemaContract schemaContract =
LanceSchemaContractBuilder.build(
+ snapshot.getTopLevelFields(), storedColumn);
+ // 7. The capture-fenced handoff: produce the execution plan by local
work only. The
+ // statement executes it outside this critical section; the plan pins
the admitted
+ // dataset version and carries the entry-internal final preflight that
re-decides the
+ // authoritative verdict against current metadata before the single
dispatch.
+ return catalogMgr.withLanceIndexAdmission(catalog, target, () -> new
LanceIndexMutationPlan(
+ catalog,
+ def.isOrReplace() ? TLanceIndexMutationType.REPLACE :
TLanceIndexMutationType.CREATE,
+ snapshot.getDatasetUri(), snapshot.getDatasetVersion(),
normalizedName, displayName,
+ storedColumn, requestedAlgorithm(def), def.getProperties(),
ifNotExists, false,
+ schemaContract,
+ () -> withAdmissionBudget(deadlineMs, () -> {
+ catalogMgr.withLanceIndexAdmission(catalog, target, () ->
null);
+ LanceIndexAdmissionSnapshot current = loader.load(
+ catalog, db.getRemoteName(),
table.getRemoteName());
+ return catalogMgr.withLanceIndexAdmission(catalog, target,
() -> {
+ requireSameDataset(snapshot, current);
+ return decideCreateAgainstSnapshot(current, def,
ifNotExists, displayName, normalizedName);
+ });
+ })));
}
/**
* Admits a top-level DROP INDEX. The static name bounds
* ({@link LanceIndexMutationValidator#validateDropIndex}) must already
have passed.
+ *
+ * @return the execution plan for a drop that would be admitted, or {@code
null} when
+ * {@code IF EXISTS} resolved the statement to a genuine no-op
(nothing is dropped;
+ * the statement completes with the default OK packet)
*/
- public static void admitDrop(LanceExternalCatalog catalog,
LanceExternalDatabase db,
- LanceExternalTable table, String indexName, boolean ifExists)
throws Exception {
- admitDrop(DEFAULT_LOADER, catalog, db, table, indexName, ifExists);
+ @Nullable
+ public static LanceIndexMutationPlan admitDrop(LanceExternalCatalog
catalog,
+ LanceExternalDatabase db, LanceExternalTable table, String
indexName, boolean ifExists)
+ throws Exception {
+ return admitDrop(DEFAULT_LOADER, catalog, db, table, indexName,
ifExists);
+ }
+
+ public static LanceIndexMutationPlan admitDrop(LanceExternalCatalog
catalog,
+ LanceExternalDatabase db, LanceExternalTable table, String
indexName,
+ boolean ifExists, long deadlineMs) throws Exception {
+ return withAdmissionBudget(deadlineMs,
+ () -> admitDrop(DEFAULT_LOADER, catalog, db, table, indexName,
ifExists));
}
- static void admitDrop(SnapshotLoader loader, LanceExternalCatalog catalog,
+ @Nullable
+ static LanceIndexMutationPlan admitDrop(SnapshotLoader loader,
LanceExternalCatalog catalog,
LanceExternalDatabase db, LanceExternalTable table, String
indexName, boolean ifExists)
throws Exception {
// Fail cheap-first: the reserved prefix is rejected before target
capture and the
// snapshot read, so it costs no remote read.
String normalizedName = LanceIndexNameNormalizer.normalize(indexName);
+ long deadlineMs = LanceIndexInspectionExecutor.statementDeadlineMs();
LanceIndexMutationValidator.rejectIfReservedIndexName(indexName);
CatalogMgr catalogMgr = Env.getCurrentEnv().getCatalogMgr();
CatalogMgr.LanceIndexTarget target =
catalogMgr.captureLanceIndexTarget(catalog);
LanceIndexAdmissionSnapshot snapshot = loader.load(catalog,
db.getRemoteName(), table.getRemoteName());
+ if (decideDropAgainstSnapshot(snapshot, indexName, normalizedName,
ifExists)
+ == PreflightResult.NO_OP) {
+ // Same deliberate empty action as the CREATE no-op above: no
local work to run
+ // inside the fence, but the target revalidation still applies.
+ catalogMgr.withLanceIndexAdmission(catalog, target, () -> null);
+ return null;
+ }
+ // The capture-fenced handoff, exactly as on the CREATE side; a DROP
carries no column,
+ // no algorithm and no schema contract.
+ return catalogMgr.withLanceIndexAdmission(catalog, target, () -> new
LanceIndexMutationPlan(
+ catalog, TLanceIndexMutationType.DROP,
snapshot.getDatasetUri(),
+ snapshot.getDatasetVersion(), normalizedName, indexName, "",
"", null, false, ifExists,
+ null, () -> withAdmissionBudget(deadlineMs, () -> {
+ catalogMgr.withLanceIndexAdmission(catalog, target, () ->
null);
+ LanceIndexAdmissionSnapshot current = loader.load(
+ catalog, db.getRemoteName(),
table.getRemoteName());
+ return catalogMgr.withLanceIndexAdmission(catalog, target,
() -> {
+ requireSameDataset(snapshot, current);
+ return decideDropAgainstSnapshot(current, indexName,
normalizedName, ifExists);
+ });
+ })));
+ }
+
+ private static <T> T withAdmissionBudget(long deadlineMs, Callable<T>
action) throws Exception {
+ try {
+ return LanceIndexInspectionExecutor.withDeadline(deadlineMs,
action);
+ } catch (LanceIndexInspectionExecutor.MetadataReadTimeoutException e) {
Review Comment:
[P2] Preserve the timeout type through production error wrappers
This catch does not see the metadata reader timeout on the production path:
LanceCatalogClient.inspectTableIndexes() catches it and
LanceErrorMessages.failure() replaces it with a RuntimeException, then
LanceExternalCatalog.loadTableIndexAdmissionSnapshot() wraps it again. With
query_timeout=300 and the default 300-second mutation cap, the inspection
executor times out after its own 60-second limit while the statement deadline
still has time left, so withDeadline() rethrows the wrapper and this mapping is
bypassed. The initial admission then reaches StmtExecutor as ERR_UNKNOWN_ERROR
(1105), instead of the confirmed non-commit error (5108). Preserve the typed
timeout across the sanitized wrappers and cover a reader timeout shorter than
the statement budget; the new one-second test only covers exhaustion of the
outer deadline.
--
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]
