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 2019/08/09 11:41:52 UTC
[carbondata] branch master updated: [CARBONDATA-3482] Fixed NPE in
Concurrent query
This is an automated email from the ASF dual-hosted git repository.
ravipesala 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 8f0724e [CARBONDATA-3482] Fixed NPE in Concurrent query
8f0724e is described below
commit 8f0724e4256e960608aa6a0d66593acd2ceaa84e
Author: kunal642 <ku...@gmail.com>
AuthorDate: Mon Jul 29 14:31:31 2019 +0530
[CARBONDATA-3482] Fixed NPE in Concurrent query
Problem: In case of concurrent queries if Q1 is loading cache and Q2 is removing from cache then Q2 may remove the segmentPropertiesIndex which Q1 has allocated and is about to access. This will cause NullPointerException .
Solution: Instead of adding index in BlockDataMap keep the reference of segmentPropertiesWrapper to be used.
This closes #3351
---
.../block/SegmentPropertiesAndSchemaHolder.java | 21 ++++++-------
.../core/indexstore/BlockletDataMapIndexStore.java | 2 +-
.../indexstore/blockletindex/BlockDataMap.java | 36 ++++++++--------------
.../indexstore/blockletindex/BlockletDataMap.java | 7 +----
...ryWithColumnMetCacheAndCacheLevelProperty.scala | 27 ++--------------
5 files changed, 26 insertions(+), 67 deletions(-)
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/block/SegmentPropertiesAndSchemaHolder.java b/core/src/main/java/org/apache/carbondata/core/datastore/block/SegmentPropertiesAndSchemaHolder.java
index f2f2d8c..056a0e7 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/block/SegmentPropertiesAndSchemaHolder.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/block/SegmentPropertiesAndSchemaHolder.java
@@ -98,7 +98,7 @@ public class SegmentPropertiesAndSchemaHolder {
* @param columnCardinality
* @param segmentId
*/
- public int addSegmentProperties(CarbonTable carbonTable,
+ public SegmentPropertiesWrapper addSegmentProperties(CarbonTable carbonTable,
List<ColumnSchema> columnsInTable, int[] columnCardinality, String segmentId) {
SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper segmentPropertiesWrapper =
new SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper(carbonTable,
@@ -137,7 +137,7 @@ public class SegmentPropertiesAndSchemaHolder {
.addMinMaxColumns(carbonTable);
}
}
- return segmentIdSetAndIndexWrapper.getSegmentPropertiesIndex();
+ return getSegmentPropertiesWrapper(segmentIdSetAndIndexWrapper.getSegmentPropertiesIndex());
}
/**
@@ -222,17 +222,14 @@ public class SegmentPropertiesAndSchemaHolder {
* Method to remove the given segment ID
*
* @param segmentId
- * @param segmentPropertiesIndex
* @param clearSegmentWrapperFromMap flag to specify whether to clear segmentPropertiesWrapper
* from Map if all the segment's using it have become stale
*/
- public void invalidate(String segmentId, int segmentPropertiesIndex,
+ public void invalidate(String segmentId, SegmentPropertiesWrapper segmentPropertiesWrapper,
boolean clearSegmentWrapperFromMap) {
- SegmentPropertiesWrapper segmentPropertiesWrapper =
- indexToSegmentPropertiesWrapperMapping.get(segmentPropertiesIndex);
- if (null != segmentPropertiesWrapper) {
- SegmentIdAndSegmentPropertiesIndexWrapper segmentIdAndSegmentPropertiesIndexWrapper =
- segmentPropWrapperToSegmentSetMap.get(segmentPropertiesWrapper);
+ SegmentIdAndSegmentPropertiesIndexWrapper segmentIdAndSegmentPropertiesIndexWrapper =
+ segmentPropWrapperToSegmentSetMap.get(segmentPropertiesWrapper);
+ if (segmentIdAndSegmentPropertiesIndexWrapper != null) {
synchronized (getOrCreateTableLock(segmentPropertiesWrapper.getTableIdentifier())) {
segmentIdAndSegmentPropertiesIndexWrapper.removeSegmentId(segmentId);
// if after removal of given SegmentId, the segmentIdSet becomes empty that means this
@@ -240,14 +237,16 @@ public class SegmentPropertiesAndSchemaHolder {
// removed from all the holders
if (clearSegmentWrapperFromMap && segmentIdAndSegmentPropertiesIndexWrapper.segmentIdSet
.isEmpty()) {
- indexToSegmentPropertiesWrapperMapping.remove(segmentPropertiesIndex);
+ indexToSegmentPropertiesWrapperMapping
+ .remove(segmentIdAndSegmentPropertiesIndexWrapper.getSegmentPropertiesIndex());
segmentPropWrapperToSegmentSetMap.remove(segmentPropertiesWrapper);
} else if (!clearSegmentWrapperFromMap
&& segmentIdAndSegmentPropertiesIndexWrapper.segmentIdSet.isEmpty()) {
// min max columns can very when cache is modified. So even though entry is not required
// to be deleted from map clear the column cache so that it can filled again
segmentPropertiesWrapper.clear();
- LOGGER.info("cleared min max for segmentProperties at index: " + segmentPropertiesIndex);
+ LOGGER.info("cleared min max for segmentProperties at index: "
+ + segmentIdAndSegmentPropertiesIndexWrapper.getSegmentPropertiesIndex());
}
}
}
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDataMapIndexStore.java b/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDataMapIndexStore.java
index ce1e8ac..32ee9cb 100644
--- a/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDataMapIndexStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDataMapIndexStore.java
@@ -230,7 +230,7 @@ public class BlockletDataMapIndexStore
// as segmentId will be same for all the dataMaps and segmentProperties cache is
// maintained at segment level so it need to be called only once for clearing
SegmentPropertiesAndSchemaHolder.getInstance()
- .invalidate(segmentId, dataMaps.get(0).getSegmentPropertiesIndex(),
+ .invalidate(segmentId, dataMaps.get(0).getSegmentPropertiesWrapper(),
tableSegmentUniqueIdentifierWrapper.isAddTableBlockToUnsafeAndLRUCache());
}
}
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockDataMap.java b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockDataMap.java
index 24ad43a..f168761 100644
--- a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockDataMap.java
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockDataMap.java
@@ -99,7 +99,8 @@ public class BlockDataMap extends CoarseGrainDataMap
/**
* index of segmentProperties in the segmentProperties holder
*/
- protected int segmentPropertiesIndex;
+ protected transient SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper
+ segmentPropertiesWrapper;
/**
* flag to check for store from 1.1 or any prior version
*/
@@ -204,10 +205,10 @@ public class BlockDataMap extends CoarseGrainDataMap
DataFileFooter fileFooter) throws IOException {
List<ColumnSchema> columnInTable = fileFooter.getColumnInTable();
int[] columnCardinality = fileFooter.getSegmentInfo().getColumnCardinality();
- segmentPropertiesIndex = SegmentPropertiesAndSchemaHolder.getInstance()
+ segmentPropertiesWrapper = SegmentPropertiesAndSchemaHolder.getInstance()
.addSegmentProperties(blockletDataMapInfo.getCarbonTable(),
columnInTable, columnCardinality, blockletDataMapInfo.getSegmentId());
- return getSegmentProperties();
+ return segmentPropertiesWrapper.getSegmentProperties();
}
/**
@@ -485,8 +486,7 @@ public class BlockDataMap extends CoarseGrainDataMap
return getTableTaskInfo(SUMMARY_INDEX_PATH);
}
// create the segment directory path
- String tablePath = SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentPropertiesWrapper(segmentPropertiesIndex).getTableIdentifier().getTablePath();
+ String tablePath = segmentPropertiesWrapper.getTableIdentifier().getTablePath();
String segmentId = getTableTaskInfo(SUMMARY_SEGMENTID);
return CarbonTablePath.getSegmentPath(tablePath, segmentId);
}
@@ -620,8 +620,7 @@ public class BlockDataMap extends CoarseGrainDataMap
}
protected List<CarbonColumn> getMinMaxCacheColumns() {
- return SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentPropertiesWrapper(segmentPropertiesIndex).getMinMaxCacheColumns();
+ return segmentPropertiesWrapper.getMinMaxCacheColumns();
}
/**
@@ -1019,18 +1018,15 @@ public class BlockDataMap extends CoarseGrainDataMap
}
protected SegmentProperties getSegmentProperties() {
- return SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentProperties(segmentPropertiesIndex);
+ return segmentPropertiesWrapper.getSegmentProperties();
}
public int[] getColumnCardinality() {
- return SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentPropertiesWrapper(segmentPropertiesIndex).getColumnCardinality();
+ return segmentPropertiesWrapper.getColumnCardinality();
}
public List<ColumnSchema> getColumnSchema() {
- return SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentPropertiesWrapper(segmentPropertiesIndex).getColumnsInTable();
+ return segmentPropertiesWrapper.getColumnsInTable();
}
protected AbstractMemoryDMStore getMemoryDMStore(boolean addToUnsafe)
@@ -1045,14 +1041,10 @@ public class BlockDataMap extends CoarseGrainDataMap
}
protected CarbonRowSchema[] getFileFooterEntrySchema() {
- return SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentPropertiesWrapper(segmentPropertiesIndex).getBlockFileFooterEntrySchema();
+ return segmentPropertiesWrapper.getBlockFileFooterEntrySchema();
}
protected CarbonRowSchema[] getTaskSummarySchema() {
- SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper segmentPropertiesWrapper =
- SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentPropertiesWrapper(segmentPropertiesIndex);
try {
return segmentPropertiesWrapper.getTaskSummarySchemaForBlock(true, isFilePathStored);
} catch (MemoryException e) {
@@ -1080,12 +1072,8 @@ public class BlockDataMap extends CoarseGrainDataMap
}
}
- public void setSegmentPropertiesIndex(int segmentPropertiesIndex) {
- this.segmentPropertiesIndex = segmentPropertiesIndex;
- }
-
- public int getSegmentPropertiesIndex() {
- return segmentPropertiesIndex;
+ public SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper getSegmentPropertiesWrapper() {
+ return segmentPropertiesWrapper;
}
@Override public int getNumberOfEntries() {
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMap.java b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMap.java
index 23d39ce..11f24f5 100644
--- a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMap.java
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMap.java
@@ -27,7 +27,6 @@ import java.util.List;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datamap.dev.DataMapModel;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
-import org.apache.carbondata.core.datastore.block.SegmentPropertiesAndSchemaHolder;
import org.apache.carbondata.core.datastore.block.TableBlockInfo;
import org.apache.carbondata.core.indexstore.BlockMetaInfo;
import org.apache.carbondata.core.indexstore.ExtendedBlocklet;
@@ -83,9 +82,6 @@ public class BlockletDataMap extends BlockDataMap implements Serializable {
if (isLegacyStore) {
return super.getTaskSummarySchema();
}
- SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper segmentPropertiesWrapper =
- SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentPropertiesWrapper(segmentPropertiesIndex);
try {
return segmentPropertiesWrapper.getTaskSummarySchemaForBlocklet(false, isFilePathStored);
} catch (MemoryException e) {
@@ -98,8 +94,7 @@ public class BlockletDataMap extends BlockDataMap implements Serializable {
if (isLegacyStore) {
return super.getFileFooterEntrySchema();
}
- return SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentPropertiesWrapper(segmentPropertiesIndex).getBlockletFileFooterEntrySchema();
+ return segmentPropertiesWrapper.getBlockletFileFooterEntrySchema();
}
/**
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/allqueries/TestQueryWithColumnMetCacheAndCacheLevelProperty.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/allqueries/TestQueryWithColumnMetCacheAndCacheLevelProperty.scala
index ac0ca8b..5c34fe4 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/allqueries/TestQueryWithColumnMetCacheAndCacheLevelProperty.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/allqueries/TestQueryWithColumnMetCacheAndCacheLevelProperty.scala
@@ -88,9 +88,8 @@ class TestQueryWithColumnMetCacheAndCacheLevelProperty extends QueryTest with Be
private def validateMinMaxColumnsCacheLength(dataMaps: List[DataMap[_ <: Blocklet]],
expectedLength: Int, storeBlockletCount: Boolean = false): Boolean = {
- val index = dataMaps(0).asInstanceOf[BlockDataMap].getSegmentPropertiesIndex
- val summarySchema = SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentPropertiesWrapper(index).getTaskSummarySchemaForBlock(storeBlockletCount, false)
+ val segmentPropertiesWrapper = dataMaps(0).asInstanceOf[BlockDataMap].getSegmentPropertiesWrapper
+ val summarySchema = segmentPropertiesWrapper.getTaskSummarySchemaForBlock(storeBlockletCount, false)
val minSchemas = summarySchema(BlockletDataMapRowIndexes.TASK_MIN_VALUES_INDEX)
.asInstanceOf[CarbonRowSchema.StructCarbonRowSchema]
.getChildSchemas
@@ -107,15 +106,10 @@ class TestQueryWithColumnMetCacheAndCacheLevelProperty extends QueryTest with Be
assert(dataMaps.nonEmpty)
assert(dataMaps(0).isInstanceOf[BlockDataMap])
assert(validateMinMaxColumnsCacheLength(dataMaps, 3, true))
- var segmentPropertyIndex = dataMaps(0).asInstanceOf[BlockDataMap].getSegmentPropertiesIndex
-
// alter table to add column_meta_cache and cache_level
sql(
"alter table metaCache set tblproperties('column_meta_cache'='c2,c1', 'CACHE_LEVEL'='BLOCKLET')")
- var wrapper = SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentPropertiesWrapper(segmentPropertyIndex)
// after alter operation cache should be cleaned and cache should be evicted
- assert(null == wrapper)
checkAnswer(sql("select * from metaCache"), Row("a", "aa", "aaa"))
// validate dataMap is non empty, its an instance of BlockletDataMap and minMaxSchema length
// is 1
@@ -125,23 +119,11 @@ class TestQueryWithColumnMetCacheAndCacheLevelProperty extends QueryTest with Be
assert(validateMinMaxColumnsCacheLength(dataMaps, 2))
// alter table to add same value as previous with order change for column_meta_cache and cache_level
- segmentPropertyIndex = dataMaps(0).asInstanceOf[BlockDataMap].getSegmentPropertiesIndex
sql(
"alter table metaCache set tblproperties('column_meta_cache'='c1,c2', 'CACHE_LEVEL'='BLOCKLET')")
- wrapper = SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentPropertiesWrapper(segmentPropertyIndex)
- // after alter operation cache should not be cleaned as value are unchanged
- assert(null != wrapper)
-
- // alter table to cache no column in column_meta_cache
- segmentPropertyIndex = dataMaps(0).asInstanceOf[BlockDataMap].getSegmentPropertiesIndex
sql(
"alter table metaCache set tblproperties('column_meta_cache'='')")
- wrapper = SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentPropertiesWrapper(segmentPropertyIndex)
-
// after alter operation cache should be cleaned and cache should be evicted
- assert(null == wrapper)
checkAnswer(sql("select * from metaCache"), Row("a", "aa", "aaa"))
// validate dataMap is non empty, its an instance of BlockletDataMap and minMaxSchema length
// is 0
@@ -151,13 +133,8 @@ class TestQueryWithColumnMetCacheAndCacheLevelProperty extends QueryTest with Be
assert(validateMinMaxColumnsCacheLength(dataMaps, 0))
// alter table to cache no column in column_meta_cache
- segmentPropertyIndex = dataMaps(0).asInstanceOf[BlockDataMap].getSegmentPropertiesIndex
sql(
"alter table metaCache unset tblproperties('column_meta_cache', 'cache_level')")
- wrapper = SegmentPropertiesAndSchemaHolder.getInstance()
- .getSegmentPropertiesWrapper(segmentPropertyIndex)
- // after alter operation cache should be cleaned and cache should be evicted
- assert(null == wrapper)
checkAnswer(sql("select * from metaCache"), Row("a", "aa", "aaa"))
// validate dataMap is non empty, its an instance of BlockletDataMap and minMaxSchema length
// is 3