[FLINK-35633][cli] Verify YAML pipeline definitions early

flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliExecutor.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.cdc.cli;
 
+import org.apache.flink.cdc.cli.guard.PipelineDefGuard;
 import org.apache.flink.cdc.cli.parser.PipelineDefinitionParser;
 import org.apache.flink.cdc.cli.parser.YamlPipelineDefinitionParser;
 import org.apache.flink.cdc.cli.utils.FlinkEnvironmentUtils;
@@ -64,6 +65,9 @@ public PipelineExecution.ExecutionInfo run() throws Exception {
         PipelineDef pipelineDef =
                 pipelineDefinitionParser.parse(pipelineDefPath, globalPipelineConfig);
 
+        // Verify pipeline definitions
+        PipelineDefGuard.verify(pipelineDef);
+
         // Create composer
         PipelineComposer composer = getComposer();
 

flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/guard/CommonDefGuard.java

@@ -0,0 +1,78 @@
+/*
+ * 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.cdc.cli.guard;
+
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.exceptions.GuardVerificationException;
+
+import org.apache.calcite.config.Lex;
+import org.apache.calcite.sql.parser.SqlParser;
+import org.apache.calcite.sql.validate.SqlConformanceEnum;
+
+/** Verification guard for common fields. */
+public class CommonDefGuard {
+    public static void verifyTableQualifier(String rawTableQualifier)
+            throws GuardVerificationException {
+        String tableQualifier = rawTableQualifier.replace("\\.", "A");
+        if (tableQualifier.isEmpty()) {
+            throw new GuardVerificationException(tableQualifier, "Empty table qualifier.");
+        }
+        if (tableQualifier.charAt(0) == '.') {
+            throw new GuardVerificationException(
+                    rawTableQualifier,
+                    "Dot (.) was used as delimiter between schema and table name, which should not present at the beginning. Other usages in RegExp should be escaped with backslash (\\).");
+        }
+        if (tableQualifier.charAt(tableQualifier.length() - 1) == '.') {
+            throw new GuardVerificationException(
+                    rawTableQualifier,
+                    "Dot (.) was used as delimiter between schema and table name, which should not present at the end. Other usages in RegExp should be escaped with backslash (\\).");
+        }
+
+        int delimiterDotCount = 0;
+        for (int i = 0; i < tableQualifier.length(); i++) {
+            // Accessing charAt(i - 1) is safe here
+            // since tableQualifier[0] can't be '.'
+            if (tableQualifier.charAt(i) == '.') {
+                delimiterDotCount++;
+            }
+        }
+
+        // delimiter dot must present exactly once
+        if (delimiterDotCount > 2) {
+            throw new GuardVerificationException(
+                    rawTableQualifier,
+                    "Dot (.) was used as delimiter between schema and table name. Other usages in RegExp should be escaped with backslash (\\).");
+        }
+
+        try {
+            TableId.parse(tableQualifier);
+        } catch (IllegalArgumentException e) {
+            throw new GuardVerificationException(
+                    rawTableQualifier, "Illegal table qualifier " + tableQualifier);
+        }
+    }
+
+    public static SqlParser getCalciteParser(String sql) {
+        return SqlParser.create(
+                sql,
+                SqlParser.Config.DEFAULT
+                        .withConformance(SqlConformanceEnum.MYSQL_5)
+                        .withCaseSensitive(true)
+                        .withLex(Lex.JAVA));
+    }
+}

flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/guard/PipelineDefGuard.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.cdc.cli.guard;
+
+import org.apache.flink.cdc.common.exceptions.GuardVerificationException;
+import org.apache.flink.cdc.composer.definition.PipelineDef;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Syntax guard for overall pipeline definition. */
+public class PipelineDefGuard {
+
+    private static final Logger LOG = LoggerFactory.getLogger(PipelineDefGuard.class);
+
+    public static void verify(PipelineDef pipelineDef) throws GuardVerificationException {
+        LOG.info("Verifying pipeline definition {}", pipelineDef);
+
+        pipelineDef.getTransforms().forEach(TransformDefGuard::verify);
+        pipelineDef.getRoute().forEach(RouteDefGuard::verify);
+    }
+}

flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/guard/RouteDefGuard.java

@@ -0,0 +1,52 @@
+/*
+ * 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.cdc.cli.guard;
+
+import org.apache.flink.cdc.common.exceptions.GuardVerificationException;
+import org.apache.flink.cdc.composer.definition.RouteDef;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+
+import static org.apache.flink.cdc.cli.guard.CommonDefGuard.verifyTableQualifier;
+
+/** Syntax guard for route definition. */
+public class RouteDefGuard {
+    private static final Logger LOG = LoggerFactory.getLogger(RouteDefGuard.class);
+
+    public static void verify(RouteDef routeDef) throws GuardVerificationException {
+        LOG.info("Verifying route definition {}", routeDef);
+
+        Optional<String> replaceSymbol = routeDef.getReplaceSymbol();
+
+        verifyTableQualifier(routeDef.getSourceTable());
+        if (replaceSymbol.isPresent()) {
+            if (!routeDef.getSinkTable().contains(replaceSymbol.get())) {
+                throw new GuardVerificationException(
+                        routeDef,
+                        "Replace symbol is specified but not present in sink definition.");
+            }
+            verifyTableQualifier(
+                    routeDef.getSinkTable().replace(replaceSymbol.get(), "SomeValidTableName"));
+        } else {
+            verifyTableQualifier(routeDef.getSinkTable());
+        }
+    }
+}

flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/guard/TransformDefGuard.java

@@ -0,0 +1,127 @@
+/*
+ * 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.cdc.cli.guard;
+
+import org.apache.flink.cdc.common.exceptions.GuardVerificationException;
+import org.apache.flink.cdc.composer.definition.TransformDef;
+
+import org.apache.flink.shaded.guava31.com.google.common.base.Strings;
+
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.parser.SqlParseException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+
+import static org.apache.flink.cdc.cli.guard.CommonDefGuard.getCalciteParser;
+import static org.apache.flink.cdc.cli.guard.CommonDefGuard.verifyTableQualifier;
+
+/** Syntax guard for transform definition. */
+public class TransformDefGuard {
+
+    private static final Logger LOG = LoggerFactory.getLogger(TransformDefGuard.class);
+
+    public static void verify(TransformDef transformDef) throws GuardVerificationException {
+        LOG.info("Verifying transform definition {}", transformDef);
+
+        verifyTableQualifier(transformDef.getSourceTable());
+
+        verifyKeys(transformDef.getPrimaryKeys());
+        verifyKeys(transformDef.getPartitionKeys());
+
+        transformDef.getProjection().ifPresent(TransformDefGuard::verifyProjectionRule);
+        transformDef.getFilter().ifPresent(TransformDefGuard::verifyFilterRule);
+    }
+
+    private static void verifyKeys(String keys) throws GuardVerificationException {
+        if (Strings.isNullOrEmpty(keys)) {
+            return;
+        }
+        Arrays.stream(keys.split(","))
+                .map(String::trim)
+                .forEach(TransformDefGuard::verifySqlIdentifier);
+    }
+
+    private static void verifyProjectionRule(String projectionRule)
+            throws GuardVerificationException {
+        try {
+            SqlNode selectNode =
+                    getCalciteParser("select " + projectionRule + " from tb").parseQuery();
+            if (selectNode instanceof SqlSelect) {
+                for (SqlNode node : ((SqlSelect) selectNode).getSelectList().getList()) {
+                    if (node instanceof SqlBasicCall) {
+                        SqlBasicCall call = (SqlBasicCall) node;
+                        if (!(SqlKind.AS.equals(call.getOperator().getKind())
+                                && call.getOperandList().size() == 2
+                                && call.getOperandList().get(1) instanceof SqlIdentifier)) {
+                            throw new GuardVerificationException(
+                                    projectionRule,
+                                    String.format(
+                                            "%s is neither a column identifier nor an aliased expression. Expected: <Expression> AS <Identifier>",
+                                            node));
+                        }
+                    } else if (!(node instanceof SqlIdentifier)) {
+                        throw new GuardVerificationException(
+                                projectionRule,
+                                String.format(
+                                        "%s is neither a column identifier nor an aliased expression. Expected: <Expression> AS <Identifier>",
+                                        node));
+                    }
+                }
+            }
+        } catch (SqlParseException e) {
+            throw new GuardVerificationException(
+                    projectionRule, projectionRule + " is not a valid projection rule");
+        }
+    }
+
+    private static void verifyFilterRule(String filterRule) throws GuardVerificationException {
+        try {
+            SqlNode selectNode =
+                    getCalciteParser("select * from tb where " + filterRule).parseQuery();
+        } catch (SqlParseException e) {
+            throw new GuardVerificationException(
+                    filterRule, filterRule + " is not a valid filter rule");
+        }
+    }
+
+    private static void verifySqlIdentifier(String identifier) throws GuardVerificationException {
+        try {
+            SqlNode selectNode = getCalciteParser("select " + identifier + " from tb").parseQuery();
+            if (selectNode instanceof SqlSelect) {
+                if (!((SqlSelect) selectNode)
+                        .getSelectList().getList().stream()
+                                .allMatch(node -> node instanceof SqlIdentifier)) {
+                    throw new GuardVerificationException(
+                            identifier, identifier + " is not a valid SQL identifier");
+                }
+            } else {
+                throw new GuardVerificationException(
+                        identifier, identifier + " is not a valid SQL identifier");
+            }
+        } catch (SqlParseException e) {
+            throw new GuardVerificationException(
+                    identifier, identifier + " is not a valid SQL identifier");
+        }
+    }
+}

flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java

@@ -18,6 +18,7 @@
 package org.apache.flink.cdc.cli.parser;
 
 import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.exceptions.GuardVerificationException;
 import org.apache.flink.cdc.common.utils.StringUtils;
 import org.apache.flink.cdc.composer.definition.PipelineDef;
 import org.apache.flink.cdc.composer.definition.RouteDef;
@@ -95,6 +96,15 @@ public PipelineDef parse(Path pipelineDefPath, Configuration globalPipelineConfi
                                 SINK_KEY));
 
         // Transforms are optional
+        Optional.ofNullable(root.get(TRANSFORM_KEY))
+                .ifPresent(
+                        node -> {
+                            if (!node.isArray()) {
+                                throw new GuardVerificationException(
+                                        node,
+                                        "Transform rules should be an array. Maybe missed a hyphen in YAML?");
+                            }
+                        });
         List<TransformDef> transformDefs = new ArrayList<>();
         Optional.ofNullable(root.get(TRANSFORM_KEY))
                 .ifPresent(
@@ -103,7 +113,18 @@ public PipelineDef parse(Path pipelineDefPath, Configuration globalPipelineConfi
                                         transform -> transformDefs.add(toTransformDef(transform))));
 
         // Routes are optional
+        Optional.ofNullable(root.get(ROUTE_KEY))
+                .ifPresent(
+                        node -> {
+                            if (!node.isArray()) {
+                                throw new GuardVerificationException(
+                                        node,
+                                        "Route rules should be an array. Maybe missed a hyphen in YAML?");
+                            }
+                        });
+
         List<RouteDef> routeDefs = new ArrayList<>();
+
         Optional.ofNullable(root.get(ROUTE_KEY))
                 .ifPresent(node -> node.forEach(route -> routeDefs.add(toRouteDef(route))));