This is an automated email from the ASF dual-hosted git repository.
AHeise pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new b53ea1b1c4d [FLINK-40481][table] Fix SHOW CREATE FROM_NOW timestamp
direction (#29030)
b53ea1b1c4d is described below
commit b53ea1b1c4d5ab0ec882559a4a050d82ae5740e0
Author: Arvid Heise <[email protected]>
AuthorDate: Fri Aug 28 17:18:24 2026 +0200
[FLINK-40481][table] Fix SHOW CREATE FROM_NOW timestamp direction (#29030)
ShowCreateUtil.extractStartMode adds the interval to now() when
annotating FROM_NOW/RESUME_OR_FROM_NOW in SHOW CREATE output,
producing a future timestamp in the "Evaluated to
FROM_TIMESTAMP(...)" comment instead of the documented past
resolution. Purely a display bug; it does not affect which data is
actually read.
Thread a Clock through extractStartMode and through a package-private
overload of buildShowCreateMaterializedTableRow (the public API keeps
its existing signatures and hardcodes Clock.systemUTC()), switch to
now(clock).minus(amount), and normalize the clock to UTC before
reading its wall-clock time so a clock in a non-UTC zone can never be
silently mislabeled by the subsequent toInstant(ZoneOffset.UTC).
ShowCreateUtilTest previously normalized the evaluated timestamp in
both the actual and expected strings via regex before comparing,
since it relied on the wall clock and the exact value couldn't be
known in advance. That normalization erased the value being tested,
so it could never have caught a sign error. Inject FIXED_CLOCK into
the parameterized materialized-table test instead and assert the
literal expected string with no normalization; that case already
covers the sign fix, so the only additional unit test kept is one
proving a non-UTC clock zone can't leak into the display.
Co-authored-by: Arvid Heise <[email protected]>
---
.../flink/table/api/internal/ShowCreateUtil.java | 65 +++++++++++++++++++++-
.../table/api/internal/ShowCreateUtilTest.java | 63 ++++++++++++---------
2 files changed, 100 insertions(+), 28 deletions(-)
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
index 9ea15b1eca8..0e690c4d0cb 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
@@ -43,6 +43,7 @@ import org.apache.flink.table.utils.EncodingUtils;
import org.apache.commons.lang3.StringUtils;
+import java.time.Clock;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
@@ -149,6 +150,32 @@ public class ShowCreateUtil {
additionalSensitiveKeys);
}
+ /**
+ * Package-private overload of the convenience {@code
buildShowCreateMaterializedTableRow}
+ * accepting a {@link Clock}, for the same reason as above.
+ */
+ static String buildShowCreateMaterializedTableRow(
+ ResolvedCatalogMaterializedTable table,
+ ObjectIdentifier tableIdentifier,
+ boolean isTemporary,
+ boolean createOrAlter,
+ ZoneId timeZoneId,
+ SqlFactory sqlFactory,
+ List<String> additionalSensitiveKeys,
+ Clock clock) {
+ return buildShowCreateMaterializedTableRow(
+ table,
+ tableIdentifier,
+ isTemporary,
+ createOrAlter,
+ timeZoneId,
+ sqlFactory,
+ true,
+ true,
+ additionalSensitiveKeys,
+ clock);
+ }
+
/** Show create materialized table statement only for materialized tables.
*/
public static String buildShowCreateMaterializedTableRow(
ResolvedCatalogMaterializedTable table,
@@ -160,6 +187,36 @@ public class ShowCreateUtil {
boolean includeFreshness,
boolean includeRefreshMode,
List<String> additionalSensitiveKeys) {
+ return buildShowCreateMaterializedTableRow(
+ table,
+ tableIdentifier,
+ isTemporary,
+ createOrAlter,
+ timeZoneId,
+ sqlFactory,
+ includeFreshness,
+ includeRefreshMode,
+ additionalSensitiveKeys,
+ Clock.systemUTC());
+ }
+
+ /**
+ * Show create materialized table statement only for materialized tables.
+ *
+ * <p>Package-private overload accepting a {@link Clock} so tests can pin
the "Evaluated to
+ * FROM_TIMESTAMP(...)" comment for FROM_NOW/RESUME_OR_FROM_NOW to a
deterministic value.
+ */
+ static String buildShowCreateMaterializedTableRow(
+ ResolvedCatalogMaterializedTable table,
+ ObjectIdentifier tableIdentifier,
+ boolean isTemporary,
+ boolean createOrAlter,
+ ZoneId timeZoneId,
+ SqlFactory sqlFactory,
+ boolean includeFreshness,
+ boolean includeRefreshMode,
+ List<String> additionalSensitiveKeys,
+ Clock clock) {
validateTableKind(table, tableIdentifier,
TableKind.MATERIALIZED_TABLE);
StringBuilder sb =
new StringBuilder()
@@ -184,7 +241,7 @@ public class ShowCreateUtil {
.ifPresent(partitionedBy ->
sb.append(formatPartitionedBy(partitionedBy)));
extractFormattedOptions(table.getOptions(), PRINT_INDENT,
additionalSensitiveKeys)
.ifPresent(v -> sb.append("WITH
(\n").append(v).append("\n)\n"));
- sb.append(extractStartMode(table, timeZoneId)).append("\n");
+ sb.append(extractStartMode(table, timeZoneId, clock)).append("\n");
if (includeFreshness) {
sb.append(extractFreshness(table)).append("\n");
}
@@ -386,7 +443,7 @@ public class ShowCreateUtil {
}
static String extractStartMode(
- ResolvedCatalogMaterializedTable materializedTable, ZoneId
timeZoneId) {
+ ResolvedCatalogMaterializedTable materializedTable, ZoneId
timeZoneId, Clock clock) {
StringBuilder sb = new StringBuilder("START_MODE = ");
StartMode startMode = materializedTable.getStartMode().get();
switch (startMode.getKind()) {
@@ -414,7 +471,9 @@ public class ShowCreateUtil {
.append(" /* Evaluated to FROM_TIMESTAMP(TIMESTAMP '")
.append(
getFormattedLocalDateTime(
-
LocalDateTime.now().plus(amount).toInstant(ZoneOffset.UTC),
+
LocalDateTime.now(clock.withZone(ZoneOffset.UTC))
+ .minus(amount)
+ .toInstant(ZoneOffset.UTC),
ZoneOffset.UTC))
.append("') at execution */");
break;
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/internal/ShowCreateUtilTest.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/internal/ShowCreateUtilTest.java
index 750818181d4..9a4391973c8 100644
---
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/internal/ShowCreateUtilTest.java
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/internal/ShowCreateUtilTest.java
@@ -51,15 +51,15 @@ import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.MethodSource;
+import java.time.Clock;
import java.time.Instant;
-import java.time.Period;
+import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.regex.Pattern;
import static org.assertj.core.api.Assertions.assertThat;
@@ -72,9 +72,13 @@ class ShowCreateUtilTest {
private static final ObjectIdentifier MATERIALIZED_TABLE_IDENTIFIER =
ObjectIdentifier.of("catalogName", "dbName",
"materializedTableName");
- private static final Pattern START_MODE_EVALUATED_TIMESTAMP =
- Pattern.compile(
- "/\\* Evaluated to FROM_TIMESTAMP\\(TIMESTAMP '[^']*'\\)
at execution \\*/");
+ /**
+ * Fixed clock used so the "Evaluated to FROM_TIMESTAMP(...)" comment for
+ * FROM_NOW/RESUME_OR_FROM_NOW resolves to a deterministic, assertable
value instead of the wall
+ * clock.
+ */
+ private static final Clock FIXED_CLOCK =
+ Clock.fixed(Instant.parse("2020-12-12T23:18:12Z"), ZoneOffset.UTC);
private static final ResolvedSchema ONE_COLUMN_SCHEMA =
ResolvedSchema.of(Column.physical("id", DataTypes.INT()));
@@ -142,12 +146,9 @@ class ShowCreateUtilTest {
createOrAlter,
ZoneOffset.UTC,
DefaultSqlFactory.INSTANCE,
- List.of());
- final String fixedTimestamp = "1970-01-02 12:34:56";
- final String normalizedMTString =
- setFixedTimestamp(createMaterializedTableString,
fixedTimestamp);
- final String normalizedExpected = setFixedTimestamp(expected,
fixedTimestamp);
- assertThat(normalizedMTString).isEqualTo(normalizedExpected);
+ List.of(),
+ FIXED_CLOCK);
+ assertThat(createMaterializedTableString).isEqualTo(expected);
}
@ParameterizedTest(name = "includeFreshness={0}, includeRefreshMode={1}")
@@ -195,6 +196,29 @@ class ShowCreateUtilTest {
assertThat(result).isEqualTo(expected.toString());
}
+ @Test
+ void extractStartModeFromNowIsUnaffectedByClockZone() {
+ final Clock fixedClock =
+ Clock.fixed(Instant.parse("2020-12-12T23:18:12Z"),
ZoneId.of("America/New_York"));
+ final ResolvedCatalogMaterializedTable materializedTable =
+ createResolvedMaterialized(
+ ONE_COLUMN_SCHEMA,
+ null,
+ List.of(),
+ null,
+ StartMode.of(StartModeKind.FROM_NOW, Interval.of(3,
TimeUnit.MINUTE)),
+ IntervalFreshness.ofMinute(1),
+ RefreshMode.CONTINUOUS,
+ "SELECT 1",
+ "SELECT 1");
+
+ assertThat(ShowCreateUtil.extractStartMode(materializedTable,
ZoneOffset.UTC, fixedClock))
+ .isEqualTo(
+ "START_MODE = FROM_NOW(INTERVAL '3' MINUTE) /*
Evaluated to"
+ + " FROM_TIMESTAMP(TIMESTAMP '2020-12-12
23:15:12') at execution"
+ + " */");
+ }
+
@ParameterizedTest(name = "{index}: {1}")
@MethodSource("argsForShowCreateCatalog")
void showCreateCatalog(CatalogDescriptor catalogDescriptor, String
expected) {
@@ -407,7 +431,7 @@ class ShowCreateUtilTest {
"%sMATERIALIZED TABLE
`catalogName`.`dbName`.`materializedTableName` (\n"
+ " `id` INT\n"
+ ")\n"
- + "START_MODE = FROM_NOW(INTERVAL '3' MINUTE) /*
Evaluated to FROM_TIMESTAMP(TIMESTAMP '2020-12-12 23:21:12') at execution */\n"
+ + "START_MODE = FROM_NOW(INTERVAL '3' MINUTE) /*
Evaluated to FROM_TIMESTAMP(TIMESTAMP '2020-12-12 23:15:12') at execution */\n"
+ "FRESHNESS = INTERVAL '1' MINUTE\n"
+ "REFRESH_MODE = CONTINUOUS\n"
+ "AS SELECT 1\n");
@@ -491,9 +515,7 @@ class ShowCreateUtilTest {
"Materialized table comment",
List.of("id"),
TableDistribution.of(TableDistribution.Kind.HASH, 5,
List.of("id")),
- StartMode.of(
- StartModeKind.FROM_NOW,
- Interval.of(Period.of(0, 1, 2),
TimeUnit.MONTH)),
+ StartMode.of(StartModeKind.FROM_NOW, Interval.of(1,
TimeUnit.MONTH)),
IntervalFreshness.ofMinute(3),
RefreshMode.FULL,
"SELECT * FROM tbl_a",
@@ -505,7 +527,7 @@ class ShowCreateUtilTest {
+ "COMMENT 'Materialized table comment'\n"
+ "DISTRIBUTED BY HASH(`id`) INTO 5 BUCKETS\n"
+ "PARTITIONED BY (`id`)\n"
- + "START_MODE = FROM_NOW(INTERVAL '1' MONTH) /*
Evaluated to FROM_TIMESTAMP(TIMESTAMP '1970-01-02 12:34:56') at execution */\n"
+ + "START_MODE = FROM_NOW(INTERVAL '1' MONTH) /*
Evaluated to FROM_TIMESTAMP(TIMESTAMP '2020-11-12 23:18:12') at execution */\n"
+ "FRESHNESS = INTERVAL '3' MINUTE\n"
+ "REFRESH_MODE = FULL\n"
+ "AS SELECT id, name FROM
`catalogName`.`dbName`.`tbl_a`\n");
@@ -652,13 +674,4 @@ class ShowCreateUtilTest {
freshness,
startMode);
}
-
- private static String setFixedTimestamp(String sql, String fixedTimestamp)
{
- return START_MODE_EVALUATED_TIMESTAMP
- .matcher(sql)
- .replaceAll(
- "/* Evaluated to FROM_TIMESTAMP(TIMESTAMP '"
- + fixedTimestamp
- + "') at execution */");
- }
}
