[Feature][Engine] Add flink engine

datavines-common/src/main/java/io/datavines/common/ConfigConstants.java

@@ -107,7 +107,7 @@ public class ConfigConstants {
     public static final String OUTPUT_TABLE = "output_table";
     public static final String TMP_TABLE = "tmp_table";
     public static final String COLUMN_SEPARATOR = "column_separator";
-    public static final String LINE_SEPERATOR = "line_separator";
+    public static final String LINE_SEPARATOR = "line_separator";
     public static final String DATA_DIR = "data_dir";
 
     public static final String ENABLE_SPARK_HIVE_SUPPORT = "enable_spark_hive_support";
@@ -169,4 +169,8 @@ public class ConfigConstants {
     public static final String SAVE_MODE = "save_mode";
 
     public static final String UPSERT = "Upsert";
+
+    public static final String FILE_NAME = "file_name";
+
+    public static final String FLINK = "flink";
 }

datavines-common/src/main/java/io/datavines/common/config/Config.java

@@ -43,6 +43,11 @@ public String getString(String key){
         return String.valueOf(config.get(key));
     }
 
+    public String getString(String key, String defaultValue) {
+        Object value = config.get(key);
+        return value != null ? String.valueOf(value) : defaultValue;
+    }
+
     public List<String> getStringList(String key){
         return (List<String>)config.get(key);
     }
@@ -51,6 +56,11 @@ public Integer getInt(String key){
         return Integer.valueOf(String.valueOf(config.get(key)));
     }
 
+    public Integer getInt(String key, Integer defaultValue) {
+        Object value = config.get(key);
+        return value != null ? Integer.valueOf(String.valueOf(value)) : defaultValue;
+    }
+
     public Boolean getBoolean(String key){
         return Boolean.valueOf(String.valueOf(config.get(key)));
     }
@@ -63,6 +73,11 @@ public Long getLong(String key){
         return Long.valueOf(String.valueOf(config.get(key)));
     }
 
+    public Long getLong(String key, Long defaultValue) {
+        Object value = config.get(key);
+        return value != null ? Long.valueOf(String.valueOf(value)) : defaultValue;
+    }
+
     public Boolean has(String key) {
         return config.get(key) != null;
     }

datavines-common/src/main/java/io/datavines/common/config/Configurations.java

@@ -57,4 +57,12 @@ public Float getFloat(String key) {
     public Float getFloat(String key,String defaultValue) {
         return Float.valueOf(configuration.getProperty(key,defaultValue));
     }
+
+    public long getLong(String key) {
+        return Long.parseLong(configuration.getProperty(key));
+    }
+
+    public long getLong(String key, long defaultValue) {
+        return Long.parseLong(configuration.getProperty(key, String.valueOf(defaultValue)));
+    }
 }

datavines-common/src/main/java/io/datavines/common/utils/ProcessUtils.java

@@ -18,14 +18,18 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.io.InputStreamReader;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Consumer;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.BufferedReader;
+
 /**
  *  mainly used to get the start command line of a process
  */
@@ -255,4 +259,43 @@ private static boolean needsEscaping(int verificationType, String arg) {
     }
     return false;
   }
+
+  /**
+   * Print process output to logger
+   * @param process process
+   * @param logHandler log handler consumer
+   * @throws IOException io exception
+   */
+  public static void printProcessOutput(Process process, Consumer<List<String>> logHandler) throws IOException {
+    if (process == null || logHandler == null) {
+      return;
+    }
+
+    try (BufferedReader inReader = new BufferedReader(new InputStreamReader(process.getInputStream()));
+         BufferedReader errReader = new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
+
+      List<String> lines = new ArrayList<>();
+      String line;
+
+      while ((line = inReader.readLine()) != null) {
+        lines.add(line);
+        if (lines.size() >= 100) { // Buffer size to avoid memory issues
+          logHandler.accept(new ArrayList<>(lines));
+          lines.clear();
+        }
+      }
+
+      while ((line = errReader.readLine()) != null) {
+        lines.add(line);
+        if (lines.size() >= 100) {
+          logHandler.accept(new ArrayList<>(lines));
+          lines.clear();
+        }
+      }
+
+      if (!lines.isEmpty()) {
+        logHandler.accept(lines);
+      }
+    }
+  }
 }

datavines-engine/datavines-engine-config/src/main/java/io/datavines/engine/config/BaseJobConfigurationBuilder.java

