carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ravipes...@apache.org
Subject [25/26] carbondata git commit: [CARBONDATA-3138] Fix random count mismatch with multi-thread block pruning
Date Fri, 30 Nov 2018 16:34:18 GMT
[CARBONDATA-3138] Fix random count mismatch with multi-thread block pruning

problem: Random count mismatch in query in multi-thread block-pruning scenario.

cause: Existing prune method not meant for multi-threading as synchronization was missing.
only in implicit filter scenario, while preparing the block ID list, synchronization was missing.
Hence pruning was giving wrong result.

solution: synchronize the implicit filter preparation, as prune now called in multi-thread

This closes #2962


Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/ceb75fe1
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/ceb75fe1
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/ceb75fe1

Branch: refs/heads/branch-1.5
Commit: ceb75fe188bdcf54897781805ca5f421bd94d797
Parents: 5dd242b
Author: ajantha-bhat <ajanthabhat@gmail.com>
Authored: Wed Nov 28 19:18:16 2018 +0530
Committer: ravipesala <ravi.pesala@gmail.com>
Committed: Fri Nov 30 21:57:21 2018 +0530

----------------------------------------------------------------------
 .../carbondata/core/datamap/TableDataMap.java    | 13 +++++++++++--
 .../core/scan/filter/ColumnFilterInfo.java       | 19 +++++++++++++------
 2 files changed, 24 insertions(+), 8 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/carbondata/blob/ceb75fe1/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java b/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
index e1b2c13..06d2cab 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
@@ -145,6 +145,7 @@ public final class TableDataMap extends OperationEventListener {
     // for filter queries
     int totalFiles = 0;
     int datamapsCount = 0;
+    int filesCountPerDatamap;
     boolean isBlockDataMapType = true;
     for (Segment segment : segments) {
       for (DataMap dataMap : dataMaps.get(segment)) {
@@ -152,7 +153,9 @@ public final class TableDataMap extends OperationEventListener {
           isBlockDataMapType = false;
           break;
         }
-        totalFiles += ((BlockDataMap) dataMap).getTotalBlocks();
+        filesCountPerDatamap = ((BlockDataMap) dataMap).getTotalBlocks();
+        // old legacy store can give 0, so consider one datamap as 1 record.
+        totalFiles += (filesCountPerDatamap == 0) ? 1 : filesCountPerDatamap;
         datamapsCount++;
       }
       if (!isBlockDataMapType) {
@@ -206,10 +209,14 @@ public final class TableDataMap extends OperationEventListener {
       List<ExtendedBlocklet> blocklets, final Map<Segment, List<DataMap>>
dataMaps,
       int totalFiles) {
     int numOfThreadsForPruning = getNumOfThreadsForPruning();
+    LOG.info(
+        "Number of threads selected for multi-thread block pruning is " + numOfThreadsForPruning
+            + ". total files: " + totalFiles + ". total segments: " + segments.size());
     int filesPerEachThread = totalFiles / numOfThreadsForPruning;
     int prev;
     int filesCount = 0;
     int processedFileCount = 0;
+    int filesCountPerDatamap;
     List<List<SegmentDataMapGroup>> segmentList = new ArrayList<>(numOfThreadsForPruning);
     List<SegmentDataMapGroup> segmentDataMapGroupList = new ArrayList<>();
     for (Segment segment : segments) {
@@ -217,7 +224,9 @@ public final class TableDataMap extends OperationEventListener {
       prev = 0;
       for (int i = 0; i < eachSegmentDataMapList.size(); i++) {
         DataMap dataMap = eachSegmentDataMapList.get(i);
-        filesCount += ((BlockDataMap) dataMap).getTotalBlocks();
+        filesCountPerDatamap = ((BlockDataMap) dataMap).getTotalBlocks();
+        // old legacy store can give 0, so consider one datamap as 1 record.
+        filesCount += (filesCountPerDatamap == 0) ? 1 : filesCountPerDatamap;
         if (filesCount >= filesPerEachThread) {
           if (segmentList.size() != numOfThreadsForPruning - 1) {
             // not the last segmentList

http://git-wip-us.apache.org/repos/asf/carbondata/blob/ceb75fe1/core/src/main/java/org/apache/carbondata/core/scan/filter/ColumnFilterInfo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/ColumnFilterInfo.java
b/core/src/main/java/org/apache/carbondata/core/scan/filter/ColumnFilterInfo.java
index 75ec35e..8677a2d 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/ColumnFilterInfo.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/ColumnFilterInfo.java
@@ -107,19 +107,26 @@ public class ColumnFilterInfo implements Serializable {
   public Set<String> getImplicitDriverColumnFilterList() {
     // this list is required to be populated only n case of driver, so in executor this check
will
     // avoid unnecessary loading of the driver filter list
-    if (null == implicitDriverColumnFilterList) {
-      populateBlockIdListForDriverBlockPruning();
+    if (implicitDriverColumnFilterList != null) {
+      return implicitDriverColumnFilterList;
+    }
+    synchronized (this) {
+      if (null == implicitDriverColumnFilterList) {
+        // populate only once. (can be called in multi-thread)
+        implicitDriverColumnFilterList = populateBlockIdListForDriverBlockPruning();
+      }
     }
     return implicitDriverColumnFilterList;
   }
 
-  private void populateBlockIdListForDriverBlockPruning() {
-    implicitDriverColumnFilterList = new HashSet<>(implicitColumnFilterList.size());
-    String blockId = null;
+  private Set<String> populateBlockIdListForDriverBlockPruning() {
+    Set<String> columnFilterList = new HashSet<>(implicitColumnFilterList.size());
+    String blockId;
     for (String blockletId : implicitColumnFilterList) {
       blockId =
           blockletId.substring(0, blockletId.lastIndexOf(CarbonCommonConstants.FILE_SEPARATOR));
-      implicitDriverColumnFilterList.add(blockId);
+      columnFilterList.add(blockId);
     }
+    return columnFilterList;
   }
 }


Mime
View raw message