carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jack...@apache.org
Subject [carbondata] branch master updated: [CARBONDATA-3893] [IUD] Fix getting block name in compacted segment with dot for horizontal compaction delta files
Date Sun, 12 Jul 2020 17:04:23 GMT
This is an automated email from the ASF dual-hosted git repository.

jackylk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/carbondata.git


The following commit(s) were added to refs/heads/master by this push:
     new 3b236a3  [CARBONDATA-3893] [IUD] Fix getting block name in compacted segment with
dot for horizontal compaction delta files
3b236a3 is described below

commit 3b236a364c7feba311b56a82f89737b0416c05b9
Author: Zhangshunyu <zhangshunyu1990@126.com>
AuthorDate: Thu Jul 9 11:03:32 2020 +0800

    [CARBONDATA-3893] [IUD] Fix getting block name in compacted segment with dot for horizontal
compaction delta files
    
    Why is this PR needed?
    Currentlly, during horizontal compaction, driver will list delta files in segment to merge
them, during this comparasion it will use block name, current code only consider the blocks
in normal segment, but not consider the compacted segement, for example, in XXXX-12.2-time.deletedelta
the block name should be XXXX-12.2 not XXXX, if use XXX it will not list the delta files and
horizontal compaction will not merge delta files.
    
    What changes were proposed in this PR?
    Correct the parse of block name
    
    Does this PR introduce any user interface change?
    No
    
    Is any new testcase added?
    No
    
    This closes #3832
---
 .../carbondata/core/mutate/CarbonUpdateUtil.java   |  23 ++--
 .../statusmanager/SegmentUpdateStatusManager.java  |   7 +-
 .../spark/testsuite/merge/MergeTestCase.scala      | 152 ++++++++++++++++++++-
 3 files changed, 167 insertions(+), 15 deletions(-)

