Doris support deducing partitions

#149

Author: beyond-up

bitsail-common/src/main/java/com/bytedance/bitsail/common/util/DateUtil.java

@@ -26,11 +26,7 @@
 import java.sql.Timestamp;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
-import java.util.Calendar;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.TimeZone;
+import java.util.*;
 
 public class DateUtil {
   public static final int LENGTH_SECOND = 10;
@@ -249,4 +245,25 @@ public static long convertStringToSeconds(Column column) {
       return stringToDate(column.asString(), null, null).toInstant().getEpochSecond();
     }
   }
+
+  public static List<Date> getDatesBetweenTwoDate(Date beginDate, Date endDate) {
+    List<Date> dataRange = new ArrayList<>();
+    dataRange.add(beginDate);
+    Calendar cal = Calendar.getInstance();
+    cal.setTime(beginDate);
+    while (endDate.after(cal.getTime())) {
+      cal.add(Calendar.DAY_OF_MONTH, 1);
+      dataRange.add(cal.getTime());
+    }
+    return dataRange;
+  }
+
+  public static Date getNDaysAfterDate(Date date, int n){
+    Calendar cal = Calendar.getInstance();
+    cal.setTime(date);
+    cal.add(Calendar.DAY_OF_MONTH, n);
+    return cal.getTime();
+  }
+
+
 }

bitsail-common/src/test/java/com/bytedance/bitsail/common/util/DateUtilTest.java

