[fix](cloud) Deduplicate pending one-shot warm up jobs

fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/VectorColumn.java

@@ -1605,6 +1605,7 @@ public byte[] getBytesVarbinary(int rowId) {
     }
 
     public void updateMeta(VectorColumn meta) {
+        meta.appendLong(isConst ? 1 : 0);
         if (columnType.isUnsupported()) {
             meta.appendLong(0);
         } else if (columnType.isStringType()) {

fe/fe-authentication/fe-authentication-handler/pom.xml

@@ -72,5 +72,11 @@ under the License.
             <version>${project.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.doris</groupId>
+            <artifactId>fe-authentication-plugin-ldap</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

fe/fe-authentication/fe-authentication-handler/src/test/java/org/apache/doris/authentication/handler/AuthenticationPluginManagerTest.java

@@ -70,7 +70,7 @@ void testPluginsAutoLoaded() {
         // Then
         Assertions.assertNotNull(pluginNames);
         Assertions.assertFalse(pluginNames.isEmpty(), "Should load at least built-in plugins");
-        Assertions.assertTrue(pluginNames.contains("oidc"), "Should include oidc plugin");
+        Assertions.assertTrue(pluginNames.contains("ldap"), "Should include ldap plugin");
         Assertions.assertTrue(pluginNames.contains("password"), "Should include password plugin");
     }
 
@@ -148,9 +148,9 @@ void testGetFactory() {
         Assertions.assertTrue(factory.isPresent());
         Assertions.assertEquals("password", factory.get().name());
 
-        Optional<AuthenticationPluginFactory> oidcFactory = pluginManager.getFactory("oidc");
-        Assertions.assertTrue(oidcFactory.isPresent());
-        Assertions.assertEquals("oidc", oidcFactory.get().name());
+        Optional<AuthenticationPluginFactory> ldapFactory = pluginManager.getFactory("ldap");
+        Assertions.assertTrue(ldapFactory.isPresent());
+        Assertions.assertEquals("ldap", ldapFactory.get().name());
     }
 
     @Test

fe/fe-core/pom.xml

@@ -731,6 +731,10 @@ under the License.
             <groupId>org.immutables</groupId>
             <artifactId>value</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.google.code.findbugs</groupId>
+            <artifactId>annotations</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.trino</groupId>
             <artifactId>trino-main</artifactId>

fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java

@@ -52,6 +52,7 @@
 import org.apache.doris.thrift.TNetworkAddress;
 import org.apache.doris.thrift.TStatusCode;
 
+import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import org.apache.logging.log4j.LogManager;
@@ -67,6 +68,7 @@
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -82,6 +84,7 @@
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.locks.ReentrantLock;
 
 public class CacheHotspotManager extends MasterDaemon {
     public static final int MAX_SHOW_ENTRIES = 2000;
@@ -109,6 +112,9 @@ public class CacheHotspotManager extends MasterDaemon {
 
     private ConcurrentMap<Long, CloudWarmUpJob> runnableCloudWarmUpJobs = Maps.newConcurrentMap();
 
+    private final ConcurrentMap<OncePendingJobKey, RefCountedPendingCreateLock> oncePendingCreateLocks
+            = Maps.newConcurrentMap();
+
     private final ThreadPoolExecutor cloudWarmUpThreadPool = ThreadPoolManager.newDaemonCacheThreadPool(
             Config.max_active_cloud_warm_up_job, "cloud-warm-up-pool", true);
 
@@ -148,10 +154,185 @@ public String toString() {
         }
     }
 
+    private static class OncePendingJobKey {
+        private final JobType jobType;
+        private final String srcName;
+        private final String dstName;
+        private final List<String> normalizedTables;
+        private final boolean force;
+
+        OncePendingJobKey(JobType jobType, String srcName, String dstName,
+                List<String> normalizedTables, boolean force) {
+            this.jobType = jobType;
+            this.srcName = normalizeNullableName(srcName);
+            this.dstName = normalizeNullableName(dstName);
+            this.normalizedTables = normalizedTables.isEmpty()
+                    ? Collections.emptyList()
+                    : Collections.unmodifiableList(new ArrayList<>(normalizedTables));
+            this.force = force;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) {
+                return true;
+            }
+            if (!(o instanceof OncePendingJobKey)) {
+                return false;
+            }
+            OncePendingJobKey jobKey = (OncePendingJobKey) o;
+            return force == jobKey.force
+                    && jobType == jobKey.jobType
+                    && Objects.equals(srcName, jobKey.srcName)
+                    && Objects.equals(dstName, jobKey.dstName)
+                    && Objects.equals(normalizedTables, jobKey.normalizedTables);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(jobType, srcName, dstName, normalizedTables, force);
+        }
+
+        @Override
+        public String toString() {
+            return "OncePendingWarmUpJob{"
+                    + "jobType=" + jobType
+                    + ", src='" + srcName + '\''
+                    + ", dst='" + dstName + '\''
+                    + ", tables=" + normalizedTables
+                    + ", force=" + force
+                    + '}';
+        }
+    }
+
+    private static class RefCountedPendingCreateLock {
+        private final ReentrantLock lock = new ReentrantLock();
+
+        // Tracks holders and waiters that retained the entry before locking.
+        private volatile int refCount = 1;
+
+        void retain() {
+            ++refCount;
+        }
+
+        int release() {
+            Preconditions.checkState(refCount > 0, "once pending create lock ref count underflow");
+            return --refCount;
+        }
+
+        void lock() {
+            lock.lock();
+        }
+
+        void unlock() {
+            lock.unlock();
+        }
+    }
+
     // Tracks long-running jobs (event-driven and periodic).
     // Ensures only one active job exists per <source, destination, sync_mode> tuple.
     private Set<JobKey> repeatJobDetectionSet = ConcurrentHashMap.newKeySet();
 
+    private static String normalizeNullableName(String value) {
+        return value == null ? "" : value;
+    }
+
+    private static String normalizeTableKey(Triple<String, String, String> tableTriple) {
+        String dbName = normalizeNullableName(tableTriple.getLeft());
+        String tableName = normalizeNullableName(tableTriple.getMiddle());
+        String partitionName = normalizeNullableName(tableTriple.getRight());
+        if (partitionName.isEmpty()) {
+            return dbName + "." + tableName;
+        }
+        return dbName + "." + tableName + "." + partitionName;
+    }
+
+    private static List<String> normalizeTables(List<Triple<String, String, String>> tables) {
+        if (tables == null || tables.isEmpty()) {
+            return Collections.emptyList();
+        }
+        HashSet<String> normalizedTables = new HashSet<>();
+        for (Triple<String, String, String> table : tables) {
+            normalizedTables.add(normalizeTableKey(table));
+        }
+        List<String> sortedTables = new ArrayList<>(normalizedTables);
+        Collections.sort(sortedTables);
+        return sortedTables;
+    }
+
+    private boolean isClusterOnceCommand(WarmUpClusterCommand command) {
+        Map<String, String> properties = command.getProperties();
+        if (properties == null) {
+            return true;
+        }
+        String syncMode = properties.get("sync_mode");
+        return !"periodic".equals(syncMode) && !"event_driven".equals(syncMode);
+    }
+
+    private OncePendingJobKey buildOncePendingJobKey(WarmUpClusterCommand command) {
+        if (command.isWarmUpWithTable()) {
+            return new OncePendingJobKey(JobType.TABLE, "", command.getDstCluster(),
+                    normalizeTables(command.getTables()), command.isForce());
+        }
+        if (!isClusterOnceCommand(command)) {
+            return null;
+        }
+        return new OncePendingJobKey(JobType.CLUSTER, command.getSrcCluster(),
+                command.getDstCluster(), Collections.emptyList(), false);
+    }
+
+    private OncePendingJobKey buildOncePendingJobKey(CloudWarmUpJob job) {
+        if (!job.isOnce()) {
+            return null;
+        }
+        if (job.getJobType() == JobType.TABLE) {
+            return new OncePendingJobKey(JobType.TABLE, "", job.getDstClusterName(),
+                    normalizeTables(job.tables), job.force);
+        }
+        if (job.getJobType() == JobType.CLUSTER) {
+            return new OncePendingJobKey(JobType.CLUSTER, job.getSrcClusterName(),
+                    job.getDstClusterName(), Collections.emptyList(), false);
+        }
+        return null;
+    }
+
+    private CloudWarmUpJob findExistingPendingOnceJob(OncePendingJobKey key) {
+        CloudWarmUpJob selectedJob = null;
+        for (CloudWarmUpJob job : cloudWarmUpJobs.values()) {
+            if (job.getJobState() != JobState.PENDING || !job.isOnce()) {
+                continue;
+            }
+            OncePendingJobKey existingKey = buildOncePendingJobKey(job);
+            if (!key.equals(existingKey)) {
+                continue;
+            }
+            if (selectedJob == null
+                    || job.getCreateTimeMs() < selectedJob.getCreateTimeMs()
+                    || (job.getCreateTimeMs() == selectedJob.getCreateTimeMs()
+                        && job.getJobId() < selectedJob.getJobId())) {
+                selectedJob = job;
+            }
+        }
+        return selectedJob;
+    }
+
+    private RefCountedPendingCreateLock retainOncePendingCreateLock(OncePendingJobKey key) {
+        return oncePendingCreateLocks.compute(key, (ignored, existingLock) -> {
+            if (existingLock == null) {
+                return new RefCountedPendingCreateLock();
+            }
+            existingLock.retain();
+            return existingLock;
+        });
+    }
+
+    private void releaseOncePendingCreateLock(OncePendingJobKey key, RefCountedPendingCreateLock lock) {
+        oncePendingCreateLocks.compute(key, (ignored, existingLock) -> {
+            Preconditions.checkState(existingLock == lock, "unexpected once pending create lock entry");
+            return existingLock.release() == 0 ? null : existingLock;
+        });
+    }
+
     private void registerJobForRepeatDetection(CloudWarmUpJob job, boolean replay) throws AnalysisException {
         if (job.isDone()) {
             return;
@@ -781,6 +962,27 @@ public Map<Long, List<Tablet>> warmUpNewClusterByTable(long jobId, String dstClu
     }
 
     public long createJob(WarmUpClusterCommand stmt) throws AnalysisException {
+        OncePendingJobKey oncePendingJobKey = buildOncePendingJobKey(stmt);
+        if (oncePendingJobKey != null) {
+            RefCountedPendingCreateLock createLock = retainOncePendingCreateLock(oncePendingJobKey);
+            createLock.lock();
+            try {
+                CloudWarmUpJob existingPendingJob = findExistingPendingOnceJob(oncePendingJobKey);
+                if (existingPendingJob != null) {
+                    LOG.info("reuse existing pending warm up job {} for key {}",
+                            existingPendingJob.getJobId(), oncePendingJobKey);
+                    return existingPendingJob.getJobId();
+                }
+                return createJobInternal(stmt);
+            } finally {
+                createLock.unlock();
+                releaseOncePendingCreateLock(oncePendingJobKey, createLock);
+            }
+        }
+        return createJobInternal(stmt);
+    }
+
+    private long createJobInternal(WarmUpClusterCommand stmt) throws AnalysisException {
         long jobId = Env.getCurrentEnv().getNextId();
         CloudWarmUpJob warmUpJob;
         if (stmt.isWarmUpWithTable()) {
@@ -800,6 +1002,9 @@ public long createJob(WarmUpClusterCommand stmt) throws AnalysisException {
                     .setJobType(JobType.CLUSTER);
 
             Map<String, String> properties = stmt.getProperties();
+            if (properties == null) {
+                properties = Collections.emptyMap();
+            }
             if ("periodic".equals(properties.get("sync_mode"))) {
                 String syncIntervalSecStr = properties.get("sync_interval_sec");
                 if (syncIntervalSecStr == null) {
@@ -831,7 +1036,6 @@ public long createJob(WarmUpClusterCommand stmt) throws AnalysisException {
             }
             warmUpJob = builder.build();
         }
-
         addCloudWarmUpJob(warmUpJob);
 
         Env.getCurrentEnv().getEditLog().logModifyCloudWarmUpJob(warmUpJob);