Repository: carbondata
Updated Branches:
  refs/heads/master 7f7ea4d75 -> 65471f2c8


[CARBONDATA-2196] Take CarbonTable from loadmodel during streaming ingestion

This closes #1991


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

Branch: refs/heads/master
Commit: 65471f2c8a9eda8c9f26d78d4640de0f1aa49b4c
Parents: 7f7ea4d
Author: rahulforallp <rahul.ku...@knoldus.in>
Authored: Thu Feb 22 18:29:57 2018 +0530
Committer: QiangCai <qiang...@qq.com>
Committed: Fri Mar 2 16:13:34 2018 +0800

----------------------------------------------------------------------
 .../carbondata/core/datamap/TableDataMap.java   |  6 ++--
 .../hadoop/api/CarbonTableInputFormat.java      |  2 +-
 .../merger/CompactionResultSortProcessor.java   |  6 ++--
 .../util/CarbonDataProcessorUtil.java           | 36 ++++++++++++++++----
 4 files changed, 35 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/carbondata/blob/65471f2c/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 6555d6c..020d6c9 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
@@ -176,16 +176,16 @@ public final class TableDataMap extends 
OperationEventListener {
    * @return
    * @throws IOException
    */
-  public List<String> pruneSegments(List<Segment> segments, FilterResolverIntf 
filterExp)
+  public List<Segment> pruneSegments(List<Segment> segments, 
FilterResolverIntf filterExp)
       throws IOException {
-    List<String> prunedSegments = new 
ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+    List<Segment> prunedSegments = new 
ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
     for (Segment segment : segments) {
       List<DataMap> dataMaps = dataMapFactory.getDataMaps(segment);
       for (DataMap dataMap : dataMaps) {
         if (dataMap.isScanRequired(filterExp)) {
           // If any one task in a given segment contains the data that means 
the segment need to
           // be scanned and we need to validate further data maps in the same 
segment
-          prunedSegments.add(segment.getSegmentNo());
+          prunedSegments.add(segment);
           break;
         }
       }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/65471f2c/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java
----------------------------------------------------------------------
diff --git 
a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java
 
b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java
index 96b0b21..3dbf04f 100644
--- 
a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java
+++ 
b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java
@@ -910,7 +910,7 @@ public class CarbonTableInputFormat<T> extends 
FileInputFormat<Void, T> {
   /**
    * return valid segment to access
    */
-  private Segment[] getSegmentsToAccess(JobContext job) {
+  public Segment[] getSegmentsToAccess(JobContext job) {
     String segmentString = job.getConfiguration().get(INPUT_SEGMENT_NUMBERS, 
"");
     if (segmentString.trim().isEmpty()) {
       return new Segment[0];

http://git-wip-us.apache.org/repos/asf/carbondata/blob/65471f2c/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java
----------------------------------------------------------------------
diff --git 
a/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java
 
b/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java
index 2fbdf4f..e7c4502 100644
--- 
a/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java
+++ 
b/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java
@@ -430,9 +430,7 @@ public class CompactionResultSortProcessor extends 
AbstractResultProcessor {
    */
   private void initTempStoreLocation() {
     tempStoreLocation = CarbonDataProcessorUtil
-        .getLocalDataFolderLocation(carbonLoadModel.getDatabaseName(), 
tableName,
-            carbonLoadModel.getTaskNo(), carbonLoadModel.getPartitionId(), 
segmentId,
-            true, false);
+        .getLocalDataFolderLocation(carbonTable, tableName, 
carbonLoadModel.getTaskNo(),
+            carbonLoadModel.getPartitionId(), segmentId, true, false);
   }
-
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/65471f2c/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
----------------------------------------------------------------------
diff --git 
a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
 
b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
index 1e648e1..cfc9fa3 100644
--- 
a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
+++ 
b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
@@ -117,22 +117,25 @@ public final class CarbonDataProcessorUtil {
       }
     }
   }
+
   /**
+   *
    * This method will form the local data folder store location
    *
-   * @param databaseName
-   * @param tableName
+   * @param carbonTable
    * @param taskId
    * @param partitionId
    * @param segmentId
+   * @param isCompactionFlow
+   * @param isAltPartitionFlow
    * @return
    */
-  public static String[] getLocalDataFolderLocation(String databaseName, 
String tableName,
+  public static String[] getLocalDataFolderLocation(CarbonTable carbonTable, 
String tableName,
       String taskId, String partitionId, String segmentId, boolean 
isCompactionFlow,
       boolean isAltPartitionFlow) {
     String tempLocationKey =
-        getTempStoreLocationKey(databaseName, tableName, segmentId, taskId, 
isCompactionFlow,
-            isAltPartitionFlow);
+        getTempStoreLocationKey(carbonTable.getDatabaseName(), tableName,
+            segmentId, taskId, isCompactionFlow, isAltPartitionFlow);
     String baseTempStorePath = CarbonProperties.getInstance()
         .getProperty(tempLocationKey);
     if (baseTempStorePath == null) {
@@ -145,7 +148,6 @@ public final class CarbonDataProcessorUtil {
     String[] baseTmpStorePathArray = StringUtils.split(baseTempStorePath, 
File.pathSeparator);
     String[] localDataFolderLocArray = new 
String[baseTmpStorePathArray.length];
 
-    CarbonTable carbonTable = 
CarbonMetadata.getInstance().getCarbonTable(databaseName, tableName);
     for (int i = 0 ; i < baseTmpStorePathArray.length; i++) {
       String tmpStore = baseTmpStorePathArray[i];
       CarbonTablePath carbonTablePath =
@@ -159,6 +161,26 @@ public final class CarbonDataProcessorUtil {
   }
 
   /**
+   * This method will form the local data folder store location
+   *
+   * @param databaseName
+   * @param tableName
+   * @param taskId
+   * @param partitionId
+   * @param segmentId
+   * @param isCompactionFlow
+   * @param isAltPartitionFlow
+   * @return
+   */
+  public static String[] getLocalDataFolderLocation(String databaseName, 
String tableName,
+      String taskId, String partitionId, String segmentId, boolean 
isCompactionFlow,
+      boolean isAltPartitionFlow) {
+    CarbonTable carbonTable = 
CarbonMetadata.getInstance().getCarbonTable(databaseName, tableName);
+    return getLocalDataFolderLocation(carbonTable, tableName, taskId, 
partitionId,
+        segmentId, isCompactionFlow, isAltPartitionFlow);
+  }
+
+  /**
    * This method will form the key for getting the temporary location set in 
carbon properties
    *
    * @param databaseName
@@ -587,4 +609,4 @@ public final class CarbonDataProcessorUtil {
     return isRawDataRequired;
   }
 
-}
\ No newline at end of file
+}

Reply via email to