luoyuxia commented on code in PR #22939:
URL: https://github.com/apache/flink/pull/22939#discussion_r1269263880
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java:
##########
@@ -125,4 +162,117 @@ public void validateColumnListParams(
// this makes it possible to ignore them in the validator and fall
back to regular row types
// see also SqlFunction#deriveType
}
+
+ @Override
+ protected void registerNamespace(
+ @Nullable SqlValidatorScope usingScope,
+ @Nullable String alias,
+ SqlValidatorNamespace ns,
+ boolean forceNullable) {
+
+ // Generate a new validator namespace for time travel scenario.
+ // Time travel only supports constant expressions, We need to ensure
that the period of
Review Comment:
```suggestion
// Since time travel only supports constant expressions, we need to
ensure that the period of
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java:
##########
@@ -229,6 +239,8 @@
* <li>Added in FLINK-28682: Lines 2277 ~ 2294
* <li>Added in FLINK-28682: Lines 2331 ~ 2359
* <li>Added in FLINK-20873: Lines 5427 ~ 5436
Review Comment:
Now, it should be `Lines 5480 ~ 5489`
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java:
##########
@@ -2927,7 +2948,39 @@ private void convertTemporalTable(Blackboard bb, SqlCall
call) {
// convert inner query, could be a table name or a derived table
SqlNode expr = snapshot.getTableRef();
- convertFrom(bb, expr);
+ SqlNode tableRef = snapshot.getTableRef();
Review Comment:
dito. Using
```
// ----- FLINK MODIFICATION BEGIN -----
xxx
// ----- FLINK MODIFICATION END -----
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java:
##########
@@ -125,4 +162,117 @@ public void validateColumnListParams(
// this makes it possible to ignore them in the validator and fall
back to regular row types
// see also SqlFunction#deriveType
}
+
+ @Override
+ protected void registerNamespace(
+ @Nullable SqlValidatorScope usingScope,
+ @Nullable String alias,
+ SqlValidatorNamespace ns,
+ boolean forceNullable) {
+
+ // Generate a new validator namespace for time travel scenario.
+ // Time travel only supports constant expressions, We need to ensure
that the period of
+ // snapshot is not an identifier.
+ Optional<SqlSnapshot> snapshot = getSnapShotNode(ns);
+ if (usingScope != null
+ && snapshot.isPresent()
+ && !(snapshot.get().getPeriod() instanceof SqlIdentifier)) {
+ SqlSnapshot sqlSnapshot = snapshot.get();
+ 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 expression: %s for
the expression can not be reduced to a constant by Flink.",
+ periodNode));
+ }
+
+ RexLiteral rexLiteral = (RexLiteral) (reducedNodes).get(0);
+ TimestampString timestampString =
rexLiteral.getValueAs(TimestampString.class);
+ checkNotNull(
+ timestampString,
+ "The time travel expression %s can not reduce to a valid
timestamp string. This is a bug. Please file an issue.",
+ periodNode);
+
+ TableConfig tableConfig =
ShortcutUtils.unwrapContext(relOptCluster).getTableConfig();
+ ZoneId zoneId = tableConfig.getLocalTimeZone();
+ long timeTravelTimestamp =
+
TimestampData.fromEpochMillis(timestampString.getMillisSinceEpoch())
+ .toLocalDateTime()
+ .atZone(zoneId)
+ .toInstant()
+ .toEpochMilli();
+
+ SchemaVersion schemaVersion =
TimestampSchemaVersion.of(timeTravelTimestamp);
+ IdentifierNamespace identifierNamespace = (IdentifierNamespace) ns;
+ ns =
+ new IdentifierSnapshotNamespace(
+ identifierNamespace,
+ schemaVersion,
+ ((DelegatingScope) usingScope).getParent());
+
+ sqlSnapshot.setOperand(
+ 1,
+ SqlLiteral.createTimestamp(
+ timestampString,
+ rexLiteral.getType().getPrecision(),
+ sqlSnapshot.getPeriod().getParserPosition()));
+ }
+
+ super.registerNamespace(usingScope, alias, ns, forceNullable);
+ }
+
+ /**
+ * Get the {@link SqlSnapshot} node in a {@link SqlValidatorNamespace}.
+ *
+ * <p>In general, if there is a snapshot expression, the enclosing node of
IdentifierNamespace
+ * is usually SqlSnapshot. However, if we encounter a situation with an
"as" operator, we need
+ * to identify whether the enclosingNode is an "as" call and if its first
operand is
+ * SqlSnapshot.
+ *
+ * @param ns Validator namespace for validator.
Review Comment:
nit:
```suggestion
* @param ns The namespace used to find SqlSnapshot
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java:
##########
@@ -2827,7 +2840,15 @@ private void convertIdentifier(
final boolean[] usedDataset = {false};
RelOptTable table =
SqlValidatorUtil.getRelOptTable(
- fromNamespace, catalogReader, datasetName,
usedDataset);
+ fromNamespace,
Review Comment:
Sorry for revisting it again. Just notice every mofication will be srounded
by
```
// ----- FLINK MODIFICATION BEGIN -----
xxx
// ----- FLINK MODIFICATION END -----
```
Could you please also follow this style?
##########
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.tables = new HashMap<>();
+ }
+
+ private final Map<ObjectPath, List<Tuple2<Long, CatalogTable>>> tables;
Review Comment:
nit:
we usually put the declaration of private field at the begining of the class.
##########
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.tables = new HashMap<>();
+ }
+
+ private final Map<ObjectPath, List<Tuple2<Long, CatalogTable>>> tables;
+
+ @Override
+ public void createTable(ObjectPath tablePath, CatalogBaseTable table,
boolean ignoreIfExists)
Review Comment:
I think this method can be removed.
##########
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.tables = new HashMap<>();
+ }
+
+ private final Map<ObjectPath, List<Tuple2<Long, CatalogTable>>> tables;
+
+ @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 {
Review Comment:
I think this method can be removed.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalSnapshot.scala:
##########
@@ -54,8 +54,8 @@ class FlinkLogicalSnapshot(
period match {
case _: RexFieldAccess =>
Review Comment:
nit:
can be simplied to `case _: RexFieldAccess | _: RexLiteral =>`?
##########
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.tables = new HashMap<>();
+ }
+
+ private final Map<ObjectPath, List<Tuple2<Long, CatalogTable>>> tables;
+
+ @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 {
+
+ if (tables.containsKey(tablePath)) {
+ List<Tuple2<Long, CatalogTable>> tableList = tables.get(tablePath);
+
+ Optional<Tuple2<Long, CatalogTable>> table =
+ tableList.stream()
+ .filter(t -> t.f0 <= timestamp)
+ .max(Comparator.comparing(t -> t.f0));
+
+ if (table.isPresent()) {
+ return table.get().f1;
+ }
+ }
+
+ return super.getTable(tablePath);
+ }
+
+ /**
+ * @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(
Review Comment:
```suggestion
public void registerTableForTableTravel(
```
?
Also, could you please add a comment for this method? I did spend somtime to
under the logic that
it'll first add to `tables` and then add it again to `tables` in
`GenericInMemoryCatalog`?
##########
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.tables = new HashMap<>();
+ }
+
+ private final Map<ObjectPath, List<Tuple2<Long, CatalogTable>>> tables;
Review Comment:
Since the name `tables` has been used by the base class
`GenericInMemoryCatalog`, rename it to `timeTravelTables`?
--
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]
