luoyuxia commented on code in PR #22939: URL: https://github.com/apache/flink/pull/22939#discussion_r1251774565
########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/IdentifierNamespaceSnapshot.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} */ +public class IdentifierNamespaceSnapshot extends IdentifierNamespace { Review Comment: ```suggestion public class IdentifierSnapshotNamespace extends IdentifierNamespace { ``` ? ########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java: ########## @@ -2924,10 +2941,33 @@ 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 = Review Comment: ```suggestion TableConfig tableConfig = ShortcutUtils.unwrapContext(relBuilder).getTableConfig(); ``` ########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/ScopeSnapshot.java: ########## @@ -0,0 +1,56 @@ +/* + * 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 ScopeSnapshot extends DelegatingScope { + private SqlValidatorSnapshot sqlValidatorSnapshot; Review Comment: nit: can be final ########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java: ########## @@ -2924,10 +2941,33 @@ 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 = + cluster.getPlanner() + .getContext() + .unwrap(FlinkContextImpl.class) + .getTableConfig(); + ZoneId zoneId = tableConfig.getLocalTimeZone(); + TimestampString timestampString = + ((RexLiteral) period).getValueAs(TimestampString.class); + + long timeTravelTimestamp = + TimestampData.fromEpochMillis(timestampString.getMillisSinceEpoch()) + .toLocalDateTime() + .atZone(zoneId) + .toEpochSecond() + * 1000; + SqlIdentifier identifier = (SqlIdentifier) ((SqlBasicCall) tableRef).operand(0); Review Comment: from my IDE(I'm not sure about that): `Casting '((SqlBasicCall)tableRef).operand(...)' to 'SqlIdentifier' is redundant ` ########## 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(); Review Comment: ```suggestion SqlSnapshot sqlSnapshot = (SqlSnapshot) ns.getEnclosingNode(); ``` ? ########## 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 = Review Comment: dito; ########## 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: nit ``` ns = new IdentifierNamespaceSnapshot( identifierNamespace, schemaVersion, ((DelegatingScope) usingScope).getParent()); ``` ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestTimeTravelCatalog.java: ########## @@ -0,0 +1,120 @@ +/* + * 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.CatalogPartitionSpec; +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.PartitionNotExistException; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.stats.CatalogColumnStatistics; +import org.apache.flink.table.catalog.stats.CatalogTableStatistics; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** Test Catalog for testing time travel. */ +public class TestTimeTravelCatalog extends GenericInMemoryCatalog { + public TestTimeTravelCatalog(String name) { + super(name); + + this.catalogs = new HashMap<>(); + } + + private Map<String, List<Tuple2<Long, CatalogTable>>> catalogs; Review Comment: nit: can be final ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestTimeTravelCatalog.java: ########## @@ -0,0 +1,120 @@ +/* + * 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.CatalogPartitionSpec; +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.PartitionNotExistException; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.stats.CatalogColumnStatistics; +import org.apache.flink.table.catalog.stats.CatalogTableStatistics; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** Test Catalog for testing time travel. */ +public class TestTimeTravelCatalog extends GenericInMemoryCatalog { + public TestTimeTravelCatalog(String name) { + super(name); + + this.catalogs = new HashMap<>(); + } + + private Map<String, List<Tuple2<Long, CatalogTable>>> catalogs; + + @Override + public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) + throws TableAlreadyExistException, DatabaseNotExistException { + super.createTable(tablePath, table, ignoreIfExists); + } + + @Override + public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException { + return super.getTable(tablePath); + } + + @Override + public CatalogBaseTable getTable(ObjectPath tablePath, long timestamp) + throws TableNotExistException { + + List<Tuple2<Long, CatalogTable>> tables = catalogs.get(tablePath.getObjectName()); + + Tuple2<Long, CatalogTable> table = + tables.stream() + .filter(t -> t.f0 <= timestamp) + .max(Comparator.comparing(t -> t.f0)) + .get(); + return table.f1; + } + + /** + * @param tableName Table name + * @param schema Table schema of the table + * @param properties Table properties to construct a table instance + * @param timestamp The snapshot of the table + */ + public void registerTable( + String tableName, Schema schema, Map<String, String> properties, long timestamp) { + CatalogTable catalogTable = + CatalogTable.of(schema, "", Collections.emptyList(), properties); + + if (!catalogs.containsKey(tableName)) { + catalogs.put(tableName, new ArrayList<>()); + } + + catalogs.get(tableName).add(Tuple2.of(timestamp, catalogTable)); + } + + @Override + public boolean tableExists(ObjectPath tablePath) { + return catalogs.keySet().stream() + .map(s -> s.split("#")[0]) + .anyMatch(s -> s.equals(tablePath.getObjectName())); + } + + @Override + public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath) Review Comment: Why override this method? ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestTimeTravelCatalog.java: ########## @@ -0,0 +1,120 @@ +/* + * 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.CatalogPartitionSpec; +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.PartitionNotExistException; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.stats.CatalogColumnStatistics; +import org.apache.flink.table.catalog.stats.CatalogTableStatistics; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** Test Catalog for testing time travel. */ +public class TestTimeTravelCatalog extends GenericInMemoryCatalog { + public TestTimeTravelCatalog(String name) { + super(name); + + this.catalogs = new HashMap<>(); + } + + private Map<String, List<Tuple2<Long, CatalogTable>>> catalogs; + + @Override + public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) + throws TableAlreadyExistException, DatabaseNotExistException { + super.createTable(tablePath, table, ignoreIfExists); + } + + @Override + public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException { + return super.getTable(tablePath); + } + + @Override + public CatalogBaseTable getTable(ObjectPath tablePath, long timestamp) + throws TableNotExistException { + + List<Tuple2<Long, CatalogTable>> tables = catalogs.get(tablePath.getObjectName()); + + Tuple2<Long, CatalogTable> table = + tables.stream() + .filter(t -> t.f0 <= timestamp) + .max(Comparator.comparing(t -> t.f0)) + .get(); + return table.f1; + } + + /** + * @param tableName Table name + * @param schema Table schema of the table + * @param properties Table properties to construct a table instance + * @param timestamp The snapshot of the table + */ + public void registerTable( + String tableName, Schema schema, Map<String, String> properties, long timestamp) { + CatalogTable catalogTable = + CatalogTable.of(schema, "", Collections.emptyList(), properties); + + if (!catalogs.containsKey(tableName)) { + catalogs.put(tableName, new ArrayList<>()); + } + + catalogs.get(tableName).add(Tuple2.of(timestamp, catalogTable)); + } + + @Override + public boolean tableExists(ObjectPath tablePath) { + return catalogs.keySet().stream() + .map(s -> s.split("#")[0]) + .anyMatch(s -> s.equals(tablePath.getObjectName())); + } + + @Override + public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath) + throws TableNotExistException { + return CatalogColumnStatistics.UNKNOWN; + } + + @Override + public CatalogColumnStatistics getPartitionColumnStatistics( + ObjectPath tablePath, CatalogPartitionSpec partitionSpec) + throws PartitionNotExistException { + return CatalogColumnStatistics.UNKNOWN; + } + + @Override + public CatalogTableStatistics getTableStatistics(ObjectPath tablePath) Review Comment: dito ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java: ########## @@ -0,0 +1,145 @@ +/* + * 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.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.planner.factories.TestTimeTravelCatalog; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.types.Row; +import org.apache.flink.util.CollectionUtil; + +import org.junit.Before; +import org.junit.Test; + +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class TimeTravelITCase { Review Comment: Can this class extend `BatchTestBase`? ########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java: ########## @@ -2924,10 +2941,33 @@ 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: dito ########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java: ########## @@ -2827,7 +2836,15 @@ private void convertIdentifier( final boolean[] usedDataset = {false}; RelOptTable table = SqlValidatorUtil.getRelOptTable( - fromNamespace, catalogReader, datasetName, usedDataset); + fromNamespace, + schemaVersion == null Review Comment: Please add which lines are added in the java doc of the class. ########## 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); Review Comment: What if the reducedNodes[0] is not instance of `RexLiteral`? ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestTimeTravelCatalog.java: ########## @@ -0,0 +1,120 @@ +/* + * 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.CatalogPartitionSpec; +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.PartitionNotExistException; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.stats.CatalogColumnStatistics; +import org.apache.flink.table.catalog.stats.CatalogTableStatistics; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** Test Catalog for testing time travel. */ +public class TestTimeTravelCatalog extends GenericInMemoryCatalog { + public TestTimeTravelCatalog(String name) { + super(name); + + this.catalogs = new HashMap<>(); + } + + private Map<String, List<Tuple2<Long, CatalogTable>>> catalogs; + + @Override + public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) + throws TableAlreadyExistException, DatabaseNotExistException { + super.createTable(tablePath, table, ignoreIfExists); + } + + @Override + public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException { + return super.getTable(tablePath); + } + + @Override + public CatalogBaseTable getTable(ObjectPath tablePath, long timestamp) + throws TableNotExistException { + + List<Tuple2<Long, CatalogTable>> tables = catalogs.get(tablePath.getObjectName()); + + Tuple2<Long, CatalogTable> table = + tables.stream() + .filter(t -> t.f0 <= timestamp) + .max(Comparator.comparing(t -> t.f0)) + .get(); Review Comment: warning from my ide: `'Optional.get()' without 'isPresent()' check ` ########## 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()); Review Comment: ```suggestion ((DelegatingScope) usingScope).getParent() ``` ? Is it possible `usingScope` is null in here? ########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/ScopeSnapshot.java: ########## @@ -0,0 +1,56 @@ +/* + * 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 ScopeSnapshot extends DelegatingScope { Review Comment: nit ```suggestion public class SnapshotScope extends DelegatingScope { ``` ? ########## 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; + } + + super.registerNamespace(usingScope, alias, ns, forceNullable); + } + + private TableConfig getTableConfig() { + return frameworkConfig.getContext().unwrap(FlinkContext.class).getTableConfig(); Review Comment: ```suggestion return ShortcutUtils.unwrapContext(relOptCluster).getTableConfig(); ``` ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java: ########## @@ -54,6 +76,12 @@ public final class FlinkCalciteSqlValidator extends SqlValidatorImpl { private SqlNode sqlNodeForExpectedOutputType; private RelDataType expectedOutputType; + private RelOptCluster relOptCluster; + + private RelOptTable.ToRelContext toRelContext; + + private FrameworkConfig frameworkConfig; + public FlinkCalciteSqlValidator( Review Comment: Seem this method can be removed then. ########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java: ########## @@ -2924,10 +2941,33 @@ 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 = + cluster.getPlanner() + .getContext() + .unwrap(FlinkContextImpl.class) + .getTableConfig(); + ZoneId zoneId = tableConfig.getLocalTimeZone(); + TimestampString timestampString = + ((RexLiteral) period).getValueAs(TimestampString.class); + + long timeTravelTimestamp = Review Comment: ``` long timeTravelTimestamp = TimestampStringUtils.toLocalDateTime(timestampString) .atZone(zoneId) .toInstant() .toEpochMilli(); ``` ? ########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/ScopeSnapshot.java: ########## @@ -0,0 +1,56 @@ +/* + * 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 ScopeSnapshot extends DelegatingScope { + private SqlValidatorSnapshot sqlValidatorSnapshot; + + public ScopeSnapshot(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) { + new EmptyScope(sqlValidatorSnapshot).resolveTable(names, nameMatcher, path, resolved); Review Comment: Please add a comment to explain why we can use `EmptyScope` directly. IIUC, the reason is. in the time travel case, the parent of the ` ScopeSnapshot` will always be `CatalogScope` with an `EmptyScope` as the parent of the `CatalogScope` ? ########## flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java: ########## @@ -463,6 +484,24 @@ private Optional<ContextResolvedTable> getPermanentTable(ObjectIdentifier object return Optional.empty(); } + private Optional<ContextResolvedTable> getPermanentTable( Review Comment: I'm just wondering can we extract a common method like : ``` getPermanentTable( ObjectIdentifier objectIdentifier, Optional<Long> timestamp) ``` Then, `getPermanentTable(ObjectIdentifier objectIdentifier)` and `getPermanentTable( ObjectIdentifier objectIdentifier, long timestamp)` can reuse the snippet code. ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java: ########## @@ -0,0 +1,145 @@ +/* + * 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.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.planner.factories.TestTimeTravelCatalog; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.types.Row; +import org.apache.flink.util.CollectionUtil; + +import org.junit.Before; +import org.junit.Test; + +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class TimeTravelITCase { Review Comment: Please add the case to cover the following case: 1: with setting different table.local-time-zone 2: jon or subquery which make a sql statement contain more than one `FOR SYSTEM_TIME AS OF TIMESTAMP` clause. ########## 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<>()); Review Comment: Do we really need to validate it? If remove these lines, the test still passes. ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/catalog/FlinkSchema.java: ########## @@ -23,13 +23,22 @@ import org.apache.calcite.schema.Schema; import org.apache.calcite.schema.SchemaVersion; +import javax.annotation.Nullable; + import java.util.Collection; import java.util.Collections; +import java.util.Optional; import java.util.Set; /** Base class for flink {@link Schema}, which provides some default implementations. */ public abstract class FlinkSchema implements Schema { + private @Nullable SchemaVersion schemaVersion; + + public FlinkSchema() { Review Comment: Can this method be removed? ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkSchemaVersion.java: ########## @@ -0,0 +1,48 @@ +/* + * 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 { + + public static SchemaVersion of(long timestamp) { + return new TimestampSchemaVersion(timestamp); + } + + /** The implementation of {@link SchemaVersion} to specify the snapshot at the specific time. */ + public static class TimestampSchemaVersion extends FlinkSchemaVersion { + + private long timestamp; + + public TimestampSchemaVersion(long timestamp) { + this.timestamp = timestamp; + } + + public long getTimestamp() { + return timestamp; + } + + @Override + public boolean isBefore(SchemaVersion other) { + return false; Review Comment: Why return false? ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkSchemaVersion.java: ########## @@ -0,0 +1,48 @@ +/* + * 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 { + + public static SchemaVersion of(long timestamp) { + return new TimestampSchemaVersion(timestamp); + } + + /** The implementation of {@link SchemaVersion} to specify the snapshot at the specific time. */ + public static class TimestampSchemaVersion extends FlinkSchemaVersion { + + private long timestamp; Review Comment: nit: can be final ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestTimeTravelCatalog.java: ########## @@ -0,0 +1,120 @@ +/* + * 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.CatalogPartitionSpec; +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.PartitionNotExistException; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.stats.CatalogColumnStatistics; +import org.apache.flink.table.catalog.stats.CatalogTableStatistics; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** Test Catalog for testing time travel. */ +public class TestTimeTravelCatalog extends GenericInMemoryCatalog { + public TestTimeTravelCatalog(String name) { + super(name); + + this.catalogs = new HashMap<>(); + } + + private Map<String, List<Tuple2<Long, CatalogTable>>> catalogs; + + @Override + public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) + throws TableAlreadyExistException, DatabaseNotExistException { + super.createTable(tablePath, table, ignoreIfExists); + } + + @Override + public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException { + return super.getTable(tablePath); + } + + @Override + public CatalogBaseTable getTable(ObjectPath tablePath, long timestamp) + throws TableNotExistException { + + List<Tuple2<Long, CatalogTable>> tables = catalogs.get(tablePath.getObjectName()); + + Tuple2<Long, CatalogTable> table = + tables.stream() + .filter(t -> t.f0 <= timestamp) + .max(Comparator.comparing(t -> t.f0)) + .get(); + return table.f1; + } + + /** + * @param tableName Table name + * @param schema Table schema of the table + * @param properties Table properties to construct a table instance + * @param timestamp The snapshot of the table + */ + public void registerTable( + String tableName, Schema schema, Map<String, String> properties, long timestamp) { + CatalogTable catalogTable = + CatalogTable.of(schema, "", Collections.emptyList(), properties); + + if (!catalogs.containsKey(tableName)) { + catalogs.put(tableName, new ArrayList<>()); + } + + catalogs.get(tableName).add(Tuple2.of(timestamp, catalogTable)); + } + + @Override + public boolean tableExists(ObjectPath tablePath) { + return catalogs.keySet().stream() + .map(s -> s.split("#")[0]) + .anyMatch(s -> s.equals(tablePath.getObjectName())); + } + + @Override + public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath) + throws TableNotExistException { + return CatalogColumnStatistics.UNKNOWN; + } + + @Override + public CatalogColumnStatistics getPartitionColumnStatistics( Review Comment: dito ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/catalog/DatabaseCalciteSchema.java: ########## @@ -69,9 +72,23 @@ public DatabaseCalciteSchema( public Table getTable(String tableName) { final ObjectIdentifier identifier = ObjectIdentifier.of(catalogName, databaseName, tableName); - return catalogManager - .getTable(identifier) - .map( + Optional<ContextResolvedTable> table; + if (getSchemaVersion().isPresent()) { + SchemaVersion schemaVersion = getSchemaVersion().get(); + if (schemaVersion instanceof FlinkSchemaVersion.TimestampSchemaVersion) { + FlinkSchemaVersion.TimestampSchemaVersion timestampSchemaVersion = + (FlinkSchemaVersion.TimestampSchemaVersion) getSchemaVersion().get(); + table = catalogManager.getTable(identifier, timestampSchemaVersion.getTimestamp()); + } else { + throw new UnsupportedOperationException( + String.format( + "Unsupported schema version type, the class is %s", Review Comment: ```suggestion "Unsupported schema version type: %s", ``` ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestTimeTravelCatalog.java: ########## @@ -0,0 +1,120 @@ +/* + * 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.CatalogPartitionSpec; +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.PartitionNotExistException; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.stats.CatalogColumnStatistics; +import org.apache.flink.table.catalog.stats.CatalogTableStatistics; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** Test Catalog for testing time travel. */ +public class TestTimeTravelCatalog extends GenericInMemoryCatalog { + public TestTimeTravelCatalog(String name) { + super(name); + + this.catalogs = new HashMap<>(); + } + + private Map<String, List<Tuple2<Long, CatalogTable>>> catalogs; + + @Override + public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) + throws TableAlreadyExistException, DatabaseNotExistException { + super.createTable(tablePath, table, ignoreIfExists); + } + + @Override + public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException { + return super.getTable(tablePath); + } + + @Override + public CatalogBaseTable getTable(ObjectPath tablePath, long timestamp) + throws TableNotExistException { + + List<Tuple2<Long, CatalogTable>> tables = catalogs.get(tablePath.getObjectName()); + + Tuple2<Long, CatalogTable> table = + tables.stream() + .filter(t -> t.f0 <= timestamp) + .max(Comparator.comparing(t -> t.f0)) + .get(); + return table.f1; + } + + /** + * @param tableName Table name + * @param schema Table schema of the table + * @param properties Table properties to construct a table instance + * @param timestamp The snapshot of the table + */ + public void registerTable( + String tableName, Schema schema, Map<String, String> properties, long timestamp) { + CatalogTable catalogTable = + CatalogTable.of(schema, "", Collections.emptyList(), properties); + + if (!catalogs.containsKey(tableName)) { + catalogs.put(tableName, new ArrayList<>()); + } + + catalogs.get(tableName).add(Tuple2.of(timestamp, catalogTable)); + } + + @Override + public boolean tableExists(ObjectPath tablePath) { + return catalogs.keySet().stream() + .map(s -> s.split("#")[0]) Review Comment: Why `s.split("#")`? ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java: ########## @@ -0,0 +1,145 @@ +/* + * 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.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.planner.factories.TestTimeTravelCatalog; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.types.Row; +import org.apache.flink.util.CollectionUtil; + +import org.junit.Before; +import org.junit.Test; + +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class TimeTravelITCase { + private StreamExecutionEnvironment env; Review Comment: nit: can be final ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java: ########## @@ -0,0 +1,145 @@ +/* + * 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.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.planner.factories.TestTimeTravelCatalog; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.types.Row; +import org.apache.flink.util.CollectionUtil; + +import org.junit.Before; +import org.junit.Test; + +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class TimeTravelITCase { + private StreamExecutionEnvironment env; + private StreamTableEnvironment tEnv; + + public static List<Tuple3<String, Schema, List<Row>>> TEST_TIME_TRAVEL_DATE = + Arrays.asList( + Tuple3.of( + "2023-01-01 01:00:00", + Schema.newBuilder().column("a", DataTypes.INT()).build(), + Arrays.asList(Row.of(1))), Review Comment: nit: Collections.singletonList ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java: ########## @@ -0,0 +1,145 @@ +/* + * 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.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.planner.factories.TestTimeTravelCatalog; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.types.Row; +import org.apache.flink.util.CollectionUtil; + +import org.junit.Before; +import org.junit.Test; + +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class TimeTravelITCase { + private StreamExecutionEnvironment env; + private StreamTableEnvironment tEnv; + + public static List<Tuple3<String, Schema, List<Row>>> TEST_TIME_TRAVEL_DATE = + Arrays.asList( + Tuple3.of( + "2023-01-01 01:00:00", + Schema.newBuilder().column("a", DataTypes.INT()).build(), + Arrays.asList(Row.of(1))), + Tuple3.of( + "2023-01-01 02:00:00", + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.INT()) + .build(), + Arrays.asList(Row.of(1, 2))), Review Comment: dito ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java: ########## @@ -0,0 +1,145 @@ +/* + * 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.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.planner.factories.TestTimeTravelCatalog; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.types.Row; +import org.apache.flink.util.CollectionUtil; + +import org.junit.Before; +import org.junit.Test; + +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class TimeTravelITCase { + private StreamExecutionEnvironment env; + private StreamTableEnvironment tEnv; + + public static List<Tuple3<String, Schema, List<Row>>> TEST_TIME_TRAVEL_DATE = + Arrays.asList( + Tuple3.of( + "2023-01-01 01:00:00", + Schema.newBuilder().column("a", DataTypes.INT()).build(), + Arrays.asList(Row.of(1))), + Tuple3.of( + "2023-01-01 02:00:00", + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.INT()) + .build(), + Arrays.asList(Row.of(1, 2))), + Tuple3.of( + "2023-01-01 03:00:00", + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.INT()) + .column("c", DataTypes.INT()) + .build(), + Arrays.asList(Row.of(1, 2, 3)))); Review Comment: dito ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java: ########## @@ -0,0 +1,145 @@ +/* + * 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.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.planner.factories.TestTimeTravelCatalog; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.types.Row; +import org.apache.flink.util.CollectionUtil; + +import org.junit.Before; +import org.junit.Test; + +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class TimeTravelITCase { Review Comment: Also, please follow the junit5 style to write the IT. ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/FlinkCalciteCatalogReaderSnapshot.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} */ +public class FlinkCalciteCatalogReaderSnapshot extends FlinkCalciteCatalogReader { Review Comment: ```suggestion public class FlinkCalciteCatalogSnapshotReader extends FlinkCalciteCatalogReader { ``` ? ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TimeTravelITCase.java: ########## @@ -0,0 +1,145 @@ +/* + * 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.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.planner.factories.TestTimeTravelCatalog; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.types.Row; +import org.apache.flink.util.CollectionUtil; + +import org.junit.Before; +import org.junit.Test; + +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class TimeTravelITCase { + private StreamExecutionEnvironment env; + private StreamTableEnvironment tEnv; + + public static List<Tuple3<String, Schema, List<Row>>> TEST_TIME_TRAVEL_DATE = + Arrays.asList( + Tuple3.of( + "2023-01-01 01:00:00", + Schema.newBuilder().column("a", DataTypes.INT()).build(), + Arrays.asList(Row.of(1))), + Tuple3.of( + "2023-01-01 02:00:00", + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.INT()) + .build(), + Arrays.asList(Row.of(1, 2))), + Tuple3.of( + "2023-01-01 03:00:00", + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.INT()) + .column("c", DataTypes.INT()) + .build(), + Arrays.asList(Row.of(1, 2, 3)))); + + public static 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]]")); + + @Before + public void setup() { + env = StreamExecutionEnvironment.getExecutionEnvironment(); + tEnv = StreamTableEnvironment.create(env, EnvironmentSettings.inBatchMode()); + TestTimeTravelCatalog catalog = new TestTimeTravelCatalog("timetravel"); + + TEST_TIME_TRAVEL_DATE.stream() Review Comment: ```suggestion TEST_TIME_TRAVEL_DATE ``` -- 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]
