u70b3 commented on code in PR #67754: URL: https://github.com/apache/doris/pull/67754#discussion_r4141831092
########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ResolveLanceIndexJobCommand.java: ########## @@ -0,0 +1,367 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.nereids.trees.plans.commands; + +import org.apache.doris.catalog.DatabaseIf; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.DdlException; +import org.apache.doris.common.ErrorCode; +import org.apache.doris.datasource.CatalogIf; +import org.apache.doris.datasource.CatalogMgr; +import org.apache.doris.datasource.ExternalDatabase; +import org.apache.doris.datasource.ExternalTable; +import org.apache.doris.datasource.lance.LanceExternalCatalog; +import org.apache.doris.datasource.lance.LanceIndexMutationValidator; +import org.apache.doris.datasource.lance.job.LanceIndexJob; +import org.apache.doris.datasource.lance.job.LanceIndexJobManager; +import org.apache.doris.datasource.lance.job.LanceIndexJobMutationState; +import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.StmtExecutor; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.nio.charset.StandardCharsets; + +/** + * RESOLVE LANCE INDEX JOB <jobId> AS FORCE_RELEASE COMMENT '<note>' — the operator + * escape hatch that durably releases a job whose mutation outcome is UNKNOWN (design section + * 7.1). RESOLVE is deliberately not gated by {@code enable_lance_index_mutation}: the gate + * controls mutation admission, while FORCE must stay available exactly when the gate is off. + * + * <p>The release protocol keeps the fence, the quota charge and the possible-live slot while + * it performs one authoritative latest-metadata read and one external-table refresh with the + * current credentials of the surviving catalog, both outside every catalog/manager lock; only + * then does the durable release transfer inside the admission critical section + * ({@code captureLanceIndexTarget} → lock-free read/refresh → {@code withLanceIndexAdmission} + * recheck → manager write lock), serialized against DROP CATALOG and identity ALTER exactly + * like admission. Any failure before the transfer is the typed + * {@code ERR_LANCE_INDEX_JOB_RESOLUTION_INCOMPLETE}: nothing is written, nothing is released, + * and the operator fixes the cause and retries the same statement. + * + * <p>Target resolution (design section 7.1 step 1) is three-valued. RESOLVED means the + * persisted names resolve and the catalog's current durable dataset locator still matches + * the job's — the same revalidation SHOW LANCE INDEX JOBS applies, so a repointed dataset + * reusing the same names never turns a stale name into table-level authorization. MISSING + * means the catalog, database or table is verifiably absent, or the locator positively + * points at a different dataset: that is the orphan family — a full orphan (catalog gone) + * has no credentials to read with and nothing to invalidate, so it is released directly + * after global ADMIN authorization, while a half-orphan skips the authoritative read and + * refreshes with {@code ignoreIfNotExists=true} as a best-effort invalidation. FAILED means + * a resolution that errors out, or a locator that cannot be resolved right now: never an + * orphan verdict — after ADMIN authorization the statement fails with the typed 5105 so the + * fence is kept when "table gone" cannot be told apart from "network down". SHOW fails the + * same uncertainty closed by hiding the row; RESOLVE fails it closed by not releasing. + * + * <p>Non-disclosure (design section 8): the job is loaded first and authorized against its + * persisted target — table-level ALTER when the target resolves, global ADMIN otherwise — and + * a missing job and an unauthorized job share the same fixed ERR_LANCE_INDEX_JOB_NOT_FOUND + * response naming only the job id. The 5104 state rejection and the 5105 resolution failure + * are only visible to an already authorized caller. + * + * <p>Success returns an OK packet carrying one warning row with {@link #LATE_COMMIT_WARNING}, + * the same text persisted as the job's durable {@code forceWarning}: the old worker may still + * overwrite, remove, or reintroduce the index name; the mutation outcome remains UNKNOWN. + * Retrying FORCE on an already released job is an idempotent success returning the existing + * release record, never an error. + */ +public class ResolveLanceIndexJobCommand extends Command implements ForwardWithSync { + /** + * The late-commit warning (design section 7.1), returned in the OK packet and persisted + * verbatim as the durable {@code forceWarning}; bounded well under + * {@link LanceIndexJob#MAX_FORCE_TEXT_BYTES}. + */ + static final String LATE_COMMIT_WARNING = + "the old worker may still overwrite, remove, or reintroduce the index name; " + + "the mutation outcome remains UNKNOWN"; + + private static final Logger LOG = LogManager.getLogger(ResolveLanceIndexJobCommand.class); + + private final long jobId; + private final String comment; + + public ResolveLanceIndexJobCommand(long jobId, String comment) { + super(PlanType.RESOLVE_LANCE_INDEX_JOB_COMMAND); + this.jobId = jobId; + this.comment = comment; + } + + public long getJobId() { + return jobId; + } + + public String getComment() { + return comment; + } + + @Override + public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { + Env env = Env.getCurrentEnv(); + LanceIndexJobManager manager = env.getLanceIndexJobManager(); + // 1. Load the job without disclosing any field (design section 7.1 step 1). + LanceIndexJob job = manager.getJob(jobId); + if (job == null) { + throw notFound(); + } + // 2. Resolve and authorize against the persisted target before any state is revealed: + // table-level ALTER when the target resolves, global ADMIN for the orphan family + // and for a target whose resolution failed outright. + CatalogMgr catalogMgr = env.getCatalogMgr(); + CatalogIf<? extends DatabaseIf<? extends TableIf>> catalog = catalogMgr.getCatalog(job.getCatalogId()); + TargetResolution resolution = resolveTarget(catalog, job); + boolean authorized = resolution == TargetResolution.RESOLVED + ? env.getAccessManager().checkTblPriv(ctx, catalog.getName(), job.getDbName(), job.getTableName(), + PrivPredicate.ALTER) + : env.getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN); + if (!authorized) { + throw notFound(); + } + // 3. Idempotent replay: a retry returns the existing release record (section 7.1). + // This deliberately precedes the resolution-failure rejection: once the release + // has landed, a retry during a provider outage is a success, not a 5105. + if (job.isForceReleased()) { + ctx.getState().setOk(0, 1, LATE_COMMIT_WARNING); + return; + } + // 4. Only UNKNOWN may be force-released; a null mutation state reads as UNKNOWN, + // same as the manager's own gate. The state rejection also precedes the + // resolution-failure rejection: for a terminal job the accurate answer is 5104, + // not a 5105 claiming the job still holds its fence. + if (job.getMutationState() != null && job.getMutationState() != LanceIndexJobMutationState.UNKNOWN) { + throw new AnalysisException(ErrorCode.ERR_LANCE_INDEX_JOB_NOT_UNKNOWN.formatErrorMsg(jobId), + ErrorCode.ERR_LANCE_INDEX_JOB_NOT_UNKNOWN); + } + if (resolution == TargetResolution.FAILED) { + // Never an orphan verdict: "table gone" cannot be told apart from "network down", + // so nothing is released and nothing beyond the typed error is disclosed; the + // operator fixes the cause and retries the same statement (design 7.1 step 4). + throw incompleteResolution("the persisted target could not be resolved with current catalog" + + " metadata; see fe.log for the cause"); + } + boolean targetResolves = resolution == TargetResolution.RESOLVED; + // 5. The grammar makes COMMENT mandatory; here the note must also be non-empty after + // trimming and fit the durable force text bound. + String note = comment == null ? "" : comment.trim(); + if (note.isEmpty()) { + throw new AnalysisException("force release note must not be empty", + ErrorCode.ERR_LANCE_INDEX_INVALID); + } + if (note.getBytes(StandardCharsets.UTF_8).length > LanceIndexJob.MAX_FORCE_TEXT_BYTES) { + throw new AnalysisException("force release note exceeds " + LanceIndexJob.MAX_FORCE_TEXT_BYTES + + " UTF-8 bytes", ErrorCode.ERR_LANCE_INDEX_INVALID); + } + // 6-9. Branch on the orphan state, then the durable release transfer. + String actor = ctx.getQualifiedUser(); + boolean released; + if (catalog == null) { + // Full orphan: no credentials survive to read with and nothing can be + // invalidated, so the release goes straight to the manager write lock. + released = manager.forceRelease(jobId, job.getRevision(), actor, note, LATE_COMMIT_WARNING); + } else { + released = releaseWithLiveCatalog(env, catalogMgr, manager, catalog, targetResolves, job, actor, note); + } + if (!released) { + // 10. The expected-revision transfer lost a race. A concurrent FORCE_RELEASE + // that already landed makes this an idempotent success; anything else means + // the job left UNKNOWN concurrently (UNKNOWN has no other outgoing + // transition), so the pinned not-UNKNOWN wording stays accurate. + LanceIndexJob reread = manager.getJob(jobId); + if (reread != null && reread.isForceReleased()) { + ctx.getState().setOk(0, 1, LATE_COMMIT_WARNING); + return; + } + throw new AnalysisException(ErrorCode.ERR_LANCE_INDEX_JOB_NOT_UNKNOWN.formatErrorMsg(jobId), + ErrorCode.ERR_LANCE_INDEX_JOB_NOT_UNKNOWN); + } + // 11. The OK packet carries the late-commit warning; it survives the forward chain + // byte-identically (proxyExecute serializes the master state). + ctx.getState().setOk(0, 1, LATE_COMMIT_WARNING); + } + + /** + * The live-catalog release path: capture the target identity, then — holding no lock — + * perform the authoritative read and the refresh, and finally transfer the release inside + * the admission critical section. Lock order stays CatalogMgr then LanceIndexJobManager. + */ + private boolean releaseWithLiveCatalog(Env env, CatalogMgr catalogMgr, LanceIndexJobManager manager, + CatalogIf<? extends DatabaseIf<? extends TableIf>> catalog, boolean targetResolves, LanceIndexJob job, + String actor, String note) throws Exception { + if (!(catalog instanceof LanceExternalCatalog) + || ((LanceExternalCatalog) catalog).isRestCatalogConfigured()) { + // Defensive: admission never targets a REST catalog and a non-Lance catalog + // cannot hold a Lance fence, so no UNKNOWN job should ever resolve here. + LanceIndexMutationValidator.rejectUnsupportedOperation( + "RESOLVE LANCE INDEX JOB AS FORCE_RELEASE", "catalog '" + catalog.getName() + "'"); + } + LanceExternalCatalog lanceCatalog = (LanceExternalCatalog) catalog; + CatalogMgr.LanceIndexTarget target; + try { + target = catalogMgr.captureLanceIndexTarget(lanceCatalog); + } catch (DdlException e) { + throw incompleteResolution(e.getMessage()); + } + if (targetResolves) { + // 7. One authoritative latest-metadata read with current credentials, proving + // the dataset is reachable before anything is released. + authoritativeRead(lanceCatalog, catalog, job); + } + // 8. Invalidate the external table and broadcast the refresh to every FE. A + // half-orphan target is invalidated best-effort (missing db/table is a no-op). + try { + env.getRefreshManager().handleRefreshTable(catalog.getName(), job.getDbName(), job.getTableName(), + !targetResolves); + } catch (DdlException e) { + throw incompleteResolution(e.getMessage()); + } + // 9. The durable transfer rechecks the catalog identity under the read lock, then + // runs the revision-checked release under the manager write lock. + try { + return catalogMgr.withLanceIndexAdmission(lanceCatalog, target, Review Comment: done: the authorized catalog NAME is captured at step 2 and rechecked inside the withLanceIndexAdmission critical section; a rename after authorization throws a retryable resolution-incomplete instead of releasing under the new name (testRenameAfterAuthorizationKeepsTheFence, 59d120c268). ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ResolveLanceIndexJobCommand.java: ########## @@ -0,0 +1,367 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.nereids.trees.plans.commands; + +import org.apache.doris.catalog.DatabaseIf; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.DdlException; +import org.apache.doris.common.ErrorCode; +import org.apache.doris.datasource.CatalogIf; +import org.apache.doris.datasource.CatalogMgr; +import org.apache.doris.datasource.ExternalDatabase; +import org.apache.doris.datasource.ExternalTable; +import org.apache.doris.datasource.lance.LanceExternalCatalog; +import org.apache.doris.datasource.lance.LanceIndexMutationValidator; +import org.apache.doris.datasource.lance.job.LanceIndexJob; +import org.apache.doris.datasource.lance.job.LanceIndexJobManager; +import org.apache.doris.datasource.lance.job.LanceIndexJobMutationState; +import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.StmtExecutor; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.nio.charset.StandardCharsets; + +/** + * RESOLVE LANCE INDEX JOB <jobId> AS FORCE_RELEASE COMMENT '<note>' — the operator + * escape hatch that durably releases a job whose mutation outcome is UNKNOWN (design section + * 7.1). RESOLVE is deliberately not gated by {@code enable_lance_index_mutation}: the gate + * controls mutation admission, while FORCE must stay available exactly when the gate is off. + * + * <p>The release protocol keeps the fence, the quota charge and the possible-live slot while + * it performs one authoritative latest-metadata read and one external-table refresh with the + * current credentials of the surviving catalog, both outside every catalog/manager lock; only + * then does the durable release transfer inside the admission critical section + * ({@code captureLanceIndexTarget} → lock-free read/refresh → {@code withLanceIndexAdmission} + * recheck → manager write lock), serialized against DROP CATALOG and identity ALTER exactly + * like admission. Any failure before the transfer is the typed + * {@code ERR_LANCE_INDEX_JOB_RESOLUTION_INCOMPLETE}: nothing is written, nothing is released, + * and the operator fixes the cause and retries the same statement. + * + * <p>Target resolution (design section 7.1 step 1) is three-valued. RESOLVED means the + * persisted names resolve and the catalog's current durable dataset locator still matches + * the job's — the same revalidation SHOW LANCE INDEX JOBS applies, so a repointed dataset + * reusing the same names never turns a stale name into table-level authorization. MISSING + * means the catalog, database or table is verifiably absent, or the locator positively + * points at a different dataset: that is the orphan family — a full orphan (catalog gone) + * has no credentials to read with and nothing to invalidate, so it is released directly + * after global ADMIN authorization, while a half-orphan skips the authoritative read and + * refreshes with {@code ignoreIfNotExists=true} as a best-effort invalidation. FAILED means + * a resolution that errors out, or a locator that cannot be resolved right now: never an + * orphan verdict — after ADMIN authorization the statement fails with the typed 5105 so the + * fence is kept when "table gone" cannot be told apart from "network down". SHOW fails the + * same uncertainty closed by hiding the row; RESOLVE fails it closed by not releasing. + * + * <p>Non-disclosure (design section 8): the job is loaded first and authorized against its + * persisted target — table-level ALTER when the target resolves, global ADMIN otherwise — and + * a missing job and an unauthorized job share the same fixed ERR_LANCE_INDEX_JOB_NOT_FOUND + * response naming only the job id. The 5104 state rejection and the 5105 resolution failure + * are only visible to an already authorized caller. + * + * <p>Success returns an OK packet carrying one warning row with {@link #LATE_COMMIT_WARNING}, + * the same text persisted as the job's durable {@code forceWarning}: the old worker may still + * overwrite, remove, or reintroduce the index name; the mutation outcome remains UNKNOWN. + * Retrying FORCE on an already released job is an idempotent success returning the existing + * release record, never an error. + */ +public class ResolveLanceIndexJobCommand extends Command implements ForwardWithSync { + /** + * The late-commit warning (design section 7.1), returned in the OK packet and persisted + * verbatim as the durable {@code forceWarning}; bounded well under + * {@link LanceIndexJob#MAX_FORCE_TEXT_BYTES}. + */ + static final String LATE_COMMIT_WARNING = + "the old worker may still overwrite, remove, or reintroduce the index name; " + + "the mutation outcome remains UNKNOWN"; + + private static final Logger LOG = LogManager.getLogger(ResolveLanceIndexJobCommand.class); + + private final long jobId; + private final String comment; + + public ResolveLanceIndexJobCommand(long jobId, String comment) { + super(PlanType.RESOLVE_LANCE_INDEX_JOB_COMMAND); + this.jobId = jobId; + this.comment = comment; + } + + public long getJobId() { + return jobId; + } + + public String getComment() { + return comment; + } + + @Override + public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { + Env env = Env.getCurrentEnv(); + LanceIndexJobManager manager = env.getLanceIndexJobManager(); + // 1. Load the job without disclosing any field (design section 7.1 step 1). + LanceIndexJob job = manager.getJob(jobId); + if (job == null) { + throw notFound(); + } + // 2. Resolve and authorize against the persisted target before any state is revealed: + // table-level ALTER when the target resolves, global ADMIN for the orphan family + // and for a target whose resolution failed outright. + CatalogMgr catalogMgr = env.getCatalogMgr(); + CatalogIf<? extends DatabaseIf<? extends TableIf>> catalog = catalogMgr.getCatalog(job.getCatalogId()); + TargetResolution resolution = resolveTarget(catalog, job); + boolean authorized = resolution == TargetResolution.RESOLVED + ? env.getAccessManager().checkTblPriv(ctx, catalog.getName(), job.getDbName(), job.getTableName(), + PrivPredicate.ALTER) + : env.getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN); + if (!authorized) { + throw notFound(); + } + // 3. Idempotent replay: a retry returns the existing release record (section 7.1). + // This deliberately precedes the resolution-failure rejection: once the release + // has landed, a retry during a provider outage is a success, not a 5105. + if (job.isForceReleased()) { + ctx.getState().setOk(0, 1, LATE_COMMIT_WARNING); + return; + } + // 4. Only UNKNOWN may be force-released; a null mutation state reads as UNKNOWN, + // same as the manager's own gate. The state rejection also precedes the + // resolution-failure rejection: for a terminal job the accurate answer is 5104, + // not a 5105 claiming the job still holds its fence. + if (job.getMutationState() != null && job.getMutationState() != LanceIndexJobMutationState.UNKNOWN) { + throw new AnalysisException(ErrorCode.ERR_LANCE_INDEX_JOB_NOT_UNKNOWN.formatErrorMsg(jobId), + ErrorCode.ERR_LANCE_INDEX_JOB_NOT_UNKNOWN); + } + if (resolution == TargetResolution.FAILED) { + // Never an orphan verdict: "table gone" cannot be told apart from "network down", + // so nothing is released and nothing beyond the typed error is disclosed; the + // operator fixes the cause and retries the same statement (design 7.1 step 4). + throw incompleteResolution("the persisted target could not be resolved with current catalog" + + " metadata; see fe.log for the cause"); + } + boolean targetResolves = resolution == TargetResolution.RESOLVED; + // 5. The grammar makes COMMENT mandatory; here the note must also be non-empty after + // trimming and fit the durable force text bound. + String note = comment == null ? "" : comment.trim(); + if (note.isEmpty()) { + throw new AnalysisException("force release note must not be empty", + ErrorCode.ERR_LANCE_INDEX_INVALID); + } + if (note.getBytes(StandardCharsets.UTF_8).length > LanceIndexJob.MAX_FORCE_TEXT_BYTES) { + throw new AnalysisException("force release note exceeds " + LanceIndexJob.MAX_FORCE_TEXT_BYTES + + " UTF-8 bytes", ErrorCode.ERR_LANCE_INDEX_INVALID); + } + // 6-9. Branch on the orphan state, then the durable release transfer. + String actor = ctx.getQualifiedUser(); + boolean released; + if (catalog == null) { + // Full orphan: no credentials survive to read with and nothing can be + // invalidated, so the release goes straight to the manager write lock. + released = manager.forceRelease(jobId, job.getRevision(), actor, note, LATE_COMMIT_WARNING); + } else { + released = releaseWithLiveCatalog(env, catalogMgr, manager, catalog, targetResolves, job, actor, note); + } + if (!released) { + // 10. The expected-revision transfer lost a race. A concurrent FORCE_RELEASE + // that already landed makes this an idempotent success; anything else means + // the job left UNKNOWN concurrently (UNKNOWN has no other outgoing + // transition), so the pinned not-UNKNOWN wording stays accurate. + LanceIndexJob reread = manager.getJob(jobId); + if (reread != null && reread.isForceReleased()) { + ctx.getState().setOk(0, 1, LATE_COMMIT_WARNING); + return; + } + throw new AnalysisException(ErrorCode.ERR_LANCE_INDEX_JOB_NOT_UNKNOWN.formatErrorMsg(jobId), + ErrorCode.ERR_LANCE_INDEX_JOB_NOT_UNKNOWN); + } + // 11. The OK packet carries the late-commit warning; it survives the forward chain + // byte-identically (proxyExecute serializes the master state). + ctx.getState().setOk(0, 1, LATE_COMMIT_WARNING); + } + + /** + * The live-catalog release path: capture the target identity, then — holding no lock — + * perform the authoritative read and the refresh, and finally transfer the release inside + * the admission critical section. Lock order stays CatalogMgr then LanceIndexJobManager. + */ + private boolean releaseWithLiveCatalog(Env env, CatalogMgr catalogMgr, LanceIndexJobManager manager, + CatalogIf<? extends DatabaseIf<? extends TableIf>> catalog, boolean targetResolves, LanceIndexJob job, + String actor, String note) throws Exception { + if (!(catalog instanceof LanceExternalCatalog) + || ((LanceExternalCatalog) catalog).isRestCatalogConfigured()) { + // Defensive: admission never targets a REST catalog and a non-Lance catalog + // cannot hold a Lance fence, so no UNKNOWN job should ever resolve here. + LanceIndexMutationValidator.rejectUnsupportedOperation( + "RESOLVE LANCE INDEX JOB AS FORCE_RELEASE", "catalog '" + catalog.getName() + "'"); + } + LanceExternalCatalog lanceCatalog = (LanceExternalCatalog) catalog; + CatalogMgr.LanceIndexTarget target; + try { + target = catalogMgr.captureLanceIndexTarget(lanceCatalog); + } catch (DdlException e) { + throw incompleteResolution(e.getMessage()); + } + if (targetResolves) { + // 7. One authoritative latest-metadata read with current credentials, proving + // the dataset is reachable before anything is released. + authoritativeRead(lanceCatalog, catalog, job); + } + // 8. Invalidate the external table and broadcast the refresh to every FE. A + // half-orphan target is invalidated best-effort (missing db/table is a no-op). + try { + env.getRefreshManager().handleRefreshTable(catalog.getName(), job.getDbName(), job.getTableName(), + !targetResolves); + } catch (DdlException e) { Review Comment: done: the refresh leg catches RuntimeException at the boundary, logs the cause, and returns the typed 5105 with the job unresolved (testRefreshRuntimeExceptionKeepsJobUnresolved, 59d120c268). The refresh is also addressed by the persisted catalog id now (the name-based resolution had its own rename race). ########## fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java: ########## @@ -7067,6 +7069,13 @@ public LogicalPlan visitShowLanceIndexJob(ShowLanceIndexJobContext ctx) { return new ShowLanceIndexJobCommand(jobId); } + @Override + public LogicalPlan visitResolveLanceIndexJob(ResolveLanceIndexJobContext ctx) { + long jobId = Long.parseLong(ctx.jobId.getText()); + String comment = stripQuotes(ctx.comment.getText()); Review Comment: done: the visitor decodes through SqlLiteralUtils.parseStringLiteral like the nearby comment builders, so doubled quotes and backslash escapes decode before the durable bound check (8c9fadc14f, parser cases included). ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ResolveLanceIndexJobCommand.java: ########## @@ -0,0 +1,367 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.nereids.trees.plans.commands; + +import org.apache.doris.catalog.DatabaseIf; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.DdlException; +import org.apache.doris.common.ErrorCode; +import org.apache.doris.datasource.CatalogIf; +import org.apache.doris.datasource.CatalogMgr; +import org.apache.doris.datasource.ExternalDatabase; +import org.apache.doris.datasource.ExternalTable; +import org.apache.doris.datasource.lance.LanceExternalCatalog; +import org.apache.doris.datasource.lance.LanceIndexMutationValidator; +import org.apache.doris.datasource.lance.job.LanceIndexJob; +import org.apache.doris.datasource.lance.job.LanceIndexJobManager; +import org.apache.doris.datasource.lance.job.LanceIndexJobMutationState; +import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.StmtExecutor; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.nio.charset.StandardCharsets; + +/** + * RESOLVE LANCE INDEX JOB <jobId> AS FORCE_RELEASE COMMENT '<note>' — the operator + * escape hatch that durably releases a job whose mutation outcome is UNKNOWN (design section + * 7.1). RESOLVE is deliberately not gated by {@code enable_lance_index_mutation}: the gate + * controls mutation admission, while FORCE must stay available exactly when the gate is off. + * + * <p>The release protocol keeps the fence, the quota charge and the possible-live slot while + * it performs one authoritative latest-metadata read and one external-table refresh with the + * current credentials of the surviving catalog, both outside every catalog/manager lock; only + * then does the durable release transfer inside the admission critical section + * ({@code captureLanceIndexTarget} → lock-free read/refresh → {@code withLanceIndexAdmission} + * recheck → manager write lock), serialized against DROP CATALOG and identity ALTER exactly + * like admission. Any failure before the transfer is the typed + * {@code ERR_LANCE_INDEX_JOB_RESOLUTION_INCOMPLETE}: nothing is written, nothing is released, + * and the operator fixes the cause and retries the same statement. + * + * <p>Target resolution (design section 7.1 step 1) is three-valued. RESOLVED means the + * persisted names resolve and the catalog's current durable dataset locator still matches + * the job's — the same revalidation SHOW LANCE INDEX JOBS applies, so a repointed dataset + * reusing the same names never turns a stale name into table-level authorization. MISSING + * means the catalog, database or table is verifiably absent, or the locator positively + * points at a different dataset: that is the orphan family — a full orphan (catalog gone) + * has no credentials to read with and nothing to invalidate, so it is released directly + * after global ADMIN authorization, while a half-orphan skips the authoritative read and + * refreshes with {@code ignoreIfNotExists=true} as a best-effort invalidation. FAILED means + * a resolution that errors out, or a locator that cannot be resolved right now: never an + * orphan verdict — after ADMIN authorization the statement fails with the typed 5105 so the + * fence is kept when "table gone" cannot be told apart from "network down". SHOW fails the + * same uncertainty closed by hiding the row; RESOLVE fails it closed by not releasing. + * + * <p>Non-disclosure (design section 8): the job is loaded first and authorized against its + * persisted target — table-level ALTER when the target resolves, global ADMIN otherwise — and + * a missing job and an unauthorized job share the same fixed ERR_LANCE_INDEX_JOB_NOT_FOUND + * response naming only the job id. The 5104 state rejection and the 5105 resolution failure + * are only visible to an already authorized caller. + * + * <p>Success returns an OK packet carrying one warning row with {@link #LATE_COMMIT_WARNING}, + * the same text persisted as the job's durable {@code forceWarning}: the old worker may still + * overwrite, remove, or reintroduce the index name; the mutation outcome remains UNKNOWN. + * Retrying FORCE on an already released job is an idempotent success returning the existing + * release record, never an error. + */ +public class ResolveLanceIndexJobCommand extends Command implements ForwardWithSync { + /** + * The late-commit warning (design section 7.1), returned in the OK packet and persisted + * verbatim as the durable {@code forceWarning}; bounded well under + * {@link LanceIndexJob#MAX_FORCE_TEXT_BYTES}. + */ + static final String LATE_COMMIT_WARNING = + "the old worker may still overwrite, remove, or reintroduce the index name; " + + "the mutation outcome remains UNKNOWN"; + + private static final Logger LOG = LogManager.getLogger(ResolveLanceIndexJobCommand.class); + + private final long jobId; + private final String comment; + + public ResolveLanceIndexJobCommand(long jobId, String comment) { + super(PlanType.RESOLVE_LANCE_INDEX_JOB_COMMAND); + this.jobId = jobId; + this.comment = comment; + } + + public long getJobId() { + return jobId; + } + + public String getComment() { + return comment; + } + + @Override + public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { + Env env = Env.getCurrentEnv(); + LanceIndexJobManager manager = env.getLanceIndexJobManager(); + // 1. Load the job without disclosing any field (design section 7.1 step 1). + LanceIndexJob job = manager.getJob(jobId); + if (job == null) { + throw notFound(); + } + // 2. Resolve and authorize against the persisted target before any state is revealed: + // table-level ALTER when the target resolves, global ADMIN for the orphan family + // and for a target whose resolution failed outright. + CatalogMgr catalogMgr = env.getCatalogMgr(); + CatalogIf<? extends DatabaseIf<? extends TableIf>> catalog = catalogMgr.getCatalog(job.getCatalogId()); + TargetResolution resolution = resolveTarget(catalog, job); + boolean authorized = resolution == TargetResolution.RESOLVED + ? env.getAccessManager().checkTblPriv(ctx, catalog.getName(), job.getDbName(), job.getTableName(), + PrivPredicate.ALTER) + : env.getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN); + if (!authorized) { + throw notFound(); + } + // 3. Idempotent replay: a retry returns the existing release record (section 7.1). + // This deliberately precedes the resolution-failure rejection: once the release + // has landed, a retry during a provider outage is a success, not a 5105. + if (job.isForceReleased()) { + ctx.getState().setOk(0, 1, LATE_COMMIT_WARNING); + return; + } + // 4. Only UNKNOWN may be force-released; a null mutation state reads as UNKNOWN, + // same as the manager's own gate. The state rejection also precedes the + // resolution-failure rejection: for a terminal job the accurate answer is 5104, + // not a 5105 claiming the job still holds its fence. + if (job.getMutationState() != null && job.getMutationState() != LanceIndexJobMutationState.UNKNOWN) { + throw new AnalysisException(ErrorCode.ERR_LANCE_INDEX_JOB_NOT_UNKNOWN.formatErrorMsg(jobId), + ErrorCode.ERR_LANCE_INDEX_JOB_NOT_UNKNOWN); + } + if (resolution == TargetResolution.FAILED) { + // Never an orphan verdict: "table gone" cannot be told apart from "network down", + // so nothing is released and nothing beyond the typed error is disclosed; the + // operator fixes the cause and retries the same statement (design 7.1 step 4). + throw incompleteResolution("the persisted target could not be resolved with current catalog" + + " metadata; see fe.log for the cause"); + } + boolean targetResolves = resolution == TargetResolution.RESOLVED; + // 5. The grammar makes COMMENT mandatory; here the note must also be non-empty after + // trimming and fit the durable force text bound. + String note = comment == null ? "" : comment.trim(); + if (note.isEmpty()) { + throw new AnalysisException("force release note must not be empty", + ErrorCode.ERR_LANCE_INDEX_INVALID); + } + if (note.getBytes(StandardCharsets.UTF_8).length > LanceIndexJob.MAX_FORCE_TEXT_BYTES) { + throw new AnalysisException("force release note exceeds " + LanceIndexJob.MAX_FORCE_TEXT_BYTES + + " UTF-8 bytes", ErrorCode.ERR_LANCE_INDEX_INVALID); + } + // 6-9. Branch on the orphan state, then the durable release transfer. + String actor = ctx.getQualifiedUser(); + boolean released; + if (catalog == null) { + // Full orphan: no credentials survive to read with and nothing can be + // invalidated, so the release goes straight to the manager write lock. + released = manager.forceRelease(jobId, job.getRevision(), actor, note, LATE_COMMIT_WARNING); + } else { + released = releaseWithLiveCatalog(env, catalogMgr, manager, catalog, targetResolves, job, actor, note); + } + if (!released) { + // 10. The expected-revision transfer lost a race. A concurrent FORCE_RELEASE + // that already landed makes this an idempotent success; anything else means + // the job left UNKNOWN concurrently (UNKNOWN has no other outgoing + // transition), so the pinned not-UNKNOWN wording stays accurate. + LanceIndexJob reread = manager.getJob(jobId); + if (reread != null && reread.isForceReleased()) { + ctx.getState().setOk(0, 1, LATE_COMMIT_WARNING); + return; + } + throw new AnalysisException(ErrorCode.ERR_LANCE_INDEX_JOB_NOT_UNKNOWN.formatErrorMsg(jobId), Review Comment: done: the !released branch re-reads the state — a job still UNKNOWN (a termination proof bumped the revision without leaving UNKNOWN) now answers the retryable resolution-incomplete instead of 5104, and the not-UNKNOWN wording is reserved for a reread that genuinely left UNKNOWN (testStaleRevisionRereadStillUnknownIsRetryable / ...LeftUnknownIsNotUnknown, 59d120c268). ########## regression-test/suites/external_table_p0/lance/test_lance_index_resolve.groovy: ########## @@ -0,0 +1,254 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// PR3E regression scope: the 3D dispatcher is not delivered yet, so a cluster cannot +// produce a genuine UNKNOWN job and every case in this suite is negative or static. +// Deliberately NOT covered here (all wait for 3D / slice 4): FORCE_RELEASE happy +// path e2e, same-name re-admission after FORCE e2e, quota reclaim after FORCE e2e, +// and expiry GC of resolved jobs e2e. + +suite("test_lance_index_resolve", "p0,external,nonConcurrent") { + // The Lance fixture is preinstalled in the MinIO container of the Iceberg + // external environment, so this suite deliberately shares its switch. + String enabled = context.config.otherConfigs.get("enableIcebergTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable Lance index resolve test because the Iceberg MinIO environment is disabled.") + return + } + + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String minioPort = context.config.otherConfigs.get("iceberg_minio_port") + // Admitted jobs are durable and stay PENDING forever in this delivery slice: + // dispatch, the FORCE happy path and job GC only land in later slices, so their + // fences and quota charges can never be released here. Every index name (and the + // filesystem catalog itself, because fence/quota keys include the persisted + // catalog id) carries this per-run suffix so that rerunning the suite on a + // shared pipeline cluster can never collide with a previous run's leftovers. + String runSuffix = "${System.currentTimeMillis()}" + String filesystemCatalog = "test_lance_index_resolve_${runSuffix}" + String user = "test_lance_index_resolve_user" + String password = "C123_567p" + String tableName = "vs_ivf_pq_f32" + String resolveIndexName = "idx_resolve_${runSuffix}" + // Job ids come from the cluster-wide Env.getNextId() allocator, so this literal + // can never name a real job. + String absentJobId = "9223372036854775806" + + // The filesystem catalog is fresh per run by construction, and once it holds + // unresolved jobs DROP CATALOG is guarded, so there is deliberately no DROP for + // it here. + try_sql "DROP USER '${user}'@'%'" + + // All three settings are masterOnly. Read them on the master even if the suite's + // ordinary JDBC connection points at a follower. SHOW uses the experimental + // display name for the gate, while ADMIN SET accepts its unprefixed alias. + def gateRows = master_sql """ADMIN SHOW FRONTEND CONFIG LIKE 'experimental_enable_lance_index_mutation'""" + def keepRows = master_sql """ADMIN SHOW FRONTEND CONFIG LIKE 'lance_index_job_keep_max_second'""" + def cleanRows = master_sql """ADMIN SHOW FRONTEND CONFIG LIKE 'lance_index_job_clean_interval_second'""" + assertEquals(1, gateRows.size()) + assertEquals(1, keepRows.size()) + assertEquals(1, cleanRows.size()) + String originalGate = gateRows[0][1].toString() + String originalKeep = keepRows[0][1].toString() + String originalClean = cleanRows[0][1].toString() + // The retention configs ship with these documented defaults; asserting them here + // fails loudly if a shared cluster has drifted instead of silently restoring a + // non-default value afterwards. + assertEquals("604800", originalKeep) + assertEquals("3600", originalClean) + Throwable suiteFailure = null + + // test { ... exception } always runs on the suite's default connection; the + // masterOnly config rejections below must be asserted on the master connection. + def expectMasterSqlException = { String stmt, String substring -> + String caught = null + try { + master_sql(stmt) + } catch (Throwable t) { + caught = t.toString() + } + assertTrue(caught != null && caught.contains(substring)) + } + + try { + // Rejections reachable with the mutation gate closed. RESOLVE is + // deliberately NOT behind enable_lance_index_mutation: it is the operator + // escape hatch and must stay usable while admission is gated, otherwise an + // unresolved job would freeze catalog DDL forever. Pin the gate closed so + // the cases in this block also prove the statement is not answered with the + // gate rejection. + master_sql """ADMIN SET FRONTEND CONFIG ("enable_lance_index_mutation" = "false")""" + + // Wrong AS literal: only FORCE_RELEASE is accepted after AS. + test { + sql """RESOLVE LANCE INDEX JOB ${absentJobId} AS FORCE COMMENT 'wrong literal'""" + exception "mismatched input" + } + + // The AS FORCE_RELEASE clause is mandatory. + test { + sql """RESOLVE LANCE INDEX JOB ${absentJobId} COMMENT 'missing as clause'""" + exception "mismatched input" + } + + // COMMENT is mandatory at grammar level. + test { + sql """RESOLVE LANCE INDEX JOB ${absentJobId} AS FORCE_RELEASE""" + exception "mismatched input" + } + + // An empty COMMENT parses but never changes the observable response: the + // note check runs after the job lookup and the UNKNOWN state gate, so with a + // missing job the fixed not-found wording answers first. The dedicated + // empty-note rejection needs a genuine UNKNOWN job and stays UT-only + // (ResolveLanceIndexJobCommandTest) until 3D can produce one. + test { + sql """RESOLVE LANCE INDEX JOB ${absentJobId} AS FORCE_RELEASE COMMENT ''""" + exception "Lance index job not found" + } + + // A well-formed RESOLVE of a missing job: the fixed non-disclosing 5103 + // wording, with no gate rejection even though the gate is closed. + test { + sql """RESOLVE LANCE INDEX JOB ${absentJobId} AS FORCE_RELEASE COMMENT 'no such job'""" + exception "Lance index job not found" + } + + // Rejections reachable with an admitted PENDING job. masterOnly configs set + // through ADMIN SET land on the master node locally, which is where + // admission reads them; the finally block below restores the gate no matter + // where the suite fails. + master_sql """ADMIN SET FRONTEND CONFIG ("enable_lance_index_mutation" = "true")""" + + sql """ + CREATE CATALOG `${filesystemCatalog}` PROPERTIES ( + "type" = "lance", + "lance.catalog.type" = "filesystem", + "warehouse" = "s3://warehouse/lance", + "s3.endpoint" = "http://${externalEnvIp}:${minioPort}";, + "s3.access_key" = "admin", + "s3.secret_key" = "password", + "s3.region" = "us-east-1", + "use_path_style" = "true" + ) + """ + + // CREATE INDEX is admitted and returns a single-column JobId result set with + // one row; the job stays PENDING because no dispatcher exists in this slice. + def createRows = sql """CREATE INDEX `${resolveIndexName}` ON `${filesystemCatalog}`.`doris`.`${tableName}` (embedding) USING ANN Review Comment: done: the suite unpauses dispatch after its PENDING assertions and waits (bounded) for the job to converge to NOT_COMMITTED via the not-implemented stub, which releases the fence and quota — a run no longer leaks a permanent unresolved job into the shared cluster quota, and the per-run catalog/name suffixes stay as belt-and-braces (59d120c268). -- 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]
