[CARBONDATA-940] alter table add/split partition for spark 2.1

add/split partition function

This closes apache#1192

Author: lionelcao

conf/carbon.properties.template

@@ -42,6 +42,9 @@ carbon.enableXXHash=true
 #carbon.max.level.cache.size=-1
 #enable prefetch of data during merge sort while reading data from sort temp files in data loading
 #carbon.merge.sort.prefetch=true
+######## Alter Partition Configuration ########
+#Number of cores to be used while alter partition
+carbon.number.of.cores.while.alterPartition=2
 ######## Compaction Configuration ########
 #Number of cores to be used while compacting
 carbon.number.of.cores.while.compacting=2

core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java

@@ -305,6 +305,10 @@ public final class CarbonCommonConstants {
    */
   @CarbonProperty
   public static final String NUM_CORES_COMPACTING = "carbon.number.of.cores.while.compacting";
+  /**
+   * Number of cores to be used while alter partition
+   */
+  public static final String NUM_CORES_ALT_PARTITION = "carbon.number.of.cores.while.altPartition";
   /**
    * Number of cores to be used for block sort
    */
@@ -966,6 +970,11 @@ public final class CarbonCommonConstants {
    */
   public static final String COMPACTION_KEY_WORD = "COMPACTION";
 
+  /**
+   * Indicates alter partition
+   */
+  public static String ALTER_PARTITION_KEY_WORD = "ALTER_PARTITION";
+
   /**
    * hdfs temporary directory key
    */

core/src/main/java/org/apache/carbondata/core/locks/LockUsage.java

@@ -26,6 +26,7 @@ public class LockUsage {
   public static final String METADATA_LOCK = "meta.lock";
   public static final String COMPACTION_LOCK = "compaction.lock";
   public static final String SYSTEMLEVEL_COMPACTION_LOCK = "system_level_compaction.lock";
+  public static final String ALTER_PARTITION_LOCK = "alter_partition.lock";
   public static final String TABLE_STATUS_LOCK = "tablestatus.lock";
   public static final String TABLE_UPDATE_STATUS_LOCK = "tableupdatestatus.lock";
   public static final String DELETE_SEGMENT_LOCK = "delete_segment.lock";

core/src/main/java/org/apache/carbondata/core/metadata/schema/PartitionInfo.java

@@ -67,6 +67,31 @@ public PartitionInfo(List<ColumnSchema> columnSchemaList, PartitionType partitio
     this.partitionIds = new ArrayList<>();
   }
 
+  /**
+   * add partition means split default partition, add in last directly
+   */
+  public void  addPartition(int addPartitionCount) {
+    for (int i = 0; i < addPartitionCount; i++) {
+      partitionIds.add(++MAX_PARTITION);
+      numPartitions++;
+    }
+  }
+
+  /**
+   * e.g. original partition[0,1,2,3,4,5]
+   * split partition 2 to partition 6,7,8 (will not reuse 2)
+   * then sourcePartitionId is 2, newPartitionNumbers is 3
+   * @param sourcePartitionIndex
+   * @param newPartitionNumbers
+   */
+  public void splitPartition(int sourcePartitionIndex, int newPartitionNumbers) {
+    partitionIds.remove(sourcePartitionIndex);
+    for (int i = 0; i < newPartitionNumbers; i++) {
+      partitionIds.add(sourcePartitionIndex + i, ++MAX_PARTITION);
+    }
+    numPartitions = numPartitions - 1 + newPartitionNumbers;
+  }
+
   public List<ColumnSchema> getColumnSchemaList() {
     return columnSchemaList;
   }

core/src/main/java/org/apache/carbondata/core/scan/filter/partition/PartitionFilterUtil.java

@@ -107,7 +107,7 @@ public static BitSet getPartitionMapForRangeFilter(PartitionInfo partitionInfo,
           }
         }
       } else {
-        // LessThanEqualTo(<)
+        // LessThan(<)
         outer4:
         for (int i = 0; i < partitions; i++) {
           for (String value : listInfo.get(i)) {

core/src/main/java/org/apache/carbondata/core/scan/result/iterator/PartitionSpliterRawResultIterator.java

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.carbondata.core.scan.result.iterator;
+
+import org.apache.carbondata.common.CarbonIterator;
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.scan.result.BatchResult;
+
+public class PartitionSpliterRawResultIterator extends CarbonIterator<Object[]> {
+
+  private CarbonIterator<BatchResult> iterator;
+  private BatchResult batch;
+  private int counter;
+
+  private static final LogService LOGGER =
+      LogServiceFactory.getLogService(PartitionSpliterRawResultIterator.class.getName());
+
+  public PartitionSpliterRawResultIterator(CarbonIterator<BatchResult> iterator) {
+    this.iterator = iterator;
+  }
+
+  @Override public boolean hasNext() {
+    if (null == batch || checkBatchEnd(batch)) {
+      if (iterator.hasNext()) {
+        batch = iterator.next();
+        counter = 0;
+      } else {
+        return false;
+      }
+    }
+
+    return !checkBatchEnd(batch);
+  }
+
+  @Override public Object[] next() {
+    if (batch == null) {
+      batch = iterator.next();
+    }
+    if (!checkBatchEnd(batch)) {
+      return batch.getRawRow(counter++);
+    } else {
+      batch = iterator.next();
+      counter = 0;
+    }
+    return batch.getRawRow(counter++);
+  }
+
+  /**
+   * To check if the batch is processed completely
+   * @param batch
+   * @return
+   */
+  private boolean checkBatchEnd(BatchResult batch) {
+    return !(counter < batch.getSize());
+  }
+
+}

core/src/main/java/org/apache/carbondata/core/scan/wrappers/ByteArrayWrapper.java

@@ -16,13 +16,15 @@
  */
 package org.apache.carbondata.core.scan.wrappers;
 
+import java.io.Serializable;
+
 import org.apache.carbondata.core.util.ByteUtil.UnsafeComparer;
 
 /**
  * This class will store the dimension column data when query is executed
  * This can be used as a key for aggregation
  */
-public class ByteArrayWrapper implements Comparable<ByteArrayWrapper> {
+public class ByteArrayWrapper implements Comparable<ByteArrayWrapper>, Serializable {
 
   /**
    * to store key which is generated using

core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java

@@ -303,6 +303,20 @@ public String getCarbonIndexFilePath(String taskId, String partitionId, String s
     }
   }
 
+  public String getCarbonIndexFilePath(String taskId, String partitionId, String segmentId,
+      int batchNo, String bucketNumber, String timeStamp,
+      ColumnarFormatVersion columnarFormatVersion) {
+    switch (columnarFormatVersion) {
+      case V1:
+      case V2:
+        return getCarbonIndexFilePath(taskId, partitionId, segmentId, bucketNumber);
+      default:
+        String segmentDir = getSegmentDir(partitionId, segmentId);
+        return segmentDir + File.separator + getCarbonIndexFileName(Integer.parseInt(taskId),
+            Integer.parseInt(bucketNumber), batchNo, timeStamp);
+    }
+  }
+
   private static String getCarbonIndexFileName(String taskNo, int bucketNumber,
       String factUpdatedtimeStamp) {
     return taskNo + "-" + bucketNumber + "-" + factUpdatedtimeStamp + INDEX_FILE_EXT;
@@ -537,6 +551,10 @@ public static int getTaskIdFromTaskNo(String taskNo) {
       return Integer.parseInt(taskNo.split(BATCH_PREFIX)[0]);
     }
 
+    public static int getBatchNoFromTaskNo(String taskNo) {
+      return Integer.parseInt(taskNo.split(BATCH_PREFIX)[1]);
+    }
+
     /**
      * Gets the file name from file path
      */

examples/spark2/src/main/resources/partition_data.csv

@@ -35,7 +35,7 @@ object CarbonPartitionExample {
     val storeLocation = s"$rootPath/examples/spark2/target/store"
     val warehouse = s"$rootPath/examples/spark2/target/warehouse"
     val metastoredb = s"$rootPath/examples/spark2/target"
-    val testData = s"$rootPath/examples/spark2/src/main/resources/partition_data.csv"
+    val testData = s"$rootPath/integration/spark-common-test/src/test/resources/partition_data.csv"
 
     CarbonProperties.getInstance()
       .addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT, "yyyy/MM/dd")
@@ -51,45 +51,70 @@ object CarbonPartitionExample {
 
     spark.sparkContext.setLogLevel("WARN")
 
-    // none partition table
+    // range partition with bucket defined
     spark.sql("DROP TABLE IF EXISTS t0")
     spark.sql("""
                 | CREATE TABLE IF NOT EXISTS t0
                 | (
+                | id Int,
                 | vin String,
-                | logdate Timestamp,
                 | phonenumber Long,
                 | country String,
-                | area String
+                | area String,
+                | salary Int
                 | )
+                | PARTITIONED BY (logdate Timestamp)
                 | STORED BY 'carbondata'
+                | TBLPROPERTIES('PARTITION_TYPE'='RANGE',
+                | 'RANGE_INFO'='2014/01/01, 2015/01/01, 2016/01/01',
+                | 'BUCKETNUMBER'='3',
+                | 'BUCKETCOLUMNS'='vin')
               """.stripMargin)
 
-    // range partition
+    // none partition table
     spark.sql("DROP TABLE IF EXISTS t1")
     spark.sql("""
                 | CREATE TABLE IF NOT EXISTS t1
                 | (
+                | id Int,
                 | vin String,
+                | logdate Timestamp,
                 | phonenumber Long,
                 | country String,
                 | area String
                 | )
-                | PARTITIONED BY (logdate Timestamp)
                 | STORED BY 'carbondata'
-                | TBLPROPERTIES('PARTITION_TYPE'='RANGE',
-                | 'RANGE_INFO'='2014/01/01, 2015/01/01, 2016/01/01')
+              """.stripMargin)
+
+    // list partition
+    spark.sql("DROP TABLE IF EXISTS t2")
+    spark.sql("""
+                | CREATE TABLE IF NOT EXISTS t2
+                | (
+                | id Int,
+                | vin String,
+                | logdate Timestamp,
+                | phonenumber Long,
+                | country String,
+                | salary Int
+                | )
+                | PARTITIONED BY (area String)
+                | STORED BY 'carbondata'
+                | TBLPROPERTIES('PARTITION_TYPE'='LIST',
+                | 'LIST_INFO'='Asia, America, Europe', 'DICTIONARY_EXCLUDE' ='area')
               """.stripMargin)
 
     // hash partition
     spark.sql("DROP TABLE IF EXISTS t3")
     spark.sql("""
                 | CREATE TABLE IF NOT EXISTS t3
                 | (
+                | id Int,
                 | logdate Timestamp,
                 | phonenumber Long,
                 | country String,
-                | area String
+                | area String,
+                | salary Int
                 | )
                 | PARTITIONED BY (vin String)
                 | STORED BY 'carbondata'
@@ -101,17 +126,40 @@ object CarbonPartitionExample {
     spark.sql("""
        | CREATE TABLE IF NOT EXISTS t5
        | (
+       | id Int,
        | vin String,
        | logdate Timestamp,
        | phonenumber Long,
-       | area String
+       | area String,
+       | salary Int
        |)
        | PARTITIONED BY (country String)
        | STORED BY 'carbondata'
        | TBLPROPERTIES('PARTITION_TYPE'='LIST',
-       | 'LIST_INFO'='(China,United States),UK ,japan,(Canada,Russia), South Korea ')
+       | 'LIST_INFO'='(China, US),UK ,Japan,(Canada,Russia, Good, NotGood), Korea ')
        """.stripMargin)
 
+    // load data into partition table
+    spark.sql(s"""
+       LOAD DATA LOCAL INPATH '$testData' into table t0 options('BAD_RECORDS_ACTION'='FORCE')
+       """)
+    spark.sql(s"""
+       LOAD DATA LOCAL INPATH '$testData' into table t5 options('BAD_RECORDS_ACTION'='FORCE')
+       """)
+
+    // alter list partition table t5 to add a partition
+    spark.sql(s"""Alter table t5 add partition ('OutSpace')""".stripMargin)
+    // alter list partition table t5 to split partition 4 into 3 independent partition
+    spark.sql(
+      s"""
+         Alter table t5 split partition(4) into ('Canada', 'Russia', '(Good, NotGood)')
+       """.stripMargin)
+
+    spark.sql("""select * from t5 where country = 'Good' """).show(100, false)
+
+    spark.sql("select * from t0 order by salary ").show(100, false)
+    spark.sql("select * from t5 order by salary ").show(100, false)
+
     // hive partition table
     spark.sql("DROP TABLE IF EXISTS t7")
     spark.sql("""
@@ -130,6 +178,7 @@ object CarbonPartitionExample {
     spark.sql(s"CREATE DATABASE partitionDB")
     spark.sql(s"""
                 | CREATE TABLE IF NOT EXISTS partitionDB.t9(
+                | id Int,
                 | logdate Timestamp,
                 | phonenumber Int,
                 | country String,
@@ -145,11 +194,11 @@ object CarbonPartitionExample {
 
     // show partitions
     try {
-      spark.sql("""SHOW PARTITIONS t0""").show(100, false)
+      spark.sql("""SHOW PARTITIONS t1""").show(100, false)
     } catch {
       case ex: AnalysisException => LOGGER.error(ex.getMessage())
     }
-    spark.sql("""SHOW PARTITIONS t1""").show(100, false)
+    spark.sql("""SHOW PARTITIONS t0""").show(100, false)
     spark.sql("""SHOW PARTITIONS t3""").show(100, false)
     spark.sql("""SHOW PARTITIONS t5""").show(100, false)
     spark.sql("""SHOW PARTITIONS t7""").show(100, false)