branch-4.2: [fix](fe) Check automatic partition limit before creation

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

@@ -3647,8 +3647,18 @@ public void addPartition(Database db, String tableName, AddPartitionClause addPa
                                              boolean writeEditLog,
                                              List<Pair<PartitionPersistInfo, Partition>> batchPartitions)
             throws DdlException {
+        addPartition(db, tableName, addPartitionClause, isCreateTable, generatedPartitionId,
+                writeEditLog, batchPartitions, false);
+    }
+
+    public void addPartition(Database db, String tableName, AddPartitionClause addPartitionClause,
+                                             boolean isCreateTable, long generatedPartitionId,
+                                             boolean writeEditLog,
+                                             List<Pair<PartitionPersistInfo, Partition>> batchPartitions,
+                                             boolean enforceAutoPartitionLimit)
+            throws DdlException {
         getInternalCatalog().addPartition(db, tableName, addPartitionClause,
-                isCreateTable, generatedPartitionId, writeEditLog, batchPartitions);
+                isCreateTable, generatedPartitionId, writeEditLog, batchPartitions, enforceAutoPartitionLimit);
     }
 
     public void addMultiPartitions(Database db, String tableName, AlterMultiPartitionClause multiPartitionClause)

fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java

@@ -1468,6 +1468,16 @@ public void addPartition(Database db, String tableName, AddPartitionClause addPa
                                              boolean writeEditLog,
                                              List<Pair<PartitionPersistInfo, Partition>> batchPartitions)
             throws DdlException {
+        addPartition(db, tableName, addPartitionClause, isCreateTable, generatedPartitionId,
+                writeEditLog, batchPartitions, false);
+    }
+
+    public void addPartition(Database db, String tableName, AddPartitionClause addPartitionClause,
+                                             boolean isCreateTable, long generatedPartitionId,
+                                             boolean writeEditLog,
+                                             List<Pair<PartitionPersistInfo, Partition>> batchPartitions,
+                                             boolean enforceAutoPartitionLimit)
+            throws DdlException {
         // in cloud mode, isCreateTable == true, create dynamic partition use, so partitionId must have been generated.
         // isCreateTable == false, other case, partitionId generate in below, must be set 0
         if (!FeConstants.runningUnitTest && Config.isCloudMode()
@@ -1747,6 +1757,18 @@ public void addPartition(Database db, String tableName, AddPartitionClause addPa
                     }
                 }
 
+                // Check at publication under the table write lock. A request may have passed the
+                // batch precheck before another request published its own auto partitions.
+                if (enforceAutoPartitionLimit) {
+                    int partitionNum = olapTable.getPartitionNum();
+                    int autoPartitionLimit = Config.max_auto_partition_num;
+                    if (partitionNum >= autoPartitionLimit) {
+                        throw new DdlException(String.format(
+                                "partition numbers %d exceeded limit of variable max_auto_partition_num %d",
+                                (long) partitionNum + 1, autoPartitionLimit));
+                    }
+                }
+
                 // check if meta changed
                 // rollup index may be added or dropped during add partition operation.
                 // schema may be changed during add partition operation.

fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java

@@ -4104,10 +4104,35 @@ public TCreatePartitionResult createPartition(TCreatePartitionRequest request) t
             return result;
         }
 
