[GitHub] [carbondata] akkio-97 opened a new pull request #3515: resolved error in timestamp during compaction

[GitBox]

akkio-97 opened a new pull request #3515: resolved error in timestamp during compaction
URL: https://github.com/apache/carbondata/pull/3515
 
 
   **Problem**:
   In Carbondata the timestamp is converted into long and stored in RDD.
   While collect(action) is applied on dataframe it gives an error due to the confliction between data types that is - long in carbonScanRDD(from which dataframe is created) and Timestamp in schema.
   
   **Solution**:
   Called Dataset.ofRows(LogicalPlan) which returns the dataframe. Used Set command which sets all segments to be compacted. And Unset once the compaction is finished.
   
   Be sure to do all of the following checklist to help us incorporate
   your contribution quickly and easily:
   
    - [ ] Any interfaces changed?
   
    - [ ] Any backward compatibility impacted?
   
    - [ ] Document update required?
   
    - [ ] 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.
         
    - [ ] 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]

jackylk commented on a change in pull request #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#discussion_r360402747
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
 ##########
 @@ -399,30 +393,24 @@ class CarbonTableCompactor(carbonLoadModel: CarbonLoadModel,
   def dataFrameOfSegments(
       sparkSession: SparkSession,
       carbonTable: CarbonTable,
-      segments: Array[Segment]
-  ): DataFrame = {
-    val columns = carbonTable
-          .getCreateOrderColumn()
-      .asScala
-      .map(_.getColName)
-      .toArray
-    val schema = SparkTypeConverter.createSparkSchema(carbonTable, columns)
-    val rdd: RDD[Row] = new CarbonScanRDD[CarbonRow](
-      sparkSession,
-      columnProjection = new CarbonProjection(columns),
-      null,
-      carbonTable.getAbsoluteTableIdentifier,
-      carbonTable.getTableInfo.serialize,
-      carbonTable.getTableInfo,
-      new CarbonInputMetrics,
-      null,
-      null,
-      classOf[CarbonRowReadSupport],
-      splitsOfSegments(sparkSession, carbonTable, segments))
-      .map { row =>
-        new GenericRow(row.getData.asInstanceOf[Array[Any]])
-      }
-    sparkSession.createDataFrame(rdd, schema)
+      segments: Array[Segment]): DataFrame = {
+    try {
+      CarbonSession
+        .threadSet(CarbonCommonConstants.CARBON_INPUT_SEGMENTS +
+                   carbonTable.getDatabaseName + CarbonCommonConstants.POINT +
+                   carbonTable.getTableName,
+          segments.map(s => s.getSegmentNo).mkString(","))
+      val logicalPlan = sparkSession
+        .sql(s"select * from ${ carbonTable.getTableName }")
 
 Review comment:
   do not construct hard coded SQL, use data frame API

----------------------------------------------------------------
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 #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#discussion_r360404061
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
 ##########
 @@ -399,30 +393,24 @@ class CarbonTableCompactor(carbonLoadModel: CarbonLoadModel,
   def dataFrameOfSegments(
       sparkSession: SparkSession,
       carbonTable: CarbonTable,
-      segments: Array[Segment]
-  ): DataFrame = {
-    val columns = carbonTable
-          .getCreateOrderColumn()
-      .asScala
-      .map(_.getColName)
-      .toArray
-    val schema = SparkTypeConverter.createSparkSchema(carbonTable, columns)
-    val rdd: RDD[Row] = new CarbonScanRDD[CarbonRow](
-      sparkSession,
-      columnProjection = new CarbonProjection(columns),
-      null,
-      carbonTable.getAbsoluteTableIdentifier,
-      carbonTable.getTableInfo.serialize,
-      carbonTable.getTableInfo,
-      new CarbonInputMetrics,
-      null,
-      null,
-      classOf[CarbonRowReadSupport],
-      splitsOfSegments(sparkSession, carbonTable, segments))
-      .map { row =>
-        new GenericRow(row.getData.asInstanceOf[Array[Any]])
-      }
-    sparkSession.createDataFrame(rdd, schema)
+      segments: Array[Segment]): DataFrame = {
+    try {
+      CarbonSession
+        .threadSet(CarbonCommonConstants.CARBON_INPUT_SEGMENTS +
 
 Review comment:
   please add comment to explain why this is needed

----------------------------------------------------------------
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]

ajantha-bhat commented on issue #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#issuecomment-568228346
 
 
   @akkio-97 : I remember this problem was there for date also. please add date data type in the test case also

----------------------------------------------------------------
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]

ajantha-bhat commented on issue #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#issuecomment-568256440
 
 
   Also please rebase

----------------------------------------------------------------
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]

akkio-97 commented on a change in pull request #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#discussion_r361670362
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
 ##########
 @@ -399,30 +393,24 @@ class CarbonTableCompactor(carbonLoadModel: CarbonLoadModel,
   def dataFrameOfSegments(
       sparkSession: SparkSession,
       carbonTable: CarbonTable,
-      segments: Array[Segment]
-  ): DataFrame = {
-    val columns = carbonTable
-          .getCreateOrderColumn()
-      .asScala
-      .map(_.getColName)
-      .toArray
-    val schema = SparkTypeConverter.createSparkSchema(carbonTable, columns)
-    val rdd: RDD[Row] = new CarbonScanRDD[CarbonRow](
-      sparkSession,
-      columnProjection = new CarbonProjection(columns),
-      null,
-      carbonTable.getAbsoluteTableIdentifier,
-      carbonTable.getTableInfo.serialize,
-      carbonTable.getTableInfo,
-      new CarbonInputMetrics,
-      null,
-      null,
-      classOf[CarbonRowReadSupport],
-      splitsOfSegments(sparkSession, carbonTable, segments))
-      .map { row =>
-        new GenericRow(row.getData.asInstanceOf[Array[Any]])
-      }
-    sparkSession.createDataFrame(rdd, schema)
+      segments: Array[Segment]): DataFrame = {
+    try {
+      CarbonSession
+        .threadSet(CarbonCommonConstants.CARBON_INPUT_SEGMENTS +
+                   carbonTable.getDatabaseName + CarbonCommonConstants.POINT +
+                   carbonTable.getTableName,
+          segments.map(s => s.getSegmentNo).mkString(","))
+      val logicalPlan = sparkSession
+        .sql(s"select * from ${ carbonTable.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]

akkio-97 commented on a change in pull request #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#discussion_r361673785
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
 ##########
 @@ -399,30 +393,24 @@ class CarbonTableCompactor(carbonLoadModel: CarbonLoadModel,
   def dataFrameOfSegments(
       sparkSession: SparkSession,
       carbonTable: CarbonTable,
-      segments: Array[Segment]
-  ): DataFrame = {
-    val columns = carbonTable
-          .getCreateOrderColumn()
-      .asScala
-      .map(_.getColName)
-      .toArray
-    val schema = SparkTypeConverter.createSparkSchema(carbonTable, columns)
-    val rdd: RDD[Row] = new CarbonScanRDD[CarbonRow](
-      sparkSession,
-      columnProjection = new CarbonProjection(columns),
-      null,
-      carbonTable.getAbsoluteTableIdentifier,
-      carbonTable.getTableInfo.serialize,
-      carbonTable.getTableInfo,
-      new CarbonInputMetrics,
-      null,
-      null,
-      classOf[CarbonRowReadSupport],
-      splitsOfSegments(sparkSession, carbonTable, segments))
-      .map { row =>
-        new GenericRow(row.getData.asInstanceOf[Array[Any]])
-      }
-    sparkSession.createDataFrame(rdd, schema)
+      segments: Array[Segment]): DataFrame = {
+    try {
+      CarbonSession
+        .threadSet(CarbonCommonConstants.CARBON_INPUT_SEGMENTS +
 
 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]

CarbonDataQA1 commented on issue #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#issuecomment-569282473
 
 
   Build Success with Spark 2.1.0, Please check CI http://121.244.95.60:12545/job/ApacheCarbonPRBuilder2.1/1302/
   

----------------------------------------------------------------
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 #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#issuecomment-569285982
 
 
   Build Failed  with Spark 2.3.4, Please check CI http://121.244.95.60:12545/job/ApacheCarbonPRBuilder2.3/1323/
   

----------------------------------------------------------------
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 #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#issuecomment-569285983
 
 
   Build Failed with Spark 2.2.1, Please check CI http://121.244.95.60:12545/job/ApacheCarbonPRBuilder2.2/1310/
   

----------------------------------------------------------------
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 #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#issuecomment-569294139
 
 
   Build Failed with Spark 2.2.1, Please check CI http://121.244.95.60:12545/job/ApacheCarbonPRBuilder2.2/1312/
   

----------------------------------------------------------------
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 #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#issuecomment-569294657
 
 
   Build Success with Spark 2.3.4, Please check CI http://121.244.95.60:12545/job/ApacheCarbonPRBuilder2.3/1325/
   

----------------------------------------------------------------
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 #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#discussion_r361772487
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
 ##########
 @@ -374,17 +365,32 @@ class CarbonTableCompactor(carbonLoadModel: CarbonLoadModel,
       sparkSession: SparkSession,
       carbonLoadModel: CarbonLoadModel,
       carbonMergerMapping: CarbonMergerMapping): Array[(String, Boolean)] = {
+    val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
     val splits = splitsOfSegments(
       sparkSession,
-      carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable,
+      carbonTable,
       carbonMergerMapping.validSegments)
-    val dataFrame = DataLoadProcessBuilderOnSpark.createInputDataFrame(
-      sparkSession,
-      carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable,
-      splits.asScala)
+    val dataFrame = try {
+      // segments to be compacted are set in the threadset() in carbon session, and unset in the end
 
 Review comment:
   please explain in comment why it is required, not just mentioning what the operation is

----------------------------------------------------------------
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]

akashrn5 commented on a change in pull request #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#discussion_r361808788
 
 

 ##########
 File path: integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala
 ##########
 @@ -442,7 +442,8 @@ object DataLoadProcessBuilderOnSpark {
       .map { row =>
         new GenericRow(row.getData.asInstanceOf[Array[Any]])
 
 Review comment:
   Here rdd creation is not required, please remove and please give the reason why we need to use this API instead of existing one in comments.

----------------------------------------------------------------
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]

akkio-97 commented on a change in pull request #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#discussion_r361925792
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
 ##########
 @@ -374,17 +365,32 @@ class CarbonTableCompactor(carbonLoadModel: CarbonLoadModel,
       sparkSession: SparkSession,
       carbonLoadModel: CarbonLoadModel,
       carbonMergerMapping: CarbonMergerMapping): Array[(String, Boolean)] = {
+    val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
     val splits = splitsOfSegments(
       sparkSession,
-      carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable,
+      carbonTable,
       carbonMergerMapping.validSegments)
-    val dataFrame = DataLoadProcessBuilderOnSpark.createInputDataFrame(
-      sparkSession,
-      carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable,
-      splits.asScala)
+    val dataFrame = try {
+      // segments to be compacted are set in the threadset() in carbon session, and unset in the end
 
 Review comment:
   During custom compaction it might so happen that all the segments might be taken into consideration. To avoid this, segments to be considered and set, are explicitly mentioned 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]

akkio-97 commented on a change in pull request #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#discussion_r361927270
 
 

 ##########
 File path: integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala
 ##########
 @@ -442,7 +442,8 @@ object DataLoadProcessBuilderOnSpark {
       .map { row =>
         new GenericRow(row.getData.asInstanceOf[Array[Any]])
 
 Review comment:
   Due to the confliction between data types that is - long in carbonScanRDD and Timestamp in schema, we cannot use the existing API with rdd.

----------------------------------------------------------------
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 #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#discussion_r361940701
 
 

 ##########
 File path: integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
 ##########
 @@ -374,17 +365,32 @@ class CarbonTableCompactor(carbonLoadModel: CarbonLoadModel,
       sparkSession: SparkSession,
       carbonLoadModel: CarbonLoadModel,
       carbonMergerMapping: CarbonMergerMapping): Array[(String, Boolean)] = {
+    val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
     val splits = splitsOfSegments(
       sparkSession,
-      carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable,
+      carbonTable,
       carbonMergerMapping.validSegments)
-    val dataFrame = DataLoadProcessBuilderOnSpark.createInputDataFrame(
-      sparkSession,
-      carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable,
-      splits.asScala)
+    val dataFrame = try {
+      // segments to be compacted are set in the threadset() in carbon session, and unset in the end
 
 Review comment:
   please add it in the comment in code

----------------------------------------------------------------
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 #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#issuecomment-569654186
 
 
   Build Success with Spark 2.1.0, Please check CI http://121.244.95.60:12545/job/ApacheCarbonPRBuilder2.1/1357/
   

----------------------------------------------------------------
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 #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#issuecomment-569665240
 
 
   Build Failed with Spark 2.2.1, Please check CI http://121.244.95.60:12545/job/ApacheCarbonPRBuilder2.2/1366/
   

----------------------------------------------------------------
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 #3515: [CARBONDATA-3623]: Fixed global sort compaction failure on timestamp column
URL: https://github.com/apache/carbondata/pull/3515#issuecomment-569669234
 
 
   Build Success with Spark 2.3.4, Please check CI http://121.244.95.60:12545/job/ApacheCarbonPRBuilder2.3/1378/
   

----------------------------------------------------------------
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