[fix](fe) Accept spaced TIMESTAMPTZ offsets in range partition bounds

fe/fe-core/src/main/java/org/apache/doris/analysis/PartitionExprUtil.java

@@ -261,6 +261,9 @@ private static PartitionValue getPartitionFromDate(Type partitionColumnType, Dat
             timeString = String.format(DATETIME_FORMATTER,
                     dateLiteral.getYear(), dateLiteral.getMonth(), dateLiteral.getDay(),
                     dateLiteral.getHour(), dateLiteral.getMinute(), dateLiteral.getSecond());
+            if (partitionColumnType.isTimeStampTz()) {
+                timeString = dateLiteral.getStringValue(partitionColumnType);
+            }
         } else {
             throw new AnalysisException(
                     "not support range partition with column type : " + partitionColumnType.toString());

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

@@ -147,11 +147,14 @@ private static Literal getDateTimeLiteral(String value, Type type) throws Analys
             } else if (type.isDatetimeV2()) {
                 return new DateTimeV2Literal(value);
             } else if (type.isTimeStampTz()) {
+                if (hasExplicitTimeZone(value)) {
+                    return new TimestampTzLiteral((TimeStampTzType) DataType.fromCatalogType(type), value);
+                }
                 DateTimeV2Literal literal = new DateTimeV2Literal(value);
-                DateTimeV2Literal dtV2Lit = (DateTimeV2Literal) (DateTimeExtractAndTransform.convertTz(
+                DateTimeV2Literal dtV2Lit = (DateTimeV2Literal) DateTimeExtractAndTransform.convertTz(
                         literal,
                         new StringLiteral(ConnectContext.get().getSessionVariable().timeZone),
-                        new StringLiteral("UTC")));
+                        new StringLiteral("UTC"));
                 return new TimestampTzLiteral((TimeStampTzType) DataType.fromCatalogType(type),
                         dtV2Lit.getYear(), dtV2Lit.getMonth(), dtV2Lit.getDay(),
                         dtV2Lit.getHour(), dtV2Lit.getMinute(), dtV2Lit.getSecond(), dtV2Lit.getMicroSecond());
@@ -164,6 +167,41 @@ private static Literal getDateTimeLiteral(String value, Type type) throws Analys
                 + "value is [" + value + "], type is [" + type + "].");
     }
 
