luoyuxia commented on code in PR #22939:
URL: https://github.com/apache/flink/pull/22939#discussion_r1262460439
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java:
##########
@@ -380,7 +380,28 @@ public Optional<ContextResolvedTable>
getTable(ObjectIdentifier objectIdentifier
resolveCatalogBaseTable(temporaryTable);
return
Optional.of(ContextResolvedTable.temporary(objectIdentifier, resolvedTable));
} else {
- return getPermanentTable(objectIdentifier);
+ return getPermanentTable(objectIdentifier, Optional.empty());
+ }
+ }
+
+ /**
+ * Retrieves a fully qualified table with a specific time. If the path is
not yet fully
Review Comment:
nit:
If the path is not yet fully qualified, use {@link
#qualifyIdentifier(UnresolvedIdentifier)} first.
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SnapshotScope.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.
+ */
+
+package org.apache.calcite.sql.validate;
+
+import org.apache.calcite.schema.SchemaVersion;
+import org.apache.calcite.sql.SqlNode;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.List;
+
+/** Represent the snapshot of the {@link SqlValidatorScope} */
+public class SnapshotScope extends DelegatingScope {
+ private final SqlValidatorSnapshot sqlValidatorSnapshot;
+
+ public SnapshotScope(SqlValidatorScope parent, SchemaVersion
schemaVersion) {
+ super(parent);
+ this.sqlValidatorSnapshot =
+ new SqlValidatorSnapshot((SqlValidatorImpl)
parent.getValidator(), schemaVersion);
+ }
+
+ @Override
+ public void resolveTable(
+ List<String> names, SqlNameMatcher nameMatcher, Path path,
Resolved resolved) {
+ // In the time travel case, the parent of the ScopeSnapshot will
always be CatalogScope
Review Comment:
```suggestion
// In the time travel case, the parent of the ScopeSnapshot will
always be CatalogScope
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java:
##########
@@ -446,12 +467,24 @@ public Optional<CatalogPartition> getPartition(
return Optional.empty();
}
- private Optional<ContextResolvedTable> getPermanentTable(ObjectIdentifier
objectIdentifier) {
+ private Optional<ContextResolvedTable> getPermanentTable(
+ ObjectIdentifier objectIdentifier, Optional<Long> timestamp) {
Review Comment:
Got warning from my IDE
`'Optional<Long>' used as type for parameter 'timestamp' `.
How about using `@Nullable Long timestamp`?
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/IdentifierSnapshotNamespace.java:
##########
@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+package org.apache.calcite.sql.validate;
+
+import org.apache.calcite.schema.SchemaVersion;
+
+/** Represent the snapshot of the {@link IdentifierNamespace} */
Review Comment:
Sorry for revisiting it again. But I found I'm still confused about the
comments.
Maybe we can follow the comment `Common base class for DML statement
namespaces` for `DmlNamespace`.
How about changing the comment to `A namespace for the statement with {@code
FOR SYSTEM_TIME AS OF TIMESTAMP} clause.`
?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/FlinkCalciteCatalogSnapshotReader.java:
##########
@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.plan;
+
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.schema.SchemaVersion;
+import org.apache.calcite.sql.validate.SqlValidatorCatalogReader;
+
+/** Represent the snapshot of {@link FlinkCalciteCatalogReader}. */
Review Comment:
```suggestion
/** Extends {@link FlinkCalciteCatalogReader} to allow to read a snapshot of
the CalciteSchema. */
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SqlValidatorSnapshot.java:
##########
@@ -0,0 +1,35 @@
+/*
+ * 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.
+ */
+
+package org.apache.calcite.sql.validate;
+
+import org.apache.flink.table.planner.plan.FlinkCalciteCatalogSnapshotReader;
+
+import org.apache.calcite.schema.SchemaVersion;
+
+/** Represent the snapshot of {@link SqlValidatorImpl} */
Review Comment:
```suggestion
/** Extends {@link SqlValidatorImpl} to allow to read the schema with a
specific version. */
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java:
##########
@@ -2924,10 +2943,29 @@ protected void afterTableFunction(
private void convertTemporalTable(Blackboard bb, SqlCall call) {
final SqlSnapshot snapshot = (SqlSnapshot) call;
final RexNode period = bb.convertExpression(snapshot.getPeriod());
-
+ final SqlNode tableRef = snapshot.getTableRef();
// convert inner query, could be a table name or a derived table
SqlNode expr = snapshot.getTableRef();
- convertFrom(bb, expr);
+ if (tableRef instanceof SqlBasicCall
Review Comment:
Please add some comments in here for the check logic.
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java:
##########
@@ -2924,10 +2943,29 @@ protected void afterTableFunction(
private void convertTemporalTable(Blackboard bb, SqlCall call) {
final SqlSnapshot snapshot = (SqlSnapshot) call;
final RexNode period = bb.convertExpression(snapshot.getPeriod());
-
+ final SqlNode tableRef = snapshot.getTableRef();
Review Comment:
Move this code line after the code line 2948 `SqlNode expr =
snapshot.getTableRef();` to make our changes group.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java:
##########
@@ -54,12 +77,24 @@ public final class FlinkCalciteSqlValidator extends
SqlValidatorImpl {
private SqlNode sqlNodeForExpectedOutputType;
private RelDataType expectedOutputType;
+ private RelOptCluster relOptCluster;
+
+ private RelOptTable.ToRelContext toRelContext;
+
+ private FrameworkConfig frameworkConfig;
Review Comment:
dito
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SnapshotScope.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.
+ */
+
+package org.apache.calcite.sql.validate;
+
+import org.apache.calcite.schema.SchemaVersion;
+import org.apache.calcite.sql.SqlNode;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.List;
+
+/** Represent the snapshot of the {@link SqlValidatorScope} */
+public class SnapshotScope extends DelegatingScope {
+ private final SqlValidatorSnapshot sqlValidatorSnapshot;
+
+ public SnapshotScope(SqlValidatorScope parent, SchemaVersion
schemaVersion) {
+ super(parent);
+ this.sqlValidatorSnapshot =
+ new SqlValidatorSnapshot((SqlValidatorImpl)
parent.getValidator(), schemaVersion);
+ }
+
+ @Override
+ public void resolveTable(
+ List<String> names, SqlNameMatcher nameMatcher, Path path,
Resolved resolved) {
+ // In the time travel case, the parent of the ScopeSnapshot will
always be CatalogScope
+ // with an EmptyScope as the parent of the CatalogScope, So we use
EmptyScope here directly.
Review Comment:
```suggestion
// with an EmptyScope as the parent of the CatalogScope, so we use
EmptyScope here directly.
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java:
##########
@@ -125,4 +160,82 @@ public void validateColumnListParams(
// this makes it possible to ignore them in the validator and fall
back to regular row types
// see also SqlFunction#deriveType
}
+
+ protected void registerNamespace(
+ @Nullable SqlValidatorScope usingScope,
+ @Nullable String alias,
+ SqlValidatorNamespace ns,
+ boolean forceNullable) {
+
+ // apply snapshot to SqlValidatorNameSpace
+ // Time travel only supports constant expressions, so we need to
investigate scenarios
+ // where the period of Snapshot is a SqlIdentifier.
+ if (usingScope != null
+ && ns instanceof IdentifierNamespace
+ && ns.getEnclosingNode() instanceof SqlSnapshot
+ && !(((SqlSnapshot) ns.getEnclosingNode()).getPeriod()
instanceof SqlIdentifier)) {
+ SqlSnapshot sqlSnapshot = (SqlSnapshot) ns.getEnclosingNode();
+ SqlNode periodNode = sqlSnapshot.getPeriod();
+ SqlToRelConverter sqlToRelConverter =
this.createSqlToRelConverter();
+ RexNode rexNode = sqlToRelConverter.convertExpression(periodNode);
+ RexNode simplifiedRexNode =
+ FlinkRexUtil.simplify(
+ sqlToRelConverter.getRexBuilder(),
+ rexNode,
+ relOptCluster.getPlanner().getExecutor());
+ List<RexNode> reducedNodes = new ArrayList<>();
+ relOptCluster
+ .getPlanner()
+ .getExecutor()
+ .reduce(
+ relOptCluster.getRexBuilder(),
+ Collections.singletonList(simplifiedRexNode),
+ reducedNodes);
+ // check whether period is the unsupported expression
+ if (!(reducedNodes.get(0) instanceof RexLiteral)) {
+ throw new UnsupportedOperationException(
+ String.format("Unsupported time travel period: %s",
periodNode));
Review Comment:
```suggestion
String.format("Unsupported time travel expression:
%s for the expresion can not be reduced to a constant by Flink.", periodNode));
```
?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java:
##########
@@ -125,4 +160,82 @@ public void validateColumnListParams(
// this makes it possible to ignore them in the validator and fall
back to regular row types
// see also SqlFunction#deriveType
}
+
+ protected void registerNamespace(
+ @Nullable SqlValidatorScope usingScope,
+ @Nullable String alias,
+ SqlValidatorNamespace ns,
+ boolean forceNullable) {
+
+ // apply snapshot to SqlValidatorNameSpace
+ // Time travel only supports constant expressions, so we need to
investigate scenarios
+ // where the period of Snapshot is a SqlIdentifier.
+ if (usingScope != null
+ && ns instanceof IdentifierNamespace
+ && ns.getEnclosingNode() instanceof SqlSnapshot
+ && !(((SqlSnapshot) ns.getEnclosingNode()).getPeriod()
instanceof SqlIdentifier)) {
+ SqlSnapshot sqlSnapshot = (SqlSnapshot) ns.getEnclosingNode();
+ SqlNode periodNode = sqlSnapshot.getPeriod();
+ SqlToRelConverter sqlToRelConverter =
this.createSqlToRelConverter();
+ RexNode rexNode = sqlToRelConverter.convertExpression(periodNode);
+ RexNode simplifiedRexNode =
+ FlinkRexUtil.simplify(
+ sqlToRelConverter.getRexBuilder(),
+ rexNode,
+ relOptCluster.getPlanner().getExecutor());
+ List<RexNode> reducedNodes = new ArrayList<>();
+ relOptCluster
+ .getPlanner()
+ .getExecutor()
+ .reduce(
+ relOptCluster.getRexBuilder(),
+ Collections.singletonList(simplifiedRexNode),
+ reducedNodes);
+ // check whether period is the unsupported expression
+ if (!(reducedNodes.get(0) instanceof RexLiteral)) {
+ throw new UnsupportedOperationException(
+ String.format("Unsupported time travel period: %s",
periodNode));
+ }
+
+ RexLiteral rexLiteral = (RexLiteral) (reducedNodes).get(0);
+ sqlSnapshot.setOperand(
+ 1,
+ SqlLiteral.createTimestamp(
+ rexLiteral.getValueAs(TimestampString.class),
+ rexLiteral.getType().getPrecision(),
+ sqlSnapshot.getPeriod().getParserPosition()));
+
+ TimestampString timestampString =
rexLiteral.getValueAs(TimestampString.class);
+ TableConfig tableConfig =
ShortcutUtils.unwrapContext(relOptCluster).getTableConfig();
+ ZoneId zoneId = tableConfig.getLocalTimeZone();
+
+ long timeTravelTimestamp =
+
TimestampData.fromEpochMillis(timestampString.getMillisSinceEpoch())
Review Comment:
dito:
Also check timestampString is not null to reduce the warning `Method
invocation 'getMillisSinceEpoch' may produce 'NullPointerException' ` from IDE
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java:
##########
@@ -0,0 +1,305 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.runtime.batch.sql;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.planner.factories.TestTimeTravelCatalog;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Test for time travel. */
+public class TimeTravelITCase extends BatchTestBase {
+ private static final List<Tuple3<String, Schema, List<Row>>>
TEST_TIME_TRAVEL_DATE =
+ Arrays.asList(
+ Tuple3.of(
+ "2023-01-01 01:00:00",
+ Schema.newBuilder().column("f1",
DataTypes.INT()).build(),
+ Collections.singletonList(Row.of(1))),
+ Tuple3.of(
+ "2023-01-01 02:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2))),
+ Tuple3.of(
+ "2023-01-01 03:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .column("f3", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2, 3))));
+
+ private static final List<Tuple2<String, String>>
EXPECTED_TIME_TRAVEL_RESULT =
+ Arrays.asList(
+ Tuple2.of("2023-01-01 01:00:00", "[+I[1]]"),
+ Tuple2.of("2023-01-01 02:00:00", "[+I[1, 2]]"),
+ Tuple2.of("2023-01-01 03:00:00", "[+I[1, 2, 3]]"));
+
+ @BeforeEach
+ public void before() {
+ TestTimeTravelCatalog catalog = new
TestTimeTravelCatalog("TimeTravelCatalog");
+
+ TEST_TIME_TRAVEL_DATE.forEach(
+ t -> {
+ String dataId = TestValuesTableFactory.registerData(t.f2);
+ Map<String, String> options = new HashMap<>();
+ options.put("connector", "values");
+ options.put("bounded", "true");
+ options.put("data-id", dataId);
+ try {
+ catalog.registerTable(
+ "t1", t.f1, options, convertStringToLong(t.f0,
ZoneId.of("UTC")));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ tEnv().registerCatalog("TimeTravelCatalog", catalog);
+ tEnv().useCatalog("TimeTravelCatalog");
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("UTC"));
+ }
+
+ @Test
+ void testTimeTravel() {
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT * FROM t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s'",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithAsExpression() {
+
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithSimpleExpression() {
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 00:00:00'+INTERVAL '60' DAY");
+ List<String> sortedResult = toSortedResults(tableResult);
+
+ assertEquals("[+I[1, 2, 3]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithDifferentTimezone() {
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ timezoneConvert(
+ res.f0,
+ ZoneId.of("UTC"),
+
ZoneId.of("Asia/Shanghai"))));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelOneTableMultiTimes() {
+ // test union all same table with different snapshot.
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " f1\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 01:00:00'\n"
+ + "UNION ALL\n"
+ + "SELECT\n"
+ + " f1\n"
Review Comment:
```suggestion
+ " f2\n"
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java:
##########
@@ -0,0 +1,305 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.runtime.batch.sql;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.planner.factories.TestTimeTravelCatalog;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Test for time travel. */
+public class TimeTravelITCase extends BatchTestBase {
+ private static final List<Tuple3<String, Schema, List<Row>>>
TEST_TIME_TRAVEL_DATE =
+ Arrays.asList(
+ Tuple3.of(
+ "2023-01-01 01:00:00",
+ Schema.newBuilder().column("f1",
DataTypes.INT()).build(),
+ Collections.singletonList(Row.of(1))),
+ Tuple3.of(
+ "2023-01-01 02:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2))),
+ Tuple3.of(
+ "2023-01-01 03:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .column("f3", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2, 3))));
+
+ private static final List<Tuple2<String, String>>
EXPECTED_TIME_TRAVEL_RESULT =
+ Arrays.asList(
+ Tuple2.of("2023-01-01 01:00:00", "[+I[1]]"),
+ Tuple2.of("2023-01-01 02:00:00", "[+I[1, 2]]"),
+ Tuple2.of("2023-01-01 03:00:00", "[+I[1, 2, 3]]"));
+
+ @BeforeEach
+ public void before() {
+ TestTimeTravelCatalog catalog = new
TestTimeTravelCatalog("TimeTravelCatalog");
+
+ TEST_TIME_TRAVEL_DATE.forEach(
+ t -> {
+ String dataId = TestValuesTableFactory.registerData(t.f2);
+ Map<String, String> options = new HashMap<>();
+ options.put("connector", "values");
+ options.put("bounded", "true");
+ options.put("data-id", dataId);
+ try {
+ catalog.registerTable(
+ "t1", t.f1, options, convertStringToLong(t.f0,
ZoneId.of("UTC")));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ tEnv().registerCatalog("TimeTravelCatalog", catalog);
+ tEnv().useCatalog("TimeTravelCatalog");
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("UTC"));
+ }
+
+ @Test
+ void testTimeTravel() {
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT * FROM t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s'",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithAsExpression() {
+
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithSimpleExpression() {
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 00:00:00'+INTERVAL '60' DAY");
+ List<String> sortedResult = toSortedResults(tableResult);
+
+ assertEquals("[+I[1, 2, 3]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithDifferentTimezone() {
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ timezoneConvert(
+ res.f0,
+ ZoneId.of("UTC"),
+
ZoneId.of("Asia/Shanghai"))));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelOneTableMultiTimes() {
+ // test union all same table with different snapshot.
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " f1\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 01:00:00'\n"
+ + "UNION ALL\n"
+ + "SELECT\n"
+ + " f1\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 02:00:00'");
+
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals("[+I[1], +I[1]]", sortedResult.toString());
Review Comment:
```suggestion
assertEquals("[+I[1], +I[2]]", sortedResult.toString());
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java:
##########
@@ -0,0 +1,305 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.runtime.batch.sql;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.planner.factories.TestTimeTravelCatalog;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Test for time travel. */
+public class TimeTravelITCase extends BatchTestBase {
+ private static final List<Tuple3<String, Schema, List<Row>>>
TEST_TIME_TRAVEL_DATE =
+ Arrays.asList(
+ Tuple3.of(
+ "2023-01-01 01:00:00",
+ Schema.newBuilder().column("f1",
DataTypes.INT()).build(),
+ Collections.singletonList(Row.of(1))),
+ Tuple3.of(
+ "2023-01-01 02:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2))),
+ Tuple3.of(
+ "2023-01-01 03:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .column("f3", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2, 3))));
+
+ private static final List<Tuple2<String, String>>
EXPECTED_TIME_TRAVEL_RESULT =
+ Arrays.asList(
+ Tuple2.of("2023-01-01 01:00:00", "[+I[1]]"),
+ Tuple2.of("2023-01-01 02:00:00", "[+I[1, 2]]"),
+ Tuple2.of("2023-01-01 03:00:00", "[+I[1, 2, 3]]"));
+
+ @BeforeEach
+ public void before() {
+ TestTimeTravelCatalog catalog = new
TestTimeTravelCatalog("TimeTravelCatalog");
+
+ TEST_TIME_TRAVEL_DATE.forEach(
+ t -> {
+ String dataId = TestValuesTableFactory.registerData(t.f2);
+ Map<String, String> options = new HashMap<>();
+ options.put("connector", "values");
+ options.put("bounded", "true");
+ options.put("data-id", dataId);
+ try {
+ catalog.registerTable(
+ "t1", t.f1, options, convertStringToLong(t.f0,
ZoneId.of("UTC")));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ tEnv().registerCatalog("TimeTravelCatalog", catalog);
+ tEnv().useCatalog("TimeTravelCatalog");
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("UTC"));
+ }
+
+ @Test
+ void testTimeTravel() {
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT * FROM t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s'",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithAsExpression() {
+
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithSimpleExpression() {
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 00:00:00'+INTERVAL '60' DAY");
+ List<String> sortedResult = toSortedResults(tableResult);
+
+ assertEquals("[+I[1, 2, 3]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithDifferentTimezone() {
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ timezoneConvert(
+ res.f0,
+ ZoneId.of("UTC"),
+
ZoneId.of("Asia/Shanghai"))));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelOneTableMultiTimes() {
+ // test union all same table with different snapshot.
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " f1\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 01:00:00'\n"
+ + "UNION ALL\n"
+ + "SELECT\n"
+ + " f1\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 02:00:00'");
+
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals("[+I[1], +I[1]]", sortedResult.toString());
+
+ // test join same table with different snapshot
+ tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " l.f1,\n"
+ + " r.f2\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 01:00:00' l\n"
+ + " LEFT JOIN t1 FOR SYSTEM_TIME AS
OF TIMESTAMP '2023-01-01 02:00:00' r ON l.f1=r.f1");
+
+ sortedResult = toSortedResults(tableResult);
+ assertEquals("[+I[1, 2]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithLookupJoin() {
+ // We must make sure time travel will not affect the lookup join and
temporal join
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " l.f1,\n"
+ + " r.f2\n"
+ + "FROM\n"
+ + " (\n"
+ + " SELECT\n"
+ + " *,\n"
+ + " proctime () as p\n"
+ + " FROM\n"
+ + " t1 FOR SYSTEM_TIME AS
OF TIMESTAMP '2023-01-01 01:00:00'\n"
+ + " ) l\n"
+ + " LEFT JOIN t1 FOR SYSTEM_TIME AS
OF l.p r ON l.f1=r.f1");
+
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals("[+I[1, 2]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithUnsupportedExpression() {
+ assertThatThrownBy(
+ () ->
+ tEnv().executeSql(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR
SYSTEM_TIME AS OF TO_TIMESTAMP_LTZ (0, 3)"))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Unsupported time travel period:
TO_TIMESTAMP_LTZ(0, 3)");
+
+ assertThatThrownBy(
+ () ->
+ tEnv().executeSql(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR
SYSTEM_TIME AS OF PROCTIME()"))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Unsupported time travel period:
PROCTIME()");
+ }
+
+ @Test
+ void testTimeTravelWithIdentifierSnapshot() {
+
Review Comment:
nit:
remove this line
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SnapshotScope.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.
+ */
+
+package org.apache.calcite.sql.validate;
+
+import org.apache.calcite.schema.SchemaVersion;
+import org.apache.calcite.sql.SqlNode;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.List;
+
+/** Represent the snapshot of the {@link SqlValidatorScope} */
Review Comment:
```suggestion
/** Scope for resolving identifiers that has a {@code FOR SYSTEM_TIME AS OF
TIMESTAMP}. */
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java:
##########
@@ -0,0 +1,305 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.runtime.batch.sql;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.planner.factories.TestTimeTravelCatalog;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Test for time travel. */
+public class TimeTravelITCase extends BatchTestBase {
+ private static final List<Tuple3<String, Schema, List<Row>>>
TEST_TIME_TRAVEL_DATE =
+ Arrays.asList(
+ Tuple3.of(
+ "2023-01-01 01:00:00",
+ Schema.newBuilder().column("f1",
DataTypes.INT()).build(),
+ Collections.singletonList(Row.of(1))),
+ Tuple3.of(
+ "2023-01-01 02:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2))),
+ Tuple3.of(
+ "2023-01-01 03:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .column("f3", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2, 3))));
+
+ private static final List<Tuple2<String, String>>
EXPECTED_TIME_TRAVEL_RESULT =
+ Arrays.asList(
+ Tuple2.of("2023-01-01 01:00:00", "[+I[1]]"),
+ Tuple2.of("2023-01-01 02:00:00", "[+I[1, 2]]"),
+ Tuple2.of("2023-01-01 03:00:00", "[+I[1, 2, 3]]"));
+
+ @BeforeEach
+ public void before() {
+ TestTimeTravelCatalog catalog = new
TestTimeTravelCatalog("TimeTravelCatalog");
+
+ TEST_TIME_TRAVEL_DATE.forEach(
+ t -> {
+ String dataId = TestValuesTableFactory.registerData(t.f2);
+ Map<String, String> options = new HashMap<>();
+ options.put("connector", "values");
+ options.put("bounded", "true");
+ options.put("data-id", dataId);
+ try {
+ catalog.registerTable(
+ "t1", t.f1, options, convertStringToLong(t.f0,
ZoneId.of("UTC")));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ tEnv().registerCatalog("TimeTravelCatalog", catalog);
+ tEnv().useCatalog("TimeTravelCatalog");
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("UTC"));
+ }
+
+ @Test
+ void testTimeTravel() {
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT * FROM t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s'",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithAsExpression() {
+
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithSimpleExpression() {
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 00:00:00'+INTERVAL '60' DAY");
+ List<String> sortedResult = toSortedResults(tableResult);
+
+ assertEquals("[+I[1, 2, 3]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithDifferentTimezone() {
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ timezoneConvert(
+ res.f0,
+ ZoneId.of("UTC"),
+
ZoneId.of("Asia/Shanghai"))));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelOneTableMultiTimes() {
+ // test union all same table with different snapshot.
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " f1\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 01:00:00'\n"
+ + "UNION ALL\n"
+ + "SELECT\n"
+ + " f1\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 02:00:00'");
+
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals("[+I[1], +I[1]]", sortedResult.toString());
+
+ // test join same table with different snapshot
+ tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " l.f1,\n"
+ + " r.f2\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 01:00:00' l\n"
+ + " LEFT JOIN t1 FOR SYSTEM_TIME AS
OF TIMESTAMP '2023-01-01 02:00:00' r ON l.f1=r.f1");
+
+ sortedResult = toSortedResults(tableResult);
+ assertEquals("[+I[1, 2]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithLookupJoin() {
+ // We must make sure time travel will not affect the lookup join and
temporal join
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " l.f1,\n"
Review Comment:
```suggestion
+ " l.f2,\n"
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SqlValidatorSnapshot.java:
##########
@@ -0,0 +1,35 @@
+/*
+ * 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.
+ */
+
+package org.apache.calcite.sql.validate;
+
+import org.apache.flink.table.planner.plan.FlinkCalciteCatalogSnapshotReader;
+
+import org.apache.calcite.schema.SchemaVersion;
+
+/** Represent the snapshot of {@link SqlValidatorImpl} */
+public class SqlValidatorSnapshot extends SqlValidatorImpl {
Review Comment:
```suggestion
public class SqlValidatorWithSnapshot extends SqlValidatorImpl {
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SqlValidatorImpl.java:
##########
@@ -2624,7 +2624,7 @@ private SqlNode registerFrom(
usingScope,
register,
operand,
- enclosingNode,
+ call,
Review Comment:
TBH, I really don't want to change this class since it's copied from
Calcite. And It's not easy to maintain these changes in Flink. Unless we have a
strong reason, it'll be better not to touch it.
I think we don't need to modify it, we can always enhance our check logic in
`FlinkCalciteSqlValidator#registerNamespace` to cover the case.
For example, if `ns.getEnclosingNode` is SqlAsOperator, we then check the
first operand of the `ns.getEnclosingNode` is SqlSnapshot not not.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java:
##########
@@ -125,4 +167,77 @@ public void validateColumnListParams(
// this makes it possible to ignore them in the validator and fall
back to regular row types
// see also SqlFunction#deriveType
}
+
+ protected void registerNamespace(
+ @Nullable SqlValidatorScope usingScope,
+ @Nullable String alias,
+ SqlValidatorNamespace ns,
+ boolean forceNullable) {
+ // apply snapshot to SqlValidatorNameSpace
+
+ if (ns instanceof IdentifierNamespace && ns.getEnclosingNode()
instanceof SqlSnapshot) {
+ SqlSnapshot sqlSnapshot = (SqlSnapshot) ((SelectScope)
usingScope).getNode().getFrom();
+ SqlNode sqlNode = sqlSnapshot.getPeriod();
+ SqlToRelConverter sqlToRelConverter =
this.createSqlToRelConverter();
+ SqlNode sqlNode1 = this.validateParameterizedExpression(sqlNode,
new HashMap<>());
+ this.getValidatedNodeType(sqlNode1);
+ RexNode rexNode = sqlToRelConverter.convertExpression(sqlNode1);
+ RexNode simplifiedRexNode =
+ FlinkRexUtil.simplify(
+ sqlToRelConverter.getRexBuilder(),
+ rexNode,
+ relOptCluster.getPlanner().getExecutor());
+ List<RexNode> reducedNodes = new ArrayList<>();
+ relOptCluster
+ .getPlanner()
+ .getExecutor()
+ .reduce(
+ relOptCluster.getRexBuilder(),
+ Collections.singletonList(simplifiedRexNode),
+ reducedNodes);
+ RexLiteral node = (RexLiteral) reducedNodes.get(0);
+ sqlSnapshot.setOperand(
+ 1,
+ SqlLiteral.createTimestamp(
+ node.getValueAs(TimestampString.class),
+ node.getType().getPrecision(),
+ sqlSnapshot.getPeriod().getParserPosition()));
+
+ TableConfig tableConfig = getTableConfig();
+ ZoneId zoneId = tableConfig.getLocalTimeZone();
+ TimestampString timestampString =
node.getValueAs(TimestampString.class);
+
+ long timeTravelTimestamp =
+
TimestampData.fromEpochMillis(timestampString.getMillisSinceEpoch())
+ .toLocalDateTime()
+ .atZone(zoneId)
+ .toEpochSecond()
+ * 1000;
+
+ SchemaVersion schemaVersion =
FlinkSchemaVersion.of(timeTravelTimestamp);
+ IdentifierNamespace identifierNamespace = (IdentifierNamespace) ns;
+ IdentifierNamespace snapshotNameSpace =
+ new IdentifierNamespaceSnapshot(
+ identifierNamespace,
+ schemaVersion,
+ ((SelectScope) usingScope).getParent());
+ ns = snapshotNameSpace;
Review Comment:
Please don't forget my comment in here.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java:
##########
@@ -446,12 +467,24 @@ public Optional<CatalogPartition> getPartition(
return Optional.empty();
}
- private Optional<ContextResolvedTable> getPermanentTable(ObjectIdentifier
objectIdentifier) {
+ private Optional<ContextResolvedTable> getPermanentTable(
+ ObjectIdentifier objectIdentifier, Optional<Long> timestamp) {
Catalog currentCatalog =
catalogs.get(objectIdentifier.getCatalogName());
ObjectPath objectPath = objectIdentifier.toObjectPath();
if (currentCatalog != null) {
try {
- final CatalogBaseTable table =
currentCatalog.getTable(objectPath);
+ final CatalogBaseTable table;
+ if (timestamp.isPresent()) {
+ table = currentCatalog.getTable(objectPath,
timestamp.get());
+ if (table.getTableKind() ==
CatalogBaseTable.TableKind.VIEW) {
+ throw new TableException(
+ String.format(
+ "Table view: %s does not support time
travel",
Review Comment:
```suggestion
" %s is a view, but time travel is
not supported for view.",
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java:
##########
@@ -54,12 +77,24 @@ public final class FlinkCalciteSqlValidator extends
SqlValidatorImpl {
private SqlNode sqlNodeForExpectedOutputType;
private RelDataType expectedOutputType;
+ private RelOptCluster relOptCluster;
Review Comment:
nit: can be final
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java:
##########
@@ -54,12 +77,24 @@ public final class FlinkCalciteSqlValidator extends
SqlValidatorImpl {
private SqlNode sqlNodeForExpectedOutputType;
private RelDataType expectedOutputType;
+ private RelOptCluster relOptCluster;
+
+ private RelOptTable.ToRelContext toRelContext;
Review Comment:
dito
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java:
##########
@@ -2924,10 +2943,29 @@ protected void afterTableFunction(
private void convertTemporalTable(Blackboard bb, SqlCall call) {
final SqlSnapshot snapshot = (SqlSnapshot) call;
final RexNode period = bb.convertExpression(snapshot.getPeriod());
-
+ final SqlNode tableRef = snapshot.getTableRef();
// convert inner query, could be a table name or a derived table
SqlNode expr = snapshot.getTableRef();
- convertFrom(bb, expr);
+ if (tableRef instanceof SqlBasicCall
Review Comment:
Also, don't forget to update in the header of the class to tell which lines
are added in the java doc of the class`
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java:
##########
@@ -0,0 +1,305 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.runtime.batch.sql;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.planner.factories.TestTimeTravelCatalog;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Test for time travel. */
+public class TimeTravelITCase extends BatchTestBase {
+ private static final List<Tuple3<String, Schema, List<Row>>>
TEST_TIME_TRAVEL_DATE =
+ Arrays.asList(
+ Tuple3.of(
+ "2023-01-01 01:00:00",
+ Schema.newBuilder().column("f1",
DataTypes.INT()).build(),
+ Collections.singletonList(Row.of(1))),
+ Tuple3.of(
+ "2023-01-01 02:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2))),
+ Tuple3.of(
+ "2023-01-01 03:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .column("f3", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2, 3))));
+
+ private static final List<Tuple2<String, String>>
EXPECTED_TIME_TRAVEL_RESULT =
+ Arrays.asList(
+ Tuple2.of("2023-01-01 01:00:00", "[+I[1]]"),
+ Tuple2.of("2023-01-01 02:00:00", "[+I[1, 2]]"),
+ Tuple2.of("2023-01-01 03:00:00", "[+I[1, 2, 3]]"));
+
+ @BeforeEach
+ public void before() {
+ TestTimeTravelCatalog catalog = new
TestTimeTravelCatalog("TimeTravelCatalog");
+
+ TEST_TIME_TRAVEL_DATE.forEach(
+ t -> {
+ String dataId = TestValuesTableFactory.registerData(t.f2);
+ Map<String, String> options = new HashMap<>();
+ options.put("connector", "values");
+ options.put("bounded", "true");
+ options.put("data-id", dataId);
+ try {
+ catalog.registerTable(
+ "t1", t.f1, options, convertStringToLong(t.f0,
ZoneId.of("UTC")));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ tEnv().registerCatalog("TimeTravelCatalog", catalog);
+ tEnv().useCatalog("TimeTravelCatalog");
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("UTC"));
+ }
+
+ @Test
+ void testTimeTravel() {
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT * FROM t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s'",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithAsExpression() {
+
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithSimpleExpression() {
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 00:00:00'+INTERVAL '60' DAY");
+ List<String> sortedResult = toSortedResults(tableResult);
+
+ assertEquals("[+I[1, 2, 3]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithDifferentTimezone() {
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ timezoneConvert(
+ res.f0,
+ ZoneId.of("UTC"),
+
ZoneId.of("Asia/Shanghai"))));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelOneTableMultiTimes() {
+ // test union all same table with different snapshot.
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " f1\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 01:00:00'\n"
+ + "UNION ALL\n"
+ + "SELECT\n"
+ + " f1\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 02:00:00'");
+
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals("[+I[1], +I[1]]", sortedResult.toString());
+
+ // test join same table with different snapshot
+ tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " l.f1,\n"
+ + " r.f2\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 01:00:00' l\n"
+ + " LEFT JOIN t1 FOR SYSTEM_TIME AS
OF TIMESTAMP '2023-01-01 02:00:00' r ON l.f1=r.f1");
+
+ sortedResult = toSortedResults(tableResult);
+ assertEquals("[+I[1, 2]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithLookupJoin() {
+ // We must make sure time travel will not affect the lookup join and
temporal join
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " l.f1,\n"
+ + " r.f2\n"
+ + "FROM\n"
+ + " (\n"
+ + " SELECT\n"
+ + " *,\n"
+ + " proctime () as p\n"
+ + " FROM\n"
+ + " t1 FOR SYSTEM_TIME AS
OF TIMESTAMP '2023-01-01 01:00:00'\n"
Review Comment:
```suggestion
+ " t1 FOR SYSTEM_TIME AS
OF TIMESTAMP '2023-01-01 02:00:00'\n"
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java:
##########
@@ -2924,10 +2943,29 @@ protected void afterTableFunction(
private void convertTemporalTable(Blackboard bb, SqlCall call) {
final SqlSnapshot snapshot = (SqlSnapshot) call;
final RexNode period = bb.convertExpression(snapshot.getPeriod());
-
+ final SqlNode tableRef = snapshot.getTableRef();
// convert inner query, could be a table name or a derived table
SqlNode expr = snapshot.getTableRef();
- convertFrom(bb, expr);
+ if (tableRef instanceof SqlBasicCall
+ && ((SqlBasicCall) tableRef).operand(0) instanceof
SqlIdentifier
+ && period instanceof RexLiteral) {
+ TableConfig tableConfig =
ShortcutUtils.unwrapContext(relBuilder).getTableConfig();
+ ZoneId zoneId = tableConfig.getLocalTimeZone();
+ TimestampString timestampString =
+ ((RexLiteral) period).getValueAs(TimestampString.class);
+
Review Comment:
Using `Preconditions.checkArgument` to check timestampString is not null?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java:
##########
@@ -125,4 +160,82 @@ public void validateColumnListParams(
// this makes it possible to ignore them in the validator and fall
back to regular row types
// see also SqlFunction#deriveType
}
+
+ protected void registerNamespace(
+ @Nullable SqlValidatorScope usingScope,
+ @Nullable String alias,
+ SqlValidatorNamespace ns,
+ boolean forceNullable) {
+
+ // apply snapshot to SqlValidatorNameSpace
+ // Time travel only supports constant expressions, so we need to
investigate scenarios
+ // where the period of Snapshot is a SqlIdentifier.
+ if (usingScope != null
Review Comment:
Extract the check logic to a new method, and please add some comments for
the new method.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestTimeTravelCatalog.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.factories;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/** Test Catalog for testing time travel. */
+public class TestTimeTravelCatalog extends GenericInMemoryCatalog {
+ public TestTimeTravelCatalog(String name) {
+ super(name);
+
+ this.catalogs = new HashMap<>();
+ }
+
+ private final Map<ObjectPath, List<Tuple2<Long, CatalogTable>>> catalogs;
Review Comment:
I think it should be `tables` instead of `catalogs`.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java:
##########
@@ -0,0 +1,305 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.runtime.batch.sql;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.planner.factories.TestTimeTravelCatalog;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Test for time travel. */
+public class TimeTravelITCase extends BatchTestBase {
+ private static final List<Tuple3<String, Schema, List<Row>>>
TEST_TIME_TRAVEL_DATE =
+ Arrays.asList(
+ Tuple3.of(
+ "2023-01-01 01:00:00",
+ Schema.newBuilder().column("f1",
DataTypes.INT()).build(),
+ Collections.singletonList(Row.of(1))),
+ Tuple3.of(
+ "2023-01-01 02:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2))),
+ Tuple3.of(
+ "2023-01-01 03:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .column("f3", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2, 3))));
+
+ private static final List<Tuple2<String, String>>
EXPECTED_TIME_TRAVEL_RESULT =
+ Arrays.asList(
+ Tuple2.of("2023-01-01 01:00:00", "[+I[1]]"),
+ Tuple2.of("2023-01-01 02:00:00", "[+I[1, 2]]"),
+ Tuple2.of("2023-01-01 03:00:00", "[+I[1, 2, 3]]"));
+
+ @BeforeEach
+ public void before() {
Review Comment:
nit:
```suggestion
void before() {
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkSchemaVersion.java:
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.calcite;
+
+import org.apache.calcite.schema.SchemaVersion;
+
+/** Schema version for time travel. */
+public abstract class FlinkSchemaVersion implements SchemaVersion {
Review Comment:
I'm wondering why we need `FlinkSchemaVersion` since we already had
`TimestampSchemaVersion`?
Can't we use `TimestampSchemaVersion` directly?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java:
##########
@@ -0,0 +1,305 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.runtime.batch.sql;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.planner.factories.TestTimeTravelCatalog;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Test for time travel. */
+public class TimeTravelITCase extends BatchTestBase {
+ private static final List<Tuple3<String, Schema, List<Row>>>
TEST_TIME_TRAVEL_DATE =
+ Arrays.asList(
+ Tuple3.of(
+ "2023-01-01 01:00:00",
+ Schema.newBuilder().column("f1",
DataTypes.INT()).build(),
+ Collections.singletonList(Row.of(1))),
+ Tuple3.of(
+ "2023-01-01 02:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2))),
+ Tuple3.of(
+ "2023-01-01 03:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .column("f3", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2, 3))));
+
+ private static final List<Tuple2<String, String>>
EXPECTED_TIME_TRAVEL_RESULT =
+ Arrays.asList(
+ Tuple2.of("2023-01-01 01:00:00", "[+I[1]]"),
+ Tuple2.of("2023-01-01 02:00:00", "[+I[1, 2]]"),
+ Tuple2.of("2023-01-01 03:00:00", "[+I[1, 2, 3]]"));
+
+ @BeforeEach
+ public void before() {
+ TestTimeTravelCatalog catalog = new
TestTimeTravelCatalog("TimeTravelCatalog");
+
+ TEST_TIME_TRAVEL_DATE.forEach(
+ t -> {
+ String dataId = TestValuesTableFactory.registerData(t.f2);
+ Map<String, String> options = new HashMap<>();
+ options.put("connector", "values");
+ options.put("bounded", "true");
+ options.put("data-id", dataId);
+ try {
+ catalog.registerTable(
+ "t1", t.f1, options, convertStringToLong(t.f0,
ZoneId.of("UTC")));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ tEnv().registerCatalog("TimeTravelCatalog", catalog);
+ tEnv().useCatalog("TimeTravelCatalog");
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("UTC"));
+ }
+
+ @Test
+ void testTimeTravel() {
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT * FROM t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s'",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithAsExpression() {
+
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithSimpleExpression() {
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 00:00:00'+INTERVAL '60' DAY");
+ List<String> sortedResult = toSortedResults(tableResult);
+
+ assertEquals("[+I[1, 2, 3]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithDifferentTimezone() {
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ timezoneConvert(
+ res.f0,
+ ZoneId.of("UTC"),
+
ZoneId.of("Asia/Shanghai"))));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelOneTableMultiTimes() {
+ // test union all same table with different snapshot.
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " f1\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 01:00:00'\n"
+ + "UNION ALL\n"
+ + "SELECT\n"
+ + " f1\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 02:00:00'");
+
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals("[+I[1], +I[1]]", sortedResult.toString());
+
+ // test join same table with different snapshot
+ tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " l.f1,\n"
+ + " r.f2\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 01:00:00' l\n"
+ + " LEFT JOIN t1 FOR SYSTEM_TIME AS
OF TIMESTAMP '2023-01-01 02:00:00' r ON l.f1=r.f1");
+
+ sortedResult = toSortedResults(tableResult);
+ assertEquals("[+I[1, 2]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithLookupJoin() {
+ // We must make sure time travel will not affect the lookup join and
temporal join
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " l.f1,\n"
+ + " r.f2\n"
Review Comment:
```suggestion
+ " r.f3\n"
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java:
##########
@@ -0,0 +1,305 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.runtime.batch.sql;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.planner.factories.TestTimeTravelCatalog;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Test for time travel. */
+public class TimeTravelITCase extends BatchTestBase {
+ private static final List<Tuple3<String, Schema, List<Row>>>
TEST_TIME_TRAVEL_DATE =
+ Arrays.asList(
+ Tuple3.of(
+ "2023-01-01 01:00:00",
+ Schema.newBuilder().column("f1",
DataTypes.INT()).build(),
+ Collections.singletonList(Row.of(1))),
+ Tuple3.of(
+ "2023-01-01 02:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2))),
+ Tuple3.of(
+ "2023-01-01 03:00:00",
+ Schema.newBuilder()
+ .column("f1", DataTypes.INT())
+ .column("f2", DataTypes.INT())
+ .column("f3", DataTypes.INT())
+ .build(),
+ Collections.singletonList(Row.of(1, 2, 3))));
+
+ private static final List<Tuple2<String, String>>
EXPECTED_TIME_TRAVEL_RESULT =
+ Arrays.asList(
+ Tuple2.of("2023-01-01 01:00:00", "[+I[1]]"),
+ Tuple2.of("2023-01-01 02:00:00", "[+I[1, 2]]"),
+ Tuple2.of("2023-01-01 03:00:00", "[+I[1, 2, 3]]"));
+
+ @BeforeEach
+ public void before() {
+ TestTimeTravelCatalog catalog = new
TestTimeTravelCatalog("TimeTravelCatalog");
+
+ TEST_TIME_TRAVEL_DATE.forEach(
+ t -> {
+ String dataId = TestValuesTableFactory.registerData(t.f2);
+ Map<String, String> options = new HashMap<>();
+ options.put("connector", "values");
+ options.put("bounded", "true");
+ options.put("data-id", dataId);
+ try {
+ catalog.registerTable(
+ "t1", t.f1, options, convertStringToLong(t.f0,
ZoneId.of("UTC")));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ tEnv().registerCatalog("TimeTravelCatalog", catalog);
+ tEnv().useCatalog("TimeTravelCatalog");
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("UTC"));
+ }
+
+ @Test
+ void testTimeTravel() {
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT * FROM t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s'",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithAsExpression() {
+
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ res.f0));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelWithSimpleExpression() {
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 00:00:00'+INTERVAL '60' DAY");
+ List<String> sortedResult = toSortedResults(tableResult);
+
+ assertEquals("[+I[1, 2, 3]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithDifferentTimezone() {
+ tEnv().getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));
+ for (Tuple2<String, String> res : EXPECTED_TIME_TRAVEL_RESULT) {
+ TableResult tableResult =
+ tEnv().executeSql(
+ String.format(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME
AS OF TIMESTAMP '%s' AS t2",
+ timezoneConvert(
+ res.f0,
+ ZoneId.of("UTC"),
+
ZoneId.of("Asia/Shanghai"))));
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals(res.f1, sortedResult.toString());
+ }
+ }
+
+ @Test
+ void testTimeTravelOneTableMultiTimes() {
+ // test union all same table with different snapshot.
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " f1\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 01:00:00'\n"
+ + "UNION ALL\n"
+ + "SELECT\n"
+ + " f1\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 02:00:00'");
+
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals("[+I[1], +I[1]]", sortedResult.toString());
+
+ // test join same table with different snapshot
+ tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " l.f1,\n"
+ + " r.f2\n"
+ + "FROM\n"
+ + " t1 FOR SYSTEM_TIME AS OF
TIMESTAMP '2023-01-01 01:00:00' l\n"
+ + " LEFT JOIN t1 FOR SYSTEM_TIME AS
OF TIMESTAMP '2023-01-01 02:00:00' r ON l.f1=r.f1");
+
+ sortedResult = toSortedResults(tableResult);
+ assertEquals("[+I[1, 2]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithLookupJoin() {
+ // We must make sure time travel will not affect the lookup join and
temporal join
+ TableResult tableResult =
+ tEnv().executeSql(
+ "SELECT\n"
+ + " l.f1,\n"
+ + " r.f2\n"
+ + "FROM\n"
+ + " (\n"
+ + " SELECT\n"
+ + " *,\n"
+ + " proctime () as p\n"
+ + " FROM\n"
+ + " t1 FOR SYSTEM_TIME AS
OF TIMESTAMP '2023-01-01 01:00:00'\n"
+ + " ) l\n"
+ + " LEFT JOIN t1 FOR SYSTEM_TIME AS
OF l.p r ON l.f1=r.f1");
+
+ List<String> sortedResult = toSortedResults(tableResult);
+ assertEquals("[+I[1, 2]]", sortedResult.toString());
+ }
+
+ @Test
+ void testTimeTravelWithUnsupportedExpression() {
+ assertThatThrownBy(
+ () ->
+ tEnv().executeSql(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR
SYSTEM_TIME AS OF TO_TIMESTAMP_LTZ (0, 3)"))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Unsupported time travel period:
TO_TIMESTAMP_LTZ(0, 3)");
+
+ assertThatThrownBy(
+ () ->
+ tEnv().executeSql(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t1 FOR
SYSTEM_TIME AS OF PROCTIME()"))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Unsupported time travel period:
PROCTIME()");
+ }
+
+ @Test
+ void testTimeTravelWithIdentifierSnapshot() {
+
+ tEnv().executeSql(
+ "CREATE TABLE\n"
+ + " t2 (f1 VARCHAR, f2 TIMESTAMP(3))\n"
+ + "WITH\n"
+ + " ('connector'='values',
'bounded'='true')");
+
+ // select snapshot with identifier only support in lookup join or
temporal join.
+ // The following query can't generate a validate execution plan.
+
+ assertThatThrownBy(
+ () ->
+ tEnv().executeSql(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " t2 FOR
SYSTEM_TIME AS OF f2"))
+ .isInstanceOf(TableException.class)
+ .hasMessageContaining("Cannot generate a valid execution plan
for the given query");
+ }
+
+ @Test
+ void testTimeTravelWithView() {
+ tEnv().executeSql("CREATE VIEW tb_view AS SELECT * FROM t1");
+
+ assertThatThrownBy(
+ () ->
+ tEnv().executeSql(
+ "SELECT\n"
+ + " *\n"
+ + "FROM\n"
+ + " tb_view FOR
SYSTEM_TIME AS OF TIMESTAMP '2013-01-01 01:00:00'"))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "Table view: TimeTravelCatalog.default.tb_view does
not support time travel");
+ }
+
+ private static Long convertStringToLong(String timestamp, ZoneId zoneId) {
+ return LocalDateTime.parse(timestamp,
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))
+ .atZone(zoneId)
+ .toInstant()
+ .toEpochMilli();
+ }
+
+ private static String timezoneConvert(
+ String timestamp, ZoneId originZoneId, ZoneId convrtedZoneId) {
+ return LocalDateTime.parse(timestamp,
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))
+ .atZone(originZoneId)
+ .toInstant()
+ .atZone(convrtedZoneId)
+ .toLocalDateTime()
+ .format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:MM:ss"));
Review Comment:
```suggestion
.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
```
--
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]
