[GitHub] [carbondata] niuge01 opened a new pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table

[GitBox]

niuge01 opened a new pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542
 
 
   This feature is actually implemented by the CarbonDataLoadCommand.
   
    - [ ] Any interfaces changed?
   NO
   
    - [ ] Any backward compatibility impacted?
    NO
   
    - [ ] Document update required?
   NA
   
    - [ ] Testing done
           Please provide details on
           - Whether new unit test cases have been added or why no new tests are required?
           - How it is tested? Please attach test report.
           - Is it a performance related change? Please attach the performance test report.
           - Any additional information to help reviewers in testing this change.
     YES
    - [ ] For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.
   
   

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

niuge01 commented on issue #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#issuecomment-569423869
 
 
   please test this

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

CarbonDataQA1 commented on issue #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#issuecomment-569426381
 
 
   Build Success with Spark 2.1.0, Please check CI http://121.244.95.60:12545/job/ApacheCarbonPRBuilder2.1/1322/
   

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

CarbonDataQA1 commented on issue #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#issuecomment-569430944
 
 
   Build Success with Spark 2.3.4, Please check CI http://121.244.95.60:12545/job/ApacheCarbonPRBuilder2.3/1345/
   

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

CarbonDataQA1 commented on issue #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#issuecomment-569436583
 
 
   Build Failed with Spark 2.2.1, Please check CI http://121.244.95.60:12545/job/ApacheCarbonPRBuilder2.2/1332/
   

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

jackylk commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361827544
 
 

 ##########
 File path: integration/flink/src/test/scala/org/apache/carbon/flink/TestCarbonPartitionWriter.scala
 ##########
 @@ -189,4 +194,20 @@ class TestCarbonPartitionWriter extends QueryTest {
     output.asScala
   }
 