diff --git a/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java b/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
index c717117..e915c66 100644
--- a/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
@@ -471,6 +471,15 @@ public class CarbonUpdateUtil {
     LoadMetadataDetails[] details =
         SegmentStatusManager.readLoadMetadata(table.getMetadataPath());
 
+    SegmentUpdateStatusManager updateStatusManager = new SegmentUpdateStatusManager(table);
+    SegmentUpdateDetails[] segmentUpdateDetails = updateStatusManager.getUpdateStatusDetails();
+    // hold all the segments updated so that wen can check the delta files in them, ne need
to
+    // check the others.
+    Set<String> updatedSegments = new HashSet<>();
+    for (SegmentUpdateDetails updateDetails : segmentUpdateDetails) {
+      updatedSegments.add(updateDetails.getSegmentName());
+    }
+
     String validUpdateStatusFile = "";
 
     boolean isAbortedFile = true;
@@ -479,22 +488,18 @@ public class CarbonUpdateUtil {
 
     List<Segment> segmentFilesToBeUpdated = new ArrayList<>();
 
+    // take the update status file name from 0th segment.
+    validUpdateStatusFile = ssm.getUpdateStatusFileName(details);
     // scan through each segment.
-
     for (LoadMetadataDetails segment : details) {
-
-      // take the update status file name from 0th segment.
-      validUpdateStatusFile = ssm.getUpdateStatusFileName(details);
-
       // if this segment is valid then only we will go for delta file deletion.
       // if the segment is mark for delete or compacted then any way it will get deleted.
-
       if (segment.getSegmentStatus() == SegmentStatus.SUCCESS
               || segment.getSegmentStatus() == SegmentStatus.LOAD_PARTIAL_SUCCESS) {
-
         // when there is no update operations done on table, then no need to go ahead. So
         // just check the update delta start timestamp and proceed if not empty
-        if (!segment.getUpdateDeltaStartTimestamp().isEmpty()) {
+        if (!segment.getUpdateDeltaStartTimestamp().isEmpty()
+                || updatedSegments.contains(segment.getLoadName())) {
           // take the list of files from this segment.
           String segmentPath = CarbonTablePath.getSegmentPath(
               table.getAbsoluteTableIdentifier().getTablePath(), segment.getLoadName());
@@ -503,8 +508,6 @@ public class CarbonUpdateUtil {
           CarbonFile[] allSegmentFiles = segDir.listFiles();
 
           // scan through the segment and find the carbondatafiles and index files.
-          SegmentUpdateStatusManager updateStatusManager = new SegmentUpdateStatusManager(table);
-
           boolean updateSegmentFile = false;
           // deleting of the aborted file scenario.
           if (deleteStaleCarbonDataFiles(segment, allSegmentFiles, updateStatusManager))
{
diff --git a/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
index cd83457..d547c3d 100644
--- a/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
+++ b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
@@ -448,10 +448,9 @@ public class SegmentUpdateStatusManager {
           @Override
           public boolean accept(CarbonFile pathName) {
             String fileName = pathName.getName();
-            if (fileName.endsWith(CarbonCommonConstants.DELETE_DELTA_FILE_EXT)
-                && pathName.getSize() > 0) {
-              String firstPart = fileName.substring(0, fileName.indexOf('.'));
-              String blkName = firstPart.substring(0, firstPart.lastIndexOf("-"));
+            if (pathName.getSize() > 0
+                && fileName.endsWith(CarbonCommonConstants.DELETE_DELTA_FILE_EXT))
{
+              String blkName = fileName.substring(0, fileName.lastIndexOf("-"));
               long timestamp =
                   Long.parseLong(CarbonTablePath.DataFileUtil.getTimeStampFromFileName(fileName));
               if (blockName.equals(blkName) && (Long.compare(timestamp, deltaEndTimeStamp)
<= 0)
diff --git a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/merge/MergeTestCase.scala
b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/merge/MergeTestCase.scala
index 6ea702d..d068507 100644
--- a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/merge/MergeTestCase.scala
+++ b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/merge/MergeTestCase.scala
@@ -17,12 +17,16 @@
 
 package org.apache.carbondata.spark.testsuite.merge
 
-import scala.collection.JavaConverters._
+import java.io.File
 import java.sql.Date
+import java.time.LocalDateTime
 
+import scala.collection.JavaConverters._
+import scala.util.Random
 import org.apache.carbondata.core.constants.CarbonCommonConstants
 import org.apache.carbondata.core.datastore.filesystem.{CarbonFile, CarbonFileFilter}
 import org.apache.carbondata.core.datastore.impl.FileFactory
+import org.apache.carbondata.core.util.CarbonProperties
 import org.apache.carbondata.core.util.path.CarbonTablePath
 import org.apache.spark.sql._
 import org.apache.spark.sql.CarbonSession._
@@ -721,6 +725,150 @@ class MergeTestCase extends QueryTest with BeforeAndAfterAll {
     checkAnswer(sql("select * from target order by key"), Seq(Row("c", "200"), Row("d", "3"),
Row("e", "100")))
   }
 
+  case class Target (id: Int, value: String, remark: String, mdt: String)
+  case class Change (id: Int, value: String, change_type: String, mdt: String)
+  private val numInitialRows = 10
+  private val numInsertPerBatch = 2
+  private val numUpdatePerBatch = 5
+  private val numDeletePerBatch = 0
+  private val numBatch = 30
+  private val random = new Random()
+  private val values =
+    (1 to 100).map { x =>
+      random.nextString(10)
+    }
+  private def pickValue = values(random.nextInt(values.size))
+  private val currentIds = new java.util.ArrayList[Int](numInitialRows * 2)
+  private def getId(index: Int) = currentIds.get(index)
+  private def getAndRemoveId(index: Int) = currentIds.remove(index)
+  private def addId(id: Int) = currentIds.add(id)
+  private def removeId(index: Int) = currentIds.remove(index)
+  private def numOfIds = currentIds.size
+  private def maxId: Int = currentIds.asScala.max
+  private val INSERT = "I"
+  private val UPDATE = "U"
+  private val DELETE = "D"
+
+  private def generateRowsForInsert(sparkSession: SparkSession) = {
+    // data for insert to the target table
+    val insertRows = (maxId + 1 to maxId + numInsertPerBatch).map { x =>
+      addId(x)
+      Change(x, pickValue, INSERT, LocalDateTime.now().toString)
+    }
+    sparkSession.createDataFrame(insertRows)
+  }
+
+  private def generateRowsForDelete(sparkSession: SparkSession) = {
+    val deletedRows = (1 to numDeletePerBatch).map { x =>
+      val idIndex = random.nextInt(numOfIds)
+      Change(getAndRemoveId(idIndex), "", DELETE, LocalDateTime.now().toString)
+    }
+    sparkSession.createDataFrame(deletedRows)
+  }
+
+  private def generateRowsForUpdate(sparkSession: SparkSession) = {
+    val updatedRows = (1 to numUpdatePerBatch).map { x =>
+      val idIndex = random.nextInt(numOfIds)
+      Change(getId(idIndex), pickValue, UPDATE, LocalDateTime.now().toString)
+    }
+    sparkSession.createDataFrame(updatedRows)
+  }
+
+  // generate initial data for target table
+  private def generateTarget(sparkSession: SparkSession): Unit = {
+    val time = timeIt { () =>
+      val insertRows = (1 to numInitialRows).map { x =>
+        addId(x)
+        Target(x, pickValue, "origin", LocalDateTime.now().toString)
+      }
+      val targetData = sparkSession.createDataFrame(insertRows)
+      targetData.repartition(8)
+        .write
+        .format("carbondata")
+        .option("tableName", "target")
+        .option("sort_scope", "global_sort")
+        .option("sort_column", "id")
+        .mode(SaveMode.Overwrite)
+        .save()
+    }
+  }
+
+  // generate change data
+  private def generateChange(sparkSession: SparkSession): Unit = {
+    val update = generateRowsForUpdate(sparkSession)
+    val delete = generateRowsForDelete(sparkSession)
+    val insert = generateRowsForInsert(sparkSession)
+    // union them so that the change contains IUD
+    update
+      .union(delete)
+      .union(insert)
+      .repartition(8)
+      .write
+      .format("carbondata")
+      .option("tableName", "change")
+      .mode(SaveMode.Overwrite)
+      .save()
+  }
+
+  private def readTargetData(sparkSession: SparkSession): Dataset[Row] =
+    sparkSession.read
+      .format("carbondata")
+      .option("tableName", "target")
+      .load()
+
+  private def readChangeData(sparkSession: SparkSession): Dataset[Row] =
+    sparkSession.read
+      .format("carbondata")
+      .option("tableName", "change")
+      .load()
+
+  private def timeIt(func: () => Unit): Long = {
+    val start = System.nanoTime()
+    func()
+    System.nanoTime() - start
+  }
+
+  test("test cdc with compaction") {
+    CarbonProperties.getInstance().addProperty("carbon.enable.auto.load.merge", "true")
+    sql("drop table if exists target")
+    sql("drop table if exists change")
+    // prepare target table
+    generateTarget(sqlContext.sparkSession)
+    // Do CDC for N batch
+    (1 to numBatch).foreach { i =>
+      // prepare for change data
+      generateChange(sqlContext.sparkSession)
+      // apply change to target table
+      import org.apache.spark.sql.CarbonSession._
+
+      // find the latest value for each key
+      val latestChangeForEachKey = readChangeData(sqlContext.sparkSession)
+        .selectExpr("id", "struct(mdt, value, change_type) as otherCols" )
+        .groupBy("id")
+        .agg(max("otherCols").as("latest"))
+        .selectExpr("id", "latest.*")
+
+      val target = readTargetData(sqlContext.sparkSession)
+      target.as("A")
+        .merge(latestChangeForEachKey.as("B"), "A.id = B.id")
+        .whenMatched("B.change_type = 'D'")
+        .delete()
+        .whenMatched("B.change_type = 'U'")
+        .updateExpr(
+          Map("id" -> "B.id", "value" -> "B.value", "remark" -> "'updated'", "mdt"
-> "B.mdt"))
+        .whenNotMatched("B.change_type = 'I'")
+        .insertExpr(
+          Map("id" -> "B.id", "value" -> "B.value", "remark" -> "'new'", "mdt" ->
"B.mdt"))
+        .execute()
+    }
+
+    sql("clean files for table target").collect()
+    assert(getDeleteDeltaFileCount("target", "12.2") == 1)
+    checkAnswer(sql("select count(*) from target"), Seq(Row(70)))
+
+    CarbonProperties.getInstance().addProperty("carbon.enable.auto.load.merge", "false")
+  }
+
   private def getDeleteDeltaFileCount(tableName: String, segment: String): Int = {
     val table = CarbonEnv.getCarbonTable(None, tableName)(sqlContext.sparkSession)
     val path = CarbonTablePath
@@ -734,5 +882,7 @@ class MergeTestCase extends QueryTest with BeforeAndAfterAll {
 
   override def afterAll {
     sql("drop table if exists order")
+    sql("drop table if exists target")
+    sql("drop table if exists change")
   }
 }


Mime
View raw message