Merge branch 'feat_1.8_3.10_25158' into 1.8_release_3.10.x

Author: gituser

db2/db2-side/db2-all-side/src/main/java/com/dtstack/flink/sql/side/db2/Db2AllSideInfo.java

@@ -22,6 +22,8 @@
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.SideTableInfo;
 import com.dtstack.flink.sql.side.rdb.all.RdbAllSideInfo;
+import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
+import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 
 import java.util.List;
@@ -38,4 +40,9 @@ public class Db2AllSideInfo extends RdbAllSideInfo {
     public Db2AllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
+
+    @Override
+    public String getTableName(RdbSideTableInfo rdbSideTableInfo) {
+        return DtStringUtil.getTableFullPath(rdbSideTableInfo.getSchema(), rdbSideTableInfo.getTableName());
+    }
 }

db2/db2-side/db2-async-side/src/main/java/com/dtstack/flink/sql/side/db2/Db2AsyncReqRow.java

@@ -29,8 +29,6 @@
 import io.vertx.ext.jdbc.JDBCClient;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.configuration.Configuration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.List;
 
@@ -43,8 +41,7 @@
  */
 
 public class Db2AsyncReqRow extends RdbAsyncReqRow {
-
-    private static final Logger LOG = LoggerFactory.getLogger(Db2AsyncReqRow.class);
+    private final static String DB2_PREFERRED_TEST_QUERY_SQL = "select 1 from sysibm.dual";
 
     private final static String DB2_DRIVER = "com.ibm.db2.jcc.DB2Driver";
 
@@ -63,7 +60,7 @@ public void open(Configuration parameters) throws Exception {
                 .put("user", rdbSideTableInfo.getUserName())
                 .put("password", rdbSideTableInfo.getPassword())
                 .put("provider_class", DT_PROVIDER_CLASS)
-                .put("preferred_test_query", PREFERRED_TEST_QUERY_SQL)
+                .put("preferred_test_query", DB2_PREFERRED_TEST_QUERY_SQL)
                 .put("idle_connection_test_period", DEFAULT_IDLE_CONNECTION_TEST_PEROID)
                 .put("test_connection_on_checkin", DEFAULT_TEST_CONNECTION_ON_CHECKIN);
 

db2/db2-side/db2-async-side/src/main/java/com/dtstack/flink/sql/side/db2/Db2AsyncSideInfo.java

@@ -22,6 +22,8 @@
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.SideTableInfo;
 import com.dtstack.flink.sql.side.rdb.async.RdbAsyncSideInfo;
+import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
+import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 
 import java.util.List;
@@ -38,4 +40,10 @@ public class Db2AsyncSideInfo extends RdbAsyncSideInfo {
     public Db2AsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
+
+    @Override
+    public String getTableName(RdbSideTableInfo rdbSideTableInfo) {
+        return DtStringUtil.getTableFullPath(rdbSideTableInfo.getSchema(), rdbSideTableInfo.getTableName());
+    }
+
 }