ahmedabu98 commented on code in PR #41304: URL: https://github.com/apache/beam/pull/41304#discussion_r4570011297
########## sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/MetadataColumnResolver.java: ########## @@ -0,0 +1,460 @@ +/* + * 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.beam.sdk.extensions.sql.impl; + +import static java.util.stream.Collectors.joining; +import static java.util.stream.Collectors.toList; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.beam.sdk.extensions.sql.impl.parser.TableClause; +import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable; +import org.apache.beam.sdk.extensions.sql.meta.ChangeRange; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.prepare.CalciteCatalogReader; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.prepare.Prepare; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.runtime.CalciteContextException; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.JoinConditionType; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlCall; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlIdentifier; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlJoin; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlKind; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlNode; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlNodeList; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlOrderBy; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlSelect; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.parser.SqlParserPos; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.validate.SqlNameMatcher; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.util.Util; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Throwables; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * Finds the metadata columns that a statement names on its {@code CHANGES(...)} reads, before + * Calcite validates it, so that queries don't have to list them in a {@code WITH} option. + * + * <p>A name binds to a read when it's one of the metadata columns that the read's table lists in + * {@link BeamSqlTable#getChangeMetadataColumns}, and isn't already a column of the read. {@code + * c._change_type} binds to the read aliased {@code c}. {@code _change_type} binds to the reads in + * its own SELECT that have it, or else to those of an enclosing SELECT. The resolver adds the names + * to the read's {@link TableClause}, which passes them to {@link BeamSqlTable#forChanges}. It also + * replaces {@code *} and {@code c.*} with the read's other columns, so {@code SELECT *} returns the + * same columns as before. + * + * <p>The resolver can't tell whether a name it can't bind is an alias, a column of another table or + * a typo, so the validator still reports unknown columns. {@link #explain} adds the metadata + * columns that such a name could have meant. + */ +final class MetadataColumnResolver { + private final CalciteCatalogReader catalogReader; + private final SqlNameMatcher nameMatcher; + private final List<ChangesRead> reads = new ArrayList<>(); + private final List<Scope> selects = new ArrayList<>(); + + /** What to add to an error about an unknown column at a position of the statement. */ + private final Map<SqlParserPos, String> hints = new HashMap<>(); + + /** Creates a resolver that looks tables up the way the validator does. */ + MetadataColumnResolver(CalciteCatalogReader catalogReader) { + this.catalogReader = catalogReader; + this.nameMatcher = catalogReader.nameMatcher(); + } + + /** Adds the metadata columns that a statement names to its reads, rewriting it in place. */ + void resolve(SqlNode statement) { + reads.clear(); + selects.clear(); + hints.clear(); + SqlKind kind = statement.getKind(); + if (kind.belongsTo(SqlKind.QUERY) || kind == SqlKind.INSERT || kind == SqlKind.EXPLAIN) { + visit(statement, null); + reads.forEach(ChangesRead::addNamedColumns); + selects.forEach(Scope::expandStars); + } + } + + /** + * Returns the message of a validation error about an unknown column, followed by the metadata + * columns that the column could have meant. Returns null for any other error. + */ + @Nullable String explain(Throwable error) { + for (Throwable cause : Throwables.getCausalChain(error)) { + if (cause instanceof CalciteContextException) { + CalciteContextException e = (CalciteContextException) cause; + @Nullable Throwable reason = e.getCause(); + @Nullable String hint = + hints.get( + new SqlParserPos( + e.getPosLine(), e.getPosColumn(), e.getEndPosLine(), e.getEndPosColumn())); + return hint == null + || reason == null + || !String.valueOf(reason.getMessage()).matches("Column '.*' not found in .*") + ? null + : e.getMessage() + ". " + hint; + } + } + return null; + } + + /** Visits a query or an expression, binding the names in it. */ + private void visit(@Nullable SqlNode node, @Nullable Scope scope) { + if (node instanceof SqlSelect) { + select((SqlSelect) node, scope); + } else if (node instanceof SqlOrderBy && ((SqlOrderBy) node).query instanceof SqlSelect) { + // The ORDER BY of a SELECT can name the columns of its FROM clause. + SqlOrderBy orderBy = (SqlOrderBy) node; + visit(orderBy.orderList, select((SqlSelect) orderBy.query, scope)); + } else if (node instanceof SqlIdentifier) { + bind((SqlIdentifier) node, scope); + } else if (node instanceof SqlCall) { + for (@Nullable SqlNode operand : ((SqlCall) node).getOperandList()) { + visit(operand, scope); + } + } else if (node instanceof SqlNodeList) { + for (SqlNode child : (SqlNodeList) node) { + visit(child, scope); + } + } + } + + private Scope select(SqlSelect select, @Nullable Scope parent) { + Scope scope = new Scope(select, parent); + List<SqlNode> joinConditions = new ArrayList<>(); + from(select.getFrom(), scope, joinConditions); + joinConditions.forEach(condition -> visit(condition, scope)); + for (@Nullable SqlNode operand : select.getOperandList()) { + if (operand != select.getFrom()) { + visit(operand, scope); + } + } + selects.add(scope); + return scope; + } + + /** Adds the items of a FROM clause to its scope, and collects the conditions of its joins. */ + private void from(@Nullable SqlNode node, Scope scope, List<SqlNode> joinConditions) { + if (node == null) { + return; + } + switch (node.getKind()) { + case JOIN: + SqlJoin join = (SqlJoin) node; Review Comment: Good catch. Naming a metadata column on a read in a FROM clause with a NATURAL JOIN now fails, and the error asks for USING or ON instead, so the join keys can't change silently. ########## sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/MetadataColumnResolver.java: ########## @@ -0,0 +1,460 @@ +/* + * 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.beam.sdk.extensions.sql.impl; + +import static java.util.stream.Collectors.joining; +import static java.util.stream.Collectors.toList; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.beam.sdk.extensions.sql.impl.parser.TableClause; +import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable; +import org.apache.beam.sdk.extensions.sql.meta.ChangeRange; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.prepare.CalciteCatalogReader; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.prepare.Prepare; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.runtime.CalciteContextException; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.JoinConditionType; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlCall; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlIdentifier; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlJoin; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlKind; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlNode; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlNodeList; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlOrderBy; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlSelect; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.parser.SqlParserPos; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.validate.SqlNameMatcher; +import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.util.Util; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Throwables; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * Finds the metadata columns that a statement names on its {@code CHANGES(...)} reads, before + * Calcite validates it, so that queries don't have to list them in a {@code WITH} option. + * + * <p>A name binds to a read when it's one of the metadata columns that the read's table lists in + * {@link BeamSqlTable#getChangeMetadataColumns}, and isn't already a column of the read. {@code + * c._change_type} binds to the read aliased {@code c}. {@code _change_type} binds to the reads in + * its own SELECT that have it, or else to those of an enclosing SELECT. The resolver adds the names + * to the read's {@link TableClause}, which passes them to {@link BeamSqlTable#forChanges}. It also + * replaces {@code *} and {@code c.*} with the read's other columns, so {@code SELECT *} returns the + * same columns as before. + * + * <p>The resolver can't tell whether a name it can't bind is an alias, a column of another table or + * a typo, so the validator still reports unknown columns. {@link #explain} adds the metadata + * columns that such a name could have meant. + */ +final class MetadataColumnResolver { + private final CalciteCatalogReader catalogReader; + private final SqlNameMatcher nameMatcher; + private final List<ChangesRead> reads = new ArrayList<>(); + private final List<Scope> selects = new ArrayList<>(); + + /** What to add to an error about an unknown column at a position of the statement. */ + private final Map<SqlParserPos, String> hints = new HashMap<>(); + + /** Creates a resolver that looks tables up the way the validator does. */ + MetadataColumnResolver(CalciteCatalogReader catalogReader) { + this.catalogReader = catalogReader; + this.nameMatcher = catalogReader.nameMatcher(); + } + + /** Adds the metadata columns that a statement names to its reads, rewriting it in place. */ + void resolve(SqlNode statement) { + reads.clear(); + selects.clear(); + hints.clear(); + SqlKind kind = statement.getKind(); + if (kind.belongsTo(SqlKind.QUERY) || kind == SqlKind.INSERT || kind == SqlKind.EXPLAIN) { + visit(statement, null); + reads.forEach(ChangesRead::addNamedColumns); + selects.forEach(Scope::expandStars); + } + } + + /** + * Returns the message of a validation error about an unknown column, followed by the metadata + * columns that the column could have meant. Returns null for any other error. + */ + @Nullable String explain(Throwable error) { + for (Throwable cause : Throwables.getCausalChain(error)) { + if (cause instanceof CalciteContextException) { + CalciteContextException e = (CalciteContextException) cause; + @Nullable Throwable reason = e.getCause(); + @Nullable String hint = + hints.get( + new SqlParserPos( + e.getPosLine(), e.getPosColumn(), e.getEndPosLine(), e.getEndPosColumn())); + return hint == null + || reason == null + || !String.valueOf(reason.getMessage()).matches("Column '.*' not found in .*") + ? null + : e.getMessage() + ". " + hint; + } + } + return null; + } + + /** Visits a query or an expression, binding the names in it. */ + private void visit(@Nullable SqlNode node, @Nullable Scope scope) { + if (node instanceof SqlSelect) { + select((SqlSelect) node, scope); + } else if (node instanceof SqlOrderBy && ((SqlOrderBy) node).query instanceof SqlSelect) { + // The ORDER BY of a SELECT can name the columns of its FROM clause. + SqlOrderBy orderBy = (SqlOrderBy) node; + visit(orderBy.orderList, select((SqlSelect) orderBy.query, scope)); + } else if (node instanceof SqlIdentifier) { + bind((SqlIdentifier) node, scope); + } else if (node instanceof SqlCall) { + for (@Nullable SqlNode operand : ((SqlCall) node).getOperandList()) { + visit(operand, scope); + } + } else if (node instanceof SqlNodeList) { + for (SqlNode child : (SqlNodeList) node) { + visit(child, scope); + } + } + } + + private Scope select(SqlSelect select, @Nullable Scope parent) { + Scope scope = new Scope(select, parent); + List<SqlNode> joinConditions = new ArrayList<>(); + from(select.getFrom(), scope, joinConditions); + joinConditions.forEach(condition -> visit(condition, scope)); + for (@Nullable SqlNode operand : select.getOperandList()) { + if (operand != select.getFrom()) { + visit(operand, scope); + } + } + selects.add(scope); + return scope; + } + + /** Adds the items of a FROM clause to its scope, and collects the conditions of its joins. */ + private void from(@Nullable SqlNode node, Scope scope, List<SqlNode> joinConditions) { + if (node == null) { + return; + } + switch (node.getKind()) { + case JOIN: + SqlJoin join = (SqlJoin) node; Review Comment: Made it fail when reading FROM clause with a NATURAL JOIN -- 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]
