snuyanzin commented on code in PR #27845:
URL: https://github.com/apache/flink/pull/27845#discussion_r3011466004
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java:
##########
@@ -68,29 +71,110 @@ public class MaterializedTableUtils {
public static IntervalFreshness getMaterializedTableFreshness(
SqlIntervalLiteral sqlIntervalLiteral) {
- if (sqlIntervalLiteral.signum() < 0) {
- throw new ValidationException(
- "Materialized table freshness doesn't support negative
value.");
+ return new IntervalFreshness(getFreshnessInterval(sqlIntervalLiteral));
+ }
+
+ private static Interval getFreshnessInterval(SqlIntervalLiteral
sqlIntervalLiteral) {
+ final IntervalValue intervalValue =
sqlIntervalLiteral.getValueAs(IntervalValue.class);
+ final SqlTypeName typeName =
intervalValue.getIntervalQualifier().typeName();
+
+ if (isDateTimeInterval(typeName)) {
+ final Interval freshnessInterval =
+ getDayTimeInterval(
+ intervalValue,
+ typeName,
+ sqlIntervalLiteral.getValueAs(BigDecimal.class),
+ "freshness");
+ final int interval = freshnessInterval.getInterval();
+ // Freshness interval might be only positive
+ if (interval <= 0) {
+ throw new ValidationException(
+ String.format(
+ "The freshness interval currently only
supports positive integer type values. But was: %d",
+ interval));
+ }
+ return freshnessInterval;
}
- if (sqlIntervalLiteral.getTypeName().getFamily() !=
SqlTypeFamily.INTERVAL_DAY_TIME) {
- throw new ValidationException(
- "Materialized table freshness only support SECOND, MINUTE,
HOUR, DAY as the time unit.");
+
+ throw new ValidationException(
+ "Materialized table freshness only supports SECOND, MINUTE,
HOUR, DAY, WEEK as the time unit.");
+ }
+
+ private static Interval intervalFrom(
+ SqlIntervalLiteral sqlIntervalLiteral, String intervalDescription)
{
+
+ final IntervalValue intervalValue =
sqlIntervalLiteral.getValueAs(IntervalValue.class);
+ final SqlTypeName typeName =
intervalValue.getIntervalQualifier().typeName();
+ if (intervalValue.getIntervalQualifier().isYearMonth()) {
+ return getYearMonthInterval(
+ intervalValue,
+ typeName,
+ sqlIntervalLiteral.getValueAs(BigDecimal.class),
+ intervalDescription);
+ }
+
+ if (isDateTimeInterval(typeName)) {
Review Comment:
this part is generic and will be executed for `START_MODE` where we have
both day-time and year-month
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
