[feature] (partition) Dynamic partition behavior changes (#33712)

fe/fe-core/src/main/java/org/apache/doris/clone/DynamicPartitionScheduler.java

@@ -421,18 +421,13 @@ public class DynamicPartitionScheduler extends MasterDaemon {
         ZonedDateTime now = ZonedDateTime.now(dynamicPartitionProperty.getTimeZone().toZoneId());
         String lowerBorder = DynamicPartitionUtil.getPartitionRangeString(dynamicPartitionProperty,
                 now, dynamicPartitionProperty.getStart(), partitionFormat);
-        String upperBorder = DynamicPartitionUtil.getPartitionRangeString(dynamicPartitionProperty,
-                now, dynamicPartitionProperty.getEnd() + 1, partitionFormat);
         PartitionValue lowerPartitionValue = new PartitionValue(lowerBorder);
-        PartitionValue upperPartitionValue = new PartitionValue(upperBorder);
         List<Range<PartitionKey>> reservedHistoryPartitionKeyRangeList = new ArrayList<Range<PartitionKey>>();
         Range<PartitionKey> reservePartitionKeyRange;
         try {
             PartitionKey lowerBound = PartitionKey.createPartitionKey(Collections.singletonList(lowerPartitionValue),
                     Collections.singletonList(partitionColumn));
-            PartitionKey upperBound = PartitionKey.createPartitionKey(Collections.singletonList(upperPartitionValue),
-                    Collections.singletonList(partitionColumn));
-            reservePartitionKeyRange = Range.closedOpen(lowerBound, upperBound);
+            reservePartitionKeyRange = Range.atLeast(lowerBound);
             reservedHistoryPartitionKeyRangeList.add(reservePartitionKeyRange);
         } catch (AnalysisException | IllegalArgumentException e) {
             // AnalysisException: keys.size is always equal to column.size, cannot reach this exception

regression-test/suites/partition_p1/dynamic_partition/test_dynamic_partition.groovy

@@ -0,0 +1,73 @@
+// 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.
+
+suite("test_dynamic_partition_with_update","nonConcurrent") {
+    def tbl = "test_dynamic_partition_with_update"
+    sql "drop table if exists ${tbl}"
+    sql """
+            CREATE TABLE IF NOT EXISTS ${tbl}
+            ( k1 date NOT NULL )
+            PARTITION BY RANGE(k1) ( )
+            DISTRIBUTED BY HASH(k1) BUCKETS 1
+            PROPERTIES (
+            "dynamic_partition.enable"="true",
+            "dynamic_partition.end"="3",
+            "dynamic_partition.buckets"="1",
+            "dynamic_partition.start"="-3",
+            "dynamic_partition.prefix"="p",
+            "dynamic_partition.time_unit"="DAY",
+            "dynamic_partition.create_history_partition"="true",
+            "dynamic_partition.replication_allocation" = "tag.location.default: 1",
+            "replication_num" = "1");
+        """
+
+    // set check interval time
+    sql """ admin set frontend config ('dynamic_partition_check_interval_seconds' = '2') """
+
+    // check table init
+    def result = sql "show partitions from ${tbl}"
+    assertEquals(7, result.size())
+    result = sql "show dynamic partition tables"
+    assertEquals("true",result.get(0).get(1))
+
+    // disable dynamic partition to insert partition
+    sql """ alter table ${tbl} set ('dynamic_partition.enable' = 'false') """
+    result = sql "show dynamic partition tables"
+    assertEquals("false",result.get(0).get(1))
+
+    // manually insert partition
+    sql """ alter table ${tbl} add partition p1 values [("2020-01-02"), ("2020-01-05")) """
+    sql """ alter table ${tbl} add partition p2 values [("2020-05-02"), ("2020-06-06")) """
+    sql """ alter table ${tbl} add partition p3 values [("2020-07-04"), ("2020-07-28")) """
+    sql """ alter table ${tbl} add partition p4 values [("2999-04-25"), ("2999-04-28")) """
+
+    // check size
+    result = sql "show partitions from ${tbl}"
+    assertEquals(11, result.size())
+    sql """ alter table ${tbl} set ('dynamic_partition.enable' = 'true') """
+    result = sql "show dynamic partition tables"
+    assertEquals("true",result.get(0).get(1))
+
+    // check and update
+    sleep(5000);
+
+    // check size
+    result = sql "show partitions from ${tbl}"
+    assertEquals(8, result.size())
+
+    sql "drop table ${tbl}"
+}