[feat-857][converter]optimize RowSizeCalculator

Author: Paddy0523

chunjun-connectors/chunjun-connector-binlog/src/main/java/com/dtstack/chunjun/connector/binlog/inputformat/BinlogInputFormatBuilder.java

@@ -60,7 +60,12 @@ public void setBinlogConf(BinlogConf binlogConf) {
     }
 
     public void setRowConverter(AbstractCDCRowConverter rowConverter) {
+        setRowConverter(rowConverter, false);
+    }
+
+    public void setRowConverter(AbstractCDCRowConverter rowConverter, boolean useAbstractColumn) {
         this.format.setRowConverter(rowConverter);
+        format.setUseAbstractColumn(useAbstractColumn);
     }
 
     @Override

chunjun-connectors/chunjun-connector-binlog/src/main/java/com/dtstack/chunjun/connector/binlog/source/BinlogSourceFactory.java

@@ -70,7 +70,7 @@ public DataStream<RowData> createSource() {
                             : TimestampFormat.ISO_8601;
             rowConverter = new BinlogRowConverter(rowType, format);
         }
-        builder.setRowConverter(rowConverter);
+        builder.setRowConverter(rowConverter, useAbstractBaseColumn);
         return createInput(builder.finish());
     }
 

chunjun-connectors/chunjun-connector-cassandra/src/main/java/com/dtstack/chunjun/connector/cassandra/sink/CassandraSinkFactory.java

@@ -66,7 +66,8 @@ public DataStreamSink<RowData> createSink(DataStream<RowData> dataSet) {
         List<FieldConf> fieldConfList = sinkConf.getColumn();
 
         final RowType rowType = TableUtil.createRowType(fieldConfList, getRawTypeConverter());
-        builder.setRowConverter(new CassandraColumnConverter(rowType, fieldConfList));
+        builder.setRowConverter(
+                new CassandraColumnConverter(rowType, fieldConfList), useAbstractBaseColumn);
 
         return createOutput(dataSet, builder.finish());
     }

chunjun-connectors/chunjun-connector-cassandra/src/main/java/com/dtstack/chunjun/connector/cassandra/source/CassandraSourceFactory.java

@@ -71,7 +71,8 @@ public DataStream<RowData> createSource() {
         fieldConfList.forEach(fieldConf -> columnNameList.add(fieldConf.getName()));
 
         final RowType rowType = TableUtil.createRowType(fieldConfList, getRawTypeConverter());
-        builder.setRowConverter(new CassandraRowConverter(rowType, columnNameList));
+        builder.setRowConverter(
+                new CassandraRowConverter(rowType, columnNameList), useAbstractBaseColumn);
 
         return createInput(builder.finish());
     }

chunjun-connectors/chunjun-connector-elasticsearch7/src/main/java/com/dtstack/chunjun/connector/elasticsearch7/sink/Elasticsearch7SinkFactory.java

@@ -91,7 +91,7 @@ public DataStreamSink<RowData> createSink(DataStream<RowData> dataSet) {
         } else {
             rowConverter = new ElasticsearchRowConverter(rowType);
         }
-        builder.setRowConverter(rowConverter);
+        builder.setRowConverter(rowConverter, useAbstractBaseColumn);
         return createOutput(dataSet, builder.finish());
     }
 

chunjun-connectors/chunjun-connector-elasticsearch7/src/main/java/com/dtstack/chunjun/connector/elasticsearch7/source/Elasticsearch7SourceFactory.java

@@ -76,7 +76,7 @@ public DataStream<RowData> createSource() {
         } else {
             rowConverter = new ElasticsearchRowConverter(rowType);
         }
-        builder.setRowConverter(rowConverter);
+        builder.setRowConverter(rowConverter, useAbstractBaseColumn);
         return createInput(builder.finish());
     }
 

chunjun-connectors/chunjun-connector-emqx/src/main/java/com/dtstack/chunjun/connector/emqx/sink/EmqxSinkFactory.java

@@ -60,7 +60,7 @@ public DataStreamSink<RowData> createSink(DataStream<RowData> dataSet) {
         }
         EmqxOutputFormatBuilder builder = new EmqxOutputFormatBuilder();
         builder.setEmqxConf(emqxConf);
-        builder.setRowConverter(new EmqxColumnConverter(emqxConf));
+        builder.setRowConverter(new EmqxColumnConverter(emqxConf), useAbstractBaseColumn);
         return createOutput(dataSet, builder.finish());
     }
 }

chunjun-connectors/chunjun-connector-emqx/src/main/java/com/dtstack/chunjun/connector/emqx/source/EmqxSourceFactory.java

@@ -60,7 +60,7 @@ public DataStream<RowData> createSource() {
         }
         EmqxInputFormatBuilder builder = new EmqxInputFormatBuilder();
         builder.setEmqxConf(emqxConf);
-        builder.setRowConverter(new EmqxColumnConverter(emqxConf));
+        builder.setRowConverter(new EmqxColumnConverter(emqxConf), useAbstractBaseColumn);
         return createInput(builder.finish());
     }
 }

chunjun-connectors/chunjun-connector-ftp/src/main/java/com/dtstack/chunjun/connector/ftp/converter/FtpColumnConverter.java

@@ -112,7 +112,7 @@ protected ISerializationConverter<List<String>> wrapIntoNullableExternalConverte
             ISerializationConverter serializationConverter, FieldConf fieldConf) {
         return (rowData, index, list) -> {
             if (rowData == null || rowData.isNullAt(index)) {
-                list.set(index, null);
+                list.add(index, null);
             } else {
                 serializationConverter.serialize(rowData, index, list);
             }

chunjun-connectors/chunjun-connector-ftp/src/main/java/com/dtstack/chunjun/connector/ftp/sink/FtpSinkFactory.java

@@ -83,7 +83,7 @@ public DataStreamSink<RowData> createSink(DataStream<RowData> dataSet) {
         ftpConfig.setColumn(fieldConfList);
         final RowType rowType =
                 TableUtil.createRowType(ftpConfig.getColumn(), getRawTypeConverter());
-        builder.setRowConverter(new FtpColumnConverter(rowType, ftpConfig));
+        builder.setRowConverter(new FtpColumnConverter(rowType, ftpConfig), useAbstractBaseColumn);
 
         return createOutput(dataSet, builder.finish());
     }