add clone ability for DataX pipeline and DataSource instance

tis-datax/tis-datax-common-plugin/src/main/java/com/qlangtech/tis/plugin/datax/DataFlowDataXProcessor.java

@@ -288,7 +288,7 @@ public boolean validateName(IFieldErrorHandler msgHandler, Context context, Stri
             if (pluginMeta.isUpdate()) {
                 return true;
             }
-            return msgHandler.validateBizLogic(IFieldErrorHandler.BizLogic.WORKFLOW_NAME_DUPLICATE, context, fieldName, value);
+            return msgHandler.validateBizLogic(IFieldErrorHandler.BizLogic.VALIDATE_WORKFLOW_NAME_DUPLICATE, context, fieldName, value);
         }
 
         @Override

tis-datax/tis-datax-common-plugin/src/main/java/com/qlangtech/tis/plugin/datax/DefaultDataxProcessor.java

@@ -20,9 +20,11 @@
 
 import com.alibaba.citrus.turbine.Context;
 import com.qlangtech.tis.config.ParamsConfig;
+import com.qlangtech.tis.datax.DefaultDataXProcessorManipulate;
 import com.qlangtech.tis.datax.IDataxGlobalCfg;
 import com.qlangtech.tis.datax.impl.DataxProcessor;
 import com.qlangtech.tis.extension.Descriptor;
+import com.qlangtech.tis.extension.IDescribableManipulate;
 import com.qlangtech.tis.extension.TISExtension;
 import com.qlangtech.tis.manage.IAppSource;
 import com.qlangtech.tis.manage.biz.dal.pojo.AppType;
@@ -31,11 +33,17 @@
 import com.qlangtech.tis.plugin.annotation.FormField;
 import com.qlangtech.tis.plugin.annotation.FormFieldType;
 import com.qlangtech.tis.plugin.annotation.Validator;
+import com.qlangtech.tis.plugin.ds.DataSourceFactoryManipulate;
 import com.qlangtech.tis.plugin.incr.TISSinkFactory;
 import com.qlangtech.tis.runtime.module.misc.IFieldErrorHandler;
 import com.qlangtech.tis.sql.parser.tuple.creator.IStreamIncrGenerateStrategy;
+import com.qlangtech.tis.util.IPluginContext;
 import com.qlangtech.tis.util.UploadPluginMeta;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.StringUtils;
 
+import java.io.File;
+import java.io.IOException;
 import java.util.Objects;
 import java.util.concurrent.Callable;
 import java.util.function.Function;
@@ -45,7 +53,7 @@
  * @author: baisui 百岁
  * @create: 2021-04-21 09:09
  **/
