fix abstract tableInfo not Serializable

Author: zoudaokoulife

core/src/main/java/com/dtstack/flink/sql/table/TableInfo.java

@@ -22,6 +22,7 @@
 
 import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
 
+import java.io.Serializable;
 import java.util.List;
 
 /**
@@ -31,7 +32,7 @@
  * @author xuchao
  */
 
-public abstract class TableInfo {
+public abstract class TableInfo implements Serializable {
 
     public static final String PARALLELISM_KEY = "parallelism";
 
@@ -121,6 +122,29 @@ public void addFieldType(String fieldType){
         fieldTypeList.add(fieldType);
     }
 
+    public void setFields(String[] fields) {
+        this.fields = fields;
+    }
+
+    public void setFieldTypes(String[] fieldTypes) {
+        this.fieldTypes = fieldTypes;
+    }
+
+    public void setFieldClasses(Class<?>[] fieldClasses) {
+        this.fieldClasses = fieldClasses;
+    }
+
+    public List<String> getFieldList() {
+        return fieldList;
+    }
+
+    public List<String> getFieldTypeList() {
+        return fieldTypeList;
+    }
+
+    public List<Class> getFieldClassList() {
+        return fieldClassList;
+    }
 
     public void finish(){
         this.fields = fieldList.toArray(new String[fieldList.size()]);

launcher/src/main/java/com/dtstack/flink/sql/launcher/LauncherMain.java

@@ -59,13 +59,15 @@ private static String getLocalCoreJarPath(String localSqlRootJar){
     }
 
     public static void main(String[] args) throws Exception {
-        if (args.length==1 && args[0].endsWith(".json")){
+        if (args.length == 1 && args[0].endsWith(".json")){
             args = parseJson(args);
         }
+
         LauncherOptionParser optionParser = new LauncherOptionParser(args);
         LauncherOptions launcherOptions = optionParser.getLauncherOptions();
         String mode = launcherOptions.getMode();
         List<String> argList = optionParser.getProgramExeArgList();
+
         if(mode.equals(ClusterMode.local.name())) {
             String[] localArgs = argList.toArray(new String[argList.size()]);
             Main.main(localArgs);

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncSideInfo.java

@@ -42,6 +42,8 @@
 
 public class RdbAsyncSideInfo extends SideInfo {
 
+    private static final long serialVersionUID = 1942629132469918611L;
+
     public RdbAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/table/RdbSideTableInfo.java

@@ -28,6 +28,7 @@
  * @author maqi
  */
 public class RdbSideTableInfo extends SideTableInfo {
+
     private static final long serialVersionUID = -1L;
 
     public static final String URL_KEY = "url";