refactor: improve extract yaml from execute pipeline command (DataLin…

…kDC#2751)

* refactor: improve extrace yaml from execute pipeline command

Signed-off-by: licho <lecho.sun@gmail.com>

* Spotless Apply

---------

Signed-off-by: licho <lecho.sun@gmail.com>
Co-authored-by: leechor <leechor@users.noreply.github.com>

dinky-core/src/main/java/org/dinky/trans/pipeline/FlinkCDCPipelineOperation.java

@@ -30,6 +30,10 @@
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.jetbrains.annotations.Nullable;
 
 import com.ververica.cdc.composer.PipelineComposer;
 import com.ververica.cdc.composer.definition.PipelineDef;
@@ -84,7 +88,7 @@ public Operation create(String statement) {
 
     @Override
     public TableResult execute(Executor executor) {
-        String yamlText = statement.substring(statement.indexOf("(") + 1, statement.lastIndexOf(")"));
+        String yamlText = getPipelineConfigure(statement);
         com.ververica.cdc.common.configuration.Configuration globalPipelineConfig =
                 com.ververica.cdc.common.configuration.Configuration.fromMap(executor.getSetConfig());
         // Parse pipeline definition file
@@ -111,6 +115,16 @@ public TableResult execute(Executor executor) {
         return null;
     }
 
+    @Nullable
+    public String getPipelineConfigure(String statement) {
+        Pattern patternYaml = Pattern.compile("(?is)^EXECUTE\\s+PIPELINE\\s+WITHYAML\\s+\\((.+)\\)");
+        Matcher matcherYaml = patternYaml.matcher(statement);
+        if (matcherYaml.find()) {
+            return matcherYaml.group(1);
+        }
+        return "";
+    }
+
     public DinkyFlinkPipelineComposer createComposer(
             boolean useMiniCluster, Configuration flinkConfig, List<Path> additionalJars) {
         if (useMiniCluster) {

dinky-core/src/test/java/org/dinky/trans/pipeline/FlinkCDCPipelineOperationTest.java

@@ -0,0 +1,35 @@
+/*
+ *
+ *  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.dinky.trans.pipeline;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import org.junit.jupiter.api.Test;
+
+class FlinkCDCPipelineOperationTest {
+
+    @Test
+    void getPipelineConfigure() {
+        FlinkCDCPipelineOperation operation = new FlinkCDCPipelineOperation();
+        String configure = "EXECUTE PIPELINE withYaml (a.b=1\n a.b.c=2)";
+        String result = operation.getPipelineConfigure(configure);
+        assertEquals("a.b=1\n a.b.c=2", result);
+    }
+}