u70b3 · Oct 9, 2026 · Oct 9, 2026 · Oct 9, 2026 · Oct 9, 2026 · Oct 9, 2026
@@ -73,6 +73,19 @@ class BaseBackendService : public BackendServiceIf {
         _agent_server->submit_tasks(return_value, tasks);
     }
 
+    // Synchronous Lance index mutation. The real worker lands in a later slice;
+    // until then the invocation is answered as definitively not executed, so
+    // the FE classifies a complete trusted pre-invocation rejection (terminal
+    // confirmed failure, never retried within the same statement) instead of
+    // an ambiguous result.
+    void lance_index_mutate(TLanceIndexMutationResult& _return,
+                            const TLanceIndexMutationRequest& request) override {
+        TStatus status;
+        status.__set_status_code(TStatusCode::NOT_IMPLEMENTED_ERROR);
+        status.__set_error_msgs({"lance index worker is not available in this build"});
+        _return.__set_status(status);
+    }
+
     void publish_cluster_state(TAgentResult& result, const TAgentPublishRequest& request) override {
         _agent_server->publish_cluster_state(result, request);
     }

@@ -0,0 +1,74 @@
+// 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.
+
+#include "service/backend_service.h"
+
+#include <gen_cpp/BackendService_types.h>
+#include <gen_cpp/Status_types.h>
+#include <gtest/gtest.h>
+
+#include "runtime/exec_env.h"
+
+namespace doris {
+namespace {
+
+class RejectingBackendService final : public BaseBackendService {
+public:
+    RejectingBackendService() : BaseBackendService(ExecEnv::GetInstance()) {}
+
+    Status start_thrift_dependencies() override {
+        dependencies_started = true;
+        return Status::OK();
+    }
+
+    bool dependencies_started = false;
+};
+
+} // namespace
+
+TEST(LanceIndexMutationTest, RejectsEveryOperationWithoutStartingWorkers) {
+    RejectingBackendService service;
+    BackendServiceIf& rpc = service;
+    for (auto operation : {TLanceIndexMutationType::CREATE, TLanceIndexMutationType::REPLACE,
+                           TLanceIndexMutationType::DROP}) {
+        TLanceIndexMutationRequest request;
+        request.__set_invocation_id("lance-stub-test");
+        request.__set_be_process_epoch(1);
+        request.__set_deadline_ms(1);
+        request.__set_mutation_type(operation);
+        request.__set_index_name("idx");
+        request.__set_column_name("embedding");
+        request.__set_index_type("IVF_PQ");
+        // A rejecting build must not open even an unavailable dataset or echo its options.
+        request.__set_dataset_uri("s3://unavailable-bucket/private-dataset");
+        request.__set_admitted_dataset_version(7);
+        request.__set_storage_options({{"aws_secret_access_key", "private-storage-secret"}});
+
+        TLanceIndexMutationResult result;
+        rpc.lance_index_mutate(result, request);
+
+        EXPECT_EQ(TStatusCode::NOT_IMPLEMENTED_ERROR, result.status.status_code);
+        ASSERT_EQ(1, result.status.error_msgs.size());
+        EXPECT_EQ("lance index worker is not available in this build", result.status.error_msgs[0]);
+        // A refusal must not fabricate a native outcome or expose provider diagnostics.
+        EXPECT_FALSE(result.__isset.lance_error_code);
+        EXPECT_FALSE(result.__isset.message);
+        EXPECT_FALSE(service.dependencies_started);
+    }
+}
+
+} // namespace doris
@@ -4253,12 +4253,14 @@ public void handle(Field field, String value) throws Exception {
     public static long lance_catalog_arrow_memory_limit_bytes = 256L * 1024 * 1024;
 
     @ConfField(mutable = true, masterOnly = true, varType = VariableAnnotation.EXPERIMENTAL, description = {
-            "是否启用 Lance 外表索引变更(CREATE/CREATE OR REPLACE/DROP INDEX)的权威校验路径。默认关闭;"
-                    + "在同步执行路径就绪前,开启本开关只会让通过校验的语句收到暂不支持的错误。",
-            "Enable the authoritative validation path of Lance index mutations "
-                    + "(CREATE/CREATE OR REPLACE/DROP INDEX). Disabled by default; until the synchronous "
-                    + "execution path lands, enabling it only makes validated statements fail with the "
-                    + "not-supported error."})
+            "是否启用 Lance 外表索引变更(CREATE/CREATE OR REPLACE/DROP INDEX)的同步执行路径。默认关闭;"
+                    + "关闭时通过校验的语句直接收到 mutation 关闭错误;开启后语句在 master 上同步完成"
+                    + "权威校验、单次派发与结果分类。",
+            "Enable the synchronous execution path of Lance index mutations "
+                    + "(CREATE/CREATE OR REPLACE/DROP INDEX). Disabled by default; while disabled a "
+                    + "validated statement fails fast with the mutation-disabled error, and when enabled "
+                    + "the statement completes authoritative validation, the single dispatch, and the "
+                    + "outcome classification synchronously on the master."})
     public static boolean enable_lance_index_mutation = false;
 
     @ConfField(mutable = true, masterOnly = true,
@@ -4273,4 +4275,34 @@ public void handle(Field field, String value) throws Exception {
                     "Static upper bound for num_sub_vectors of Lance IVF_PQ indexes."})
     public static int lance_index_max_num_sub_vectors = 256;
 