-public class DefaultDataxProcessor extends DataxProcessor  {
+public class DefaultDataxProcessor extends DataxProcessor {
 
     public static final String KEY_FIELD_NAME = "globalCfg";
 
@@ -75,7 +83,22 @@ public Application buildApp() {
         return app;
     }
 
-
+    @Override
+    public void copy(String newIdentityVal) {
+        if (StringUtils.isEmpty(newIdentityVal)) {
+            throw new IllegalArgumentException("param newIdentityVal can not be empty");
+        }
+        try {
+            File workDir = this.getDataXWorkDir(null);
+            if (!workDir.exists()) {
+                throw new IllegalStateException("workDir:" + workDir.getAbsolutePath() + " is not exist ");
+            }
+            File newWorkDir = new File(workDir.getParentFile(), newIdentityVal);
+            FileUtils.copyDirectory(workDir, newWorkDir);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
 
     @Override
     public String identityValue() {
@@ -93,14 +116,6 @@ public IStreamTemplateResource getFlinkStreamGenerateTplResource() {
 
         return writerPluginOverwrite((d) -> d.getFlinkStreamGenerateTplResource(),
                 () -> DefaultDataxProcessor.super.getFlinkStreamGenerateTplResource());
-
-        //        TISSinkFactory sinKFactory = TISSinkFactory.getIncrSinKFactory(this.identityValue());
-        //        Objects.requireNonNull(sinKFactory, "writer plugin can not be null");
-        //        if (sinKFactory instanceof IStreamIncrGenerateStrategy) {
-        //            return ((IStreamIncrGenerateStrategy) sinKFactory).getFlinkStreamGenerateTemplateFileName();
-        //        }
-        //
-        //        return super.getFlinkStreamGenerateTemplateFileName();
     }
 
     @Override
@@ -123,7 +138,7 @@ private <T> T writerPluginOverwrite(Function<IStreamIncrGenerateStrategy, T> fun
 
 
     @TISExtension()
-    public static class DescriptorImpl extends Descriptor<IAppSource> {
+    public static class DescriptorImpl extends Descriptor<IAppSource> implements IDescribableManipulate<DefaultDataXProcessorManipulate> {
 
         private static final Pattern PATTERN_START_WITH_NUMBER = Pattern.compile("^\\d.{0,}");
 
@@ -144,14 +159,19 @@ public boolean validateName(IFieldErrorHandler msgHandler, Context context, Stri
             if (pluginMeta.isUpdate()) {
                 return true;
             }
-            return msgHandler.validateBizLogic(IFieldErrorHandler.BizLogic.APP_NAME_DUPLICATE, context, fieldName,
+            return msgHandler.validateBizLogic(IFieldErrorHandler.BizLogic.VALIDATE_APP_NAME_DUPLICATE, context, fieldName,
                     value);
         }
 
         @Override
         public String getDisplayName() {
             return DataxProcessor.DEFAULT_DATAX_PROCESSOR_NAME;
         }
+
+        @Override
+        public Class<DefaultDataXProcessorManipulate> getManipulateExtendPoint() {
+            return DefaultDataXProcessorManipulate.class;
+        }
     }
 
 

tis-datax/tis-datax-common-plugin/src/main/java/com/qlangtech/tis/plugin/datax/common/BasicDataXRdbmsReader.java

@@ -93,15 +93,24 @@ public final List<SelectedTab> getSelectedTabs() {
         if (this.preSelectedTabsHash == selectedTabs.hashCode()) {
             return selectedTabs;
         }
-        this.selectedTabs = fillSelectedTabMeta(this.selectedTabs);
+        this.selectedTabs = fillSelectedTabMeta(false, this.selectedTabs);
         this.preSelectedTabsHash = selectedTabs.hashCode();
         return this.selectedTabs;
 
     }
 
     @Override
     public List<SelectedTab> fillSelectedTabMeta(List<SelectedTab> tabs) {
-        boolean shallFillSelectedTabMeta = shallFillSelectedTabMeta();
+        return fillSelectedTabMeta(true, tabs);
+    }
+
+    /**
+     * @param forceFill 忽视缓存的存在，每次都填充
+     * @param tabs
+     * @return
+     */
+    public List<SelectedTab> fillSelectedTabMeta(boolean forceFill, List<SelectedTab> tabs) {
+        boolean shallFillSelectedTabMeta = forceFill || shallFillSelectedTabMeta();
 
         if (shallFillSelectedTabMeta) {
             try (TableColsMeta tabsMeta = getTabsMeta()) {

tis-datax/tis-datax-elasticsearch-plugin/src/main/java/com/qlangtech/tis/plugin/datax/DataXElasticsearchWriter.java

@@ -140,8 +140,15 @@ public ElasticEndpoint getToken() {
     @Override
     public boolean hasDifferWithSource(IPluginContext pluginCtx, ISelectedTab esTab, TableAlias tableAlias) {
         List<IColMetaGetter> cols = esTab.overwriteCols(pluginCtx);
+//        IColMetaGetter col = null;
+//        ISchemaField schemaCol = null;
         ISchema schema = convert2Schema(tableAlias);
-        return schema.getSchemaFields().size() != cols.size();
+        List<ISchemaField> schemaFields = schema.getSchemaFields();
+        if (schemaFields.size() != cols.size()) {
+            return true;
+        }
+
+        return false;
     }
 
     @Override

tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/plugins/incr/flink/cdc/RowFieldGetterFactory.java

@@ -243,6 +243,9 @@ public final Object getFieldOrNull(RowData rowData) {
         }
 
         private Object getVal(RowData rowData) {
+            if (rowData.isNullAt(this.colIndex)) {
+                return null;
+            }
             try {
                 return getObject((GenericRowData) rowData);
             } catch (ClassCastException e) {

tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/plugins/incr/flink/common/FlinkCluster.java

@@ -148,7 +148,7 @@ private void checkUseable(FlinkCluster cluster) throws TisException {
                     Collection<JobStatusMessage> jobStatus = status.get();
                 }
             } catch (Throwable e) {
-                throw TisException.create("Please check link is valid:" + cluster.getJobManagerAddress().getURL(), e);
+                throw TisException.create("Please check link is valid:" + cluster.getJobManagerAddress().getUrl(), e);
             }
         }
 

tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/plugins/incr/flink/launch/FlinkTaskNodeController.java

@@ -239,7 +239,7 @@ public IDeploymentDetail getRCDeployment(TargetResName collection) {
         } catch (TimeoutException e) {
             ClusterType clusterCfg = this.factory.getClusterCfg();
             throw TisException.create(//"flinkClusterId:" + clusterCfg.getClusterId()
-                    ",Address:" + clusterCfg.getJobManagerAddress().getURL() + "连接超时，请检查相应配置是否正确", e);
+                    ",Address:" + clusterCfg.getJobManagerAddress().getUrl() + "连接超时，请检查相应配置是否正确", e);
         } catch (ExecutionException e) {
             Throwable cause = e.getCause();
             if (isNotFoundException(cause)) {

tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/plugins/incr/flink/launch/clustertype/KubernetesApplication.java

@@ -260,7 +260,7 @@ public JobManagerAddress getJobManagerAddress() {
         JSONObject meta = launchTokenMeta != null ? this.launchTokenMeta : this.getLaunchToken().readLaunchedToken();
         return new JobManagerAddress(null, -1) {
             @Override
-            public String getURL() {
+            public String getUrl() {
                 String url = meta.getString(FlinkClusterTokenManager.JSON_KEY_WEB_INTERFACE_URL);
                 if (StringUtils.isEmpty(url)) {
                     throw new IllegalStateException("key:" + FlinkClusterTokenManager.JSON_KEY_WEB_INTERFACE_URL + " relevant value can not be empty");

tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/plugins/incr/flink/launch/clustertype/KubernetesSession.java

@@ -78,7 +78,7 @@ public ClusterClient createRestClusterClient() {
     public JobManagerAddress getJobManagerAddress() {
         return new JobManagerAddress(null, -1) {
             @Override
-            public String getURL() {
+            public String getUrl() {
                 return createRestClusterClient().getWebInterfaceURL();
             }
         };

tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/tis/plugins/incr/flink/cdc/ReocrdTransformerMapper.java

@@ -21,16 +21,12 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.qlangtech.plugins.incr.flink.cdc.FlinkCol;
-import com.qlangtech.tis.async.message.client.consumer.IFlinkColCreator;
-import com.qlangtech.tis.plugin.datax.transformer.OutputParameter;
 import com.qlangtech.tis.plugin.datax.transformer.RecordTransformerRules;
 import com.qlangtech.tis.plugin.datax.transformer.UDFDefinition;
-import com.qlangtech.tis.plugin.datax.transformer.jdbcprop.TargetColType;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.flink.api.common.functions.MapFunction;
 
 import java.io.Serializable;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -48,8 +44,7 @@ public abstract class ReocrdTransformerMapper<Type> implements MapFunction<Type,
     private final FlinkCol2Index col2IdxMapper;
 
     public ReocrdTransformerMapper(List<FlinkCol> cols, RecordTransformerRules transformerRules) {
-
-        this.cols = Collections.unmodifiableList(Objects.requireNonNull(cols, "cols can not be null"));
+        this.cols = Lists.newArrayList(Objects.requireNonNull(cols, "cols can not be null"));
         this.transformerUDF
                 = Objects.requireNonNull(transformerRules, "param transformerRules can not be null")
                 .rules.stream().map((t) -> t.getUdf()).collect(Collectors.toList());

tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/tis/plugins/incr/flink/cdc/impl/RowDataTransformerMapper.java

@@ -27,7 +27,6 @@
 import java.util.List;
 
 /**
- *
  * @author: 百岁（baisui@qlangtech.com）
  * @create: 2024-06-20 17:09
  **/
@@ -36,6 +35,7 @@ public RowDataTransformerMapper(List<FlinkCol> cols, RecordTransformerRules tran
         super(cols, transformerRules);
     }
 
+
     @Override
     protected AbstractTransformerRecord<RowData> createDelegate(RowData row) {
         return new TransformerRowData(row, this.cols);

tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/tis/plugins/incr/flink/cdc/impl/TransformerRowData.java

@@ -22,6 +22,7 @@
 import com.qlangtech.tis.plugins.incr.flink.cdc.AbstractTransformerRecord;
 import org.apache.flink.table.data.ArrayData;
 import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericRowData;
 import org.apache.flink.table.data.MapData;
 import org.apache.flink.table.data.RawValueData;
 import org.apache.flink.table.data.RowData;
@@ -41,28 +42,34 @@ public class TransformerRowData extends AbstractTransformerRecord<RowData> imple
 
     public TransformerRowData(RowData row, List<FlinkCol> cols) {
         super(row);
+        if (!(row instanceof GenericRowData)) {
+            throw new IllegalArgumentException("param row must be type of " + GenericRowData.class);
+        }
         this.cols = cols;
         int newSize = cols.size();
         this.rewriteVals = new Object[newSize];
     }
 
     @Override
     public void setString(String field, String val) {
-        setColumn(field, (val == null) ? null : StringData.fromString(val));
+        setColumn(field, (val == null) ? null : val);
     }
 
     @Override
     public void setColumn(String field, Object val) {
-        rewriteVals[getPos(field)] = (val == null ? NULL : val);
+        Integer pos = getPos(field);
+        FlinkCol flinkCol = cols.get(pos);
+        rewriteVals[pos] = (val == null ? NULL : flinkCol.rowDataProcess.apply(val));
     }
 
     @Override
     public Object getColumn(String field) {
         Integer pos = getPos(field);
+        FlinkCol flinkCol = cols.get(pos);
         if (rewriteVals[pos] != null) {
-            return rewriteVals[pos];
+            Object overWrite = rewriteVals[pos];
+            return overWrite == NULL ? null : flinkCol.getRowDataVal(this);
         }
-        FlinkCol flinkCol = cols.get(pos);
         return getColVal(flinkCol);
     }
 
@@ -77,10 +84,23 @@ public String getString(String field) {
 
     @Override
     public RowData getDelegate() {
-        return this;
+        final int rewriteValsLength = this.rewriteVals.length;
+        GenericRowData old = (GenericRowData) this.row;
+        GenericRowData rowData = new GenericRowData(this.row.getRowKind(), rewriteValsLength);
+        Object rewriteVal = null;
+        for (int i = 0; i < rewriteValsLength; i++) {
+            if ((rewriteVal = rewriteVals[i]) != null && rewriteVal != NULL) {
+                rowData.setField(i, rewriteVal);
+            } else {
+                if (i < old.getArity()) {
+                    rowData.setField(i, old.getField(i));
+                }
+            }
+        }
+        return rowData;
     }
 
-    protected Object getColVal(FlinkCol flinkCol) {
+    private Object getColVal(FlinkCol flinkCol) {
         return flinkCol.getRowDataVal(this.row);
     }
 

tis-split-table-strategy-plugin/src/main/java/com/qlangtech/tis/plugin/ds/manipulate/CloneDataSourceFactory.java

@@ -0,0 +1,75 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 com.qlangtech.tis.plugin.ds.manipulate;
+
+import com.alibaba.citrus.turbine.Context;
+import com.qlangtech.tis.extension.TISExtension;
+import com.qlangtech.tis.plugin.IPluginStore.AfterPluginSaved;
+import com.qlangtech.tis.plugin.IdentityName;
+import com.qlangtech.tis.plugin.annotation.FormField;
+import com.qlangtech.tis.plugin.annotation.FormFieldType;
+import com.qlangtech.tis.plugin.annotation.Validator;
+import com.qlangtech.tis.plugin.ds.DBIdentity;
+import com.qlangtech.tis.plugin.ds.DataSourceFactoryManipulate;
+import com.qlangtech.tis.util.IPluginContext;
+import com.qlangtech.tis.util.IPluginItemsProcessor;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Optional;
+
+/**
+ * @author: 百岁（baisui@qlangtech.com）
+ * @create: 2024-07-09 18:40
+ **/
+public class CloneDataSourceFactory extends DataSourceFactoryManipulate implements AfterPluginSaved, IdentityName {
+
+    @FormField(identity = true, ordinal = 0, type = FormFieldType.INPUTTEXT, validate = {Validator.require, Validator.identity})
+    public String name;
+
+    @Override
+    public void afterSaved(IPluginContext pluginContext, Optional<Context> context) {
+        if (StringUtils.isEmpty(this.name)) {
+            throw new IllegalArgumentException("property name can not be null");
+        }
+        IPluginItemsProcessor itemsProcessor = ManipuldateUtils.cloneInstance(pluginContext, context.get(), this.name, (meta) -> {
+            meta.putExtraParams(DBIdentity.KEY_UPDATE, Boolean.FALSE.toString());
+            meta.putExtraParams(DBIdentity.KEY_DB_NAME, this.name);
+        }, (oldIdentity) -> {
+
+        });
+        itemsProcessor.save(context.get());
+    }
+
+    @Override
+    public String identityValue() {
+        return this.name;
+    }
+
+    @TISExtension
+    public static class DefaultDesc extends BasicDesc {
+        public DefaultDesc() {
+            super();
+        }
+
+        @Override
+        public String getDisplayName() {
+            return "Clone";
+        }
+    }
+}