[INLONG-10109][SDK] Support to transform from Json protocol to CSV/KV…

… protocol by single SQL

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/CsvSourceData.java

@@ -51,10 +51,10 @@ public int getRowCount() {
 
     @Override
     public String getField(int rowNum, String fieldName) {
-        if (rowNum > this.rows.size()) {
+        if (rowNum >= this.rows.size()) {
             return null;
         }
-        Map<String, String> targetRow = this.rows.get(rowNum - 1);
+        Map<String, String> targetRow = this.rows.get(rowNum);
         return targetRow.get(fieldName);
     }
 }

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/JsonNode.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.inlong.sdk.transform.decode;
+
+import lombok.Data;
+import org.apache.commons.lang.math.NumberUtils;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * JsonNode
+ * 
+ */
+@Data
+public class JsonNode {
+
+    private String name;
+    private boolean isArray = false;
+    private int arrayIndex = -1;
+
+    public JsonNode(String nodeString) {
+        int beginIndex = nodeString.indexOf('[');
+        if (beginIndex < 0) {
+            this.name = nodeString;
+        } else {
+            this.name = StringUtils.trim(nodeString.substring(0, beginIndex));
+            int endIndex = nodeString.lastIndexOf(']');
+            if (endIndex >= 0) {
+                this.isArray = true;
+                this.arrayIndex = NumberUtils.toInt(nodeString.substring(beginIndex + 1, endIndex), -1);
+                if (this.arrayIndex < 0) {
+                    this.arrayIndex = 0;
+                }
+            }
+        }
+    }
+}

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/JsonSourceData.java

@@ -0,0 +1,134 @@
+/*
+ * 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.inlong.sdk.transform.decode;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * JsonSourceData
+ * 
+ */
+public class JsonSourceData implements SourceData {
+
+    public static final String ROOT_KEY = "$root";
+
+    public static final String CHILD_KEY = "$child";
+
+    private JsonObject root;
+
+    private JsonArray childRoot;
+
+    /**
+     * Constructor
+     * @param root
+     * @param childRoot
+     */
+    public JsonSourceData(JsonObject root, JsonArray childRoot) {
+        this.root = root;
+        this.childRoot = childRoot;
+    }
+
+    /**
+     * getRowCount
+     * @return
+     */
+    @Override
+    public int getRowCount() {
+        if (this.childRoot == null) {
+            return 1;
+        } else {
+            return this.childRoot.size();
+        }
+    }
+
+    /**
+     * getField
+     * @param rowNum
+     * @param fieldName
+     * @return
+     */
+    @Override
+    public String getField(int rowNum, String fieldName) {
+        try {
+            List<JsonNode> childNodes = new ArrayList<>();
+            String[] nodeStrings = fieldName.split("\\.");
+            for (String nodeString : nodeStrings) {
+                childNodes.add(new JsonNode(nodeString));
+            }
+            // parse
+            if (childNodes.size() == 0) {
+                return "";
+            }
+            // first node
+            JsonNode firstNode = childNodes.get(0);
+            JsonElement current = root;
+            if (StringUtils.equals(ROOT_KEY, firstNode.getName())) {
+                current = root;
+            } else if (StringUtils.equals(CHILD_KEY, firstNode.getName())) {
+                if (rowNum < childRoot.size()) {
+                    current = childRoot.get(rowNum);
+                } else {
+                    return "";
+                }
+            } else {
+                // error data
+                return "";
+            }
+            if (current == null) {
+                // error data
+                return "";
+            }
+            // parse other node
+            for (int i = 1; i < childNodes.size(); i++) {
+                JsonNode node = childNodes.get(i);
+                if (!current.isJsonObject()) {
+                    // error data
+                    return "";
+                }
+                JsonElement newElement = current.getAsJsonObject().get(node.getName());
+                if (newElement == null) {
+                    // error data
+                    return "";
+                }
+                if (!node.isArray()) {
+                    current = newElement;
+                } else {
+                    if (!newElement.isJsonArray()) {
+                        // error data
+                        return "";
+                    }
+                    JsonArray newArray = newElement.getAsJsonArray();
+                    if (node.getArrayIndex() >= newArray.size()) {
+                        // error data
+                        return "";
+                    }
+                    current = newArray.get(node.getArrayIndex());
+                }
+            }
+            return current.getAsString();
+        } catch (Exception e) {
+            return "";
+        }
+    }
+}

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/JsonSourceDecoder.java

@@ -0,0 +1,123 @@
+/*
+ * 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.inlong.sdk.transform.decode;
+
+import org.apache.inlong.sdk.transform.pojo.JsonSourceInfo;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import org.apache.commons.lang3.StringUtils;
+
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * JsonSourceDecoder
+ * 
+ */
+public class JsonSourceDecoder implements SourceDecoder {
+
+    protected JsonSourceInfo sourceInfo;
+    private Charset srcCharset = Charset.defaultCharset();
+    private String rowsNodePath;
+    private List<JsonNode> childNodes;
+
+    private Gson gson = new Gson();
+
+    /**
+     * Constructor
+     * @param sourceInfo
+     */
+    public JsonSourceDecoder(JsonSourceInfo sourceInfo) {
+        this.sourceInfo = sourceInfo;
+        if (!StringUtils.isBlank(sourceInfo.getCharset())) {
+            this.srcCharset = Charset.forName(sourceInfo.getCharset());
+        }
+        this.rowsNodePath = sourceInfo.getRowsNodePath();
+        if (!StringUtils.isBlank(rowsNodePath)) {
+            this.childNodes = new ArrayList<>();
+            String[] nodeStrings = this.rowsNodePath.split("\\.");
+            for (String nodeString : nodeStrings) {
+                this.childNodes.add(new JsonNode(nodeString));
+            }
+        }
+    }
+
+    /**
+     * decode
+     * @param srcBytes
+     * @param extParams
+     * @return
+     */
+    @Override
+    public SourceData decode(byte[] srcBytes, Map<String, Object> extParams) {
+        String srcString = new String(srcBytes, srcCharset);
+        return this.decode(srcString, extParams);
+    }
+
+    /**
+     * decode
+     * @param srcString
+     * @param extParams
+     * @return
+     */
+    @Override
+    public SourceData decode(String srcString, Map<String, Object> extParams) {
+        JsonObject root = gson.fromJson(srcString, JsonObject.class);
+        JsonArray childRoot = null;
+        if (this.childNodes != null && this.childNodes.size() > 0) {
+            JsonElement current = root;
+            for (JsonNode node : childNodes) {
+                if (!current.isJsonObject()) {
+                    // error data
+                    return new JsonSourceData(root, childRoot);
+                }
+                JsonElement newElement = current.getAsJsonObject().get(node.getName());
+                if (newElement == null) {
+                    // error data
+                    return new JsonSourceData(root, childRoot);
+                }
+                if (!node.isArray()) {
+                    current = newElement;
+                } else {
+                    if (!newElement.isJsonArray()) {
+                        // error data
+                        return new JsonSourceData(root, childRoot);
+                    }
+                    JsonArray newArray = newElement.getAsJsonArray();
+                    if (node.getArrayIndex() >= newArray.size()) {
+                        // error data
+                        return new JsonSourceData(root, childRoot);
+                    }
+                    current = newArray.get(node.getArrayIndex());
+                }
+            }
+            if (!current.isJsonArray()) {
+                // error data
+                return new JsonSourceData(root, childRoot);
+            }
+            childRoot = current.getAsJsonArray();
+        }
+        SourceData sourceData = new JsonSourceData(root, childRoot);
+        return sourceData;
+    }
+}

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/KvSourceData.java

@@ -50,10 +50,10 @@ public int getRowCount() {
 
     @Override
     public String getField(int rowNum, String fieldName) {
-        if (rowNum > this.rows.size()) {
+        if (rowNum >= this.rows.size()) {
             return null;
         }
-        Map<String, String> targetRow = this.rows.get(rowNum - 1);
+        Map<String, String> targetRow = this.rows.get(rowNum);
         return targetRow.get(fieldName);
     }
 }

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/encode/CsvSinkEncoder.java

@@ -74,4 +74,12 @@ public String encode(SinkData sinkData) {
         }
         return builder.substring(0, builder.length() - 1);
     }
+
+    /**
+     * get fields
+     * @return the fields
+     */
+    public List<FieldInfo> getFields() {
+        return fields;
+    }
 }

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/encode/KvSinkEncoder.java

@@ -61,4 +61,12 @@ public String encode(SinkData sinkData) {
         }
         return builder.substring(0, builder.length() - 1);
     }
+
+    /**
+     * get fields
+     * @return the fields
+     */
+    public List<FieldInfo> getFields() {
+        return fields;
+    }
 }

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/encode/SinkEncoder.java

@@ -17,10 +17,16 @@
 
 package org.apache.inlong.sdk.transform.encode;
 
+import org.apache.inlong.sdk.transform.pojo.FieldInfo;
+
+import java.util.List;
+
 /**
  * SinkEncoder
  */
 public interface SinkEncoder {
 
     String encode(SinkData sinkData);
+
+    List<FieldInfo> getFields();
 }