+    @ConfField(mutable = true, masterOnly = true,
+            callback = LanceIndexConfigValidator.PositiveIntConfigHandler.class,
+            description = {"Lance 索引同步变更语句的最大并发执行数（独立有界池，超出者进入有限等待队列，"
+                    + "池与队列均满则快速失败）。运行中的池不可动态扩缩，修改需重启生效。",
+                    "Maximum concurrently executing synchronous Lance index mutation statements"
+                            + " (a dedicated bounded pool; excess statements enter the finite wait"
+                            + " queue, and a full pool or queue fails fast). The live pool cannot"
+                            + " be resized; changes take effect after a restart."})
+    public static int lance_index_mutation_max_concurrency = 2;
+
+    @ConfField(mutable = true, masterOnly = true,
+            callback = LanceIndexConfigValidator.PositiveIntConfigHandler.class,
+            description = {"Lance 索引同步变更语句有界准入池的有限等待队列长度；队列满即快速失败，不无界等待。"
+                    + "运行中的池不可动态扩缩，修改需重启生效。",
+                    "Length of the bounded admission pool's finite wait queue for synchronous"
+                            + " Lance index mutation statements; a full queue fails fast instead of"
+                            + " waiting unboundedly. The live pool cannot be resized; changes take"
+                            + " effect after a restart."})
+    public static int lance_index_mutation_max_queued = 16;
+
+    @ConfField(mutable = true, masterOnly = true,
+            callback = LanceIndexConfigValidator.PositiveIntConfigHandler.class,
+            description = {"单条 Lance 索引同步变更语句的预算上限（秒）；实际预算取其与语句 query_timeout 的较小值，"
+                    + "从语句入口起算，派发与等待共享。",
+                    "Upper bound in seconds of one synchronous Lance index mutation statement's"
+                            + " total budget; the effective budget is the smaller of this cap and the"
+                            + " statement's query_timeout, measured from statement entry and shared"
+                            + " by dispatch and wait."})
+    public static int lance_index_mutation_budget_cap_seconds = 300;
+
 }
@@ -39,6 +39,9 @@ public void testReviewedDefaults() {
         Assertions.assertFalse(Config.enable_lance_index_mutation);
         Assertions.assertEquals(4096, Config.lance_index_max_num_partitions);
         Assertions.assertEquals(256, Config.lance_index_max_num_sub_vectors);
+        Assertions.assertEquals(2, Config.lance_index_mutation_max_concurrency);
+        Assertions.assertEquals(16, Config.lance_index_mutation_max_queued);
+        Assertions.assertEquals(300, Config.lance_index_mutation_budget_cap_seconds);
     }
 
     @Test
@@ -47,6 +50,12 @@ public void testConfFieldWiring() throws Exception {
                 LanceIndexConfigValidator.PositiveIntConfigHandler.class);
         assertCallbackWiring("lance_index_max_num_sub_vectors", true,
                 LanceIndexConfigValidator.PositiveIntConfigHandler.class);
+        assertCallbackWiring("lance_index_mutation_max_concurrency", true,
+                LanceIndexConfigValidator.PositiveIntConfigHandler.class);
+        assertCallbackWiring("lance_index_mutation_max_queued", true,
+                LanceIndexConfigValidator.PositiveIntConfigHandler.class);
+        assertCallbackWiring("lance_index_mutation_budget_cap_seconds", true,
+                LanceIndexConfigValidator.PositiveIntConfigHandler.class);
 
         ConfigBase.ConfField gate = Config.class.getField("enable_lance_index_mutation")
                 .getAnnotation(ConfigBase.ConfField.class);