+        olapTable.readLock();
+        try {
+            int partitionNum = olapTable.getPartitionNum();
+            int newPartitionNum = 0;
+            for (String partitionName : addPartitionClauseMap.keySet()) {
+                if (!olapTable.checkPartitionNameExist(partitionName)) {
+                    newPartitionNum++;
+                }
+            }
+            long expectedPartitionNum = (long) partitionNum + newPartitionNum;
+            int autoPartitionLimit = Config.max_auto_partition_num;
+            if (expectedPartitionNum > autoPartitionLimit) {
+                String errorMessage = String.format(
+                        "partition numbers %d exceeded limit of variable max_auto_partition_num %d",
+                        expectedPartitionNum, autoPartitionLimit);
+                LOG.warn(errorMessage);
+                errorStatus.setErrorMsgs(Lists.newArrayList(errorMessage));
+                result.setStatus(errorStatus);
+                LOG.warn("send create partition error status: {}", result);
+                return result;
+            }
+        } finally {
+            olapTable.readUnlock();
+        }
+
         for (AddPartitionClause addPartitionClause : addPartitionClauseMap.values()) {
             try {
-                // here maybe check and limit created partitions num
-                Env.getCurrentEnv().addPartition(db, olapTable.getName(), addPartitionClause, false, 0, true, null);
+                Env.getCurrentEnv().addPartition(db, olapTable.getName(), addPartitionClause, false, 0, true,
+                        null, true);
             } catch (DdlException e) {
                 LOG.warn(e);
                 errorStatus.setErrorMsgs(
@@ -4133,9 +4158,7 @@ public TCreatePartitionResult createPartition(TCreatePartitionRequest request) t
 
         olapTable.readLock();
         try {
-            // check partition's number limit. because partitions in addPartitionClauseMap may be duplicated with
-            // existing partitions, which would lead to false positive. so we should check the partition number AFTER
-            // adding new partitions using its ACTUAL NUMBER, rather than the sum of existing and requested partitions.
+            // Retain the actual-count check for concurrent changes and configuration updates during creation.
             int partitionNum = olapTable.getPartitionNum();
             int autoPartitionLimit = Config.max_auto_partition_num;
             if (partitionNum > autoPartitionLimit) {

fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java

@@ -412,6 +412,108 @@ public void testCreatePartitionRange() throws Exception {
         Assert.assertNotNull(p20230807);
     }
 
+    @Test
+    public void testCreatePartitionRangeLimitDoesNotLeavePartitions() throws Exception {
+        createTable("CREATE TABLE test.partition_range_limit (event_day DATETIME NOT NULL, id INT) "
+                + "DUPLICATE KEY(event_day, id) "
+                + "AUTO PARTITION BY RANGE (date_trunc(event_day, 'day')) () "
+                + "DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES('replication_num' = '1')");
+        Database db = Env.getCurrentInternalCatalog().getDbOrAnalysisException("test");
+        OlapTable table = (OlapTable) db.getTableOrAnalysisException("partition_range_limit");
+        int oldLimit = Config.max_auto_partition_num;
+        try {
+            Config.max_auto_partition_num = 1;
+            TCreatePartitionResult result = new FrontendServiceImpl(exeEnv).createPartition(
+                    createPartitionRequest(db, table, "2023-08-07 00:00:00", "2023-08-08 00:00:00"));
+            Assert.assertEquals(TStatusCode.RUNTIME_ERROR, result.getStatus().getStatusCode());
+            Assert.assertTrue(result.getStatus().getErrorMsgs().get(0).contains("max_auto_partition_num"));
+            Assert.assertEquals(0, table.getPartitionNum());
+        } finally {
+            Config.max_auto_partition_num = oldLimit;
+        }
+    }
+
+    @Test
+    public void testCreatePartitionListLimitCountsOnlyNewPartitions() throws Exception {
+        createTable("CREATE TABLE test.partition_list_limit (city_code VARCHAR(100) NOT NULL, id INT) "
+                + "DUPLICATE KEY(city_code, id) "
+                + "AUTO PARTITION BY LIST (city_code) () "
+                + "DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES('replication_num' = '1')");
+        Database db = Env.getCurrentInternalCatalog().getDbOrAnalysisException("test");
+        OlapTable table = (OlapTable) db.getTableOrAnalysisException("partition_list_limit");
+        int oldLimit = Config.max_auto_partition_num;
+        try {
+            Config.max_auto_partition_num = 1;
+            FrontendServiceImpl impl = new FrontendServiceImpl(exeEnv);
+            TCreatePartitionResult first = impl.createPartition(createPartitionRequest(db, table, "BEIJING"));
+            Assert.assertEquals(TStatusCode.OK, first.getStatus().getStatusCode());
+            Assert.assertEquals(1, table.getPartitionNum());
+
+            TCreatePartitionResult existing = impl.createPartition(createPartitionRequest(db, table, "BEIJING"));
+            Assert.assertEquals(TStatusCode.OK, existing.getStatus().getStatusCode());
+            Assert.assertEquals(1, table.getPartitionNum());
+
+            TCreatePartitionResult overflow = impl.createPartition(
+                    createPartitionRequest(db, table, "BEIJING", "SHANGHAI"));
+            Assert.assertEquals(TStatusCode.RUNTIME_ERROR, overflow.getStatus().getStatusCode());
+            Assert.assertTrue(overflow.getStatus().getErrorMsgs().get(0).contains("max_auto_partition_num"));
+            Assert.assertEquals(1, table.getPartitionNum());
+        } finally {
+            Config.max_auto_partition_num = oldLimit;
+        }
+    }
+
+    @Test
+    public void testCreatePartitionLimitIsRecheckedAtPublication() throws Exception {
+        createTable("CREATE TABLE test.partition_list_concurrent_limit (city_code VARCHAR(100) NOT NULL, id INT) "
+                + "DUPLICATE KEY(city_code, id) "
+                + "AUTO PARTITION BY LIST (city_code) () "
+                + "DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES('replication_num' = '1')");
+        Database db = Env.getCurrentInternalCatalog().getDbOrAnalysisException("test");
+        OlapTable table = (OlapTable) db.getTableOrAnalysisException("partition_list_concurrent_limit");
+        int oldLimit = Config.max_auto_partition_num;
+        try {
+            Config.max_auto_partition_num = 1;
+            FrontendServiceImpl impl = new FrontendServiceImpl(exeEnv);
+            TCreatePartitionResult first = impl.createPartition(createPartitionRequest(db, table, "BEIJING"));
+            Assert.assertEquals(TStatusCode.OK, first.getStatus().getStatusCode());
+
+            // Model a concurrent publication after the request's read-lock precheck.
+            OlapTable spyTable = Mockito.spy(table);
+            AtomicBoolean stalePrecheck = new AtomicBoolean(true);
+            db.unregisterTable(table.getName());
+            db.registerTable(spyTable);
+            try {
+                Mockito.doAnswer(invocation -> stalePrecheck.compareAndSet(true, false)
+                        ? 0 : invocation.callRealMethod()).when(spyTable).getPartitionNum();
+                TCreatePartitionResult result = impl.createPartition(createPartitionRequest(db, spyTable, "SHANGHAI"));
+                Assert.assertEquals(TStatusCode.RUNTIME_ERROR, result.getStatus().getStatusCode());
+                Assert.assertTrue(result.getStatus().getErrorMsgs().get(0).contains("max_auto_partition_num"));
+                Assert.assertFalse(stalePrecheck.get());
+                Assert.assertEquals(1, spyTable.getPartitionNum());
+            } finally {
+                db.unregisterTable(spyTable.getName());
+                db.registerTable(table);
+            }
+        } finally {
+            Config.max_auto_partition_num = oldLimit;
+        }
+    }
+
+    private static TCreatePartitionRequest createPartitionRequest(Database db, OlapTable table, String... values) {
+        List<List<TNullableStringLiteral>> partitionValues = new ArrayList<>();
+        for (String value : values) {
+            TNullableStringLiteral literal = new TNullableStringLiteral();
+            literal.setValue(value);
+            partitionValues.add(Collections.singletonList(literal));
+        }
+        TCreatePartitionRequest request = new TCreatePartitionRequest();
+        request.setDbId(db.getId());
+        request.setTableId(table.getId());
+        request.setPartitionValues(partitionValues);
+        return request;
+    }
+
     @Test
     public void testCreatePartitionReturnsRetryErrorWhenResultPartitionIsMissing() throws Exception {
         String createOlapTblStmt = "CREATE TABLE test.partition_dropped_before_result_snapshot(\n"