+    private static boolean hasExplicitTimeZone(String value) {
+        String trimmedValue = value.trim();
+        int timeSeparatorIndex = -1;
+        for (int index = 0; index < trimmedValue.length(); index++) {
+            if (trimmedValue.charAt(index) == ' ' || trimmedValue.charAt(index) == 'T') {
+                timeSeparatorIndex = index;
+                break;
+            }
+        }
+
+        int timeEndIndex;
+        if (timeSeparatorIndex != -1) {
+            timeEndIndex = timeSeparatorIndex + 1;
+            while (timeEndIndex < trimmedValue.length() && (Character.isDigit(trimmedValue.charAt(timeEndIndex))
+                    || trimmedValue.charAt(timeEndIndex) == ':' || trimmedValue.charAt(timeEndIndex) == '.'
+                    || trimmedValue.charAt(timeEndIndex) == ' ')) {
+                timeEndIndex++;
+            }
+        } else {
+            timeEndIndex = 0;
+            while (timeEndIndex < trimmedValue.length() && (Character.isDigit(trimmedValue.charAt(timeEndIndex))
+                    || trimmedValue.charAt(timeEndIndex) == '.')) {
+                timeEndIndex++;
+            }
+        }
+
+        if (timeEndIndex >= trimmedValue.length() || timeEndIndex < 12) {
+            return false;
+        }
+
+        char nextChar = trimmedValue.charAt(timeEndIndex);
+        return nextChar == '+' || nextChar == '-' || Character.isLetter(nextChar)
+                || nextChar == '/' || nextChar == '_';
+    }
+
     public static PartitionKey createListPartitionKeyWithTypes(List<PartitionValue> values, List<Type> types,
             boolean isHive)
             throws AnalysisException {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/PartitionDefinition.java

@@ -194,7 +194,7 @@ public void setPartitionTypes(List<DataType> partitionTypes) {
      */
     protected PartitionValue toLegacyPartitionValueStmt(Expression e) {
         if (e.isLiteral()) {
-            return new PartitionValue(((Literal) e).getStringValue(), e.isNullLiteral());
+            return new PartitionValue(((Literal) e).toLegacyLiteral().getStringValue(), e.isNullLiteral());
         } else if (e instanceof MaxValue) {
             return PartitionValue.MAX_VALUE;
         }

fe/fe-core/src/test/java/org/apache/doris/analysis/PartitionExprUtilTest.java

@@ -224,4 +224,48 @@ public void testAutoBucketReusePrevPartitionBuckets() throws Exception {
         Assertions.assertNotNull(p2);
         Assertions.assertEquals(p1.getDistributionInfo().getBucketNum(), p2.getDistributionInfo().getBucketNum());
     }
+
+    @Test
+    public void testAutoPartitionDateTruncTimestampTzKeepsUtcBoundary() throws Exception {
+        String originalTimeZone = connectContext.getSessionVariable().getTimeZone();
+        try {
+            connectContext.getSessionVariable().setTimeZone("+00:00");
+            String createOlapTblStmt = "CREATE TABLE test.auto_timestamptz_day_partition (\n"
+                    + "  id INT,\n"
+                    + "  ts_tz TIMESTAMPTZ(6) NOT NULL\n"
+                    + ")\n"
+                    + "DUPLICATE KEY(id)\n"
+                    + "AUTO PARTITION BY RANGE (date_trunc(ts_tz, 'day')) ()\n"
+                    + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+                    + "PROPERTIES(\"replication_num\"=\"1\");";
+
+            createTable(createOlapTblStmt);
+            Database db = Env.getCurrentInternalCatalog().getDbOrAnalysisException("test");
+            OlapTable table = (OlapTable) db.getTableOrAnalysisException("auto_timestamptz_day_partition");
+
+            List<List<TNullableStringLiteral>> partitionValues = new ArrayList<>();
+            List<TNullableStringLiteral> values = new ArrayList<>();
+            TNullableStringLiteral truncatedValue = new TNullableStringLiteral();
+            truncatedValue.setValue("2024-06-15 00:00:00");
+            values.add(truncatedValue);
+            partitionValues.add(values);
+
+            FrontendServiceImpl impl = new FrontendServiceImpl(exeEnv);
+            TCreatePartitionRequest request = new TCreatePartitionRequest();
+            request.setDbId(db.getId());
+            request.setTableId(table.getId());
+            request.setPartitionValues(partitionValues);
+
+            TCreatePartitionResult result = impl.createPartition(request);
+            Assertions.assertEquals(TStatusCode.OK, result.getStatus().getStatusCode());
+            Assertions.assertNotNull(table.getPartition("p20240615000000"));
+
+            List<String> createTableStmt = new ArrayList<>();
+            Env.getDdlStmt(table, createTableStmt, null, null, false, true, -1L);
+            Assertions.assertTrue(createTableStmt.get(0).contains(
+                    "PARTITION p20240615000000 VALUES [('2024-06-15 00:00:00.000000+00:00'), ('2024-06-16 00:00:00.000000+00:00'))"));
+        } finally {
+            connectContext.getSessionVariable().setTimeZone(originalTimeZone);
+        }
+    }
 }

fe/fe-core/src/test/java/org/apache/doris/catalog/CreateTableLikeTest.java

@@ -95,6 +95,12 @@ private static void createTableLike(String sql) throws Exception {
         command.run(connectContext, new StmtExecutor(connectContext, sql));
     }
 
+    private static String getCreateTableStmt(Table table) {
+        List<String> createTableStmt = Lists.newArrayList();
+        Env.getDdlStmt(table, createTableStmt, null, null, false, true /* hide password */, -1L);
+        return createTableStmt.get(0);
+    }
+
     private static void checkTableEqual(Table newTable, Table existedTable, int rollupSize) {
         List<String> newCreateTableStmt = Lists.newArrayList();
         List<String> newAddRollupStmt = Lists.newArrayList();
@@ -384,6 +390,157 @@ public void testAbnormal() {
                         existedDbName3, newTblName3, existedTblName3, 1));
     }
 
+    @Test
+    public void testTimestampTzLessThanPartitionCreateTableAcceptsExplicitOffset() throws Exception {
+        String tableName = "test_timestamptz_less_than_create";
+        String createTableSql = "CREATE TABLE test." + tableName + " (\n"
+                + "  `ts` TIMESTAMPTZ(6) NOT NULL,\n"
+                + "  `seq` INT NOT NULL\n"
+                + ")\n"
+                + "UNIQUE KEY(`ts`)\n"
+                + "PARTITION BY RANGE(`ts`) (\n"
+                + "  PARTITION p0 VALUES LESS THAN ('2024-01-15 13:00:00 +00:00'),\n"
+                + "  PARTITION p1 VALUES LESS THAN ('2024-01-15 14:00:00 +00:00')\n"
+                + ")\n"
+                + "DISTRIBUTED BY HASH(`ts`) BUCKETS 1\n"
+                + "PROPERTIES (\n"
+                + "  \"replication_num\" = \"1\",\n"
+                + "  \"enable_unique_key_merge_on_write\" = \"true\"\n"
+                + ");";
+
+        String originalTimeZone = connectContext.getSessionVariable().getTimeZone();
+        try {
+            connectContext.getSessionVariable().setTimeZone("+00:00");
+            createTable(createTableSql);
+
+            Database db = Env.getCurrentInternalCatalog().getDbOrDdlException("test");
+            OlapTable table = (OlapTable) db.getTableOrDdlException(tableName);
+
+            String createStmt = getCreateTableStmt(table);
+            Assert.assertTrue(createStmt, createStmt.contains(
+                    "PARTITION p0 VALUES [('0000-01-01 00:00:00.000000+00:00'), ('2024-01-15 13:00:00.000000+00:00'))"));
+            Assert.assertTrue(createStmt, createStmt.contains(
+                    "PARTITION p1 VALUES [('2024-01-15 13:00:00.000000+00:00'), ('2024-01-15 14:00:00.000000+00:00'))"));
+        } finally {
+            connectContext.getSessionVariable().setTimeZone(originalTimeZone);
+        }
+    }
+
+    @Test
+    public void testTimestampTzRangePartitionCreateTableAcceptsNamedAndLowercaseZones() throws Exception {
+        String tableName = "test_timestamptz_named_lowercase_zone_create";
+        String createTableSql = "CREATE TABLE test." + tableName + " (\n"
+                + "  `ts` TIMESTAMPTZ(6) NOT NULL,\n"
+                + "  `seq` INT NOT NULL\n"
+                + ")\n"
+                + "UNIQUE KEY(`ts`)\n"
+                + "PARTITION BY RANGE(`ts`) (\n"
+                + "  PARTITION p1 VALUES [('2024-01-15 20:00:00Asia/Shanghai'), ('2024-01-15 13:00:00    uTc')),\n"
+                + "  PARTITION p2 VALUES [('2024-01-15 13:00:00    uTc'), ('2024-01-15 22:00:00 Asia/Shanghai'))\n"
+                + ")\n"
+                + "DISTRIBUTED BY HASH(`ts`) BUCKETS 1\n"
+                + "PROPERTIES (\n"
+                + "  \"replication_num\" = \"1\",\n"
+                + "  \"enable_unique_key_merge_on_write\" = \"true\"\n"
+                + ");";
+
+        String originalTimeZone = connectContext.getSessionVariable().getTimeZone();
+        try {
+            connectContext.getSessionVariable().setTimeZone("America/New_York");
+            createTable(createTableSql);
+
+            Database db = Env.getCurrentInternalCatalog().getDbOrDdlException("test");
+            OlapTable table = (OlapTable) db.getTableOrDdlException(tableName);
+
+            String createStmt = getCreateTableStmt(table);
+            Assert.assertTrue(createStmt, createStmt.contains(
+                    "PARTITION p1 VALUES [('2024-01-15 12:00:00.000000+00:00'), ('2024-01-15 13:00:00.000000+00:00'))"));
+            Assert.assertTrue(createStmt, createStmt.contains(
+                    "PARTITION p2 VALUES [('2024-01-15 13:00:00.000000+00:00'), ('2024-01-15 14:00:00.000000+00:00'))"));
+        } finally {
+            connectContext.getSessionVariable().setTimeZone(originalTimeZone);
+        }
+    }
+
+    @Test
+    public void testTimestampTzLessThanPartitionCreateTableUsesSessionTimezoneWithoutOffset() throws Exception {
+        String tableName = "test_timestamptz_less_than_session_tz";
+        String createTableSql = "CREATE TABLE test." + tableName + " (\n"
+                + "  `ts` TIMESTAMPTZ(6) NOT NULL,\n"
+                + "  `seq` INT NOT NULL\n"
+                + ")\n"
+                + "UNIQUE KEY(`ts`)\n"
+                + "PARTITION BY RANGE(`ts`) (\n"
+                + "  PARTITION p0 VALUES LESS THAN ('2024-01-15 13:00:00'),\n"
+                + "  PARTITION p1 VALUES LESS THAN ('2024-01-15 14:00:00')\n"
+                + ")\n"
+                + "DISTRIBUTED BY HASH(`ts`) BUCKETS 1\n"
+                + "PROPERTIES (\n"
+                + "  \"replication_num\" = \"1\",\n"
+                + "  \"enable_unique_key_merge_on_write\" = \"true\"\n"
+                + ");";
+
+        String originalTimeZone = connectContext.getSessionVariable().getTimeZone();
+        try {
+            connectContext.getSessionVariable().setTimeZone("America/New_York");
+            createTable(createTableSql);
+
+            Database db = Env.getCurrentInternalCatalog().getDbOrDdlException("test");
+            OlapTable table = (OlapTable) db.getTableOrDdlException(tableName);
+
+            String createStmt = getCreateTableStmt(table);
+            Assert.assertTrue(createStmt, createStmt.contains(
+                    "PARTITION p0 VALUES [('0000-01-01 00:00:00.000000+00:00'), ('2024-01-15 18:00:00.000000+00:00'))"));
+            Assert.assertTrue(createStmt, createStmt.contains(
+                    "PARTITION p1 VALUES [('2024-01-15 18:00:00.000000+00:00'), ('2024-01-15 19:00:00.000000+00:00'))"));
+        } finally {
+            connectContext.getSessionVariable().setTimeZone(originalTimeZone);
+        }
+    }
+
+    @Test
+    public void testTimestampTzRangePartitionCreateLikeKeepsUtcBoundary() throws Exception {
+        String sourceTableName = "test_timestamptz_range_like_src";
+        String cloneTableName = "test_timestamptz_range_like_clone";
+        String createSourceTableSql = "CREATE TABLE test." + sourceTableName + " (\n"
+                + "  `ts` TIMESTAMPTZ(6) NOT NULL,\n"
+                + "  `seq` INT NOT NULL\n"
+                + ")\n"
+                + "UNIQUE KEY(`ts`)\n"
+                + "PARTITION BY RANGE(`ts`) (\n"
+                + "  PARTITION p1 VALUES [('2024-01-15 12:00:00 +00:00'), ('2024-01-15 13:00:00 +00:00')),\n"
+                + "  PARTITION p2 VALUES [('2024-01-15 13:00:00 +00:00'), ('2024-01-15 14:00:00 +00:00'))\n"
+                + ")\n"
+                + "DISTRIBUTED BY HASH(`ts`) BUCKETS 1\n"
+                + "PROPERTIES (\n"
+                + "  \"replication_num\" = \"1\",\n"
+                + "  \"enable_unique_key_merge_on_write\" = \"true\"\n"
+                + ");";
+        String createLikeSql = "CREATE TABLE test." + cloneTableName + " LIKE test." + sourceTableName;
+
+        String originalTimeZone = connectContext.getSessionVariable().getTimeZone();
+        try {
+            connectContext.getSessionVariable().setTimeZone("+00:00");
+            createTable(createSourceTableSql);
+
+            connectContext.getSessionVariable().setTimeZone("Asia/Shanghai");
+            createTableLike(createLikeSql);
+
+            Database db = Env.getCurrentInternalCatalog().getDbOrDdlException("test");
+            OlapTable sourceTable = (OlapTable) db.getTableOrDdlException(sourceTableName);
+            OlapTable cloneTable = (OlapTable) db.getTableOrDdlException(cloneTableName);
+
+            String sourceCreateStmt = getCreateTableStmt(sourceTable);
+            Assert.assertTrue(sourceCreateStmt.contains(
+                    "PARTITION p1 VALUES [('2024-01-15 12:00:00.000000+00:00'), ('2024-01-15 13:00:00.000000+00:00'))"));
+            Assert.assertTrue(sourceCreateStmt.contains(
+                    "PARTITION p2 VALUES [('2024-01-15 13:00:00.000000+00:00'), ('2024-01-15 14:00:00.000000+00:00'))"));
+            checkTableEqual(cloneTable, sourceTable, 0);
+        } finally {
+            connectContext.getSessionVariable().setTimeZone(originalTimeZone);
+        }
+    }
+
     @Test
     public void checkSyncedTableWithRollup() throws Exception {
         String createTableWithRollup = "CREATE TABLE IF NOT EXISTS test.table_with_rollup_synced\n" + "(\n"