[Feature][Transform][python] add data on thread

Author: FuYouJ

Diff for: seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/python/PythonOperationProxy.java

@@ -17,11 +17,17 @@
 
 package org.apache.seatunnel.transform.python;
 
+import org.apache.seatunnel.api.table.type.SeaTunnelRowAccessor;
 import py4j.GatewayServer;
 
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
 public class PythonOperationProxy implements RowOperation {
 
     private GatewayServer javaServer;
+    private final Map<Long,SeaTunnelRowAccessor> threadDataMap = new ConcurrentHashMap<>();
+    private final Map<Long,Long> threadDataOffsetMap = new ConcurrentHashMap<>();
     private PythonOperationProxy(){
         if (INSTANCE != null) {
             throw new RuntimeException("Please use newInstance() method for getting the single instance of this class.");
@@ -43,4 +49,16 @@ public static PythonOperationProxy newInstance(Integer javaServerPort) {
     public void shutdown(){
         this.javaServer.shutdown();
     }
+
+    public void putThreadData(long threadId, SeaTunnelRowAccessor inputRow) {
+        this.threadDataMap.put(threadId,inputRow);
+        this.threadDataOffsetMap.compute(threadId,(thread,offset) ->{
+            if (offset == null){
+                offset = 0L;
+            }else {
+                offset = offset + 1;
+            }
+            return offset;
+        });
+    }
 }

Diff for: seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/python/PythonTransform.java

@@ -11,14 +11,14 @@ public class PythonTransform extends MultipleFieldOutputTransform {
 
     public static final String PLUGIN_NAME = "Python";
 
-    private final PythonTransformConfig config;
+    private final PythonTransformConfig transformConfig;
 
     private final PythonOperationProxy pythonOperationProxy;
 
     public PythonTransform(@NonNull CatalogTable inputCatalogTable, PythonTransformConfig transformConfig) {
         super(inputCatalogTable, transformConfig.getErrorHandleWay());
-        this.config = transformConfig;
-        pythonOperationProxy = initLocalSingletonJavaServer(config.getJavaServerPort());
+        this.transformConfig = transformConfig;
+        this.pythonOperationProxy = initLocalSingletonJavaServer(transformConfig.getJavaServerPort());
     }
 
     private PythonOperationProxy initLocalSingletonJavaServer(Integer javaServerPort) {
@@ -32,6 +32,8 @@ public String getPluginName() {
 
     @Override
     protected Object[] getOutputFieldValues(SeaTunnelRowAccessor inputRow) {
+        long threadId = Thread.currentThread().getId();
+        pythonOperationProxy.putThreadData(threadId,inputRow);
         return new Object[0];
     }