branch-4.1: [feature](lance) Add Lance index job dispatcher with thrift dispatch boundary

be/src/service/backend_service.h

@@ -73,6 +73,16 @@ class BaseBackendService : public BackendServiceIf {
         _agent_server->submit_tasks(return_value, tasks);
     }
 
+    // One-shot Lance index mutation dispatch. The isolated worker lands in a later
+    // slice; until then the request is answered as definitively NOT enqueued, so the
+    // FE classifies a trusted pre-invocation rejection (terminal NOT_COMMITTED)
+    // instead of an ambiguous result.
+    void submit_lance_index_job(TStatus& _return,
+                                const TLanceIndexJobDispatch& dispatch) override {
+        _return.__set_status_code(TStatusCode::NOT_IMPLEMENTED_ERROR);
+        _return.__set_error_msgs({"lance index worker is not available in this build"});
+    }
+
     void publish_cluster_state(TAgentResult& result, const TAgentPublishRequest& request) override {
         _agent_server->publish_cluster_state(result, request);
     }

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -4265,4 +4265,48 @@ 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 索引 job 派发器(含 deadline/possible-live 扫掠与 refresh 驱动)的轮询周期(秒)。",
+                    "Polling interval in seconds of the Lance index job dispatcher "
+                    + "(dispatch sweep, deadline/possible-live sweeps, and refresh driver)."})
+    public static int lance_index_job_dispatch_interval_second = 10;
+
+    @ConfField(mutable = true, masterOnly = true,
+            callback = LanceIndexConfigValidator.PositiveLongConfigHandler.class,
+            description = {"单个 Lance 索引 job 派发后的结果等待上限(秒)。到期仍无完整可信结果即收敛为 UNKNOWN;"
+                    + "该期限只限定等待,不证明终止,也不释放 possible-live 槽位。",
+                    "Wait bound in seconds for the result of one dispatched Lance index job. Expiry without "
+                    + "a complete trusted result converges the job to UNKNOWN; the deadline bounds the wait "
+                    + "only, never proves termination, and never releases a possible-live slot."})
+    public static long lance_index_job_execute_deadline_second = 3600;
+
+    @ConfField(mutable = true, masterOnly = true,
+            callback = LanceIndexConfigValidator.PositiveIntConfigHandler.class,
+            description = {"派发器单轮最多新派发的 Lance 索引 job 数(背压上限)。",
+                    "Maximum number of Lance index jobs newly dispatched per dispatcher round (backpressure)."})
+    public static int lance_index_job_max_dispatch_per_round = 16;
+
+    @ConfField(mutable = true, masterOnly = true,
+            callback = LanceIndexConfigValidator.PositiveIntConfigHandler.class,
+            description = {"单个 BE 上允许同时在途(RUNNING)的 Lance 索引 job 数上限。",
+                    "Maximum number of in-flight (RUNNING) Lance index jobs per backend."})
+    public static int lance_index_job_max_inflight_per_backend = 2;
+
+    @ConfField(mutable = true, masterOnly = true,
+            callback = LanceIndexConfigValidator.PositiveIntConfigHandler.class,
+            description = {"refresh 失败的 Lance 索引 job 的最小重试间隔(秒);首次刷新不受此间隔限制。",
+                    "Minimum retry interval in seconds for a terminal Lance index job whose metadata "
+                    + "refresh FAILED; the first refresh attempt is never delayed by this interval."})
+    public static int lance_index_job_refresh_retry_second = 300;
+
+    @ConfField(mutable = true, masterOnly = true, description = {
+            "是否允许 file:// 本地路径上的 Lance 索引变更派发(运维断言,默认关闭)。开启后派发仍要求"
+                    + "集群恰一台 FE 且目标 BE 是唯一存活 BE;对象存储是生产形态。",
+            "Operator assertion allowing dispatch of Lance index mutations on local file:// datasets "
+                    + "(disabled by default). When enabled, dispatch still requires exactly one FE in the "
+                    + "cluster and the target backend to be the only alive backend. Object storage is the "
+                    + "production mode."})
+    public static boolean enable_lance_index_local_file_mutation = false;
+
 }

fe/fe-common/src/test/java/org/apache/doris/common/LanceIndexConfigValidatorTest.java

@@ -44,6 +44,16 @@ public void testReviewedDefaults() {
         Assertions.assertEquals(256, Config.lance_index_max_num_sub_vectors);
     }
 
