[FLINK-35872][table] Fix the incorrect partition generation for mater…

…ialized table period refresh in Full Mode

This closes #25108

(cherry picked from commit af7d2b3)

flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

@@ -28,6 +28,7 @@
 import org.apache.flink.table.api.config.TableConfigOptions;
 import org.apache.flink.table.catalog.CatalogMaterializedTable;
 import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.IntervalFreshness;
 import org.apache.flink.table.catalog.ObjectIdentifier;
 import org.apache.flink.table.catalog.ResolvedCatalogBaseTable;
 import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
@@ -102,8 +103,9 @@
 import static org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactoryUtils.getEndpointConfig;
 import static org.apache.flink.table.gateway.service.utils.Constants.CLUSTER_INFO;
 import static org.apache.flink.table.gateway.service.utils.Constants.JOB_ID;
-import static org.apache.flink.table.utils.DateTimeUtils.formatTimestampString;
+import static org.apache.flink.table.utils.DateTimeUtils.formatTimestampStringWithOffset;
 import static org.apache.flink.table.utils.IntervalFreshnessUtils.convertFreshnessToCron;
+import static org.apache.flink.table.utils.IntervalFreshnessUtils.convertFreshnessToDuration;
 
 /** Manager is responsible for execute the {@link MaterializedTableOperation}. */
 @Internal
@@ -615,6 +617,7 @@ public ResultFetcher refreshMaterializedTable(
                 isPeriodic
                         ? getPeriodRefreshPartition(
                                 scheduleTime,
+                                materializedTable.getDefinitionFreshness(),
                                 materializedTableIdentifier,
                                 materializedTable.getOptions(),
                                 operationExecutor
@@ -686,13 +689,14 @@ public ResultFetcher refreshMaterializedTable(
     @VisibleForTesting
     static Map<String, String> getPeriodRefreshPartition(
             String scheduleTime,
+            IntervalFreshness freshness,
             ObjectIdentifier materializedTableIdentifier,
             Map<String, String> materializedTableOptions,
             ZoneId localZoneId) {
         if (scheduleTime == null) {
             throw new ValidationException(
                     String.format(
-                            "Scheduler time not properly set for periodic refresh of materialized table %s.",
+                            "The scheduler time must not be null during the periodic refresh of the materialized table %s.",
                             materializedTableIdentifier));
         }
 
@@ -707,12 +711,14 @@ static Map<String, String> getPeriodRefreshPartition(
                             PARTITION_FIELDS.length() + 1,
                             partKey.length() - (DATE_FORMATTER.length() + 1));
             String partFieldFormatter = materializedTableOptions.get(partKey);
+
             String partFiledValue =
-                    formatTimestampString(
+                    formatTimestampStringWithOffset(
                             scheduleTime,
                             SCHEDULE_TIME_DATE_FORMATTER_DEFAULT,
                             partFieldFormatter,
-                            TimeZone.getTimeZone(localZoneId));
+                            TimeZone.getTimeZone(localZoneId),
+                            -convertFreshnessToDuration(freshness).toMillis());
             if (partFiledValue == null) {
                 throw new SqlExecutionException(
                         String.format(

flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java

@@ -1392,22 +1392,21 @@ void testPeriodicRefreshMaterializedTableWithoutPartitionOptions() throws Except
     @Test
     void testPeriodicRefreshMaterializedTableWithPartitionOptions() throws Exception {
         List<Row> data = new ArrayList<>();
-        data.add(Row.of(1L, 1L, 1L, "2024-01-01"));
-        data.add(Row.of(2L, 2L, 2L, "2024-01-02"));
 
         // create materialized table with partition formatter
         createAndVerifyCreateMaterializedTableWithData(
                 "my_materialized_table",
                 data,
                 Collections.singletonMap("ds", "yyyy-MM-dd"),
-                RefreshMode.CONTINUOUS);
+                RefreshMode.FULL);
 
         ObjectIdentifier materializedTableIdentifier =
                 ObjectIdentifier.of(
                         fileSystemCatalogName, TEST_DEFAULT_DATABASE, "my_materialized_table");
 
         // add more data to all data list
-        data.add(Row.of(3L, 3L, 3L, "2024-01-01"));
+        data.add(Row.of(1L, 1L, 1L, "2024-01-01"));
+        data.add(Row.of(2L, 2L, 2L, "2024-01-02"));
         data.add(Row.of(4L, 4L, 4L, "2024-01-02"));
 
         // refresh the materialized table with period schedule
@@ -1417,7 +1416,7 @@ void testPeriodicRefreshMaterializedTableWithPartitionOptions() throws Exception
                         sessionHandle,
                         materializedTableIdentifier.asSerializableString(),
                         true,
-                        "2024-01-02 00:00:00",
+                        "2024-01-03 00:00:00",
                         Collections.emptyMap(),
                         Collections.emptyMap(),
                         Collections.emptyMap());
@@ -1485,7 +1484,7 @@ void testRefreshMaterializedTableWithInvalidParameterInContinuousMode() throws E
                 .isInstanceOf(ValidationException.class)
                 .hasMessage(
                         String.format(
-                                "Scheduler time not properly set for periodic refresh of materialized table %s.",
+                                "The scheduler time must not be null during the periodic refresh of the materialized table %s.",
                                 ObjectIdentifier.of(
                                                 fileSystemCatalogName,
                                                 TEST_DEFAULT_DATABASE,