@@ -21,6 +21,10 @@
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
 public class DateUtilTest {
   @Test
   public void testConvertStringToSeconds() {
@@ -29,4 +33,26 @@ public void testConvertStringToSeconds() {
     Assert.assertEquals(1575734400L, DateUtil.convertStringToSeconds(new StringColumn("2020-08-01")));
     Assert.assertEquals(1596214923L, DateUtil.convertStringToSeconds(new StringColumn("2020-08-01 01:02:03")));
   }
+
+  @Test
+  public void testGetDatesBetweenTwoDate(){
+    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
+    try {
+      Date dStart = sdf.parse("2022-12-26");
+      Date dEnd = sdf.parse("2022-12-26");
+      Assert.assertEquals(1, DateUtil.getDatesBetweenTwoDate(dStart, dEnd).size());
+    } catch (ParseException e) {
+      e.printStackTrace();
+    }
+  }
+  @Test
+  public void testGetNDaysAfterDate(){
+    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
+    try {
+      Date date = sdf.parse("2022-12-31");
+      Assert.assertEquals("2023-01-01", sdf.format(DateUtil.getNDaysAfterDate(date, 1)));
+    } catch (ParseException e) {
+      e.printStackTrace();
+    }
+  }
 }

bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/option/DorisWriterOptions.java

@@ -21,8 +21,8 @@
 import com.bytedance.bitsail.common.option.WriterOptions;
 
 import com.alibaba.fastjson.TypeReference;
+import com.bytedance.bitsail.connector.doris.partition.DorisPartitionTemplate;
 
-import java.util.List;
 import java.util.Map;
 
 import static com.bytedance.bitsail.common.option.ConfigOptions.key;
@@ -64,9 +64,9 @@ public interface DorisWriterOptions extends WriterOptions.BaseWriterOptions {
       key(WRITER_PREFIX + "table_has_partition")
           .defaultValue(true);
 
-  ConfigOption<List<Map<String, Object>>> PARTITIONS =
+  ConfigOption<DorisPartitionTemplate> PARTITIONS =
       key(WRITER_PREFIX + "partitions")
-          .onlyReference(new TypeReference<List<Map<String, Object>>>() {});
+          .onlyReference(new TypeReference<DorisPartitionTemplate>() {});
 
   ConfigOption<Integer> SINK_FLUSH_INTERVAL_MS =
       key(WRITER_PREFIX + "sink_flush_interval_ms")

bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/partition/DorisPartitionTemplate.java

@@ -0,0 +1,25 @@
+package com.bytedance.bitsail.connector.doris.partition;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+
+@AllArgsConstructor
+@NoArgsConstructor
+@Data
+public class DorisPartitionTemplate {
+
+    @JsonProperty(value = "prefix", required = false, defaultValue = "p")
+    private String prefix;
+
+    @JsonProperty(value = "start_range", required = true)
+    private String startRange;
+
+    @JsonProperty(value = "end_range", required = true)
+    private String endRange;
+
+    @JsonProperty(value = "pattern", required = false, defaultValue = "yyyy-MM-dd")
+    private String pattern;
+}

bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/sink/DorisSink.java

@@ -21,11 +21,13 @@
 import com.bytedance.bitsail.base.connector.writer.v1.WriterCommitter;
 import com.bytedance.bitsail.base.serializer.BinarySerializer;
 import com.bytedance.bitsail.base.serializer.SimpleBinarySerializer;
+import com.bytedance.bitsail.common.BitSailException;
 import com.bytedance.bitsail.common.configuration.BitSailConfiguration;
 import com.bytedance.bitsail.common.exception.CommonErrorCode;
 import com.bytedance.bitsail.common.model.ColumnInfo;
 import com.bytedance.bitsail.common.type.TypeInfoConverter;
 import com.bytedance.bitsail.common.type.filemapping.FileMappingTypeInfoConverter;
+import com.bytedance.bitsail.common.util.DateUtil;
 import com.bytedance.bitsail.connector.doris.DorisConnectionHolder;
 import com.bytedance.bitsail.connector.doris.committer.DorisCommittable;
 import com.bytedance.bitsail.connector.doris.committer.DorisCommittableSerializer;
@@ -36,6 +38,7 @@
 import com.bytedance.bitsail.connector.doris.option.DorisWriterOptions;
 import com.bytedance.bitsail.connector.doris.partition.DorisPartition;
 import com.bytedance.bitsail.connector.doris.partition.DorisPartitionManager;
+import com.bytedance.bitsail.connector.doris.partition.DorisPartitionTemplate;
 import com.bytedance.bitsail.connector.doris.sink.ddl.DorisSchemaManagerGenerator;
 
 import com.alibaba.fastjson.JSON;
@@ -44,11 +47,9 @@
 
 import java.io.IOException;
 import java.sql.SQLException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.*;
 import java.util.stream.Collectors;
 
 public class DorisSink<InputT> implements Sink<InputT, DorisCommittable, DorisWriterState> {
@@ -137,16 +138,38 @@ private void initDorisOptions(BitSailConfiguration writerConfiguration) {
     // Need partition info in batch replace modes.
     if (isHasPartition && this.writeMode.equals(DorisExecutionOptions.WRITE_MODE.BATCH_REPLACE)) {
       //BATCH and REPLACE mode need the partition infos
-      List<Map<String, Object>> partitionList = writerConfiguration.getNecessaryOption(DorisWriterOptions.PARTITIONS, CommonErrorCode.CONFIG_ERROR);
-      builder.partitions(
-          partitionList.stream()
-              .map(partition -> JSON.parseObject(JSON.toJSONString(partition), DorisPartition.class))
-              .collect(Collectors.toList())
-      );
+      DorisPartitionTemplate dorisPartitionTemplate = writerConfiguration.getNecessaryOption(DorisWriterOptions.PARTITIONS, CommonErrorCode.CONFIG_ERROR);
+      List<DorisPartition> dorisPartitions = parseTemplateToDorisPartitions(dorisPartitionTemplate);
+      builder.partitions(dorisPartitions);
     }
     dorisOptions = builder.build();
   }
 
+  private List<DorisPartition> parseTemplateToDorisPartitions(DorisPartitionTemplate dorisPartitionTemplate){
+    String pattern = dorisPartitionTemplate.getPattern();
+    String prefix = dorisPartitionTemplate.getPrefix();
+    String start = dorisPartitionTemplate.getStartRange();
+    String end = dorisPartitionTemplate.getEndRange();
+
+    SimpleDateFormat sdf = new SimpleDateFormat(pattern);
+    try {
+      Date dStart = sdf.parse(start);
+      Date dEnd = sdf.parse(end);
+      List<DorisPartition> dorisPartitions = new ArrayList<>();
+      List<Date> listDate = DateUtil.getDatesBetweenTwoDate(dStart, dEnd);
+      listDate.forEach(date->{
+        DorisPartition dorisPartition = new DorisPartition();
+        dorisPartition.setName(prefix + sdf.format(date));
+        dorisPartition.setStartRange(Collections.singletonList(sdf.format(date)));
+        dorisPartition.setEndRange(Collections.singletonList(sdf.format(DateUtil.getNDaysAfterDate(date, 1))));
+        dorisPartitions.add(dorisPartition);
+      });
+      return dorisPartitions;
+    } catch (ParseException e) {
+      throw BitSailException.asBitSailException(CommonErrorCode.CONFIG_ERROR, "Can't parse configuration info: " + dorisPartitionTemplate, e);
+    }
+  }
+
   private void initDorisExecutionOptions(BitSailConfiguration writerConfiguration) {
     LOG.info("Start to init DorisExecutionOptions!");
     final DorisExecutionOptions.DorisExecutionOptionsBuilder builder = DorisExecutionOptions.builder();