Repository: apache/flink
File: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java (Java)

Code before:
        return new PlannerQueryOperation(
                relational.project(), () -> context.toQuotedSqlString(validated));
    }

    /** convert the query part of a VIEW statement into a {@link CatalogView}. */
    static CatalogView toCatalogView(
            SqlNode query,
            List<SqlNode> viewFields,
            Map<String, String> viewOptions,
            String viewComment,
            ConvertContext context) {
        // Put the sql string unparse (getQuotedSqlString()) in front of
        // the node conversion (toQueryOperation()),
        // because before Calcite 1.22.0, during sql-to-rel conversion, the SqlWindow
        // bounds state would be mutated as default when they are null (not specified).

        // This bug is fixed in CALCITE-3877 of Calcite 1.23.0.
        String originalQuery = context.toQuotedSqlString(query);
        SqlNode validateQuery = context.getSqlValidator().validate(query);

        // Check name is unique.
        // Don't rely on the calcite because if the field names are duplicate, calcite will add
        // index to identify the duplicate names.
        SqlValidatorNamespace validatedNamespace =
                context.getSqlValidator().getNamespace(validateQuery);
        validateDuplicatedColumnNames(query, viewFields, validatedNamespace);

        String expandedQuery = context.toQuotedSqlString(validateQuery);

        PlannerQueryOperation operation = toQueryOperation(validateQuery, context);
        ResolvedSchema schema = operation.getResolvedSchema();

        // the view column list in CREATE VIEW is optional, if it's not empty, we should update
        // the column name with the names in view column list.
        if (!viewFields.isEmpty()) {
            // alias column names:
            List<String> inputFieldNames = schema.getColumnNames();
            List<String> aliasFieldNames =
                    viewFields.stream().map(SqlNode::toString).collect(Collectors.toList());

            if (inputFieldNames.size() != aliasFieldNames.size()) {
                throw new ValidationException(
                        String.format(
                                "VIEW definition and input fields not match:\n\tDef fields: %s.\n\tInput fields: %s.",
                                aliasFieldNames, inputFieldNames));
            }

            schema = ResolvedSchema.physical(aliasFieldNames, schema.getColumnDataTypes());
        }

        return new ResolvedCatalogView(

Review comment (suggestion):
looks like in this line we also need to use `validatedQuery` given the fact that `query` might be invalid

Code after:
        return new PlannerQueryOperation(
                relational.project(), () -> context.toQuotedSqlString(validated));
    }

    /** convert the query part of a VIEW statement into a {@link CatalogView}. */
    static CatalogView toCatalogView(
            SqlNode query,
            List<SqlNode> viewFields,
            Map<String, String> viewOptions,
            String viewComment,
            ConvertContext context) {
        // Put the sql string unparse (getQuotedSqlString()) in front of
        // the node conversion (toQueryOperation()),
        // because before Calcite 1.22.0, during sql-to-rel conversion, the SqlWindow
        // bounds state would be mutated as default when they are null (not specified).

        // This bug is fixed in CALCITE-3877 of Calcite 1.23.0.
        String originalQuery = context.toQuotedSqlString(query);
        SqlNode validateQuery = context.getSqlValidator().validate(query);
        // FLINK-38950: SqlValidator.validate() mutates its input parameter. Always use the
        // returned validateQuery instead of the mutated query for all subsequent operations.

        // Check name is unique.
        // Don't rely on the calcite because if the field names are duplicate, calcite will add
        // index to identify the duplicate names.
        SqlValidatorNamespace validatedNamespace =
                context.getSqlValidator().getNamespace(validateQuery);
        validateDuplicatedColumnNames(validateQuery, viewFields, validatedNamespace);

        String expandedQuery = context.toQuotedSqlString(validateQuery);

        PlannerQueryOperation operation = toQueryOperation(validateQuery, context);
        ResolvedSchema schema = operation.getResolvedSchema();

        // the view column list in CREATE VIEW is optional, if it's not empty, we should update
        // the column name with the names in view column list.
        if (!viewFields.isEmpty()) {
            // alias column names:
            List<String> inputFieldNames = schema.getColumnNames();
            List<String> aliasFieldNames =
                    viewFields.stream().map(SqlNode::toString).collect(Collectors.toList());

            if (inputFieldNames.size() != aliasFieldNames.size()) {
                throw new ValidationException(
                        String.format(
                                "VIEW definition and input fields not match:\n\tDef fields: %s.\n\tInput fields: %s.",
                                aliasFieldNames, inputFieldNames));
            }

            schema = ResolvedSchema.physical(aliasFieldNames, schema.getColumnDataTypes());
        }