luoyuxia commented on code in PR #22939: URL: https://github.com/apache/flink/pull/22939#discussion_r1266727886
########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SnapshotScope.java: ########## @@ -0,0 +1,59 @@ +/* + * 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; + +/** Scope for resolving identifiers that has a {@code FOR SYSTEM_TIME AS OF TIMESTAMP}. */ +public class SnapshotScope extends DelegatingScope { + private final SqlValidatorWithSnapshot sqlValidatorWithSnapshot; + + public SnapshotScope(SqlValidatorScope parent, SchemaVersion schemaVersion) { + super(parent); + this.sqlValidatorWithSnapshot = + new SqlValidatorWithSnapshot( + (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 SnapshotScope will always be CatalogScope ``` ########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java: ########## @@ -2927,7 +2948,37 @@ 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(); + // Since we have simplified the SqlSnapshot in the validate phase, we only need to check Review Comment: Got confused in here. What do you mean `simplified the SqlSnapshot`? ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java: ########## @@ -125,4 +162,118 @@ 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, so we need to investigate scenarios + // where the period of Snapshot is a SqlIdentifier. Review Comment: ```suggestion // where the period of Snapshot isn't a SqlIdentifier. ``` ? ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java: ########## @@ -125,4 +162,118 @@ 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, so we need to investigate scenarios + // where the period of Snapshot is a SqlIdentifier. + Optional<SqlSnapshot> timeTravelNode = getTimeTravelNode(ns); + if (usingScope != null + && timeTravelNode.isPresent() + && !(timeTravelNode.get().getPeriod() instanceof SqlIdentifier)) { + SqlSnapshot sqlSnapshot = timeTravelNode.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; + IdentifierNamespace snapshotNameSpace = + new IdentifierSnapshotNamespace( + identifierNamespace, + schemaVersion, + ((DelegatingScope) usingScope).getParent()); + ns = snapshotNameSpace; Review Comment: Can be simpified to ``` ns = new IdentifierSnapshotNamespace( identifierNamespace, schemaVersion, ((DelegatingScope) usingScope).getParent()); ``` ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java: ########## @@ -125,4 +162,118 @@ 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, so we need to investigate scenarios + // where the period of Snapshot is a SqlIdentifier. + Optional<SqlSnapshot> timeTravelNode = getTimeTravelNode(ns); + if (usingScope != null + && timeTravelNode.isPresent() + && !(timeTravelNode.get().getPeriod() instanceof SqlIdentifier)) { + SqlSnapshot sqlSnapshot = timeTravelNode.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; + IdentifierNamespace snapshotNameSpace = + new IdentifierSnapshotNamespace( + identifierNamespace, + schemaVersion, + ((DelegatingScope) usingScope).getParent()); + ns = snapshotNameSpace; + + sqlSnapshot.setOperand( Review Comment: Just notice this code is added in recent commit. I'm wondering why add in here? ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java: ########## @@ -125,4 +162,118 @@ 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, so we need to investigate scenarios + // where the period of Snapshot is a SqlIdentifier. + Optional<SqlSnapshot> timeTravelNode = getTimeTravelNode(ns); + if (usingScope != null + && timeTravelNode.isPresent() + && !(timeTravelNode.get().getPeriod() instanceof SqlIdentifier)) { + SqlSnapshot sqlSnapshot = timeTravelNode.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; + IdentifierNamespace snapshotNameSpace = + new IdentifierSnapshotNamespace( + identifierNamespace, + schemaVersion, + ((DelegatingScope) usingScope).getParent()); + ns = snapshotNameSpace; + + sqlSnapshot.setOperand( + 1, + SqlLiteral.createTimestamp( + timestampString, + rexLiteral.getType().getPrecision(), + sqlSnapshot.getPeriod().getParserPosition())); + } + + super.registerNamespace(usingScope, alias, ns, forceNullable); + } + + /** + * For the time travel scenario, we need to build a SchemaVersion based on the Snapshot node for Review Comment: I think we don't need to talk time travel in the java foc for this method. I think may just be some thing like `Get the SqlSnapShot node in a SqlValidatorNamespace`. ########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SnapshotScope.java: ########## @@ -0,0 +1,59 @@ +/* + * 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; + +/** Scope for resolving identifiers that has a {@code FOR SYSTEM_TIME AS OF TIMESTAMP}. */ Review Comment: Sorry for visiting it again. After reading the java doc for `OrderScope`, etc, I would like to suggestion to ``` Represents the name-resolution context for expressions in an FOR SYSTEM_TIME AS OF TIMESTAMP clause. ``` ########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java: ########## @@ -2927,7 +2948,37 @@ 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(); + // Since we have simplified the SqlSnapshot in the validate phase, we only need to check + // whether the period is a RexLiteral and tableRef's first operand is a SqlIdentifier. + if ((tableRef instanceof SqlBasicCall Review Comment: I think ``` ((tableRef instanceof SqlBasicCall && ((SqlBasicCall) tableRef).operand(0) instanceof SqlIdentifier) || tableRef instanceof SqlTableRef) ``` may be more clear althogh more brackets, but more clear for which expression are a group. ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java: ########## @@ -125,4 +162,118 @@ 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, so we need to investigate scenarios + // where the period of Snapshot is a SqlIdentifier. + Optional<SqlSnapshot> timeTravelNode = getTimeTravelNode(ns); + if (usingScope != null + && timeTravelNode.isPresent() + && !(timeTravelNode.get().getPeriod() instanceof SqlIdentifier)) { + SqlSnapshot sqlSnapshot = timeTravelNode.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; + IdentifierNamespace snapshotNameSpace = + new IdentifierSnapshotNamespace( + identifierNamespace, + schemaVersion, + ((DelegatingScope) usingScope).getParent()); + ns = snapshotNameSpace; + + sqlSnapshot.setOperand( + 1, + SqlLiteral.createTimestamp( + timestampString, + rexLiteral.getType().getPrecision(), + sqlSnapshot.getPeriod().getParserPosition())); + } + + super.registerNamespace(usingScope, alias, ns, forceNullable); + } + + /** + * For the time travel scenario, we need to build a SchemaVersion based on the Snapshot node for + * IdentifierNamespace in order to generate a new namespace. + * + * <p>In general, 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. + * @return snapshot node for generating a new namespace for time travel. Review Comment: ```suggestion * @return SqlSnapshot found in {@param ns}, empty if can't find. ``` ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java: ########## @@ -125,4 +162,118 @@ 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, so we need to investigate scenarios + // where the period of Snapshot is a SqlIdentifier. + Optional<SqlSnapshot> timeTravelNode = getTimeTravelNode(ns); + if (usingScope != null + && timeTravelNode.isPresent() + && !(timeTravelNode.get().getPeriod() instanceof SqlIdentifier)) { + SqlSnapshot sqlSnapshot = timeTravelNode.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; + IdentifierNamespace snapshotNameSpace = + new IdentifierSnapshotNamespace( + identifierNamespace, + schemaVersion, + ((DelegatingScope) usingScope).getParent()); + ns = snapshotNameSpace; + + sqlSnapshot.setOperand( + 1, + SqlLiteral.createTimestamp( + timestampString, + rexLiteral.getType().getPrecision(), + sqlSnapshot.getPeriod().getParserPosition())); + } + + super.registerNamespace(usingScope, alias, ns, forceNullable); + } + + /** + * For the time travel scenario, we need to build a SchemaVersion based on the Snapshot node for + * IdentifierNamespace in order to generate a new namespace. + * + * <p>In general, 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. + * @return snapshot node for generating a new namespace for time travel. + */ + private Optional<SqlSnapshot> getTimeTravelNode(SqlValidatorNamespace ns) { Review Comment: I think the naming `getTimeTravelNode` is to speicifed, also, I think when it's not for timetravel like lookup join, it will still return a `SqlSnapshot`, Right? I'd like just rename to `getSnapShotNode`. If you think so, aslo, don't forget to update the java docs, ########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java: ########## @@ -2927,7 +2948,37 @@ 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(); + // Since we have simplified the SqlSnapshot in the validate phase, we only need to check Review Comment: Maybe we need more clear while commenting here. ########## flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java: ########## @@ -2927,7 +2948,37 @@ 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(); + // Since we have simplified the SqlSnapshot in the validate phase, we only need to check + // whether the period is a RexLiteral and tableRef's first operand is a SqlIdentifier. + if ((tableRef instanceof SqlBasicCall Review Comment: Also, could you please also comment `(tableRef instanceof SqlBasicCall && ((SqlBasicCall) tableRef).operand(0) instanceof SqlIdentifier)` is for what case and `tableRef instanceof SqlTableRef` is for what case? -- 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]
