[BitSail-486][Connector] Kudu source refactor

bitsail-common/src/main/java/com/bytedance/bitsail/common/row/Row.java

@@ -153,4 +153,16 @@ public boolean isNullAt(int pos) {
     return this.fields[pos] == null;
   }
 
+  public java.time.LocalTime getLocalTime(int pos) {
+    return (java.time.LocalTime) this.fields[pos];
+  }
+
+  public java.time.LocalDateTime getLocalDateTime(int pos) {
+    return (java.time.LocalDateTime) this.fields[pos];
+  }
+
+  public java.time.LocalDate getLocalDate(int pos) {
+    return (java.time.LocalDate) this.fields[pos];
+  }
+
 }

bitsail-connectors/connector-fake/src/main/java/com/bytedance/bitsail/connector/fake/source/FakeRowGenerator.java

@@ -103,6 +103,9 @@ private Object fakeRawValue(TypeInfo<?> typeInfo) {
     } else if (TypeInfos.SHORT_TYPE_INFO.getTypeClass() == typeInfo.getTypeClass()) {
       return Long.valueOf(faker.number().randomNumber()).shortValue();
 
+    } else if (TypeInfos.BYTE_TYPE_INFO.getTypeClass() == typeInfo.getTypeClass()) {
+      return Long.valueOf(faker.number().randomNumber()).byteValue();
+
     } else if (TypeInfos.STRING_TYPE_INFO.getTypeClass() == typeInfo.getTypeClass()) {
       return faker.name().fullName();
 

bitsail-connectors/connector-kudu/src/main/java/com/bytedance/bitsail/connector/kudu/core/KuduFactory.java

@@ -127,6 +127,7 @@ public KuduSession getSession() {
 
       if (sessionConfig.getMutationBufferSize() != null) {
         kuduSession.setMutationBufferSpace(sessionConfig.getMutationBufferSize());
+        kuduSession.setErrorCollectorSpace(sessionConfig.getMutationBufferSize());
       }
 
       if (sessionConfig.getFlushInterval() != null) {

bitsail-connectors/connector-kudu/src/main/java/com/bytedance/bitsail/connector/kudu/error/KuduErrorCode.java

@@ -27,7 +27,9 @@ public enum KuduErrorCode implements ErrorCode {
   UNSUPPORTED_TYPE("Kudu-04", "Type is not supported"),
   ILLEGAL_VALUE("Kudu-05", "Value type is illegal"),
   SPLIT_ERROR("Kudu-06", "Something wrong with creating splits."),
-  PREDICATE_ERROR("Kudu-07", "Something wrong with kudu predicates.");
+  PREDICATE_ERROR("Kudu-07", "Something wrong with kudu predicates."),
+
+  SCANNER_ERROR("Kudu-08", "Something wrong with kudu scanner.");
 
   private final String code;
 

bitsail-connectors/connector-kudu/src/main/java/com/bytedance/bitsail/connector/kudu/option/KuduReaderOptions.java

@@ -113,22 +113,18 @@ public interface KuduReaderOptions extends ReaderOptions.BaseReaderOptions {
 
   ConfigOption<Boolean> CACHE_BLOCKS =
       key(READER_PREFIX + "enable_cache_blocks")
-          .defaultValue(true);
+          .defaultValue(false);
 
   ConfigOption<Long> SCAN_TIMEOUT =
       key(READER_PREFIX + "scan_timeout_ms")
           .defaultValue(30000L);
-
   ConfigOption<Long> SCAN_ALIVE_PERIOD_MS =
       key(READER_PREFIX + "scan_keep_alive_period_ms")
           .noDefaultValue(Long.class);
-
-  // Split configurations.
-  ConfigOption<String> SPLIT_STRATEGY =
-      key(READER_PREFIX + "split_strategy")
-          .defaultValue("SIMPLE_DIVIDE");
-
-  ConfigOption<String> SPLIT_CONFIGURATION =
-      key(READER_PREFIX + "split_config")
+  ConfigOption<Long> SCAN_SPLIT_SIZE_BYTES =
+      key(READER_PREFIX + "scan_split_size_bytes")
+          .defaultValue(-1L);
+  ConfigOption<String> PREDICATES_CONFIGURATION =
+      key(READER_PREFIX + "predicates")
           .noDefaultValue(String.class);
 }

bitsail-connectors/connector-kudu/src/main/java/com/bytedance/bitsail/connector/kudu/option/KuduWriterOptions.java

@@ -104,7 +104,7 @@ public interface KuduWriterOptions extends WriterOptions.BaseWriterOptions {
    */
   ConfigOption<Integer> MUTATION_BUFFER_SIZE =
       key(WRITER_PREFIX + "kudu_mutation_buffer_size")
-          .noDefaultValue(Integer.class);
+          .defaultValue(1024);
 
   ConfigOption<Integer> SESSION_FLUSH_INTERVAL =
       key(WRITER_PREFIX + "kudu_session_flush_interval")

bitsail-connectors/connector-kudu/src/main/java/com/bytedance/bitsail/connector/kudu/sink/KuduRowBuilder.java

@@ -28,8 +28,11 @@
 import org.apache.kudu.client.PartialRow;
 
 import java.math.BigDecimal;
+import java.math.RoundingMode;
 import java.sql.Date;
 import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.BiConsumer;
@@ -39,16 +42,28 @@ public class KuduRowBuilder {
   private final List<BiConsumer<PartialRow, Object>> rowInserters;
   private final int fieldSize;
 
+  private final Schema schema;
+
   public KuduRowBuilder(BitSailConfiguration jobConf, Schema schema) {
     List<ColumnInfo> columnInfos = jobConf.get(KuduWriterOptions.COLUMNS);
-    this.fieldSize = columnInfos.size();
-
-    this.rowInserters = new ArrayList<>(fieldSize);
-    columnInfos.forEach(columnInfo -> {
-      ColumnSchema columnSchema = schema.getColumn(columnInfo.getName());
-      rowInserters.add(columnSchema.isNullable() ?
-          initNullableRowInserter(columnInfo) : initRowInserter(columnInfo));
-    });
+    this.schema = schema;
+    if (columnInfos != null) {
+      this.fieldSize = columnInfos.size();
+      this.rowInserters = new ArrayList<>(fieldSize);
+      columnInfos.forEach(columnInfo -> {
+        ColumnSchema columnSchema = schema.getColumn(columnInfo.getName());
+        rowInserters.add(columnSchema.isNullable() ?
+            initNullableRowInserter(columnInfo) : initRowInserter(columnInfo));
+      });
+    } else {
+      this.fieldSize = this.schema.getColumnCount();
+      this.rowInserters = new ArrayList<>(fieldSize);
+      this.schema.getColumns().forEach(columnSchema -> {
+        ColumnInfo columnInfo = new ColumnInfo(columnSchema.getName(), columnSchema.getType().getName());
+        rowInserters.add(columnSchema.isNullable() ?
+            initNullableRowInserter(columnInfo) : initRowInserter(columnInfo));
+      });
+    }
   }
 
   /**
@@ -78,36 +93,89 @@ private BiConsumer<PartialRow, Object> initRowInserter(ColumnInfo columnInfo) {
     String typeName = columnInfo.getType().trim().toUpperCase();
 
     switch (typeName) {
+      case "BOOL":
       case "BOOLEAN": // BOOLEAN_TYPE_INFO
         return (PartialRow kuduRow, Object value) -> kuduRow.addBoolean(columnName, (boolean) value);
       case "INT8":    // BYTE_TYPE_INFO
-        return (PartialRow kuduRow, Object value) -> kuduRow.addByte(columnName, (byte) value);
+        return (PartialRow kuduRow, Object value) -> {
+          if (value instanceof Long) {
+            kuduRow.addByte(columnName, ((Long) value).byteValue());
+          } else {
+            kuduRow.addByte(columnName, (byte) value);
+          }
+        };
       case "INT16":   // SHORT_TYPE_INFO
-        return (PartialRow kuduRow, Object value) -> kuduRow.addShort(columnName, (short) value);
+        return (PartialRow kuduRow, Object value) -> {
+          if (value instanceof Long) {
+            kuduRow.addShort(columnName, ((Long) value).shortValue());
+          } else {
+            kuduRow.addShort(columnName, (short) value);
+          }
+        };
       case "INT32":
       case "INT":     // INT_TYPE_INFO
-        return (PartialRow kuduRow, Object value) -> kuduRow.addInt(columnName, (int) value);
+        return (PartialRow kuduRow, Object value) -> {
+          if (value instanceof Long) {
+            kuduRow.addInt(columnName, ((Long) value).intValue());
+          } else {
+            kuduRow.addInt(columnName, (int) value);
+          }
+        };
       case "INT64":
       case "LONG":    // LONG_TYPE_INFO
         return (PartialRow kuduRow, Object value) -> kuduRow.addLong(columnName, (long) value);
       case "DATE":    // SQL_DATE_TYPE_INFO
-        return (PartialRow kuduRow, Object value) -> kuduRow.addDate(columnName, (Date) value);
+        return (PartialRow kuduRow, Object value) -> {
+          if (value instanceof Timestamp) {
+            kuduRow.addDate(columnName, new Date(((Timestamp) value).getTime()));
+          } else if (value instanceof LocalDate) {
+            kuduRow.addDate(columnName, Date.valueOf((LocalDate) value));
+          } else if (value instanceof LocalDateTime) {
+            kuduRow.addDate(columnName, Date.valueOf(((LocalDateTime) value).toLocalDate()));
+          } else if (value instanceof java.util.Date) {
+            kuduRow.addDate(columnName, new Date(((java.util.Date) value).getTime()));
+          } else {
+            kuduRow.addDate(columnName, (Date) value);
+          }
+        };
+      case "UNIXTIME_MICROS":
       case "TIMESTAMP":   // SQL_TIMESTAMP_TYPE_INFO
         return (PartialRow kuduRow, Object value) -> {
           if (value instanceof Timestamp) {
             kuduRow.addTimestamp(columnName, (Timestamp) value);
+          } else if (value instanceof LocalDateTime) {
+            kuduRow.addTimestamp(columnName, Timestamp.valueOf((LocalDateTime) value));
+          } else if (value instanceof LocalDate) {
+            kuduRow.addTimestamp(columnName, Timestamp.valueOf(((LocalDate) value).atStartOfDay()));
+          } else if (value instanceof java.util.Date) {
+            kuduRow.addTimestamp(columnName, new Timestamp(((java.util.Date) value).getTime()));
           } else if (value instanceof Long) {
             kuduRow.addLong(columnName, (Long) value);
           } else {
             throw new BitSailException(KuduErrorCode.ILLEGAL_VALUE, "Value " + value + " is not Long or Timestamp.");
           }
         };
       case "FLOAT":   // FLOAT_TYPE_INFO
-        return (PartialRow kuduRow, Object value) -> kuduRow.addFloat(columnName, (float) value);
+        return (PartialRow kuduRow, Object value) -> {
+          if (value instanceof Double) {
+            kuduRow.addFloat(columnName, ((Double) value).floatValue());
+          } else {
+            kuduRow.addFloat(columnName, (float) value);
+          }
+        };
       case "DOUBLE":  // DOUBLE_TYPE_INFO
         return (PartialRow kuduRow, Object value) -> kuduRow.addDouble(columnName, (double) value);
       case "DECIMAL": // BIG_DECIMAL_TYPE_INFO
-        return (PartialRow kuduRow, Object value) -> kuduRow.addDecimal(columnName, (BigDecimal) value);
+        // scala align
+        final ColumnSchema columnSchema = schema.getColumn(columnInfo.getName());
+        final int scala = columnSchema.getTypeAttributes().getScale();
+        return (PartialRow kuduRow, Object value) -> {
+          if (value instanceof Double) {
+            kuduRow.addDecimal(columnName, BigDecimal.valueOf((double) value).setScale(scala, RoundingMode.DOWN));
+          } else {
+            kuduRow.addDecimal(columnName, ((BigDecimal) value).setScale(scala, RoundingMode.DOWN));
+          }
+        };
       case "VARCHAR": // STRING_TYPE_INFO
         return (PartialRow kuduRow, Object value) -> kuduRow.addVarchar(columnName, value.toString());
       case "STRING":  // STRING_TYPE_INFO

bitsail-connectors/connector-kudu/src/main/java/com/bytedance/bitsail/connector/kudu/sink/KuduWriter.java

@@ -83,6 +83,8 @@ public void write(Row element) throws IOException {
     }
 
     handleOperationResponse(response);
+    // if back_ground_flush need check
+    handleSessionPendingErrors(kuduSession);
   }
 
   @Override

bitsail-connectors/connector-kudu/src/main/java/com/bytedance/bitsail/connector/kudu/source/KuduSource.java

@@ -27,22 +27,17 @@
 import com.bytedance.bitsail.common.configuration.BitSailConfiguration;
 import com.bytedance.bitsail.common.row.Row;
 import com.bytedance.bitsail.common.type.TypeInfoConverter;
-import com.bytedance.bitsail.common.type.filemapping.FileMappingTypeInfoConverter;
 import com.bytedance.bitsail.connector.kudu.core.KuduConstants;
-import com.bytedance.bitsail.connector.kudu.core.KuduFactory;
 import com.bytedance.bitsail.connector.kudu.option.KuduReaderOptions;
 import com.bytedance.bitsail.connector.kudu.source.reader.KuduSourceReader;
-import com.bytedance.bitsail.connector.kudu.source.split.AbstractKuduSplitConstructor;
 import com.bytedance.bitsail.connector.kudu.source.split.KuduSourceSplit;
-import com.bytedance.bitsail.connector.kudu.source.split.KuduSplitFactory;
 import com.bytedance.bitsail.connector.kudu.source.split.coordinator.KuduSourceSplitCoordinator;
+import com.bytedance.bitsail.connector.kudu.source.split.strategy.SimpleDivideSplitConstructor;
+import com.bytedance.bitsail.connector.kudu.type.KuduTypeConverter;
 
-import org.apache.kudu.client.KuduClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.IOException;
-
 public class KuduSource implements Source<Row, KuduSourceSplit, EmptyState>, ParallelismComputable {
   private static final Logger LOG = LoggerFactory.getLogger(KuduSource.class);
 
@@ -75,7 +70,7 @@ public String getReaderName() {
 
   @Override
   public TypeInfoConverter createTypeInfoConverter() {
-    return new FileMappingTypeInfoConverter(getReaderName());
+    return new KuduTypeConverter();
   }
 
   @Override
@@ -84,14 +79,7 @@ public ParallelismAdvice getParallelismAdvice(BitSailConfiguration commonConf, B
     if (selfConf.fieldExists(KuduReaderOptions.READER_PARALLELISM_NUM)) {
       parallelism = selfConf.get(KuduReaderOptions.READER_PARALLELISM_NUM);
     } else {
-      try (KuduFactory kuduFactory = KuduFactory.initReaderFactory(jobConf)) {
-        KuduClient client = kuduFactory.getClient();
-        AbstractKuduSplitConstructor splitConstructor = KuduSplitFactory.getSplitConstructor(jobConf, client);
-        parallelism = splitConstructor.estimateSplitNum();
-      } catch (IOException e) {
-        parallelism = 1;
-        LOG.warn("Failed to compute splits for computing parallelism, will use default 1.");
-      }
+      parallelism = SimpleDivideSplitConstructor.getInstance(jobConf).estimateSplitNum();
     }
 
     return ParallelismAdvice.builder()

bitsail-connectors/connector-kudu/src/main/java/com/bytedance/bitsail/connector/kudu/source/reader/KuduRowDeserializer.java

@@ -73,11 +73,11 @@ private Function<RowResult, Object> initConverter(String columnName, TypeInfo<?>
     if (TypeInfos.LONG_TYPE_INFO.getTypeClass() == curClass) {
       return rowResult -> rowResult.getLong(columnName);
     }
-    if (TypeInfos.SQL_DATE_TYPE_INFO.getTypeClass() == curClass) {
-      return rowResult -> rowResult.getDate(columnName);
+    if (TypeInfos.LOCAL_DATE_TYPE_INFO.getTypeClass() == curClass) {
+      return rowResult -> rowResult.getDate(columnName).toLocalDate();
     }
-    if (TypeInfos.SQL_TIMESTAMP_TYPE_INFO.getTypeClass() == curClass) {
-      return rowResult -> rowResult.getTimestamp(columnName);
+    if (TypeInfos.LOCAL_DATE_TIME_TYPE_INFO.getTypeClass() == curClass) {
+      return rowResult -> rowResult.getTimestamp(columnName).toLocalDateTime();
     }
     if (TypeInfos.FLOAT_TYPE_INFO.getTypeClass() == curClass) {
       return rowResult -> rowResult.getFloat(columnName);