+  private def delDir(dir: File): Boolean = {
+    if (dir.isDirectory) {
+      val children = dir.list
+      if (children != null) {
+        var i = 0
+        while (i < children.length) {
 
 Review comment:
   declare a val like `val len = children.length`

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

jackylk commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361827632
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
 ##########
 @@ -269,11 +281,90 @@ case class CarbonInsertFromStageCommand(
       SparkSQLUtil.sessionState(spark).newHadoopConf()
     ).map { row =>
         (row._1, FailureCauses.NONE == row._2._2.failureCauses)
-      }
+    }
+
+    LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
+  }
+
+  /**
+   * Start global sort loading of partition table
+   */
+  private def startLoadingWithPartition(
+      spark: SparkSession,
+      table: CarbonTable,
+      loadModel: CarbonLoadModel,
+      stageInput: Seq[StageInput]
+    ): Unit = {
+    val partitionDataList = listPartitionFiles(stageInput)
+    val start = System.currentTimeMillis()
+    var index = 0
+    partitionDataList.map(
 
 Review comment:
   formatted like:
   ```
   partitoinDataList.map { partitionData =>
       ...
   }
   

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

jackylk commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361827656
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
 ##########
 @@ -269,11 +281,90 @@ case class CarbonInsertFromStageCommand(
       SparkSQLUtil.sessionState(spark).newHadoopConf()
     ).map { row =>
         (row._1, FailureCauses.NONE == row._2._2.failureCauses)
-      }
+    }
+
+    LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
+  }
+
+  /**
+   * Start global sort loading of partition table
+   */
+  private def startLoadingWithPartition(
+      spark: SparkSession,
+      table: CarbonTable,
+      loadModel: CarbonLoadModel,
+      stageInput: Seq[StageInput]
+    ): Unit = {
+    val partitionDataList = listPartitionFiles(stageInput)
+    val start = System.currentTimeMillis()
+    var index = 0
+    partitionDataList.map(
+      partitionData => {
+        index = index + 1
+        val partition = partitionData._1
+        val splits = partitionData._2
+        LOGGER.info(s"start to load ${splits.size} files into " +
+          s"${table.getDatabaseName}.${table.getTableName}")
+        val dataFrame = createInputDataFrameOfInternalRow(spark, table, splits)
+        val columns = dataFrame.columns
+        val header = columns.mkString(",")
+        val selectColumns = columns.filter(!partition.contains(_))
+        val selectedDataFrame = dataFrame.select(selectColumns.head, selectColumns.tail : _*)
 
+        val loadCommand = CarbonLoadDataCommand(
+          databaseNameOp = Option(table.getDatabaseName),
+          tableName = table.getTableName,
+          factPathFromUser = null,
+          dimFilesPath = Seq(),
+          options = scala.collection.immutable.Map("fileheader" -> header),
+          isOverwriteTable = false,
+          inputSqlString = null,
+          dataFrame = Some(selectedDataFrame),
+          updateModel = None,
+          tableInfoOp = None,
+          internalOptions = Map.empty,
+          partition = partition
+        )
+        loadCommand.run(spark)
+      }
+    )
     LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
   }
 
+  private def listPartitionFiles(stageInputs : Seq[StageInput]):
 
 Review comment:
   move `:` to next line

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

jackylk commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361827670
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
 ##########
 @@ -269,11 +281,90 @@ case class CarbonInsertFromStageCommand(
       SparkSQLUtil.sessionState(spark).newHadoopConf()
     ).map { row =>
         (row._1, FailureCauses.NONE == row._2._2.failureCauses)
-      }
+    }
+
+    LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
+  }
+
+  /**
+   * Start global sort loading of partition table
+   */
+  private def startLoadingWithPartition(
+      spark: SparkSession,
+      table: CarbonTable,
+      loadModel: CarbonLoadModel,
+      stageInput: Seq[StageInput]
+    ): Unit = {
+    val partitionDataList = listPartitionFiles(stageInput)
+    val start = System.currentTimeMillis()
+    var index = 0
+    partitionDataList.map(
+      partitionData => {
+        index = index + 1
+        val partition = partitionData._1
+        val splits = partitionData._2
+        LOGGER.info(s"start to load ${splits.size} files into " +
+          s"${table.getDatabaseName}.${table.getTableName}")
+        val dataFrame = createInputDataFrameOfInternalRow(spark, table, splits)
+        val columns = dataFrame.columns
+        val header = columns.mkString(",")
+        val selectColumns = columns.filter(!partition.contains(_))
+        val selectedDataFrame = dataFrame.select(selectColumns.head, selectColumns.tail : _*)
 
+        val loadCommand = CarbonLoadDataCommand(
+          databaseNameOp = Option(table.getDatabaseName),
+          tableName = table.getTableName,
+          factPathFromUser = null,
+          dimFilesPath = Seq(),
+          options = scala.collection.immutable.Map("fileheader" -> header),
+          isOverwriteTable = false,
+          inputSqlString = null,
+          dataFrame = Some(selectedDataFrame),
+          updateModel = None,
+          tableInfoOp = None,
+          internalOptions = Map.empty,
+          partition = partition
+        )
+        loadCommand.run(spark)
+      }
+    )
     LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
   }
 
+  private def listPartitionFiles(stageInputs : Seq[StageInput]):
 
 Review comment:
   describe the return tuple in comment

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

jackylk commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361827670
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
 ##########
 @@ -269,11 +281,90 @@ case class CarbonInsertFromStageCommand(
       SparkSQLUtil.sessionState(spark).newHadoopConf()
     ).map { row =>
         (row._1, FailureCauses.NONE == row._2._2.failureCauses)
-      }
+    }
+
+    LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
+  }
+
+  /**
+   * Start global sort loading of partition table
+   */
+  private def startLoadingWithPartition(
+      spark: SparkSession,
+      table: CarbonTable,
+      loadModel: CarbonLoadModel,
+      stageInput: Seq[StageInput]
+    ): Unit = {
+    val partitionDataList = listPartitionFiles(stageInput)
+    val start = System.currentTimeMillis()
+    var index = 0
+    partitionDataList.map(
+      partitionData => {
+        index = index + 1
+        val partition = partitionData._1
+        val splits = partitionData._2
+        LOGGER.info(s"start to load ${splits.size} files into " +
+          s"${table.getDatabaseName}.${table.getTableName}")
+        val dataFrame = createInputDataFrameOfInternalRow(spark, table, splits)
+        val columns = dataFrame.columns
+        val header = columns.mkString(",")
+        val selectColumns = columns.filter(!partition.contains(_))
+        val selectedDataFrame = dataFrame.select(selectColumns.head, selectColumns.tail : _*)
 
+        val loadCommand = CarbonLoadDataCommand(
+          databaseNameOp = Option(table.getDatabaseName),
+          tableName = table.getTableName,
+          factPathFromUser = null,
+          dimFilesPath = Seq(),
+          options = scala.collection.immutable.Map("fileheader" -> header),
+          isOverwriteTable = false,
+          inputSqlString = null,
+          dataFrame = Some(selectedDataFrame),
+          updateModel = None,
+          tableInfoOp = None,
+          internalOptions = Map.empty,
+          partition = partition
+        )
+        loadCommand.run(spark)
+      }
+    )
     LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
   }
 
+  private def listPartitionFiles(stageInputs : Seq[StageInput]):
 
 Review comment:
   describe the return value in comment, otherwise it is not easy to understand

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

jackylk commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361827985
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
 ##########
 @@ -269,11 +281,90 @@ case class CarbonInsertFromStageCommand(
       SparkSQLUtil.sessionState(spark).newHadoopConf()
     ).map { row =>
         (row._1, FailureCauses.NONE == row._2._2.failureCauses)
-      }
+    }
+
+    LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
+  }
+
+  /**
+   * Start global sort loading of partition table
+   */
+  private def startLoadingWithPartition(
+      spark: SparkSession,
+      table: CarbonTable,
+      loadModel: CarbonLoadModel,
+      stageInput: Seq[StageInput]
+    ): Unit = {
+    val partitionDataList = listPartitionFiles(stageInput)
+    val start = System.currentTimeMillis()
+    var index = 0
+    partitionDataList.map(
+      partitionData => {
+        index = index + 1
+        val partition = partitionData._1
+        val splits = partitionData._2
+        LOGGER.info(s"start to load ${splits.size} files into " +
+          s"${table.getDatabaseName}.${table.getTableName}")
 
 Review comment:
   suggest mention it is for which partition in the log

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

jackylk commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361828062
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
 ##########
 @@ -269,11 +281,90 @@ case class CarbonInsertFromStageCommand(
       SparkSQLUtil.sessionState(spark).newHadoopConf()
     ).map { row =>
         (row._1, FailureCauses.NONE == row._2._2.failureCauses)
-      }
+    }
+
+    LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
+  }
+
+  /**
+   * Start global sort loading of partition table
+   */
+  private def startLoadingWithPartition(
+      spark: SparkSession,
+      table: CarbonTable,
+      loadModel: CarbonLoadModel,
+      stageInput: Seq[StageInput]
+    ): Unit = {
+    val partitionDataList = listPartitionFiles(stageInput)
+    val start = System.currentTimeMillis()
+    var index = 0
+    partitionDataList.map(
+      partitionData => {
+        index = index + 1
+        val partition = partitionData._1
+        val splits = partitionData._2
+        LOGGER.info(s"start to load ${splits.size} files into " +
+          s"${table.getDatabaseName}.${table.getTableName}")
+        val dataFrame = createInputDataFrameOfInternalRow(spark, table, splits)
+        val columns = dataFrame.columns
+        val header = columns.mkString(",")
+        val selectColumns = columns.filter(!partition.contains(_))
+        val selectedDataFrame = dataFrame.select(selectColumns.head, selectColumns.tail : _*)
 
+        val loadCommand = CarbonLoadDataCommand(
+          databaseNameOp = Option(table.getDatabaseName),
+          tableName = table.getTableName,
+          factPathFromUser = null,
+          dimFilesPath = Seq(),
+          options = scala.collection.immutable.Map("fileheader" -> header),
+          isOverwriteTable = false,
+          inputSqlString = null,
+          dataFrame = Some(selectedDataFrame),
+          updateModel = None,
+          tableInfoOp = None,
+          internalOptions = Map.empty,
+          partition = partition
+        )
+        loadCommand.run(spark)
+      }
+    )
     LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
   }
 
+  private def listPartitionFiles(stageInputs : Seq[StageInput]):
+  Seq[(Map[String, Option[String]], Seq[InputSplit])] = {
+    val partitionMap = new util.HashMap[Map[String, Option[String]], util.List[InputSplit]]()
+    stageInputs.foreach(
+      stageInput => {
+        val locations = stageInput.getLocations.asScala
+        locations.foreach(
+          location => {
+            val partition = location.getPartitions.asScala.map(t => (t._1, Option(t._2))).toMap
 
 Review comment:
   Why Option is required? Can't we remove it?

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

jackylk commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361828078
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
 ##########
 @@ -794,8 +792,8 @@ case class CarbonLoadDataCommand(
         sparkSession,
         operationContext)
       val logicalPlan = if (sortScope == SortScopeOptions.SortScope.GLOBAL_SORT) {
-        var numPartitions =
-          CarbonDataProcessorUtil.getGlobalSortPartitions(carbonLoadModel.getGlobalSortPartitions)
+        var numPartitions = 1
+        //  CarbonDataProcessorUtil.getGlobalSortPartitions(carbonLoadModel.getGlobalSortPartitions)
 
 Review comment:
   remove it

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

jackylk commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361828090
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
 ##########
 @@ -898,12 +896,32 @@ case class CarbonLoadDataCommand(
       sortScope: SortScopeOptions.SortScope,
       isDataFrame: Boolean): (LogicalPlan, Int, Option[RDD[InternalRow]]) = {
     // Converts the data as per the loading steps before give it to writer or sorter
-    val updatedRdd = convertData(
+    val convertedRdd = convertData(
       rdd,
       sparkSession,
       loadModel,
       isDataFrame,
       partitionValues)
+    val updatedRdd = if (isDataFrame) {
+      val columnCount = loadModel.getCsvHeaderColumns.length
+      convertedRdd.map{ row =>
 
 Review comment:
   add space before `{`

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

jackylk commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361828111
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
 ##########
 @@ -375,5 +466,37 @@ case class CarbonInsertFromStageCommand(
     }
   }
 
+  /**
+   * create DataFrame basing on specified splits
+   */
+  private def createInputDataFrameOfInternalRow(
+      sparkSession: SparkSession,
+      carbonTable: CarbonTable,
+      splits: Seq[InputSplit]
+    ): DataFrame = {
+    val columns = carbonTable
+      .getCreateOrderColumn
+      .asScala
+      .map(_.getColName)
+      .toArray
+    val schema = SparkTypeConverter.createSparkSchema(carbonTable, columns)
+    val rdd: RDD[Row] = new CarbonScanRDD[InternalRow](
+      sparkSession,
+      columnProjection = new CarbonProjection(columns),
+      null,
+      carbonTable.getAbsoluteTableIdentifier,
+      carbonTable.getTableInfo.serialize,
+      carbonTable.getTableInfo,
+      new CarbonInputMetrics,
+      null,
+      null,
+      classOf[SparkRowReadSupportImpl],
+      splits.asJava)
 
 Review comment:
   move `)` to next line

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

jackylk commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361827632
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
 ##########
 @@ -269,11 +281,90 @@ case class CarbonInsertFromStageCommand(
       SparkSQLUtil.sessionState(spark).newHadoopConf()
     ).map { row =>
         (row._1, FailureCauses.NONE == row._2._2.failureCauses)
-      }
+    }
+
+    LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
+  }
+
+  /**
+   * Start global sort loading of partition table
+   */
+  private def startLoadingWithPartition(
+      spark: SparkSession,
+      table: CarbonTable,
+      loadModel: CarbonLoadModel,
+      stageInput: Seq[StageInput]
+    ): Unit = {
+    val partitionDataList = listPartitionFiles(stageInput)
+    val start = System.currentTimeMillis()
+    var index = 0
+    partitionDataList.map(
 
 Review comment:
   formatted like:
   ```
   partitoinDataList.map { case (partitionMap, splits) =>
       ...
   }
   

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

niuge01 commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361839004
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
 ##########
 @@ -269,11 +281,90 @@ case class CarbonInsertFromStageCommand(
       SparkSQLUtil.sessionState(spark).newHadoopConf()
     ).map { row =>
         (row._1, FailureCauses.NONE == row._2._2.failureCauses)
-      }
+    }
+
+    LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
+  }
+
+  /**
+   * Start global sort loading of partition table
+   */
+  private def startLoadingWithPartition(
+      spark: SparkSession,
+      table: CarbonTable,
+      loadModel: CarbonLoadModel,
+      stageInput: Seq[StageInput]
+    ): Unit = {
+    val partitionDataList = listPartitionFiles(stageInput)
+    val start = System.currentTimeMillis()
+    var index = 0
+    partitionDataList.map(
 
 Review comment:
   done

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

niuge01 commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361839061
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
 ##########
 @@ -269,11 +281,90 @@ case class CarbonInsertFromStageCommand(
       SparkSQLUtil.sessionState(spark).newHadoopConf()
     ).map { row =>
         (row._1, FailureCauses.NONE == row._2._2.failureCauses)
-      }
+    }
+
+    LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
+  }
+
+  /**
+   * Start global sort loading of partition table
+   */
+  private def startLoadingWithPartition(
+      spark: SparkSession,
+      table: CarbonTable,
+      loadModel: CarbonLoadModel,
+      stageInput: Seq[StageInput]
+    ): Unit = {
+    val partitionDataList = listPartitionFiles(stageInput)
+    val start = System.currentTimeMillis()
+    var index = 0
+    partitionDataList.map(
+      partitionData => {
+        index = index + 1
+        val partition = partitionData._1
+        val splits = partitionData._2
+        LOGGER.info(s"start to load ${splits.size} files into " +
+          s"${table.getDatabaseName}.${table.getTableName}")
 
 Review comment:
   done

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

niuge01 commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361839104
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
 ##########
 @@ -269,11 +281,90 @@ case class CarbonInsertFromStageCommand(
       SparkSQLUtil.sessionState(spark).newHadoopConf()
     ).map { row =>
         (row._1, FailureCauses.NONE == row._2._2.failureCauses)
-      }
+    }
+
+    LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
+  }
+
+  /**
+   * Start global sort loading of partition table
+   */
+  private def startLoadingWithPartition(
+      spark: SparkSession,
+      table: CarbonTable,
+      loadModel: CarbonLoadModel,
+      stageInput: Seq[StageInput]
+    ): Unit = {
+    val partitionDataList = listPartitionFiles(stageInput)
+    val start = System.currentTimeMillis()
+    var index = 0
+    partitionDataList.map(
+      partitionData => {
+        index = index + 1
+        val partition = partitionData._1
+        val splits = partitionData._2
+        LOGGER.info(s"start to load ${splits.size} files into " +
+          s"${table.getDatabaseName}.${table.getTableName}")
+        val dataFrame = createInputDataFrameOfInternalRow(spark, table, splits)
+        val columns = dataFrame.columns
+        val header = columns.mkString(",")
+        val selectColumns = columns.filter(!partition.contains(_))
+        val selectedDataFrame = dataFrame.select(selectColumns.head, selectColumns.tail : _*)
 
+        val loadCommand = CarbonLoadDataCommand(
+          databaseNameOp = Option(table.getDatabaseName),
+          tableName = table.getTableName,
+          factPathFromUser = null,
+          dimFilesPath = Seq(),
+          options = scala.collection.immutable.Map("fileheader" -> header),
+          isOverwriteTable = false,
+          inputSqlString = null,
+          dataFrame = Some(selectedDataFrame),
+          updateModel = None,
+          tableInfoOp = None,
+          internalOptions = Map.empty,
+          partition = partition
+        )
+        loadCommand.run(spark)
+      }
+    )
     LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")
   }
 
+  private def listPartitionFiles(stageInputs : Seq[StageInput]):
+  Seq[(Map[String, Option[String]], Seq[InputSplit])] = {
+    val partitionMap = new util.HashMap[Map[String, Option[String]], util.List[InputSplit]]()
+    stageInputs.foreach(
+      stageInput => {
+        val locations = stageInput.getLocations.asScala
+        locations.foreach(
+          location => {
+            val partition = location.getPartitions.asScala.map(t => (t._1, Option(t._2))).toMap
 
 Review comment:
   The CarbonLoadDataCommand required Option, so we cann't remove it here.

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

niuge01 commented on a change in pull request #3542: [CARBONDATA-3640] Insert from stage command support partition table
URL: https://github.com/apache/carbondata/pull/3542#discussion_r361839109
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
 ##########
 @@ -794,8 +792,8 @@ case class CarbonLoadDataCommand(
         sparkSession,
         operationContext)
       val logicalPlan = if (sortScope == SortScopeOptions.SortScope.GLOBAL_SORT) {
-        var numPartitions =
-          CarbonDataProcessorUtil.getGlobalSortPartitions(carbonLoadModel.getGlobalSortPartitions)
+        var numPartitions = 1
+        //  CarbonDataProcessorUtil.getGlobalSortPartitions(carbonLoadModel.getGlobalSortPartitions)
 
 Review comment:
   done

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services