@@ -70,13 +70,13 @@ public void init(Map<String, String> inputParameter, JobExecutionInfo jobExecuti
             this.inputParameter.put(COLUMN_SEPARATOR,
                     errorDataParameterMap.get(CommonPropertyUtils.COLUMN_SEPARATOR) == null ?
                             CommonPropertyUtils.COLUMN_SEPARATOR_DEFAULT : errorDataParameterMap.get(CommonPropertyUtils.COLUMN_SEPARATOR));
-            this.inputParameter.put(LINE_SEPERATOR,
+            this.inputParameter.put(LINE_SEPARATOR,
                     errorDataParameterMap.get(CommonPropertyUtils.LINE_SEPARATOR) == null ?
                             CommonPropertyUtils.LINE_SEPARATOR_DEFAULT : errorDataParameterMap.get(CommonPropertyUtils.LINE_SEPARATOR));
         } else {
             this.inputParameter.put(ERROR_DATA_DIR, CommonPropertyUtils.getString(CommonPropertyUtils.ERROR_DATA_DIR, CommonPropertyUtils.ERROR_DATA_DIR_DEFAULT));
             this.inputParameter.put(COLUMN_SEPARATOR, CommonPropertyUtils.getString(CommonPropertyUtils.COLUMN_SEPARATOR, CommonPropertyUtils.COLUMN_SEPARATOR_DEFAULT));
-            this.inputParameter.put(LINE_SEPERATOR, CommonPropertyUtils.getString(CommonPropertyUtils.LINE_SEPARATOR, CommonPropertyUtils.LINE_SEPARATOR_DEFAULT));
+            this.inputParameter.put(LINE_SEPARATOR, CommonPropertyUtils.getString(CommonPropertyUtils.LINE_SEPARATOR, CommonPropertyUtils.LINE_SEPARATOR_DEFAULT));
         }
 
         if (FILE.equalsIgnoreCase(jobExecutionInfo.getValidateResultDataStorageType())) {

datavines-engine/datavines-engine-executor/src/main/java/io/datavines/engine/executor/core/executor/BaseCommandProcess.java

@@ -43,7 +43,7 @@ public abstract class BaseCommandProcess {
     /**
      *  process
      */
-    private Process process;
+    protected Process process;
 
     /**
      *  log handler
@@ -60,7 +60,7 @@ public abstract class BaseCommandProcess {
      */
     protected Logger logger;
 
-    private final Configurations configurations;
+    protected final Configurations configurations;
 
     /**
      *  log list
@@ -163,7 +163,7 @@ private void buildProcess(String commandFile) throws IOException{
      * @param process process
      * @return processId
      */
-    private int getProcessId(Process process){
+    protected int getProcessId(Process process){
         int processId = 0;
 
         try {
@@ -266,7 +266,7 @@ private void hardKill(int processId) {
      * get the standard output of the process
      * @param process process
      */
-    private void parseProcessOutput(Process process) {
+    protected void parseProcessOutput(Process process) {
         String threadLoggerInfoName = String.format(LoggerUtils.JOB_LOGGER_THREAD_NAME + "-%s", jobExecutionRequest.getJobExecutionName());
         ExecutorService parseProcessOutputExecutorService = ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName);
         parseProcessOutputExecutorService.submit(new Runnable(){
@@ -357,4 +357,8 @@ protected List<String> commandOptions() {
      */
     protected abstract void createCommandFileIfNotExists(String execCommand, String commandFile) throws IOException;
 
+    protected Consumer<List<String>> getLogHandler() {
+        return logHandler;
+    }
+
 }

datavines-engine/datavines-engine-plugins/datavines-engine-flink/HEADER

@@ -0,0 +1,14 @@
+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.

datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-api/pom.xml

@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>datavines-engine-flink</artifactId>
+        <groupId>io.datavines</groupId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>datavines-engine-flink-api</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.datavines</groupId>
+            <artifactId>datavines-engine-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.datavines</groupId>
+            <artifactId>datavines-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+</project>

datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-api/src/main/java/io/datavines/engine/flink/api/FlinkRuntimeEnvironment.java

@@ -0,0 +1,89 @@
+/*
+ * 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 io.datavines.engine.flink.api;
+
+import io.datavines.common.config.CheckResult;
+import io.datavines.common.config.Config;
+import io.datavines.common.exception.DataVinesException;
+import io.datavines.engine.api.env.Execution;
+import io.datavines.engine.api.env.RuntimeEnvironment;
+import io.datavines.engine.flink.api.stream.FlinkStreamExecution;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+
+public class FlinkRuntimeEnvironment implements RuntimeEnvironment {
+
+    private StreamExecutionEnvironment env;
+    private StreamTableEnvironment tableEnv;
+    private Config config;
+    private FlinkStreamExecution execution;
+
+    public FlinkRuntimeEnvironment() {
+        this.config = new Config();
+        this.execution = new FlinkStreamExecution(this);
+    }
+
+    public void setConfig(Config config) {
+        if (config != null) {
+            this.config = config;
+        }
+    }
+
+    public Config getConfig() {
+        return config;
+    }
+
+    public CheckResult checkConfig() {
+        return new CheckResult(true, "Configuration check passed");
+    }
+
+    public Execution getExecution() {
+        return execution;
+    }
+
+    public void prepare() {
+        try {
+            env = StreamExecutionEnvironment.getExecutionEnvironment();
+            tableEnv = StreamTableEnvironment.create(env);
+        } catch (Exception e) {
+            throw new DataVinesException("Failed to prepare Flink environment", e);
+        }
+    }
+
+    public void stop() {
+        try {
+            if (env != null) {
+                // Flink's environment doesn't have a direct cancel method, 
+                // we need to handle job cancellation through JobClient
+            }
+        } catch (Exception e) {
+            throw new DataVinesException("Failed to stop Flink environment", e);
+        }
+    }
+
+    public String getType() {
+        return "flink";
+    }
+
+    public StreamExecutionEnvironment getEnv() {
+        return env;
+    }
+
+    public StreamTableEnvironment getTableEnv() {
+        return tableEnv;
+    }
+}