[CARBONDATA-3640][CARBONDATA-3557] Support flink ingest carbon partition table

 Add support for flink carbon sink to write partitioned carbondata files
 as stage files.
 Add support for INSERT STAGE command to load stage files into CarbonData
 table.

 This closes apache#3542

Author: niuge01

Diff for: core/src/main/java/org/apache/carbondata/core/statusmanager/StageInput.java

@@ -39,6 +39,12 @@ public class StageInput {
    */
   private Map<String, Long> files;
 
+  /**
+   * this list of partition data information in this StageInput
+   * @see PartitionLocation
+   */
+  private List<PartitionLocation> locations;
+
   public StageInput() {
 
   }
@@ -48,6 +54,11 @@ public StageInput(String base, Map<String, Long> files) {
     this.files = files;
   }
 
+  public StageInput(String base, List<PartitionLocation> locations) {
+    this.base = base;
+    this.locations = locations;
+  }
+
   public String getBase() {
     return base;
   }
@@ -64,6 +75,14 @@ public void setFiles(Map<String, Long> files) {
     this.files = files;
   }
 
+  public List<PartitionLocation> getLocations() {
+    return this.locations;
+  }
+
+  public void setLocations(final List<PartitionLocation> locations) {
+    this.locations = locations;
+  }
+
   public List<InputSplit> createSplits() {
     return
         files.entrySet().stream().filter(
@@ -75,4 +94,43 @@ public List<InputSplit> createSplits() {
         ).collect(Collectors.toList());
   }
 
+  public static final class PartitionLocation {
+
+    public PartitionLocation() {
+
+    }
+
+    public PartitionLocation(final Map<String, String> partitions, final Map<String, Long> files) {
+      this.partitions = partitions;
+      this.files = files;
+    }
+
+    /**
+     * the list of (partitionColumn, partitionValue) of this partition.
+     */
+    private Map<String, String> partitions;
+
+    /**
+     * the list of (file, length) in this partition.
+     */
+    private Map<String, Long> files;
+
+    public Map<String, String> getPartitions() {
+      return this.partitions;
+    }
+
+    public void setPartitions(final Map<String, String> partitions) {
+      this.partitions = partitions;
+    }
+
+    public Map<String, Long> getFiles() {
+      return this.files;
+    }
+
+    public void setFiles(final Map<String, Long> files) {
+      this.files = files;
+    }
+
+  }
+
 }

Diff for: core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java

@@ -75,7 +75,7 @@ protected DateFormat initialValue() {
   /**
    * DataType converter for different computing engines
    */
-  private static DataTypeConverter converter;
+  private static final ThreadLocal<DataTypeConverter> converter = new ThreadLocal<>();
 
   /**
    * This method will convert a given value to its specific type
@@ -105,7 +105,7 @@ public static Object getMeasureValueBasedOnDataType(String msrValue, DataType da
           new BigDecimal(msrValue).setScale(scale, RoundingMode.HALF_UP);
       BigDecimal decimal = normalizeDecimalValue(bigDecimal, precision);
       if (useConverter) {
-        return converter.convertFromBigDecimalToDecimal(decimal);
+        return getDataTypeConverter().convertFromBigDecimalToDecimal(decimal);
       } else {
         return decimal;
       }
@@ -144,7 +144,7 @@ public static Object getNoDictionaryValueBasedOnDataType(String dimValue, DataTy
           new BigDecimal(dimValue).setScale(scale, RoundingMode.HALF_UP);
       BigDecimal decimal = normalizeDecimalValue(bigDecimal, precision);
       if (useConverter) {
-        return converter.convertFromBigDecimalToDecimal(decimal);
+        return getDataTypeConverter().convertFromBigDecimalToDecimal(decimal);
       } else {
         return decimal;
       }
@@ -457,7 +457,7 @@ public static Object getDataDataTypeForNoDictionaryColumn(String dimensionValue,
       }
     } else {
       // Default action for String/Varchar
-      return converter.convertFromStringToUTF8String(dimensionValue);
+      return getDataTypeConverter().convertFromStringToUTF8String(dimensionValue);
     }
   }
 
@@ -974,7 +974,7 @@ private static String parseStringToBigDecimal(String value, ColumnSchema columnS
    */
   public static void setDataTypeConverter(DataTypeConverter converterLocal) {
     if (converterLocal != null) {
-      converter = converterLocal;
+      converter.set(converterLocal);
       timeStampformatter.remove();
       dateformatter.remove();
     }
@@ -989,10 +989,17 @@ public static void clearFormatter() {
   }
 
   public static DataTypeConverter getDataTypeConverter() {
-    if (converter == null) {
-      converter = new DataTypeConverterImpl();
+    DataTypeConverter dataTypeConverter = converter.get();
+    if (dataTypeConverter == null) {
+      synchronized (converter) {
+        dataTypeConverter = converter.get();
+        if (dataTypeConverter == null) {
+          dataTypeConverter = new DataTypeConverterImpl();
+          converter.set(dataTypeConverter);
+        }
+      }
     }
-    return converter;
+    return dataTypeConverter;
   }
 
   public static DataType valueOf(String name) {

Diff for: integration/flink-proxy/src/main/java/org/apache/carbon/flink/ProxyFileWriter.java

@@ -25,7 +25,9 @@ public abstract class ProxyFileWriter<OUT> implements BulkWriter<OUT> {
 
   public abstract ProxyFileWriterFactory getFactory();
 
-  public abstract String getPartition();
+  public abstract String getIdentifier();
+
+  public abstract String getPath();
 
   public abstract void commit() throws IOException;
 

Diff for: integration/flink-proxy/src/main/java/org/apache/carbon/flink/ProxyFileWriterFactory.java

@@ -74,7 +74,7 @@ public void setConfiguration(final Configuration configuration) {
     this.configuration = configuration;
   }
 
-  public abstract ProxyFileWriter<OUT> create(String partition) throws IOException;
+  public abstract ProxyFileWriter<OUT> create(String identifier, String path) throws IOException;
 
   public static class Configuration implements Serializable {
 

Diff for: integration/flink-proxy/src/main/java/org/apache/carbon/flink/ProxyRecoverable.java

@@ -25,18 +25,22 @@ public final class ProxyRecoverable
   public ProxyRecoverable(
       final String writerType,
       final ProxyFileWriterFactory.Configuration writerConfiguration,
-      final String partition
+      final String writerIdentifier,
+      final String writePath
   ) {
     this.writerType = writerType;
     this.writerConfiguration = writerConfiguration;
-    this.partition = partition;
+    this.writerIdentifier = writerIdentifier;
+    this.writePath = writePath;
   }
 
   private final String writerType;
 
   private final ProxyFileWriterFactory.Configuration writerConfiguration;
 
-  private final String partition;
+  private final String writerIdentifier;
+
+  private final String writePath;
 
   public String getWriterType() {
     return this.writerType;
@@ -46,8 +50,12 @@ public ProxyFileWriterFactory.Configuration getWriterConfiguration() {
     return this.writerConfiguration;
   }
 
-  public String getPartition() {
-    return this.partition;
+  public String getWriterIdentifier() {
+    return this.writerIdentifier;
+  }
+
+  public String getWritePath() {
+    return this.writePath;
   }
 
 }

Diff for: integration/flink-proxy/src/main/java/org/apache/carbon/flink/ProxyRecoverableOutputStream.java

@@ -82,7 +82,8 @@ public Committer closeForCommit() {
         new ProxyRecoverable(
             this.writer.getFactory().getType(),
             this.writer.getFactory().getConfiguration(),
-            this.writer.getPartition()
+            this.writer.getIdentifier(),
+            this.writer.getPath()
         )
     );
   }
@@ -118,7 +119,8 @@ private ProxyFileWriter<?> newWriter() throws IOException {
         throw new UnsupportedOperationException();
       }
       writerFactory.setConfiguration(this.recoverable.getWriterConfiguration());
-      return writerFactory.create(this.recoverable.getPartition());
+      return writerFactory.create(this.recoverable.getWriterIdentifier(),
+          this.recoverable.getWritePath());
     }
 
   }

Diff for: integration/flink-proxy/src/main/java/org/apache/carbon/flink/ProxyRecoverableSerializer.java

@@ -53,7 +53,8 @@ public byte[] serialize(final ProxyRecoverable proxyRecoverable) {
     final ByteBuffer byteBuffer = ByteBuffer.allocate(BUFFER_SIZE);
     serializeString(byteBuffer, proxyRecoverable.getWriterType());
     serializeConfiguration(byteBuffer, proxyRecoverable.getWriterConfiguration());
-    serializeString(byteBuffer, proxyRecoverable.getPartition());
+    serializeString(byteBuffer, proxyRecoverable.getWriterIdentifier());
+    serializeString(byteBuffer, proxyRecoverable.getWritePath());
     final byte[] bytes = new byte[byteBuffer.position()];
     byteBuffer.position(0);
     byteBuffer.get(bytes);
@@ -113,8 +114,9 @@ public ProxyRecoverable deserialize(final int version, final byte[] bytes) {
     final String writerType = deserializeString(byteBuffer);
     final ProxyFileWriterFactory.Configuration writerConfiguration =
         deserializeConfiguration(byteBuffer);
-    final String partition = deserializeString(byteBuffer);
-    return new ProxyRecoverable(writerType, writerConfiguration, partition);
+    final String writerIdentifier = deserializeString(byteBuffer);
+    final String writePath = deserializeString(byteBuffer);
+    return new ProxyRecoverable(writerType, writerConfiguration, writerIdentifier, writePath);
   }
 
   private static ProxyFileWriterFactory.Configuration deserializeConfiguration(

Diff for: integration/flink/src/main/java/org/apache/carbon/flink/CarbonLocalProperty.java

@@ -23,6 +23,8 @@ public final class CarbonLocalProperty {
 
   public static final String DATA_PATH = "carbon.writer.local.data.path";
 
+  static final String COMMIT_THRESHOLD = "carbon.writer.local.commit.threshold";
+
   private CarbonLocalProperty() {
     // private constructor.
   }