@@ -69,6 +78,9 @@ private static void assertCallbackWiring(String fieldName, boolean masterOnly, C
     public void testPositiveIntHandlerAssignsAcceptedValue() throws Exception {
         assertIntAssigns("lance_index_max_num_partitions");
         assertIntAssigns("lance_index_max_num_sub_vectors");
+        assertIntAssigns("lance_index_mutation_max_concurrency");
+        assertIntAssigns("lance_index_mutation_max_queued");
+        assertIntAssigns("lance_index_mutation_budget_cap_seconds");
     }
 
     private static void assertIntAssigns(String fieldName) throws Exception {
@@ -89,6 +101,10 @@ public void testPositiveIntHandlerRejectsInvalidValues() throws Exception {
         assertIntRejected("lance_index_max_num_partitions", "-4096");
         assertIntRejected("lance_index_max_num_partitions", "1.5");
         assertIntRejected("lance_index_max_num_sub_vectors", "abc");
+        assertIntRejected("lance_index_mutation_max_concurrency", "0");
+        assertIntRejected("lance_index_mutation_max_concurrency", "-2");
+        assertIntRejected("lance_index_mutation_max_queued", "0");
+        assertIntRejected("lance_index_mutation_budget_cap_seconds", "-1");
     }
 
     private static void assertIntRejected(String fieldName, String value) throws Exception {

@@ -1245,6 +1245,29 @@ public enum ErrorCode {
     // 5103 (ERR_LANCE_INDEX_JOB_NOT_FOUND) was removed with the durable Lance index job
     // framework; the number stays unused so old client-visible codes keep their meaning.
 
+    ERR_LANCE_INDEX_MUTATION_INDETERMINATE(5104, new byte[]{'4', '2', '0', '0', '0'},
+            "Lance index mutation may or may not have committed (%s). No complete trusted result"
+            + " was received, so the outcome is unknown and must not be blindly retried; inspect"
+            + " authoritative metadata with SHOW INDEX FROM <table> and lance_index_entries()"),
+
+    ERR_LANCE_INDEX_MUTATION_COMMITTED_REFRESH_INCOMPLETE(5105, new byte[]{'4', '2', '0', '0', '0'},
+            "Lance index mutation committed successfully, but the metadata refresh did not complete"
+            + " (%s). This is not a build failure; inspect authoritative metadata with"
+            + " SHOW INDEX FROM <table> and lance_index_entries()"),
+
+    ERR_LANCE_INDEX_MUTATION_BUSY(5106, new byte[]{'4', '2', '0', '0', '0'},
+            "Lance index mutation is rejected because the mutation path is busy (%s);"
+            + " the statement was not queued, retry it later"),
+
+    ERR_LANCE_INDEX_MUTATION_REFRESH_FAILED(5107, new byte[]{'4', '2', '0', '0', '0'},
+            "Lance index mutation completed as a no-op, but the required local metadata cache"
+            + " invalidation failed (%s); no mutation was executed and this is not a build failure"),
+
+    ERR_LANCE_INDEX_MUTATION_REJECTED(5108, new byte[]{'4', '2', '0', '0', '0'},
+            "Lance index mutation was rejected and nothing was committed (%s); either nothing"
+            + " was sent to the backend, or the complete trusted result proves this invocation"
+            + " made no change"),
+
     ERR_NOT_CLOUD_MODE(6000, new byte[]{'4', '2', '0', '0', '0'},
             "Command only support in cloud mode.");
 

@@ -142,6 +142,15 @@ synchronized Lease acquire() {
         return new Lease(this);
     }
 
+    /** A queued read may extend the lease of a generation already selected by its caller. */
+    private synchronized Lease retainForRead() {
+        if (retired && activeOperations == 0) {
+            throw new IllegalStateException("Lance catalog resources have been closed");
+        }
+        activeOperations++;
+        return new Lease(this);
+    }
+
     /** Retires this generation immediately; its last active operation performs resource cleanup. */
     @Override
     public void close() {
@@ -846,32 +855,34 @@ public LanceIndexAdmissionSnapshot loadTableIndexAdmissionSnapshot(String dbName
     }
 
     private <T> T inspectTableIndexes(String dbName, String tableName, BiFunction<Dataset, String, T> inspection) {
-        LanceTableAccess tableAccess = null;
         try {
-            // The worker owns its Dataset and Session even if the caller releases its lease on timeout.
-            // Index admission must verify the current target even while query access is cached.
-            tableAccess = namespaceClient.resolveTableAccessUncached(dbName, tableName);
-            // Index paths open the dataset by URI at storage-latest. They are only reachable for
-            // filesystem catalogs, whose Directory namespace never manages versions; a managed
-            // table here would bypass the namespace.
-            if (tableAccess.isManagedVersioning()) {
-                throw new IllegalStateException("Lance index inspection does not support namespace-managed tables");
-            }
-            LanceTableAccess access = tableAccess;
             return LanceIndexInspectionExecutor.execute(() -> {
-                // The caller can time out while JNI is running; the worker must own resource cleanup.
-                try (BufferAllocator allocator = new RootAllocator(LanceMetadataLoader.READ_ALLOCATOR_LIMIT);
-                        Dataset dataset = Dataset.open().allocator(allocator).uri(access.getDatasetUri())
-                                .readOptions(LanceReadOptions.forIndependentRead(
-                                        access.getStorageOptions(), OptionalLong.empty()))
-                                .build()) {
-                    return inspection.apply(dataset, access.getDatasetUri());
+                LanceTableAccess access = null;
+                // The read owns a generation lease until all namespace/JNI work returns,
+                // even when its statement has already timed out and released its own lease.
+                try (Lease ignored = retainForRead()) {
+                    access = namespaceClient.resolveTableAccessUncached(dbName, tableName);
+                    if (access.isManagedVersioning()) {
+                        throw new IllegalStateException(
+                                "Lance index inspection does not support namespace-managed tables");
+                    }
+                    try (BufferAllocator allocator = new RootAllocator(LanceMetadataLoader.READ_ALLOCATOR_LIMIT);
+                            Dataset dataset = Dataset.open().allocator(allocator).uri(access.getDatasetUri())
+                                    .readOptions(LanceReadOptions.forIndependentRead(
+                                            access.getStorageOptions(), OptionalLong.empty()))
+                                    .build()) {
+                        return inspection.apply(dataset, access.getDatasetUri());
+                    }
+                } catch (Exception e) {
+                    throw LanceErrorMessages.failure(
+                            "Failed to load Lance index metadata for " + dbName + "." + tableName,
+                            e, access == null ? null : access.getDatasetUri(),
+                            access == null ? namespaceStorageOptions : access.getStorageOptions(), catalogSecrets);
                 }
             });
         } catch (Exception e) {
             throw LanceErrorMessages.failure("Failed to load Lance index metadata for " + dbName + "." + tableName, e,
-                    tableAccess == null ? null : tableAccess.getDatasetUri(),
-                    tableAccess == null ? namespaceStorageOptions : tableAccess.getStorageOptions(), catalogSecrets);
+                    null, namespaceStorageOptions, catalogSecrets);
         }
     }
 

@@ -17,6 +17,8 @@
 
 package org.apache.doris.datasource.lance;
 
+import org.apache.doris.datasource.lance.index.LanceIndexInspectionExecutor.MetadataReadTimeoutException;
+
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.exception.ExceptionUtils;
 
@@ -43,6 +45,11 @@ private LanceErrorMessages() {
     static RuntimeException failure(String prefix, Throwable error, String uri,
             Map<String, String> options, List<String> catalogSecrets) {
         String message = sanitize(error, uri, options, catalogSecrets);
+        if (error instanceof MetadataReadTimeoutException) {
+            // Preserve classification across every sanitized wrapper, without retaining
+            // the provider exception or its potentially credential-bearing cause chain.
+            return new MetadataReadTimeoutException(prefix + ": " + message);
+        }
         RuntimeException cause = error instanceof IllegalArgumentException
                 ? new IllegalArgumentException(message) : new RuntimeException(message);
         return new RuntimeException(prefix + ": " + message, cause);
@@ -66,14 +73,18 @@ static String sanitize(Throwable throwable, String datasetUri,
         return truncateUtf8(removeControlCharacters(message), MAX_PROVIDER_MESSAGE_BYTES);
     }
 
-    private static String removeControlCharacters(String value) {
+    // Package-private: LanceIndexMutationOutcome applies the same bounded-sanitized-text
+    // contract to its carried message, so both sides share one truncation/control-strip
+    // implementation instead of drifting apart.
+    static String removeControlCharacters(String value) {
         StringBuilder sanitized = new StringBuilder(value.length());
         value.codePoints().filter(codePoint -> !Character.isISOControl(codePoint))
                 .forEach(sanitized::appendCodePoint);
         return sanitized.toString();
     }
 
-    private static String truncateUtf8(String value, int maxBytes) {
+    // Package-private for the same reason as removeControlCharacters above.
+    static String truncateUtf8(String value, int maxBytes) {
         if (value.getBytes(StandardCharsets.UTF_8).length <= maxBytes) {
             return value;
         }