You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ra...@apache.org on 2017/06/13 03:39:46 UTC
[1/2] carbondata git commit: IUD Performance Changes
Repository: carbondata
Updated Branches:
refs/heads/master 0ad92b6a0 -> 47d07808e
IUD Performance Changes
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/7656ad2a
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/7656ad2a
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/7656ad2a
Branch: refs/heads/master
Commit: 7656ad2a549ddb29dd95bcf7381b7aee16200ff9
Parents: 0ad92b6
Author: sounakr <so...@gmail.com>
Authored: Thu Jun 8 20:28:50 2017 +0530
Committer: sounakr <so...@gmail.com>
Committed: Tue Jun 13 08:28:50 2017 +0530
----------------------------------------------------------------------
.../core/datastore/SegmentTaskIndexStore.java | 27 +++++++++----
.../SegmentUpdateStatusManager.java | 1 +
.../carbondata/hadoop/CarbonInputFormat.java | 37 ++++++++++++-----
.../presto/impl/CarbonTableReader.java | 42 +++++++++++++++-----
.../iud/HorizontalCompactionTestCase.scala | 2 +-
.../carbondata/spark/rdd/CarbonMergerRDD.scala | 10 +++--
.../store/CarbonFactDataHandlerModel.java | 3 +-
7 files changed, 88 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7656ad2a/core/src/main/java/org/apache/carbondata/core/datastore/SegmentTaskIndexStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/SegmentTaskIndexStore.java b/core/src/main/java/org/apache/carbondata/core/datastore/SegmentTaskIndexStore.java
index b1eaa84..734aaaf 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/SegmentTaskIndexStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/SegmentTaskIndexStore.java
@@ -187,19 +187,25 @@ public class SegmentTaskIndexStore
TaskBucketHolder taskBucketHolder = null;
try {
while (iteratorOverSegmentBlocksInfos.hasNext()) {
+ // Initialize the UpdateVO to Null for each segment.
+ UpdateVO updateVO = null;
// segment id to table block mapping
Map.Entry<String, List<TableBlockInfo>> next = iteratorOverSegmentBlocksInfos.next();
// group task id to table block info mapping for the segment
Map<TaskBucketHolder, List<TableBlockInfo>> taskIdToTableBlockInfoMap =
mappedAndGetTaskIdToTableBlockInfo(segmentToTableBlocksInfos);
segmentId = next.getKey();
- // get the existing map of task id to table segment map
- UpdateVO updateVO = updateStatusManager.getInvalidTimestampRange(segmentId);
+ // updateVO is only required when Updates Or Delete performed on the Table.
+ if (updateStatusManager.getUpdateStatusDetails().length != 0) {
+ // get the existing map of task id to table segment map
+ updateVO = updateStatusManager.getInvalidTimestampRange(segmentId);
+ }
// check if segment is already loaded, if segment is already loaded
//no need to load the segment block
String lruCacheKey = tableSegmentUniqueIdentifier.getUniqueTableSegmentIdentifier();
segmentTaskIndexWrapper = (SegmentTaskIndexWrapper) lruCache.get(lruCacheKey);
- if (segmentTaskIndexWrapper == null || tableSegmentUniqueIdentifier.isSegmentUpdated()) {
+ if ((segmentTaskIndexWrapper == null) || ((null != updateVO)
+ && (tableSegmentUniqueIdentifier.isSegmentUpdated()))) {
// get the segment loader lock object this is to avoid
// same segment is getting loaded multiple times
// in case of concurrent query
@@ -210,8 +216,8 @@ public class SegmentTaskIndexStore
// acquire lock to lod the segment
synchronized (segmentLoderLockObject) {
segmentTaskIndexWrapper = (SegmentTaskIndexWrapper) lruCache.get(lruCacheKey);
- if (null == segmentTaskIndexWrapper || tableSegmentUniqueIdentifier
- .isSegmentUpdated()) {
+ if ((null == segmentTaskIndexWrapper) || ((null != updateVO)
+ && (tableSegmentUniqueIdentifier.isSegmentUpdated()))) {
// if the segment is updated then get the existing block task id map details
// so that the same can be updated after loading the btree.
if (tableSegmentUniqueIdentifier.isSegmentUpdated()
@@ -245,9 +251,14 @@ public class SegmentTaskIndexStore
"Can not load the segment. No Enough space available.");
}
- // set the latest timestamp.
- segmentTaskIndexWrapper
- .setRefreshedTimeStamp(updateVO.getCreatedOrUpdatedTimeStamp());
+ // Refresh the Timestamp for those tables which underwent through IUD Operations.
+ if (null != updateVO) {
+ // set the latest timestamp.
+ segmentTaskIndexWrapper
+ .setRefreshedTimeStamp(updateVO.getCreatedOrUpdatedTimeStamp());
+ } else {
+ segmentTaskIndexWrapper.setRefreshedTimeStamp(0L);
+ }
// tableSegmentMapTemp.put(next.getKey(), taskIdToSegmentIndexMap);
// removing from segment lock map as once segment is loaded
// if concurrent query is coming for same segment
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7656ad2a/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
----------------------------------------------------------------------
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 5a93bc5..3e0a2cb 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
@@ -923,6 +923,7 @@ public class SegmentUpdateStatusManager {
range.setLatestUpdateTimestamp(
CarbonUpdateUtil.getTimeStampAsLong(segment.getUpdateDeltaEndTimestamp()));
}
+ return range;
}
}
return range;
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7656ad2a/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonInputFormat.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonInputFormat.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonInputFormat.java
index f5fab10..1e69648 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonInputFormat.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonInputFormat.java
@@ -340,23 +340,37 @@ public class CarbonInputFormat<T> extends FileInputFormat<Void, T> {
List<InputSplit> result = new LinkedList<InputSplit>();
FilterExpressionProcessor filterExpressionProcessor = new FilterExpressionProcessor();
+ UpdateVO invalidBlockVOForSegmentId = null;
+ Boolean isIUDTable = false;
AbsoluteTableIdentifier absoluteTableIdentifier =
getCarbonTable(job.getConfiguration()).getAbsoluteTableIdentifier();
SegmentUpdateStatusManager updateStatusManager =
new SegmentUpdateStatusManager(absoluteTableIdentifier);
+
+ isIUDTable = (updateStatusManager.getUpdateStatusDetails().length != 0);
+
//for each segment fetch blocks matching filter in Driver BTree
for (String segmentNo : getSegmentsToAccess(job)) {
List<DataRefNode> dataRefNodes =
getDataBlocksOfSegment(job, filterExpressionProcessor, absoluteTableIdentifier,
filterResolver, matchedPartitions, segmentNo, cacheClient, updateStatusManager);
+
+ // Get the UpdateVO for those tables on which IUD operations being performed.
+ if (isIUDTable) {
+ invalidBlockVOForSegmentId =
+ updateStatusManager.getInvalidTimestampRange(segmentNo);
+ }
for (DataRefNode dataRefNode : dataRefNodes) {
BlockBTreeLeafNode leafNode = (BlockBTreeLeafNode) dataRefNode;
TableBlockInfo tableBlockInfo = leafNode.getTableBlockInfo();
- if (CarbonUtil.isInvalidTableBlock(tableBlockInfo,
- updateStatusManager.getInvalidTimestampRange(tableBlockInfo.getSegmentId()),
- updateStatusManager)) {
- continue;
+ if (isIUDTable) {
+ // In case IUD is not performed in this table avoid searching for
+ // invalidated blocks.
+ if (CarbonUtil.isInvalidTableBlock(tableBlockInfo, invalidBlockVOForSegmentId,
+ updateStatusManager)) {
+ continue;
+ }
}
String[] deleteDeltaFilePath = null;
try {
@@ -535,16 +549,21 @@ public class CarbonInputFormat<T> extends FileInputFormat<Void, T> {
CacheClient cacheClient, SegmentUpdateStatusManager updateStatusManager) throws IOException {
Map<SegmentTaskIndexStore.TaskBucketHolder, AbstractIndex> segmentIndexMap = null;
SegmentTaskIndexWrapper segmentTaskIndexWrapper = null;
+ UpdateVO updateDetails = null;
boolean isSegmentUpdated = false;
Set<SegmentTaskIndexStore.TaskBucketHolder> taskKeys = null;
TableSegmentUniqueIdentifier tableSegmentUniqueIdentifier =
new TableSegmentUniqueIdentifier(absoluteTableIdentifier, segmentId);
segmentTaskIndexWrapper =
cacheClient.getSegmentAccessClient().getIfPresent(tableSegmentUniqueIdentifier);
- UpdateVO updateDetails = updateStatusManager.getInvalidTimestampRange(segmentId);
+
+ if (updateStatusManager.getUpdateStatusDetails().length != 0) {
+ updateDetails = updateStatusManager.getInvalidTimestampRange(segmentId);
+ }
+
if (null != segmentTaskIndexWrapper) {
segmentIndexMap = segmentTaskIndexWrapper.getTaskIdToTableSegmentMap();
- if (isSegmentUpdate(segmentTaskIndexWrapper, updateDetails)) {
+ if (null != updateDetails && isSegmentUpdate(segmentTaskIndexWrapper, updateDetails)) {
taskKeys = segmentIndexMap.keySet();
isSegmentUpdated = true;
}
@@ -629,9 +648,9 @@ public class CarbonInputFormat<T> extends FileInputFormat<Void, T> {
private boolean isSegmentUpdate(SegmentTaskIndexWrapper segmentTaskIndexWrapper,
UpdateVO updateDetails) {
- if (null != updateDetails.getLatestUpdateTimestamp()
- && updateDetails.getLatestUpdateTimestamp() > segmentTaskIndexWrapper
- .getRefreshedTimeStamp()) {
+ Long refreshedTime = segmentTaskIndexWrapper.getRefreshedTimeStamp();
+ Long updateTimeStamp = updateDetails.getLatestUpdateTimestamp();
+ if (null != refreshedTime && null != updateTimeStamp && updateTimeStamp > refreshedTime) {
return true;
}
return false;
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7656ad2a/integration/presto/src/main/java/org/apache/carbondata/presto/impl/CarbonTableReader.java
----------------------------------------------------------------------
diff --git a/integration/presto/src/main/java/org/apache/carbondata/presto/impl/CarbonTableReader.java b/integration/presto/src/main/java/org/apache/carbondata/presto/impl/CarbonTableReader.java
index 0461603..c328a64 100755
--- a/integration/presto/src/main/java/org/apache/carbondata/presto/impl/CarbonTableReader.java
+++ b/integration/presto/src/main/java/org/apache/carbondata/presto/impl/CarbonTableReader.java
@@ -251,6 +251,8 @@ public class CarbonTableReader {
// need apply filters to segment
FilterExpressionProcessor filterExpressionProcessor = new FilterExpressionProcessor();
+ UpdateVO invalidBlockVOForSegmentId = null;
+ Boolean IUDTable = false;
AbsoluteTableIdentifier absoluteTableIdentifier =
tableCacheModel.carbonTable.getAbsoluteTableIdentifier();
@@ -289,9 +291,17 @@ public class CarbonTableReader {
FilterResolverIntf filterInterface = CarbonInputFormatUtil
.resolveFilter(filters, tableCacheModel.carbonTable.getAbsoluteTableIdentifier());
+ IUDTable = (updateStatusManager.getUpdateStatusDetails().length != 0);
List<CarbonLocalInputSplit> result = new ArrayList<>();
// for each segment fetch blocks matching filter in Driver BTree
for (String segmentNo : tableCacheModel.segments) {
+
+ if (IUDTable) {
+ // update not being performed on this table.
+ invalidBlockVOForSegmentId =
+ updateStatusManager.getInvalidTimestampRange(segmentNo);
+ }
+
try {
List<DataRefNode> dataRefNodes =
getDataBlocksOfSegment(filterExpressionProcessor, absoluteTableIdentifier,
@@ -301,10 +311,11 @@ public class CarbonTableReader {
BlockBTreeLeafNode leafNode = (BlockBTreeLeafNode) dataRefNode;
TableBlockInfo tableBlockInfo = leafNode.getTableBlockInfo();
- if (CarbonUtil.isInvalidTableBlock(tableBlockInfo,
- updateStatusManager.getInvalidTimestampRange(tableBlockInfo.getSegmentId()),
- updateStatusManager)) {
- continue;
+ if (IUDTable) {
+ if (CarbonUtil.isInvalidTableBlock(tableBlockInfo, invalidBlockVOForSegmentId,
+ updateStatusManager)) {
+ continue;
+ }
}
result.add(new CarbonLocalInputSplit(segmentNo, tableBlockInfo.getFilePath(),
tableBlockInfo.getBlockOffset(), tableBlockInfo.getBlockLength(),
@@ -361,9 +372,9 @@ public class CarbonTableReader {
private boolean isSegmentUpdate(SegmentTaskIndexWrapper segmentTaskIndexWrapper,
UpdateVO updateDetails) {
- if (null != updateDetails.getLatestUpdateTimestamp()
- && updateDetails.getLatestUpdateTimestamp() > segmentTaskIndexWrapper
- .getRefreshedTimeStamp()) {
+ Long refreshedTime = segmentTaskIndexWrapper.getRefreshedTimeStamp();
+ Long updateTime = updateDetails.getLatestUpdateTimestamp();
+ if (null != refreshedTime && null != updateTime && updateTime > refreshedTime) {
return true;
}
return false;
@@ -374,16 +385,25 @@ public class CarbonTableReader {
CacheClient cacheClient, SegmentUpdateStatusManager updateStatusManager) throws IOException {
Map<SegmentTaskIndexStore.TaskBucketHolder, AbstractIndex> segmentIndexMap = null;
SegmentTaskIndexWrapper segmentTaskIndexWrapper = null;
+ UpdateVO updateDetails = null;
boolean isSegmentUpdated = false;
Set<SegmentTaskIndexStore.TaskBucketHolder> taskKeys = null;
TableSegmentUniqueIdentifier tableSegmentUniqueIdentifier =
new TableSegmentUniqueIdentifier(absoluteTableIdentifier, segmentId);
segmentTaskIndexWrapper =
cacheClient.getSegmentAccessClient().getIfPresent(tableSegmentUniqueIdentifier);
- UpdateVO updateDetails = updateStatusManager.getInvalidTimestampRange(segmentId);
+
+ // Until Updates or Deletes being performed on the table Invalid Blocks will not
+ // be formed. So it is unnecessary to get the InvalidTimeStampRange.
+ if (updateStatusManager.getUpdateStatusDetails().length != 0) {
+ updateDetails = updateStatusManager.getInvalidTimestampRange(segmentId);
+ }
+
if (null != segmentTaskIndexWrapper) {
segmentIndexMap = segmentTaskIndexWrapper.getTaskIdToTableSegmentMap();
- if (isSegmentUpdate(segmentTaskIndexWrapper, updateDetails)) {
+ // IUD operations should be performed on the table in order to mark the segment as Updated.
+ // For Normal table no need to check for invalided blocks as there will be none of them.
+ if ((null != updateDetails) && isSegmentUpdate(segmentTaskIndexWrapper, updateDetails)) {
taskKeys = segmentIndexMap.keySet();
isSegmentUpdated = true;
}
@@ -413,8 +433,8 @@ public class CarbonTableReader {
List<TableBlockInfo> tableBlockInfoList = new ArrayList<>();
for (FileSplit inputSplit : carbonSplits) {
- if (isValidBlockBasedOnUpdateDetails(taskKeys, inputSplit, updateDetails,
- updateStatusManager, segmentId)) {
+ if ((null == updateDetails) || ((null != updateDetails) && isValidBlockBasedOnUpdateDetails(
+ taskKeys, inputSplit, updateDetails, updateStatusManager, segmentId))) {
BlockletInfos blockletInfos = new BlockletInfos(0, 0,
0);//this level we do not need blocklet info!!!! Is this a trick?
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7656ad2a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/HorizontalCompactionTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/HorizontalCompactionTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/HorizontalCompactionTestCase.scala
index 9c3b261..d8310da 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/HorizontalCompactionTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/HorizontalCompactionTestCase.scala
@@ -53,7 +53,7 @@ class HorizontalCompactionTestCase extends QueryTest with BeforeAndAfterAll {
- test("test IUD Horizontal Compaction Update Alter Clean") {
+ test("test IUD Horizontal Compaction Update Alter Clean.") {
sql("""drop database if exists iud4 cascade""")
sql("""create database iud4""")
sql("""use iud4""")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7656ad2a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala
index 2898870..caa389a 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala
@@ -261,6 +261,7 @@ class CarbonMergerRDD[K, V](
val jobConf: JobConf = new JobConf(new Configuration)
val job: Job = new Job(jobConf)
val format = CarbonInputFormatUtil.createCarbonInputFormat(absoluteTableIdentifier, job)
+ var updateDetails: UpdateVO = null
// initialise query_id for job
job.getConfiguration.set("query.id", queryId)
var defaultParallelism = sparkContext.defaultParallelism
@@ -287,8 +288,11 @@ class CarbonMergerRDD[K, V](
// map for keeping the relation of a task and its blocks.
job.getConfiguration.set(CarbonInputFormat.INPUT_SEGMENT_NUMBERS, eachSeg)
- val updateDetails: UpdateVO = updateStatusManager.getInvalidTimestampRange(eachSeg)
+ if (updateStatusManager.getUpdateStatusDetails.length != 0) {
+ updateDetails = updateStatusManager.getInvalidTimestampRange(eachSeg)
+ }
+ var updated: Boolean = updateStatusManager.getUpdateStatusDetails.length != 0
// get splits
val splits = format.getSplits(job)
@@ -303,8 +307,8 @@ class CarbonMergerRDD[K, V](
entry.getLocations, entry.getLength, entry.getVersion,
updateStatusManager.getDeleteDeltaFilePath(entry.getPath.toString)
)
- !CarbonUtil
- .isInvalidTableBlock(blockInfo, updateDetails, updateStatusManager)
+ ((!updated) || ((updated) && (!CarbonUtil
+ .isInvalidTableBlock(blockInfo, updateDetails, updateStatusManager))))
})
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7656ad2a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
index d400a6d..763274c 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
@@ -297,8 +297,7 @@ public class CarbonFactDataHandlerModel {
.checkAndCreateCarbonStoreLocation(loadModel.getStorePath(), loadModel.getDatabaseName(),
tableName, loadModel.getPartitionId(), loadModel.getSegmentId());
carbonFactDataHandlerModel.setCarbonDataDirectoryPath(carbonDataDirectoryPath);
- List<CarbonDimension> dimensionByTableName =
- loadModel.getCarbonDataLoadSchema().getCarbonTable().getDimensionByTableName(tableName);
+ List<CarbonDimension> dimensionByTableName = carbonTable.getDimensionByTableName(tableName);
boolean[] isUseInvertedIndexes = new boolean[dimensionByTableName.size()];
int index = 0;
for (CarbonDimension dimension : dimensionByTableName) {
[2/2] carbondata git commit: [CARBONDATA-1154] IUD Performance
Changes. This closes #1013
Posted by ra...@apache.org.
[CARBONDATA-1154] IUD Performance Changes. This closes #1013
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/47d07808
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/47d07808
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/47d07808
Branch: refs/heads/master
Commit: 47d07808ec131056d42b87007973e76e1a12e9a8
Parents: 0ad92b6 7656ad2
Author: ravipesala <ra...@gmail.com>
Authored: Tue Jun 13 09:09:21 2017 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Tue Jun 13 09:09:21 2017 +0530
----------------------------------------------------------------------
.../core/datastore/SegmentTaskIndexStore.java | 27 +++++++++----
.../SegmentUpdateStatusManager.java | 1 +
.../carbondata/hadoop/CarbonInputFormat.java | 37 ++++++++++++-----
.../presto/impl/CarbonTableReader.java | 42 +++++++++++++++-----
.../iud/HorizontalCompactionTestCase.scala | 2 +-
.../carbondata/spark/rdd/CarbonMergerRDD.scala | 10 +++--
.../store/CarbonFactDataHandlerModel.java | 3 +-
7 files changed, 88 insertions(+), 34 deletions(-)
----------------------------------------------------------------------