+    @Test
+    public void testDispatcherReviewedDefaults() {
+        Assertions.assertEquals(10, Config.lance_index_job_dispatch_interval_second);
+        Assertions.assertEquals(3600L, Config.lance_index_job_execute_deadline_second);
+        Assertions.assertEquals(16, Config.lance_index_job_max_dispatch_per_round);
+        Assertions.assertEquals(2, Config.lance_index_job_max_inflight_per_backend);
+        Assertions.assertEquals(300, Config.lance_index_job_refresh_retry_second);
+        Assertions.assertFalse(Config.enable_lance_index_local_file_mutation);
+    }
+
     @Test
     public void testConfFieldWiring() throws Exception {
         assertCallbackWiring("lance_index_job_max_unresolved_per_table", true,
@@ -65,6 +75,29 @@ public void testConfFieldWiring() throws Exception {
         Assertions.assertEquals(VariableAnnotation.EXPERIMENTAL, gate.varType());
     }
 
+    @Test
+    public void testDispatcherConfFieldWiring() throws Exception {
+        assertCallbackWiring("lance_index_job_dispatch_interval_second", true,
+                LanceIndexConfigValidator.PositiveIntConfigHandler.class);
+        assertCallbackWiring("lance_index_job_execute_deadline_second", true,
+                LanceIndexConfigValidator.PositiveLongConfigHandler.class);
+        assertCallbackWiring("lance_index_job_max_dispatch_per_round", true,
+                LanceIndexConfigValidator.PositiveIntConfigHandler.class);
+        assertCallbackWiring("lance_index_job_max_inflight_per_backend", true,
+                LanceIndexConfigValidator.PositiveIntConfigHandler.class);
+        assertCallbackWiring("lance_index_job_refresh_retry_second", true,
+                LanceIndexConfigValidator.PositiveIntConfigHandler.class);
+
+        // The local-file operator assertion is a plain mutable master-only boolean: no
+        // numeric validator is attached, so any boolean the ADMIN SET path accepts is legal.
+        ConfigBase.ConfField gate = Config.class.getField("enable_lance_index_local_file_mutation")
+                .getAnnotation(ConfigBase.ConfField.class);
+        Assertions.assertNotNull(gate);
+        Assertions.assertTrue(gate.mutable());
+        Assertions.assertTrue(gate.masterOnly());
+        Assertions.assertEquals(ConfigBase.DefaultConfHandler.class, gate.callback());
+    }
+
     private static void assertCallbackWiring(String fieldName, boolean masterOnly, Class<?> callback)
             throws Exception {
         ConfigBase.ConfField anno = Config.class.getField(fieldName).getAnnotation(ConfigBase.ConfField.class);
@@ -160,6 +193,38 @@ private static void assertIntRejected(String fieldName, String value) throws Exc
      * End-to-end through the ADMIN SET FRONTEND CONFIG machinery: the annotation callback
      * must both validate and assign, and a rejected value must leave the field untouched.
      */
+    @Test
+    public void testDispatcherPositiveHandlersAssignAcceptedValues() throws Exception {
+        assertIntAssigns("lance_index_job_dispatch_interval_second");
+        assertIntAssigns("lance_index_job_max_dispatch_per_round");
+        assertIntAssigns("lance_index_job_max_inflight_per_backend");
+        assertIntAssigns("lance_index_job_refresh_retry_second");
+        // The long handler must also be exercised on its one long dispatcher field.
+        Field deadlineField = Config.class.getField("lance_index_job_execute_deadline_second");
+        long originalDeadline = deadlineField.getLong(null);
+        try {
+            new LanceIndexConfigValidator.PositiveLongConfigHandler().handle(deadlineField, " 7200 ");
+            Assertions.assertEquals(7200L, deadlineField.getLong(null));
+        } finally {
+            deadlineField.setLong(null, originalDeadline);
+        }
+    }
+
+    @Test
+    public void testDispatcherPositiveHandlersRejectInvalidValues() throws Exception {
+        assertIntRejected("lance_index_job_dispatch_interval_second", "0");
+        assertIntRejected("lance_index_job_dispatch_interval_second", "-10");
+        assertIntRejected("lance_index_job_max_dispatch_per_round", "0");
+        assertIntRejected("lance_index_job_max_dispatch_per_round", "-1");
+        assertIntRejected("lance_index_job_max_inflight_per_backend", "0");
+        assertIntRejected("lance_index_job_max_inflight_per_backend", "-3");
+        assertIntRejected("lance_index_job_refresh_retry_second", "0");
+        assertIntRejected("lance_index_job_refresh_retry_second", "-300");
+        assertLongRejected("lance_index_job_execute_deadline_second", "0");
+        assertLongRejected("lance_index_job_execute_deadline_second", "-3600");
+        assertLongRejected("lance_index_job_execute_deadline_second", "soon");
+    }
+
     @Test
     public void testSetMutableConfigPath() throws Exception {
         Config config = new Config();
@@ -170,6 +235,8 @@ public void testSetMutableConfigPath() throws Exception {
         long originalQuota = Config.lance_index_job_max_unresolved_per_catalog;
         int originalBound = Config.lance_index_max_num_partitions;
         boolean originalGate = Config.enable_lance_index_mutation;
+        int originalInterval = Config.lance_index_job_dispatch_interval_second;
+        boolean originalLocalFile = Config.enable_lance_index_local_file_mutation;
         try {
             ConfigBase.setMutableConfig("lance_index_job_max_unresolved_per_catalog", "96");
             Assertions.assertEquals(96L, Config.lance_index_job_max_unresolved_per_catalog);
@@ -187,10 +254,23 @@ public void testSetMutableConfigPath() throws Exception {
 
             ConfigBase.setMutableConfig("enable_lance_index_mutation", "true");
             Assertions.assertTrue(Config.enable_lance_index_mutation);
+
+            ConfigBase.setMutableConfig("lance_index_job_dispatch_interval_second", "60");
+            Assertions.assertEquals(60, Config.lance_index_job_dispatch_interval_second);
+            Assertions.assertThrows(ConfigException.class,
+                    () -> ConfigBase.setMutableConfig("lance_index_job_dispatch_interval_second", "0"));
+            Assertions.assertEquals(60, Config.lance_index_job_dispatch_interval_second);
+
+            ConfigBase.setMutableConfig("enable_lance_index_local_file_mutation", "true");
+            Assertions.assertTrue(Config.enable_lance_index_local_file_mutation);
+            ConfigBase.setMutableConfig("enable_lance_index_local_file_mutation", "false");
+            Assertions.assertFalse(Config.enable_lance_index_local_file_mutation);
         } finally {
             Config.lance_index_job_max_unresolved_per_catalog = originalQuota;
             Config.lance_index_max_num_partitions = originalBound;
             Config.enable_lance_index_mutation = originalGate;
+            Config.lance_index_job_dispatch_interval_second = originalInterval;
+            Config.enable_lance_index_local_file_mutation = originalLocalFile;
         }
     }
 }

fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java

@@ -111,6 +111,7 @@
 import org.apache.doris.datasource.iceberg.IcebergExternalTable;
 import org.apache.doris.datasource.iceberg.IcebergSysExternalTable;
 import org.apache.doris.datasource.jdbc.JdbcExternalTable;
+import org.apache.doris.datasource.lance.job.LanceIndexJobDispatcher;
 import org.apache.doris.datasource.lance.job.LanceIndexJobManager;
 import org.apache.doris.datasource.paimon.PaimonExternalTable;
 import org.apache.doris.datasource.paimon.PaimonSysExternalTable;
@@ -573,6 +574,8 @@ public class Env {
 
     private LanceIndexJobManager lanceIndexJobManager;
 
+    private LanceIndexJobDispatcher lanceIndexJobDispatcher;
+
     private DNSCache dnsCache;
 
     private final NereidsSqlCacheManager sqlCacheManager;
@@ -859,6 +862,7 @@ public Env(boolean isCheckpointCatalog) {
         this.eventProcessor = new EventProcessor(mtmvService);
         this.insertOverwriteManager = new InsertOverwriteManager();
         this.lanceIndexJobManager = new LanceIndexJobManager();
+        this.lanceIndexJobDispatcher = new LanceIndexJobDispatcher(lanceIndexJobManager);
         this.dnsCache = new DNSCache();
         this.sqlCacheManager = new NereidsSqlCacheManager();
         this.sortedPartitionsCacheManager = new NereidsSortedPartitionsCacheManager();
@@ -2030,6 +2034,8 @@ protected void startMasterOnlyDaemonThreads() {
             keyManager.init();
         }
         agentTaskCleanupDaemon.start();
+        // lance index job dispatcher: dispatch sweep, deadline/possible-live sweeps, refresh driver
+        lanceIndexJobDispatcher.start();
     }
 
     // start threads that should run on all FE