[Feature][Transform] add json path transform (#247)

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/domain/request/job/transform/JsonPath.java

@@ -0,0 +1,39 @@
+/*
+ * 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.seatunnel.app.domain.request.job.transform;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.util.List;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class JsonPath extends TransformOption {
+
+    private List<JsonPathColumn> columns;
+}
+
+@Data
+class JsonPathColumn {
+    private String src_field;
+    private String path;
+    private String destField;
+    private String destType;
+    private String columnErrorHandleWay;
+}

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/domain/request/job/transform/JsonPathTransformOptions.java

@@ -0,0 +1,26 @@
+/*
+ * 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.seatunnel.app.domain.request.job.transform;
+
+import lombok.Data;
+
+@Data
+public class JsonPathTransformOptions implements TransformOptions {
+
+    private JsonPath jsonPath;
+}

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/domain/request/job/transform/Transform.java

@@ -24,5 +24,6 @@ public enum Transform {
     FIELDMAPPER,
     FILTERROWKIND,
     SPLIT,
-    SQL
+    SQL,
+    JSONPATH
 }

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/service/impl/ConnectorServiceImpl.java

@@ -123,6 +123,7 @@ public List<ConnectorInfo> listTransformsForJob(Long jobId) {
                                 .toUpperCase());
 
         if (businessMode.equals(BusinessMode.DATA_INTEGRATION)) {
+
             return connectorCache.getTransform().stream()
                     .filter(
                             connectorInfo -> {
@@ -133,7 +134,8 @@ public List<ConnectorInfo> listTransformsForJob(Long jobId) {
                                         || pluginName.equals("Replace")
                                         || pluginName.equals("Copy")
                                         || pluginName.equals("MultiFieldSplit")
-                                        || pluginName.equals("Sql");
+                                        || pluginName.equals("Sql")
+                                        || pluginName.equals("JsonPath");
                             })
                     .collect(Collectors.toList());
         }

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/service/impl/JobTaskServiceImpl.java

@@ -350,6 +350,8 @@ private JobTaskCheckRes checkNextTaskSchema(
                                 Collections.singletonList(sqlTransformOptions.getSql()));
                     }
                     break;
+                case JSONPATH:
+                    break;
                 case FILTERROWKIND:
                 case REPLACE:
                 default:

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/thirdparty/transfrom/impl/JsonPathTransformSwitcher.java

@@ -0,0 +1,51 @@
+/*
+ * 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.seatunnel.app.thirdparty.transfrom.impl;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.app.domain.request.job.TableSchemaReq;
+import org.apache.seatunnel.app.domain.request.job.transform.Transform;
+import org.apache.seatunnel.app.domain.request.job.transform.TransformOptions;
+import org.apache.seatunnel.app.dynamicforms.FormStructure;
+import org.apache.seatunnel.app.thirdparty.framework.SeaTunnelOptionRuleWrapper;
+import org.apache.seatunnel.app.thirdparty.transfrom.TransformConfigSwitcher;
+
+import com.google.auto.service.AutoService;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@AutoService(TransformConfigSwitcher.class)
+public class JsonPathTransformSwitcher implements TransformConfigSwitcher {
+    @Override
+    public Transform getTransform() {
+        return Transform.JSONPATH;
+    }
+
+    @Override
+    public FormStructure getFormStructure(OptionRule transformOptionRule) {
+        return SeaTunnelOptionRuleWrapper.wrapper(transformOptionRule, this.getTransform().name());
+    }
+
+    @Override
+    public Config mergeTransformConfig(
+            Config transformConfig, TransformOptions transformOption, TableSchemaReq inputSchema) {
+        return transformConfig;
+    }
+}

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/utils/TaskOptionUtils.java

@@ -19,6 +19,7 @@
 
 import org.apache.seatunnel.app.domain.request.job.transform.CopyTransformOptions;
 import org.apache.seatunnel.app.domain.request.job.transform.FieldMapperTransformOptions;
+import org.apache.seatunnel.app.domain.request.job.transform.JsonPathTransformOptions;
 import org.apache.seatunnel.app.domain.request.job.transform.SQLTransformOptions;
 import org.apache.seatunnel.app.domain.request.job.transform.SplitTransformOptions;
 import org.apache.seatunnel.app.domain.request.job.transform.Transform;
@@ -47,6 +48,9 @@ public static <T extends TransformOptions> T getTransformOption(
                         transformOptionsStr, CopyTransformOptions.class);
             case SQL:
                 return convertTransformStrToOptions(transformOptionsStr, SQLTransformOptions.class);
+            case JSONPATH:
+                return convertTransformStrToOptions(
+                        transformOptionsStr, JsonPathTransformOptions.class);
             case FILTERROWKIND:
             case REPLACE:
             default:

seatunnel-ui/src/views/task/synchronization-definition/dag/canvas/node.tsx

@@ -23,6 +23,7 @@ import SinkImg from '../images/sink.png'
 import FieldMapperImg from '../images/field-mapper.png'
 import FilterEventTypeImg from '../images/filter-event-type.png'
 import ReplaceImg from '../images/replace.png'
+import JsonPathImg from '../images/json-path.png'
 import SplitImg from '../images/spilt.png'
 import CopyImg from '../images/copy.png'
 import SqlImg from '../images/sql.png'
@@ -52,6 +53,8 @@ const Node = defineComponent({
       icon.value = CopyImg
     } else if (type === 'transform' && connectorType === 'Sql') {
       icon.value = SqlImg
+    } else if (type === 'transform' && connectorType === 'JsonPath') {
+      icon.value = JsonPathImg
     }
 
     return () => (

seatunnel-ui/src/views/task/synchronization-definition/dag/sidebar/index.tsx

@@ -25,6 +25,7 @@ import SinkImg from '../images/sink.png'
 import FieldMapperImg from '../images/field-mapper.png'
 import FilterEventTypeImg from '../images/filter-event-type.png'
 import ReplaceImg from '../images/replace.png'
+import JsonPathImg from '../images/json-path.png'
 import SplitImg from '../images/spilt.png'
 import CopyImg from '../images/copy.png'
 import SqlImg from '../images/sql.png'
@@ -112,6 +113,8 @@ const DagSidebar = defineComponent({
                 item.icon = CopyImg
               } else if (item.name === 'Sql') {
                 item.icon = SqlImg
+              } else if (item.name === 'JsonPath') {
+                item.icon = JsonPathImg
               }
 
               return (

seatunnel-ui/src/views/task/synchronization-definition/dag/use-model.ts

@@ -263,8 +263,30 @@ export function useNodeModel(
           } else if(transformType === 'Sql'){
             let table = await getSqlTransformOutputData()
             state.outputTableData = table
-
-          } else {
+          } else if(transformType === 'JsonPath'){
+            // extract the dest_field in columns
+            const fixedInput = refForm.value.getValues().columns.replace(/=/g, ':');
+            const destFieldMatches = fixedInput.match(/"dest_field"\s*:\s*"([^"]*)"/g);
+            const destFields = destFieldMatches.map(match => match.match(/"([^"]+)"$/)[1]);
+
+            state.outputTableData = state.inputTableData.map(item => ({ ...item }));
+            destFields.map(item => {
+              state.outputTableData.push(
+                {
+                  "format":"",
+                  "type": "TEXT",
+                  "name": item,
+                  "comment": "",
+                  "primaryKey": false,
+                  "defaultValue": '',
+                  "nullable": false,
+                  "unSupport": false,
+                  "outputDataType": "TEXT"
+              }
+              )
+            });
+          }
+          else {
             state.outputTableData = t.fields
           }
         }

seatunnel-ui/src/views/task/synchronization-definition/dag/use-node-setting.ts

@@ -241,6 +241,18 @@ export function useNodeSettingModal(
           tableName: tableInfo[0].tableName,
           fields: resultSchema.outputTableData
         }]
+      }else if(props.nodeInfo.connectorType === 'JsonPath') {
+        const resultSchema = modelRef.value.getOutputSchema()
+        const tableInfo = resultSchema.allTableData[0].tableInfos
+        transformOptions.columns = {
+          "sourceFieldName": resultSchema.outputTableData[0]?.name || null,
+          "columns": values.columns
+        }
+        modelOutputTableData = [{
+          database: tableInfo[0].database,
+          tableName: tableInfo[0].tableName,
+          fields: resultSchema.outputTableData
+        }]
       }
 
       await saveTaskDefinitionItem(