You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by gv...@apache.org on 2017/08/22 14:21:30 UTC
[1/2] carbondata git commit: [CARBONDATA-1308] Added tableProvider to
supply carbonTable wherever needed
Repository: carbondata
Updated Branches:
refs/heads/master e3f98fa43 -> d3a09e279
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImpl.java b/core/src/main/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImpl.java
index 64ff202..ed54a3b 100644
--- a/core/src/main/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImpl.java
@@ -25,11 +25,11 @@ import java.util.List;
import org.apache.carbondata.common.logging.LogService;
import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
-import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.reader.CarbonDictionaryColumnMetaChunk;
import org.apache.carbondata.core.reader.CarbonDictionaryMetadataReader;
import org.apache.carbondata.core.reader.CarbonDictionaryMetadataReaderImpl;
@@ -77,7 +77,7 @@ public class CarbonDictionaryWriterImpl implements CarbonDictionaryWriter {
/**
* column identifier
*/
- protected ColumnIdentifier columnIdentifier;
+ protected DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier;
/**
* carbon dictionary data store path
@@ -132,12 +132,12 @@ public class CarbonDictionaryWriterImpl implements CarbonDictionaryWriter {
*
* @param storePath carbon dictionary data store path
* @param carbonTableIdentifier table identifier which will give table name and database name
- * @param columnIdentifier column unique identifier
+ * @param dictionaryColumnUniqueIdentifier column unique identifier
*/
- public CarbonDictionaryWriterImpl(String storePath,
- CarbonTableIdentifier carbonTableIdentifier, ColumnIdentifier columnIdentifier) {
+ public CarbonDictionaryWriterImpl(String storePath, CarbonTableIdentifier carbonTableIdentifier,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
this.carbonTableIdentifier = carbonTableIdentifier;
- this.columnIdentifier = columnIdentifier;
+ this.dictionaryColumnUniqueIdentifier = dictionaryColumnUniqueIdentifier;
this.storePath = storePath;
this.isFirstTime = true;
}
@@ -254,11 +254,13 @@ public class CarbonDictionaryWriterImpl implements CarbonDictionaryWriter {
protected void initPaths() {
PathService pathService = CarbonCommonFactory.getPathService();
- CarbonTablePath carbonTablePath = pathService.getCarbonTablePath(
- this.storePath, carbonTableIdentifier);
- this.dictionaryFilePath = carbonTablePath.getDictionaryFilePath(columnIdentifier.getColumnId());
- this.dictionaryMetaFilePath =
- carbonTablePath.getDictionaryMetaFilePath(columnIdentifier.getColumnId());
+ CarbonTablePath carbonTablePath = pathService
+ .getCarbonTablePath(this.storePath, carbonTableIdentifier,
+ dictionaryColumnUniqueIdentifier);
+ this.dictionaryFilePath = carbonTablePath.getDictionaryFilePath(
+ dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId());
+ this.dictionaryMetaFilePath = carbonTablePath.getDictionaryMetaFilePath(
+ dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId());
}
/**
@@ -290,17 +292,19 @@ public class CarbonDictionaryWriterImpl implements CarbonDictionaryWriter {
int bytesToTruncate = 0;
if (null != chunkMetaObjectForLastSegmentEntry) {
bytesToTruncate =
- (int) (chunk_start_offset - chunkMetaObjectForLastSegmentEntry.getEnd_offset());
+ (int) (chunk_start_offset - chunkMetaObjectForLastSegmentEntry.getEnd_offset());
}
if (bytesToTruncate > 0) {
- LOGGER.info("some inconsistency in dictionary file for column " + this.columnIdentifier);
+ LOGGER.info("some inconsistency in dictionary file for column "
+ + this.dictionaryColumnUniqueIdentifier.getColumnIdentifier());
// truncate the dictionary data till chunk meta end offset
FileFactory.FileType fileType = FileFactory.getFileType(this.dictionaryFilePath);
CarbonFile carbonFile = FileFactory.getCarbonFile(this.dictionaryFilePath, fileType);
boolean truncateSuccess = carbonFile
.truncate(this.dictionaryFilePath, chunkMetaObjectForLastSegmentEntry.getEnd_offset());
if (!truncateSuccess) {
- LOGGER.info("Diction file not truncated successfully for column " + this.columnIdentifier);
+ LOGGER.info("Diction file not truncated successfully for column "
+ + this.dictionaryColumnUniqueIdentifier.getColumnIdentifier());
}
}
}
@@ -344,7 +348,8 @@ public class CarbonDictionaryWriterImpl implements CarbonDictionaryWriter {
// write dictionary metadata file
writeThriftObject(dictionaryChunkMeta);
LOGGER.info("Dictionary metadata file written successfully for column "
- + this.columnIdentifier + " at path " + this.dictionaryMetaFilePath);
+ + this.dictionaryColumnUniqueIdentifier.getColumnIdentifier() + " at path "
+ + this.dictionaryMetaFilePath);
} finally {
closeThriftWriter();
}
@@ -426,7 +431,7 @@ public class CarbonDictionaryWriterImpl implements CarbonDictionaryWriter {
*/
protected CarbonDictionaryMetadataReader getDictionaryMetadataReader() {
return new CarbonDictionaryMetadataReaderImpl(storePath, carbonTableIdentifier,
- columnIdentifier);
+ dictionaryColumnUniqueIdentifier);
}
@Override public void commit() throws IOException {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImpl.java b/core/src/main/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImpl.java
index d4750da..b10da11 100644
--- a/core/src/main/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImpl.java
@@ -23,11 +23,11 @@ import java.util.List;
import org.apache.carbondata.common.logging.LogService;
import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
-import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.service.CarbonCommonFactory;
import org.apache.carbondata.core.service.PathService;
import org.apache.carbondata.core.util.CarbonProperties;
@@ -50,7 +50,7 @@ public class CarbonDictionarySortIndexWriterImpl implements CarbonDictionarySort
/**
* column name
*/
- protected ColumnIdentifier columnIdentifier;
+ protected DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier;
/**
* carbon store location
@@ -79,12 +79,13 @@ public class CarbonDictionarySortIndexWriterImpl implements CarbonDictionarySort
/**
* @param carbonStorePath Carbon store path
* @param carbonTableIdentifier table identifier which will give table name and database name
- * @param columnIdentifier column unique identifier
+ * @param dictionaryColumnUniqueIdentifier column unique identifier
*/
public CarbonDictionarySortIndexWriterImpl(final CarbonTableIdentifier carbonTableIdentifier,
- final ColumnIdentifier columnIdentifier, final String carbonStorePath) {
+ final DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier,
+ final String carbonStorePath) {
this.carbonTableIdentifier = carbonTableIdentifier;
- this.columnIdentifier = columnIdentifier;
+ this.dictionaryColumnUniqueIdentifier = dictionaryColumnUniqueIdentifier;
this.carbonStorePath = carbonStorePath;
}
@@ -150,11 +151,14 @@ public class CarbonDictionarySortIndexWriterImpl implements CarbonDictionarySort
protected void initPath() {
PathService pathService = CarbonCommonFactory.getPathService();
CarbonTablePath carbonTablePath = pathService
- .getCarbonTablePath(carbonStorePath, carbonTableIdentifier);
- String dictionaryPath = carbonTablePath.getDictionaryFilePath(columnIdentifier.getColumnId());
+ .getCarbonTablePath(carbonStorePath, carbonTableIdentifier,
+ dictionaryColumnUniqueIdentifier);
+ String dictionaryPath = carbonTablePath.getDictionaryFilePath(
+ dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId());
long dictOffset = CarbonUtil.getFileSize(dictionaryPath);
- this.sortIndexFilePath =
- carbonTablePath.getSortIndexFilePath(columnIdentifier.getColumnId(), dictOffset);
+ this.sortIndexFilePath = carbonTablePath
+ .getSortIndexFilePath(dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId(),
+ dictOffset);
cleanUpOldSortIndex(carbonTablePath, dictionaryPath);
}
@@ -166,9 +170,8 @@ public class CarbonDictionarySortIndexWriterImpl implements CarbonDictionarySort
protected void cleanUpOldSortIndex(CarbonTablePath carbonTablePath, String dictPath) {
CarbonFile dictFile =
FileFactory.getCarbonFile(dictPath, FileFactory.getFileType(dictPath));
- CarbonFile[] files =
- carbonTablePath.getSortIndexFiles(dictFile.getParentFile(),
- columnIdentifier.getColumnId());
+ CarbonFile[] files = carbonTablePath.getSortIndexFiles(dictFile.getParentFile(),
+ dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId());
int maxTime;
try {
maxTime = Integer.parseInt(CarbonProperties.getInstance()
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/test/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCacheTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCacheTest.java b/core/src/test/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCacheTest.java
index 2fa2b93..957ea22 100644
--- a/core/src/test/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCacheTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCacheTest.java
@@ -105,7 +105,8 @@ public class AbstractDictionaryCacheTest {
String columnId) {
ColumnIdentifier columnIdentifier = new ColumnIdentifier(columnId, null, DataType.STRING);
return new DictionaryColumnUniqueIdentifier(carbonTableIdentifier, columnIdentifier,
- DataType.STRING);
+ DataType.STRING,
+ CarbonStorePath.getCarbonTablePath(carbonStorePath, carbonTableIdentifier));
}
/**
@@ -126,8 +127,12 @@ public class AbstractDictionaryCacheTest {
protected void prepareWriterAndWriteData(List<String> data, String columnId)
throws IOException {
ColumnIdentifier columnIdentifier = new ColumnIdentifier(columnId, null, null);
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier =
+ new DictionaryColumnUniqueIdentifier(carbonTableIdentifier, columnIdentifier,
+ columnIdentifier.getDataType(),
+ CarbonStorePath.getCarbonTablePath(carbonStorePath, carbonTableIdentifier));
CarbonDictionaryWriter carbonDictionaryWriter =
- new CarbonDictionaryWriterImpl(carbonStorePath, carbonTableIdentifier, columnIdentifier);
+ new CarbonDictionaryWriterImpl(carbonStorePath, carbonTableIdentifier, dictionaryColumnUniqueIdentifier);
CarbonTablePath carbonTablePath =
CarbonStorePath.getCarbonTablePath(carbonStorePath, carbonTableIdentifier);
CarbonUtil.checkAndCreateFolder(carbonTablePath.getMetadataDirectoryPath());
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/test/java/org/apache/carbondata/core/cache/dictionary/DictionaryCacheLoaderImplTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/cache/dictionary/DictionaryCacheLoaderImplTest.java b/core/src/test/java/org/apache/carbondata/core/cache/dictionary/DictionaryCacheLoaderImplTest.java
index 0e17600..a751120 100644
--- a/core/src/test/java/org/apache/carbondata/core/cache/dictionary/DictionaryCacheLoaderImplTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/cache/dictionary/DictionaryCacheLoaderImplTest.java
@@ -29,6 +29,7 @@ import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.reader.CarbonDictionaryReaderImpl;
import org.apache.carbondata.core.reader.sortindex.CarbonDictionarySortIndexReaderImpl;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
import org.apache.carbondata.format.ColumnDictionaryChunk;
import mockit.Mock;
@@ -43,10 +44,19 @@ public class DictionaryCacheLoaderImplTest {
private static DictionaryCacheLoaderImpl dictionaryCacheLoader;
private static DictionaryInfo dictionaryInfo;
private static ColumnIdentifier columnIdentifier;
+ private static DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier;
@BeforeClass public static void setUp() {
CarbonTableIdentifier carbonTableIdentifier = new CarbonTableIdentifier("db", "table1", "1");
- dictionaryCacheLoader = new DictionaryCacheLoaderImpl(carbonTableIdentifier, "/tmp/");
+ Map<String, String> columnProperties = new HashMap<>();
+ columnProperties.put("prop1", "value1");
+ columnProperties.put("prop2", "value2");
+ columnIdentifier = new ColumnIdentifier("1", columnProperties, DataType.STRING);
+ dictionaryColumnUniqueIdentifier =
+ new DictionaryColumnUniqueIdentifier(carbonTableIdentifier, columnIdentifier,
+ columnIdentifier.getDataType(), CarbonStorePath.getCarbonTablePath("/tmp", carbonTableIdentifier));
+ dictionaryCacheLoader = new DictionaryCacheLoaderImpl(carbonTableIdentifier, "/tmp/",
+ dictionaryColumnUniqueIdentifier);
dictionaryInfo = new ColumnDictionaryInfo(DataType.STRING);
new MockUp<CarbonDictionaryReaderImpl>() {
@Mock @SuppressWarnings("unused") Iterator<byte[]> read(long startOffset, long endOffset)
@@ -68,10 +78,6 @@ public class DictionaryCacheLoaderImplTest {
return Arrays.asList(1, 2);
}
};
- Map<String, String> columnProperties = new HashMap<>();
- columnProperties.put("prop1", "value1");
- columnProperties.put("prop2", "value2");
- columnIdentifier = new ColumnIdentifier("1", columnProperties, DataType.STRING);
}
@Test public void testToLoad() throws IOException {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/test/java/org/apache/carbondata/core/cache/dictionary/DictionaryColumnUniqueIdentifierTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/cache/dictionary/DictionaryColumnUniqueIdentifierTest.java b/core/src/test/java/org/apache/carbondata/core/cache/dictionary/DictionaryColumnUniqueIdentifierTest.java
index 028ae81..0e2eed9 100644
--- a/core/src/test/java/org/apache/carbondata/core/cache/dictionary/DictionaryColumnUniqueIdentifierTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/cache/dictionary/DictionaryColumnUniqueIdentifierTest.java
@@ -47,13 +47,13 @@ public class DictionaryColumnUniqueIdentifierTest {
ColumnIdentifier columnIdentifier2 = new ColumnIdentifier("1", properties, DataType.INT);
dictionaryColumnUniqueIdentifier1 =
new DictionaryColumnUniqueIdentifier(carbonTableIdentifier1, columnIdentifier,
- DataType.MAP);
+ DataType.MAP, null);
dictionaryColumnUniqueIdentifier2 =
new DictionaryColumnUniqueIdentifier(carbonTableIdentifier2, columnIdentifier2,
- DataType.MAP);
+ DataType.MAP, null);
dictionaryColumnUniqueIdentifier3 =
new DictionaryColumnUniqueIdentifier(carbonTableIdentifier2, columnIdentifier,
- DataType.MAP);
+ DataType.MAP, null);
}
@Test public void testToGetDataType() {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/test/java/org/apache/carbondata/core/cache/dictionary/ForwardDictionaryCacheTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/cache/dictionary/ForwardDictionaryCacheTest.java b/core/src/test/java/org/apache/carbondata/core/cache/dictionary/ForwardDictionaryCacheTest.java
index 4efb093..34aed8a 100644
--- a/core/src/test/java/org/apache/carbondata/core/cache/dictionary/ForwardDictionaryCacheTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/cache/dictionary/ForwardDictionaryCacheTest.java
@@ -32,6 +32,7 @@ import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.util.CarbonProperties;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
import org.apache.carbondata.core.writer.sortindex.CarbonDictionarySortIndexWriter;
import org.apache.carbondata.core.writer.sortindex.CarbonDictionarySortIndexWriterImpl;
import org.junit.After;
@@ -209,6 +210,10 @@ public class ForwardDictionaryCacheTest extends AbstractDictionaryCacheTest {
*/
private void writeSortIndexFile(List<String> data, String columnId) throws IOException {
ColumnIdentifier columnIdentifier = new ColumnIdentifier(columnId, null, null);
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier =
+ new DictionaryColumnUniqueIdentifier(carbonTableIdentifier, columnIdentifier,
+ columnIdentifier.getDataType(),
+ CarbonStorePath.getCarbonTablePath(carbonStorePath, carbonTableIdentifier));
Map<String, Integer> dataToSurrogateKeyMap = new HashMap<>(data.size());
int surrogateKey = 0;
List<Integer> invertedIndexList = new ArrayList<>(data.size());
@@ -228,7 +233,7 @@ public class ForwardDictionaryCacheTest extends AbstractDictionaryCacheTest {
invertedIndexList.add(invertedIndexArray[i]);
}
CarbonDictionarySortIndexWriter dictionarySortIndexWriter =
- new CarbonDictionarySortIndexWriterImpl(carbonTableIdentifier, columnIdentifier,
+ new CarbonDictionarySortIndexWriterImpl(carbonTableIdentifier, dictionaryColumnUniqueIdentifier,
carbonStorePath);
try {
dictionarySortIndexWriter.writeSortIndex(sortedIndexList);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/test/java/org/apache/carbondata/core/reader/CarbonDictionaryReaderImplTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/reader/CarbonDictionaryReaderImplTest.java b/core/src/test/java/org/apache/carbondata/core/reader/CarbonDictionaryReaderImplTest.java
index dcd1780..b3fbdd6 100644
--- a/core/src/test/java/org/apache/carbondata/core/reader/CarbonDictionaryReaderImplTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/reader/CarbonDictionaryReaderImplTest.java
@@ -23,11 +23,13 @@ import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.service.impl.PathFactory;
import org.apache.carbondata.core.service.CarbonCommonFactory;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.service.PathService;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
import mockit.Mock;
import mockit.MockUp;
@@ -47,8 +49,12 @@ public class CarbonDictionaryReaderImplTest {
columnIdentifier = new ColumnIdentifier("1", null, null);
carbonTableIdentifier =
new CarbonTableIdentifier("dbName", "tableName", UUID.randomUUID().toString());
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier =
+ new DictionaryColumnUniqueIdentifier(carbonTableIdentifier, columnIdentifier,
+ columnIdentifier.getDataType(),
+ CarbonStorePath.getCarbonTablePath("storePath", carbonTableIdentifier));
carbonDictionaryReaderImpl =
- new CarbonDictionaryReaderImpl("storePath", carbonTableIdentifier, columnIdentifier);
+ new CarbonDictionaryReaderImpl("storePath", carbonTableIdentifier, dictionaryColumnUniqueIdentifier);
}
@Test public void testRead() throws Exception {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/test/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImplTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImplTest.java b/core/src/test/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImplTest.java
index 2a2551b..2a4c290 100644
--- a/core/src/test/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImplTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImplTest.java
@@ -22,11 +22,13 @@ import java.util.Arrays;
import java.util.List;
import java.util.UUID;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
import org.apache.carbondata.core.writer.CarbonDictionaryWriter;
import org.apache.carbondata.core.writer.CarbonDictionaryWriterImpl;
import org.apache.carbondata.core.writer.sortindex.CarbonDictionarySortIndexWriter;
@@ -62,12 +64,14 @@ public class CarbonDictionarySortIndexReaderImplTest {
CarbonTableIdentifier carbonTableIdentifier = new CarbonTableIdentifier("testSchema", "carbon",
UUID.randomUUID().toString());
ColumnIdentifier columnIdentifier = new ColumnIdentifier("Name", null, null);
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier = new DictionaryColumnUniqueIdentifier(carbonTableIdentifier, columnIdentifier, columnIdentifier.getDataType(),
+ CarbonStorePath.getCarbonTablePath(storePath, carbonTableIdentifier));
CarbonDictionaryWriter dictionaryWriter = new CarbonDictionaryWriterImpl(storePath,
- carbonTableIdentifier, columnIdentifier);
+ carbonTableIdentifier, dictionaryColumnUniqueIdentifier);
String metaFolderPath =storePath+File.separator+carbonTableIdentifier.getDatabaseName()+File.separator+carbonTableIdentifier.getTableName()+File.separator+"Metadata";
CarbonUtil.checkAndCreateFolder(metaFolderPath);
CarbonDictionarySortIndexWriter dictionarySortIndexWriter =
- new CarbonDictionarySortIndexWriterImpl(carbonTableIdentifier, columnIdentifier, storePath);
+ new CarbonDictionarySortIndexWriterImpl(carbonTableIdentifier, dictionaryColumnUniqueIdentifier, storePath);
List<int[]> expectedData = prepareExpectedData();
int[] data = expectedData.get(0);
for(int i=0;i<data.length;i++) {
@@ -81,7 +85,7 @@ public class CarbonDictionarySortIndexReaderImplTest {
dictionarySortIndexWriter.writeInvertedSortIndex(invertedSortIndex);
dictionarySortIndexWriter.close();
CarbonDictionarySortIndexReader dictionarySortIndexReader =
- new CarbonDictionarySortIndexReaderImpl(carbonTableIdentifier, columnIdentifier, storePath);
+ new CarbonDictionarySortIndexReaderImpl(carbonTableIdentifier, dictionaryColumnUniqueIdentifier, storePath);
List<Integer> actualSortIndex = dictionarySortIndexReader.readSortIndex();
List<Integer> actualInvertedSortIndex = dictionarySortIndexReader.readInvertedSortIndex();
for (int i = 0; i < actualSortIndex.size(); i++) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/test/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImplTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImplTest.java b/core/src/test/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImplTest.java
index 55abac3..8bb2052 100644
--- a/core/src/test/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImplTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImplTest.java
@@ -32,6 +32,7 @@ import java.util.List;
import java.util.Properties;
import java.util.UUID;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.util.path.CarbonStorePath;
@@ -72,6 +73,8 @@ public class CarbonDictionaryWriterImplTest {
private ColumnIdentifier columnIdentifier;
+ private DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier;
+
private Properties props;
/**
@@ -97,6 +100,10 @@ public class CarbonDictionaryWriterImplTest {
this.carbonStorePath = props.getProperty("storePath", "carbonStore");
this.columnIdentifier = new ColumnIdentifier("Name", null, null);
carbonTableIdentifier = new CarbonTableIdentifier(databaseName, tableName, UUID.randomUUID().toString());
+ this.dictionaryColumnUniqueIdentifier =
+ new DictionaryColumnUniqueIdentifier(carbonTableIdentifier, columnIdentifier,
+ columnIdentifier.getDataType(),
+ CarbonStorePath.getCarbonTablePath(carbonStorePath, carbonTableIdentifier));
deleteStorePath();
prepareDataSet();
}
@@ -177,7 +184,7 @@ public class CarbonDictionaryWriterImplTest {
private CarbonDictionaryWriterImpl prepareWriter() throws IOException {
initDictionaryDirPaths();
return new CarbonDictionaryWriterImpl(this.carbonStorePath, carbonTableIdentifier,
- columnIdentifier);
+ dictionaryColumnUniqueIdentifier);
}
/**
@@ -432,7 +439,7 @@ public class CarbonDictionaryWriterImplTest {
private List<CarbonDictionaryColumnMetaChunk> readDictionaryMetadataFile() throws IOException {
CarbonDictionaryMetadataReaderImpl columnMetadataReaderImpl =
new CarbonDictionaryMetadataReaderImpl(this.carbonStorePath, this.carbonTableIdentifier,
- this.columnIdentifier);
+ this.dictionaryColumnUniqueIdentifier);
List<CarbonDictionaryColumnMetaChunk> dictionaryMetaChunkList = null;
// read metadata file
try {
@@ -451,7 +458,7 @@ public class CarbonDictionaryWriterImplTest {
throws IOException {
CarbonDictionaryReaderImpl dictionaryReader =
new CarbonDictionaryReaderImpl(this.carbonStorePath, this.carbonTableIdentifier,
- this.columnIdentifier);
+ this.dictionaryColumnUniqueIdentifier);
List<byte[]> dictionaryValues = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
try {
if (0 == dictionaryEndOffset) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/test/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImplTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImplTest.java b/core/src/test/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImplTest.java
index 24083b7..d04d8a2 100644
--- a/core/src/test/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImplTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImplTest.java
@@ -22,11 +22,13 @@ import java.util.Arrays;
import java.util.List;
import java.util.UUID;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.reader.sortindex.CarbonDictionarySortIndexReader;
import org.apache.carbondata.core.reader.sortindex.CarbonDictionarySortIndexReaderImpl;
import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
import org.apache.carbondata.core.writer.CarbonDictionaryWriter;
import org.apache.carbondata.core.writer.CarbonDictionaryWriterImpl;
@@ -53,12 +55,14 @@ public class CarbonDictionarySortIndexWriterImplTest {
carbonTableIdentifier =
new CarbonTableIdentifier("testSchema", "carbon", UUID.randomUUID().toString());
columnIdentifier = new ColumnIdentifier("Name", null, null);
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier = new DictionaryColumnUniqueIdentifier(carbonTableIdentifier, columnIdentifier, columnIdentifier.getDataType(),
+ CarbonStorePath.getCarbonTablePath(storePath, carbonTableIdentifier));
dictionaryWriter =
- new CarbonDictionaryWriterImpl(storePath, carbonTableIdentifier, columnIdentifier);
+ new CarbonDictionaryWriterImpl(storePath, carbonTableIdentifier, dictionaryColumnUniqueIdentifier);
dictionarySortIndexWriter =
- new CarbonDictionarySortIndexWriterImpl(carbonTableIdentifier, columnIdentifier, storePath);
+ new CarbonDictionarySortIndexWriterImpl(carbonTableIdentifier, dictionaryColumnUniqueIdentifier, storePath);
carbonDictionarySortIndexReader =
- new CarbonDictionarySortIndexReaderImpl(carbonTableIdentifier, columnIdentifier, storePath);
+ new CarbonDictionarySortIndexReaderImpl(carbonTableIdentifier, dictionaryColumnUniqueIdentifier, storePath);
}
/**
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/examples/spark/src/main/scala/org/apache/carbondata/examples/GenerateDictionaryExample.scala
----------------------------------------------------------------------
diff --git a/examples/spark/src/main/scala/org/apache/carbondata/examples/GenerateDictionaryExample.scala b/examples/spark/src/main/scala/org/apache/carbondata/examples/GenerateDictionaryExample.scala
index 9aed2ea..8b2ceba 100644
--- a/examples/spark/src/main/scala/org/apache/carbondata/examples/GenerateDictionaryExample.scala
+++ b/examples/spark/src/main/scala/org/apache/carbondata/examples/GenerateDictionaryExample.scala
@@ -77,7 +77,9 @@ object GenerateDictionaryExample {
println(s"dictionary of dimension: ${dimension.getColName}")
println(s"Key\t\t\tValue")
val columnIdentifier = new DictionaryColumnUniqueIdentifier(carbonTableIdentifier,
- dimension.getColumnIdentifier, dimension.getDataType)
+ dimension.getColumnIdentifier, dimension.getDataType,
+ CarbonStorePath
+ .getCarbonTablePath(carbonTable.getStorePath, carbonTable.getCarbonTableIdentifier))
val dict = CarbonLoaderUtil.getDictionary(columnIdentifier, cc.storePath)
var index: Int = 1
var distinctValue = dict.getDictionaryValueForKey(index)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/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 249543e..fb3a637 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonInputFormat.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonInputFormat.java
@@ -54,6 +54,8 @@ import org.apache.carbondata.core.mutate.data.BlockMappingVO;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.scan.filter.FilterExpressionProcessor;
import org.apache.carbondata.core.scan.filter.FilterUtil;
+import org.apache.carbondata.core.scan.filter.SingleTableProvider;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.scan.model.CarbonQueryPlan;
import org.apache.carbondata.core.scan.model.QueryModel;
@@ -348,6 +350,7 @@ public class CarbonInputFormat<T> extends FileInputFormat<Void, T> {
// process and resolve the expression
Expression filter = getFilterPredicates(job.getConfiguration());
CarbonTable carbonTable = getOrCreateCarbonTable(job.getConfiguration());
+ TableProvider tableProvider = new SingleTableProvider(carbonTable);
CarbonInputFormatUtil.processFilterExpression(filter, carbonTable);
BitSet matchedPartitions = null;
PartitionInfo partitionInfo = carbonTable.getPartitionInfo(carbonTable.getFactTableName());
@@ -365,7 +368,8 @@ public class CarbonInputFormat<T> extends FileInputFormat<Void, T> {
}
}
- FilterResolverIntf filterInterface = CarbonInputFormatUtil.resolveFilter(filter, identifier);
+ FilterResolverIntf filterInterface = CarbonInputFormatUtil
+ .resolveFilter(filter, carbonTable.getAbsoluteTableIdentifier(), tableProvider);
// do block filtering and get split
List<InputSplit> splits = getSplits(job, filterInterface, matchedPartitions, cacheClient,
@@ -794,6 +798,7 @@ public class CarbonInputFormat<T> extends FileInputFormat<Void, T> {
throws IOException {
Configuration configuration = taskAttemptContext.getConfiguration();
CarbonTable carbonTable = getOrCreateCarbonTable(configuration);
+ TableProvider tableProvider = new SingleTableProvider(carbonTable);
// getting the table absoluteTableIdentifier from the carbonTable
// to avoid unnecessary deserialization
AbsoluteTableIdentifier identifier = carbonTable.getAbsoluteTableIdentifier();
@@ -807,7 +812,8 @@ public class CarbonInputFormat<T> extends FileInputFormat<Void, T> {
// set the filter to the query model in order to filter blocklet before scan
Expression filter = getFilterPredicates(configuration);
CarbonInputFormatUtil.processFilterExpression(filter, carbonTable);
- FilterResolverIntf filterIntf = CarbonInputFormatUtil.resolveFilter(filter, identifier);
+ FilterResolverIntf filterIntf = CarbonInputFormatUtil
+ .resolveFilter(filter, carbonTable.getAbsoluteTableIdentifier(), tableProvider);
queryModel.setFilterExpressionResolverTree(filterIntf);
// update the file level index store if there are invalid segment
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/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 6ae346f..c69e19f 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
@@ -47,6 +47,8 @@ import org.apache.carbondata.core.mutate.UpdateVO;
import org.apache.carbondata.core.mutate.data.BlockMappingVO;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.scan.filter.FilterExpressionProcessor;
+import org.apache.carbondata.core.scan.filter.SingleTableProvider;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.scan.model.CarbonQueryPlan;
import org.apache.carbondata.core.scan.model.QueryModel;
@@ -280,6 +282,7 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
// process and resolve the expression
Expression filter = getFilterPredicates(job.getConfiguration());
CarbonTable carbonTable = getOrCreateCarbonTable(job.getConfiguration());
+ TableProvider tableProvider = new SingleTableProvider(carbonTable);
// this will be null in case of corrupt schema file.
if (null == carbonTable) {
throw new IOException("Missing/Corrupt schema file for table.");
@@ -300,7 +303,8 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
}
}
- FilterResolverIntf filterInterface = CarbonInputFormatUtil.resolveFilter(filter, identifier);
+ FilterResolverIntf filterInterface = CarbonInputFormatUtil
+ .resolveFilter(filter, carbonTable.getAbsoluteTableIdentifier(), tableProvider);
// do block filtering and get split
List<InputSplit> splits =
@@ -346,6 +350,7 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
CarbonInputFormatUtil.processFilterExpression(filter, carbonTable);
+ TableProvider tableProvider = new SingleTableProvider(carbonTable);
// prune partitions for filter query on partition table
String partitionIds = job.getConfiguration().get(ALTER_PARTITION_ID);
BitSet matchedPartitions = null;
@@ -360,7 +365,8 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
}
}
- FilterResolverIntf filterInterface = CarbonInputFormatUtil.resolveFilter(filter, identifier);
+ FilterResolverIntf filterInterface =
+ CarbonInputFormatUtil.resolveFilter(filter, identifier, tableProvider);
// do block filtering and get split
List<InputSplit> splits = getSplits(job, filterInterface, segmentList, matchedPartitions,
partitionInfo, oldPartitionIdList);
@@ -543,6 +549,7 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
throws IOException {
Configuration configuration = taskAttemptContext.getConfiguration();
CarbonTable carbonTable = getOrCreateCarbonTable(configuration);
+ TableProvider tableProvider = new SingleTableProvider(carbonTable);
// getting the table absoluteTableIdentifier from the carbonTable
// to avoid unnecessary deserialization
AbsoluteTableIdentifier identifier = carbonTable.getAbsoluteTableIdentifier();
@@ -556,7 +563,8 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
// set the filter to the query model in order to filter blocklet before scan
Expression filter = getFilterPredicates(configuration);
CarbonInputFormatUtil.processFilterExpression(filter, carbonTable);
- FilterResolverIntf filterIntf = CarbonInputFormatUtil.resolveFilter(filter, identifier);
+ FilterResolverIntf filterIntf = CarbonInputFormatUtil
+ .resolveFilter(filter, carbonTable.getAbsoluteTableIdentifier(), tableProvider);
queryModel.setFilterExpressionResolverTree(filterIntf);
// update the file level index store if there are invalid segment
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/hadoop/src/main/java/org/apache/carbondata/hadoop/readsupport/impl/DictionaryDecodeReadSupport.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/readsupport/impl/DictionaryDecodeReadSupport.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/readsupport/impl/DictionaryDecodeReadSupport.java
index 0239bce..32d879f 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/readsupport/impl/DictionaryDecodeReadSupport.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/readsupport/impl/DictionaryDecodeReadSupport.java
@@ -28,6 +28,7 @@ import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.encoder.Encoding;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
import org.apache.carbondata.hadoop.readsupport.CarbonReadSupport;
/**
@@ -64,7 +65,8 @@ public class DictionaryDecodeReadSupport<T> implements CarbonReadSupport<T> {
dataTypes[i] = carbonColumns[i].getDataType();
dictionaries[i] = forwardDictionaryCache.get(new DictionaryColumnUniqueIdentifier(
absoluteTableIdentifier.getCarbonTableIdentifier(),
- carbonColumns[i].getColumnIdentifier(), dataTypes[i]));
+ carbonColumns[i].getColumnIdentifier(), dataTypes[i],
+ CarbonStorePath.getCarbonTablePath(absoluteTableIdentifier)));
} else {
dataTypes[i] = carbonColumns[i].getDataType();
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java
index 0dc79fa..a559cc4 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java
@@ -26,6 +26,7 @@ import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.scan.filter.FilterExpressionProcessor;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.filter.intf.FilterOptimizer;
import org.apache.carbondata.core.scan.filter.intf.FilterOptimizerBasic;
import org.apache.carbondata.core.scan.filter.optimizer.RangeFilterOptmizer;
@@ -131,11 +132,12 @@ public class CarbonInputFormatUtil {
* @return
*/
public static FilterResolverIntf resolveFilter(Expression filterExpression,
- AbsoluteTableIdentifier absoluteTableIdentifier) {
+ AbsoluteTableIdentifier absoluteTableIdentifier, TableProvider tableProvider) {
try {
FilterExpressionProcessor filterExpressionProcessor = new FilterExpressionProcessor();
//get resolved filter
- return filterExpressionProcessor.getFilterResolver(filterExpression, absoluteTableIdentifier);
+ return filterExpressionProcessor
+ .getFilterResolver(filterExpression, absoluteTableIdentifier, tableProvider);
} catch (Exception e) {
throw new RuntimeException("Error while resolving filter expression", e);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/hadoop/src/test/java/org/apache/carbondata/hadoop/test/util/StoreCreator.java
----------------------------------------------------------------------
diff --git a/hadoop/src/test/java/org/apache/carbondata/hadoop/test/util/StoreCreator.java b/hadoop/src/test/java/org/apache/carbondata/hadoop/test/util/StoreCreator.java
index cf44c6f..beca50d 100644
--- a/hadoop/src/test/java/org/apache/carbondata/hadoop/test/util/StoreCreator.java
+++ b/hadoop/src/test/java/org/apache/carbondata/hadoop/test/util/StoreCreator.java
@@ -325,9 +325,13 @@ public class StoreCreator {
.createCache(CacheType.REVERSE_DICTIONARY, absoluteTableIdentifier.getStorePath());
for (int i = 0; i < set.length; i++) {
ColumnIdentifier columnIdentifier = new ColumnIdentifier(dims.get(i).getColumnId(), null, null);
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier =
+ new DictionaryColumnUniqueIdentifier(table.getCarbonTableIdentifier(), columnIdentifier,
+ columnIdentifier.getDataType(), CarbonStorePath
+ .getCarbonTablePath(table.getStorePath(), table.getCarbonTableIdentifier()));
CarbonDictionaryWriter writer =
new CarbonDictionaryWriterImpl(absoluteTableIdentifier.getStorePath(),
- absoluteTableIdentifier.getCarbonTableIdentifier(), columnIdentifier);
+ absoluteTableIdentifier.getCarbonTableIdentifier(), dictionaryColumnUniqueIdentifier);
for (String value : set[i]) {
writer.write(value);
}
@@ -335,7 +339,8 @@ public class StoreCreator {
writer.commit();
Dictionary dict = (Dictionary) dictCache.get(
new DictionaryColumnUniqueIdentifier(absoluteTableIdentifier.getCarbonTableIdentifier(),
- columnIdentifier, dims.get(i).getDataType()));
+ columnIdentifier, dims.get(i).getDataType(),CarbonStorePath
+ .getCarbonTablePath(table.getStorePath(), table.getCarbonTableIdentifier())));
CarbonDictionarySortInfoPreparator preparator =
new CarbonDictionarySortInfoPreparator();
List<String> newDistinctValues = new ArrayList<String>();
@@ -343,7 +348,7 @@ public class StoreCreator {
preparator.getDictionarySortInfo(newDistinctValues, dict, dims.get(i).getDataType());
CarbonDictionarySortIndexWriter carbonDictionaryWriter =
new CarbonDictionarySortIndexWriterImpl(
- absoluteTableIdentifier.getCarbonTableIdentifier(), columnIdentifier,
+ absoluteTableIdentifier.getCarbonTableIdentifier(), dictionaryColumnUniqueIdentifier,
absoluteTableIdentifier.getStorePath());
try {
carbonDictionaryWriter.writeSortIndex(dictionarySortInfo.getSortIndex());
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonDictionaryDecodeReadSupport.java
----------------------------------------------------------------------
diff --git a/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonDictionaryDecodeReadSupport.java b/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonDictionaryDecodeReadSupport.java
index f08b92b..5eae253 100644
--- a/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonDictionaryDecodeReadSupport.java
+++ b/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonDictionaryDecodeReadSupport.java
@@ -33,6 +33,7 @@ import org.apache.carbondata.core.metadata.encoder.Encoding;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
import org.apache.carbondata.hadoop.readsupport.CarbonReadSupport;
@@ -88,7 +89,8 @@ public class CarbonDictionaryDecodeReadSupport<T> implements CarbonReadSupport<T
dataTypes[i] = carbonColumns[i].getDataType();
dictionaries[i] = forwardDictionaryCache.get(
new DictionaryColumnUniqueIdentifier(absoluteTableIdentifier.getCarbonTableIdentifier(),
- carbonColumns[i].getColumnIdentifier(), dataTypes[i]));
+ carbonColumns[i].getColumnIdentifier(), dataTypes[i],
+ CarbonStorePath.getCarbonTablePath(absoluteTableIdentifier)));
} else {
dataTypes[i] = carbonColumns[i].getDataType();
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/hive/src/main/java/org/apache/carbondata/hive/MapredCarbonInputFormat.java
----------------------------------------------------------------------
diff --git a/integration/hive/src/main/java/org/apache/carbondata/hive/MapredCarbonInputFormat.java b/integration/hive/src/main/java/org/apache/carbondata/hive/MapredCarbonInputFormat.java
index 86ebc0d..8f3fdce 100644
--- a/integration/hive/src/main/java/org/apache/carbondata/hive/MapredCarbonInputFormat.java
+++ b/integration/hive/src/main/java/org/apache/carbondata/hive/MapredCarbonInputFormat.java
@@ -24,6 +24,8 @@ import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.scan.expression.Expression;
+import org.apache.carbondata.core.scan.filter.SingleTableProvider;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.scan.model.CarbonQueryPlan;
import org.apache.carbondata.core.scan.model.QueryModel;
@@ -120,6 +122,7 @@ public class MapredCarbonInputFormat extends CarbonInputFormat<ArrayWritable>
private QueryModel getQueryModel(Configuration configuration, String path) throws IOException {
CarbonTable carbonTable = getCarbonTable(configuration, path);
+ TableProvider tableProvider = new SingleTableProvider(carbonTable);
// getting the table absoluteTableIdentifier from the carbonTable
// to avoid unnecessary deserialization
@@ -133,7 +136,8 @@ public class MapredCarbonInputFormat extends CarbonInputFormat<ArrayWritable>
// set the filter to the query model in order to filter blocklet before scan
Expression filter = getFilterPredicates(configuration);
CarbonInputFormatUtil.processFilterExpression(filter, carbonTable);
- FilterResolverIntf filterIntf = CarbonInputFormatUtil.resolveFilter(filter, identifier);
+ FilterResolverIntf filterIntf =
+ CarbonInputFormatUtil.resolveFilter(filter, identifier, tableProvider);
queryModel.setFilterExpressionResolverTree(filterIntf);
return queryModel;
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/presto/src/main/java/org/apache/carbondata/presto/CarbondataRecordSetProvider.java
----------------------------------------------------------------------
diff --git a/integration/presto/src/main/java/org/apache/carbondata/presto/CarbondataRecordSetProvider.java b/integration/presto/src/main/java/org/apache/carbondata/presto/CarbondataRecordSetProvider.java
index 3385170..8aacf88 100755
--- a/integration/presto/src/main/java/org/apache/carbondata/presto/CarbondataRecordSetProvider.java
+++ b/integration/presto/src/main/java/org/apache/carbondata/presto/CarbondataRecordSetProvider.java
@@ -40,6 +40,8 @@ import org.apache.carbondata.core.scan.expression.LiteralExpression;
import org.apache.carbondata.core.scan.expression.conditional.*;
import org.apache.carbondata.core.scan.expression.logical.AndExpression;
import org.apache.carbondata.core.scan.expression.logical.OrExpression;
+import org.apache.carbondata.core.scan.filter.SingleTableProvider;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.model.CarbonQueryPlan;
import org.apache.carbondata.core.scan.model.QueryModel;
import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil;
@@ -241,10 +243,11 @@ public class CarbondataRecordSetProvider implements ConnectorRecordSetProvider {
} else if (tmp.size() == 1) finalFilters = tmp.get(0);
else return;
+ TableProvider tableProvider = new SingleTableProvider(carbonTable);
// todo set into QueryModel
CarbonInputFormatUtil.processFilterExpression(finalFilters, carbonTable);
- queryModel.setFilterExpressionResolverTree(
- CarbonInputFormatUtil.resolveFilter(finalFilters, queryModel.getAbsoluteTableIdentifier()));
+ queryModel.setFilterExpressionResolverTree(CarbonInputFormatUtil
+ .resolveFilter(finalFilters, queryModel.getAbsoluteTableIdentifier(), tableProvider));
}
public static DataType spi2CarbondataTypeMapper(CarbondataColumnHandle carbondataColumnHandle) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/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 6ddb8be..d78f786 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
@@ -61,6 +61,8 @@ import org.apache.carbondata.core.reader.ThriftReader;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.scan.filter.FilterExpressionProcessor;
import org.apache.carbondata.core.scan.filter.FilterUtil;
+import org.apache.carbondata.core.scan.filter.SingleTableProvider;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.service.impl.PathFactory;
import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
@@ -298,8 +300,8 @@ public class CarbonTableReader {
new CarbonTableIdentifier(table.getSchemaName(), table.getTableName(),
UUID.randomUUID().toString());
// get the store path of the table.
- cache.carbonTablePath =
- PathFactory.getInstance().getCarbonTablePath(storePath, cache.carbonTableIdentifier);
+ cache.carbonTablePath = PathFactory.getInstance()
+ .getCarbonTablePath(storePath, cache.carbonTableIdentifier, null);
// cache the table
cc.put(table, cache);
@@ -385,10 +387,13 @@ public class CarbonTableReader {
cacheClient.getSegmentAccessClient().invalidateAll(invalidSegmentsIds);
}
+ TableProvider tableProvider = new SingleTableProvider(tableCacheModel.carbonTable);
+
// get filter for segment
CarbonInputFormatUtil.processFilterExpression(filters, tableCacheModel.carbonTable);
FilterResolverIntf filterInterface = CarbonInputFormatUtil
- .resolveFilter(filters, tableCacheModel.carbonTable.getAbsoluteTableIdentifier());
+ .resolveFilter(filters, tableCacheModel.carbonTable.getAbsoluteTableIdentifier(),
+ tableProvider);
IUDTable = (updateStatusManager.getUpdateStatusDetails().length != 0);
List<CarbonLocalInputSplit> result = new ArrayList<>();
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/presto/src/main/scala/org/apache/carbondata/presto/CarbonDictionaryDecodeSupport.scala
----------------------------------------------------------------------
diff --git a/integration/presto/src/main/scala/org/apache/carbondata/presto/CarbonDictionaryDecodeSupport.scala b/integration/presto/src/main/scala/org/apache/carbondata/presto/CarbonDictionaryDecodeSupport.scala
index fbdfebd..a3244ae 100644
--- a/integration/presto/src/main/scala/org/apache/carbondata/presto/CarbonDictionaryDecodeSupport.scala
+++ b/integration/presto/src/main/scala/org/apache/carbondata/presto/CarbonDictionaryDecodeSupport.scala
@@ -22,7 +22,7 @@ import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.metadata.datatype.DataType
import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn
-
+import org.apache.carbondata.core.util.path.CarbonStorePath
class CarbonDictionaryDecodeReaderSupport[T] {
@@ -38,7 +38,8 @@ class CarbonDictionaryDecodeReaderSupport[T] {
val dict: Dictionary = forwardDictionaryCache
.get(new DictionaryColumnUniqueIdentifier(absoluteTableIdentifier.getCarbonTableIdentifier,
carbonColumn.getColumnIdentifier,
- carbonColumn.getDataType))
+ carbonColumn.getColumnIdentifier.getDataType,
+ CarbonStorePath.getCarbonTablePath(absoluteTableIdentifier)))
(carbonColumn.getDataType, dict, index)
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java
index 24c09ca..7c2d157 100644
--- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java
+++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java
@@ -423,7 +423,8 @@ public final class CarbonLoaderUtil {
ColumnIdentifier columnIdentifier, String carbonStorePath, DataType dataType)
throws IOException {
return getDictionary(
- new DictionaryColumnUniqueIdentifier(tableIdentifier, columnIdentifier, dataType),
+ new DictionaryColumnUniqueIdentifier(tableIdentifier, columnIdentifier, dataType,
+ CarbonStorePath.getCarbonTablePath(carbonStorePath, tableIdentifier)),
carbonStorePath);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala
index a7b8143..c7ed1c7 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala
@@ -34,7 +34,7 @@ import org.apache.spark.rdd.RDD
import org.apache.spark.sql.Row
import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.cache.dictionary.Dictionary
+import org.apache.carbondata.core.cache.dictionary.{Dictionary, DictionaryColumnUniqueIdentifier}
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.locks.{CarbonLockFactory, LockUsage}
@@ -42,7 +42,7 @@ import org.apache.carbondata.core.metadata.{CarbonTableIdentifier, ColumnIdentif
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension
import org.apache.carbondata.core.service.{CarbonCommonFactory, PathService}
import org.apache.carbondata.core.util.{CarbonProperties, CarbonTimeStatisticsFactory, CarbonUtil}
-import org.apache.carbondata.core.util.path.CarbonTablePath
+import org.apache.carbondata.core.util.path.{CarbonStorePath, CarbonTablePath}
import org.apache.carbondata.processing.model.CarbonLoadModel
import org.apache.carbondata.spark.load.CarbonLoaderUtil
import org.apache.carbondata.spark.tasks.{DictionaryWriterTask, SortIndexWriterTask}
@@ -341,9 +341,16 @@ class CarbonGlobalDictionaryGenerateRDD(
var dictionaryForDistinctValueLookUp: Dictionary = _
var dictionaryForSortIndexWriting: Dictionary = _
var dictionaryForDistinctValueLookUpCleared: Boolean = false
+ val dictionaryColumnUniqueIdentifier: DictionaryColumnUniqueIdentifier = new
+ DictionaryColumnUniqueIdentifier(
+ model.table,
+ model.columnIdentifier(split.index),
+ model.columnIdentifier(split.index).getDataType,
+ CarbonStorePath.getCarbonTablePath(model.hdfsLocation, model.table))
val pathService: PathService = CarbonCommonFactory.getPathService
val carbonTablePath: CarbonTablePath =
- pathService.getCarbonTablePath(model.hdfsLocation, model.table)
+ pathService
+ .getCarbonTablePath(model.hdfsLocation, model.table, dictionaryColumnUniqueIdentifier)
if (StringUtils.isNotBlank(model.hdfsTempLocation)) {
CarbonProperties.getInstance.addProperty(CarbonCommonConstants.HDFS_TEMP_LOCATION,
model.hdfsTempLocation)
@@ -402,7 +409,7 @@ class CarbonGlobalDictionaryGenerateRDD(
val dictWriteTask = new DictionaryWriterTask(valuesBuffer,
dictionaryForDistinctValueLookUp,
model.table,
- model.columnIdentifier(split.index),
+ dictionaryColumnUniqueIdentifier,
model.hdfsLocation,
model.primDimensions(split.index).getColumnSchema,
isDictFileExists
@@ -414,7 +421,7 @@ class CarbonGlobalDictionaryGenerateRDD(
// if new data came than rewrite sort index file
if (distinctValues.size() > 0) {
val sortIndexWriteTask = new SortIndexWriterTask(model.table,
- model.columnIdentifier(split.index),
+ dictionaryColumnUniqueIdentifier,
model.primDimensions(split.index).getDataType,
model.hdfsLocation,
dictionaryForDistinctValueLookUp,
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala
index 2b1ccdf..a9ac9f1 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala
@@ -20,7 +20,7 @@ import java.io.IOException
import scala.collection.mutable
-import org.apache.carbondata.core.cache.dictionary.Dictionary
+import org.apache.carbondata.core.cache.dictionary.{Dictionary, DictionaryColumnUniqueIdentifier}
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.metadata.{CarbonTableIdentifier, ColumnIdentifier}
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
@@ -33,7 +33,7 @@ import org.apache.carbondata.core.writer.CarbonDictionaryWriter
* @param valuesBuffer
* @param dictionary
* @param carbonTableIdentifier
- * @param columnIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @param carbonStoreLocation
* @param columnSchema
* @param isDictionaryFileExist
@@ -42,7 +42,7 @@ import org.apache.carbondata.core.writer.CarbonDictionaryWriter
class DictionaryWriterTask(valuesBuffer: mutable.HashSet[String],
dictionary: Dictionary,
carbonTableIdentifier: CarbonTableIdentifier,
- columnIdentifier: ColumnIdentifier,
+ dictionaryColumnUniqueIdentifier: DictionaryColumnUniqueIdentifier,
carbonStoreLocation: String,
columnSchema: ColumnSchema,
isDictionaryFileExist: Boolean,
@@ -59,7 +59,7 @@ class DictionaryWriterTask(valuesBuffer: mutable.HashSet[String],
val dictService = CarbonCommonFactory.getDictionaryService
writer = dictService.getDictionaryWriter(
carbonTableIdentifier,
- columnIdentifier,
+ dictionaryColumnUniqueIdentifier,
carbonStoreLocation)
val distinctValues: java.util.List[String] = new java.util.ArrayList()
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/SortIndexWriterTask.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/SortIndexWriterTask.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/SortIndexWriterTask.scala
index c0aa0f9..27f9418 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/SortIndexWriterTask.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/SortIndexWriterTask.scala
@@ -16,7 +16,7 @@
*/
package org.apache.carbondata.spark.tasks
-import org.apache.carbondata.core.cache.dictionary.Dictionary
+import org.apache.carbondata.core.cache.dictionary.{Dictionary, DictionaryColumnUniqueIdentifier}
import org.apache.carbondata.core.metadata.{CarbonTableIdentifier, ColumnIdentifier}
import org.apache.carbondata.core.metadata.datatype.DataType
import org.apache.carbondata.core.service.CarbonCommonFactory
@@ -26,7 +26,7 @@ import org.apache.carbondata.core.writer.sortindex.{CarbonDictionarySortIndexWri
* This task writes sort index file
*
* @param carbonTableIdentifier
- * @param columnIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @param dataType
* @param carbonStoreLocation
* @param dictionary
@@ -35,7 +35,7 @@ import org.apache.carbondata.core.writer.sortindex.{CarbonDictionarySortIndexWri
*/
class SortIndexWriterTask(
carbonTableIdentifier: CarbonTableIdentifier,
- columnIdentifier: ColumnIdentifier,
+ dictionaryColumnUniqueIdentifier: DictionaryColumnUniqueIdentifier,
dataType: DataType,
carbonStoreLocation: String,
dictionary: Dictionary,
@@ -50,7 +50,8 @@ class SortIndexWriterTask(
preparator.getDictionarySortInfo(distinctValues, dictionary,
dataType)
carbonDictionarySortIndexWriter =
- dictService.getDictionarySortIndexWriter(carbonTableIdentifier, columnIdentifier,
+ dictService
+ .getDictionarySortIndexWriter(carbonTableIdentifier, dictionaryColumnUniqueIdentifier,
carbonStoreLocation)
carbonDictionarySortIndexWriter.writeSortIndex(dictionarySortInfo.getSortIndex)
carbonDictionarySortIndexWriter
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
index e1c564d..1f7862b 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
@@ -39,7 +39,7 @@ import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.util.FileUtils
import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.cache.dictionary.Dictionary
+import org.apache.carbondata.core.cache.dictionary.{Dictionary, DictionaryColumnUniqueIdentifier}
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.locks.{CarbonLockFactory, LockUsage}
@@ -169,9 +169,15 @@ object GlobalDictionaryUtil {
columnIndex: Int,
iter: Iterator[String]): Unit = {
val dictService = CarbonCommonFactory.getDictionaryService
- val writer: CarbonDictionaryWriter = dictService.getDictionaryWriter(
+ val dictionaryColumnUniqueIdentifier: DictionaryColumnUniqueIdentifier = new
+ DictionaryColumnUniqueIdentifier(
model.table,
model.columnIdentifier(columnIndex),
+ model.columnIdentifier(columnIndex).getDataType,
+ CarbonStorePath.getCarbonTablePath(model.hdfsLocation, model.table))
+ val writer: CarbonDictionaryWriter = dictService.getDictionaryWriter(
+ model.table,
+ dictionaryColumnUniqueIdentifier,
model.hdfsLocation
)
try {
@@ -207,10 +213,16 @@ object GlobalDictionaryUtil {
val dictMap = new HashMap[String, HashSet[String]]
val dictService = CarbonCommonFactory.getDictionaryService
for (i <- model.primDimensions.indices) {
+ val dictionaryColumnUniqueIdentifier: DictionaryColumnUniqueIdentifier = new
+ DictionaryColumnUniqueIdentifier(
+ model.table,
+ model.columnIdentifier(i),
+ model.columnIdentifier(i).getDataType,
+ CarbonStorePath.getCarbonTablePath(model.hdfsLocation, model.table))
val set = new HashSet[String]
if (model.dictFileExists(i)) {
val reader: CarbonDictionaryReader = dictService.getDictionaryReader(model.table,
- model.columnIdentifier(i), model.hdfsLocation
+ dictionaryColumnUniqueIdentifier, model.hdfsLocation
)
val values = reader.read
if (values != null) {
@@ -835,6 +847,12 @@ object GlobalDictionaryUtil {
val columnIdentifier = new ColumnIdentifier(columnSchema.getColumnUniqueId,
null,
columnSchema.getDataType)
+ val dictionaryColumnUniqueIdentifier: DictionaryColumnUniqueIdentifier = new
+ DictionaryColumnUniqueIdentifier(
+ tableIdentifier,
+ columnIdentifier,
+ columnIdentifier.getDataType,
+ carbonTablePath)
val parsedValue = DataTypeUtil.normalizeColumnValueForItsDataType(defaultValue, columnSchema)
val valuesBuffer = new mutable.HashSet[String]
if (null != parsedValue) {
@@ -843,7 +861,7 @@ object GlobalDictionaryUtil {
val dictWriteTask = new DictionaryWriterTask(valuesBuffer,
dictionary,
tableIdentifier,
- columnIdentifier,
+ dictionaryColumnUniqueIdentifier,
storePath,
columnSchema,
false
@@ -855,7 +873,7 @@ object GlobalDictionaryUtil {
if (distinctValues.size() > 0) {
val sortIndexWriteTask = new SortIndexWriterTask(tableIdentifier,
- columnIdentifier,
+ dictionaryColumnUniqueIdentifier,
columnSchema.getDataType,
storePath,
dictionary,
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
index 621a960..37505d0 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
@@ -35,8 +35,8 @@ import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, ColumnIdent
import org.apache.carbondata.core.metadata.datatype.DataType
import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension
-import org.apache.carbondata.core.stats._
import org.apache.carbondata.core.util.{CarbonTimeStatisticsFactory, DataTypeUtil}
+import org.apache.carbondata.core.util.path.CarbonStorePath
import org.apache.carbondata.spark.CarbonAliasDecoderRelation
/**
@@ -224,7 +224,8 @@ case class CarbonDictionaryDecoder(
if (dictionaryId._2 != null) {
new DictionaryColumnUniqueIdentifier(
atiMap(dictionaryId._1).getCarbonTableIdentifier,
- dictionaryId._2, dictionaryId._3)
+ dictionaryId._2, dictionaryId._3,
+ CarbonStorePath.getCarbonTablePath(atiMap(dictionaryId._1)))
} else {
null
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/spark/src/test/scala/org/apache/carbondata/spark/util/DictionaryTestCaseUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/test/scala/org/apache/carbondata/spark/util/DictionaryTestCaseUtil.scala b/integration/spark/src/test/scala/org/apache/carbondata/spark/util/DictionaryTestCaseUtil.scala
index c0dfc68..e2b185e 100644
--- a/integration/spark/src/test/scala/org/apache/carbondata/spark/util/DictionaryTestCaseUtil.scala
+++ b/integration/spark/src/test/scala/org/apache/carbondata/spark/util/DictionaryTestCaseUtil.scala
@@ -23,6 +23,7 @@ import org.apache.spark.sql.test.TestQueryExecutor
import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier
import org.apache.carbondata.core.metadata.CarbonTableIdentifier
import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.util.path.CarbonStorePath
import org.apache.carbondata.spark.load.CarbonLoaderUtil
/**
@@ -41,7 +42,8 @@ object DictionaryTestCaseUtil {
val dimension = table.getDimensionByName(table.getFactTableName, columnName)
val tableIdentifier = new CarbonTableIdentifier(table.getDatabaseName, table.getFactTableName, "uniqueid")
val columnIdentifier = new DictionaryColumnUniqueIdentifier(tableIdentifier,
- dimension.getColumnIdentifier, dimension.getDataType
+ dimension.getColumnIdentifier, dimension.getDataType,
+ CarbonStorePath.getCarbonTablePath(table.getStorePath, table.getCarbonTableIdentifier)
)
val dict = CarbonLoaderUtil.getDictionary(columnIdentifier, TestQueryExecutor.storeLocation)
assert(dict.getSurrogateKey(value) != CarbonCommonConstants.INVALID_SURROGATE_KEY)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/spark/src/test/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtilConcurrentTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/test/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtilConcurrentTestCase.scala b/integration/spark/src/test/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtilConcurrentTestCase.scala
index a35f88b..a126686 100644
--- a/integration/spark/src/test/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtilConcurrentTestCase.scala
+++ b/integration/spark/src/test/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtilConcurrentTestCase.scala
@@ -24,10 +24,13 @@ import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.spark.sql.test.util.QueryTest
import org.apache.spark.sql.{CarbonEnv, CarbonRelation}
import org.scalatest.BeforeAndAfterAll
+
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier
import org.apache.carbondata.core.service.impl.PathFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.util.CarbonProperties
+import org.apache.carbondata.core.util.path.CarbonStorePath
import org.apache.carbondata.processing.constants.TableOptionConstant
import org.apache.carbondata.processing.model.{CarbonDataLoadSchema, CarbonLoadModel}
@@ -130,8 +133,13 @@ class GlobalDictionaryUtilConcurrentTestCase extends QueryTest with BeforeAndAft
}
val carbonTableIdentifier = sampleRelation.tableMeta.carbonTable.getCarbonTableIdentifier
val columnIdentifier = sampleRelation.tableMeta.carbonTable.getDimensionByName("employee", "empid").getColumnIdentifier
+ val dictionaryColumnUniqueIdentifier = new DictionaryColumnUniqueIdentifier(
+ carbonTableIdentifier,
+ columnIdentifier,
+ columnIdentifier.getDataType,
+ CarbonStorePath.getCarbonTablePath(storeLocation, carbonTableIdentifier))
val carbonTablePath = PathFactory.getInstance()
- .getCarbonTablePath(sampleRelation.tableMeta.storePath, carbonTableIdentifier)
+ .getCarbonTablePath(sampleRelation.tableMeta.storePath, carbonTableIdentifier, dictionaryColumnUniqueIdentifier)
val dictPath = carbonTablePath.getDictionaryFilePath(columnIdentifier.getColumnId)
val dictFile = FileFactory.getCarbonFile(dictPath, FileFactory.getFileType(dictPath))
val offSet = dictFile.getSize
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
index 7e27d1b..1addd03 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
@@ -39,6 +39,7 @@ import org.apache.carbondata.core.metadata.datatype.DataType
import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension
import org.apache.carbondata.core.util.DataTypeUtil
+import org.apache.carbondata.core.util.path.CarbonStorePath
import org.apache.carbondata.spark.CarbonAliasDecoderRelation
import org.apache.carbondata.spark.rdd.{CarbonRDD, CarbonRDDWithTableInfo}
import org.apache.carbondata.spark.util.SparkDataTypeConverterImpl
@@ -248,7 +249,8 @@ case class CarbonDictionaryDecoder(
try {
cache.get(new DictionaryColumnUniqueIdentifier(
atiMap(f._1).getCarbonTableIdentifier,
- f._2, f._3.getDataType))
+ f._2, f._3.getDataType,
+ CarbonStorePath.getCarbonTablePath(atiMap(f._1))))
} catch {
case _: Throwable => null
}
@@ -268,7 +270,8 @@ case class CarbonDictionaryDecoder(
try {
val dictionaryColumnUniqueIdentifier = new DictionaryColumnUniqueIdentifier(
atiMap(tableName).getCarbonTableIdentifier,
- columnIdentifier, carbonDimension.getDataType)
+ columnIdentifier, carbonDimension.getDataType,
+ CarbonStorePath.getCarbonTablePath(atiMap(tableName)))
allDictIdentifiers += dictionaryColumnUniqueIdentifier;
new ForwardDictionaryWrapper(
storePath,
@@ -566,7 +569,8 @@ class CarbonDecoderRDD(
try {
cache.get(new DictionaryColumnUniqueIdentifier(
atiMap(f._1).getCarbonTableIdentifier,
- f._2, f._3.getDataType))
+ f._2, f._3.getDataType,
+ CarbonStorePath.getCarbonTablePath(atiMap(f._1))))
} catch {
case _: Throwable => null
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/DictionaryTestCaseUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/DictionaryTestCaseUtil.scala b/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/DictionaryTestCaseUtil.scala
index 62b0aff..e7eb422 100644
--- a/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/DictionaryTestCaseUtil.scala
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/DictionaryTestCaseUtil.scala
@@ -23,6 +23,7 @@ import org.apache.spark.sql.test.TestQueryExecutor
import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.metadata.CarbonTableIdentifier
+import org.apache.carbondata.core.util.path.CarbonStorePath
import org.apache.carbondata.spark.load.CarbonLoaderUtil
/**
@@ -41,7 +42,8 @@ object DictionaryTestCaseUtil {
val dimension = table.getDimensionByName(table.getFactTableName, columnName)
val tableIdentifier = new CarbonTableIdentifier(table.getDatabaseName, table.getFactTableName, "uniqueid")
val columnIdentifier = new DictionaryColumnUniqueIdentifier(tableIdentifier,
- dimension.getColumnIdentifier, dimension.getDataType
+ dimension.getColumnIdentifier, dimension.getDataType,
+ CarbonStorePath.getCarbonTablePath(table.getAbsoluteTableIdentifier)
)
val dict = CarbonLoaderUtil.getDictionary(columnIdentifier, TestQueryExecutor.storeLocation)
assert(dict.getSurrogateKey(value) != CarbonCommonConstants.INVALID_SURROGATE_KEY)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/processing/src/main/java/org/apache/carbondata/processing/datatypes/PrimitiveDataType.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/datatypes/PrimitiveDataType.java b/processing/src/main/java/org/apache/carbondata/processing/datatypes/PrimitiveDataType.java
index 729f9e3..8373b58 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/datatypes/PrimitiveDataType.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/datatypes/PrimitiveDataType.java
@@ -41,6 +41,7 @@ import org.apache.carbondata.core.metadata.encoder.Encoding;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.DataTypeUtil;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
import org.apache.carbondata.processing.newflow.dictionary.DictionaryServerClientDictionary;
import org.apache.carbondata.processing.newflow.dictionary.DirectDictionary;
import org.apache.carbondata.processing.newflow.dictionary.PreCreatedDictionary;
@@ -119,7 +120,8 @@ public class PrimitiveDataType implements GenericDataType<Object> {
this.carbonDimension = carbonDimension;
DictionaryColumnUniqueIdentifier identifier =
new DictionaryColumnUniqueIdentifier(carbonTableIdentifier,
- carbonDimension.getColumnIdentifier(), carbonDimension.getDataType());
+ carbonDimension.getColumnIdentifier(), carbonDimension.getDataType(),
+ CarbonStorePath.getCarbonTablePath(storePath, carbonTableIdentifier));
try {
if (carbonDimension.hasEncoding(Encoding.DIRECT_DICTIONARY)) {
dictionaryGenerator = new DirectDictionary(DirectDictionaryKeyGeneratorFactory
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/processing/src/main/java/org/apache/carbondata/processing/newflow/converter/impl/DictionaryFieldConverterImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/newflow/converter/impl/DictionaryFieldConverterImpl.java b/processing/src/main/java/org/apache/carbondata/processing/newflow/converter/impl/DictionaryFieldConverterImpl.java
index 5f7bd02..2614e17 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/newflow/converter/impl/DictionaryFieldConverterImpl.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/newflow/converter/impl/DictionaryFieldConverterImpl.java
@@ -35,6 +35,7 @@ import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.DataTypeUtil;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
import org.apache.carbondata.processing.newflow.DataField;
import org.apache.carbondata.processing.newflow.converter.BadRecordLogHolder;
import org.apache.carbondata.processing.newflow.dictionary.DictionaryServerClientDictionary;
@@ -69,7 +70,8 @@ public class DictionaryFieldConverterImpl extends AbstractDictionaryFieldConvert
this.isEmptyBadRecord = isEmptyBadRecord;
DictionaryColumnUniqueIdentifier identifier =
new DictionaryColumnUniqueIdentifier(carbonTableIdentifier,
- dataField.getColumn().getColumnIdentifier(), dataField.getColumn().getDataType());
+ dataField.getColumn().getColumnIdentifier(), dataField.getColumn().getDataType(),
+ CarbonStorePath.getCarbonTablePath(storePath, carbonTableIdentifier));
// if use one pass, use DictionaryServerClientDictionary
if (useOnePass) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/processing/src/test/java/org/apache/carbondata/processing/StoreCreator.java
----------------------------------------------------------------------
diff --git a/processing/src/test/java/org/apache/carbondata/processing/StoreCreator.java b/processing/src/test/java/org/apache/carbondata/processing/StoreCreator.java
index aad0d3f..3bb186e 100644
--- a/processing/src/test/java/org/apache/carbondata/processing/StoreCreator.java
+++ b/processing/src/test/java/org/apache/carbondata/processing/StoreCreator.java
@@ -317,9 +317,11 @@ public class StoreCreator {
.createCache(CacheType.REVERSE_DICTIONARY, absoluteTableIdentifier.getStorePath());
for (int i = 0; i < set.length; i++) {
ColumnIdentifier columnIdentifier = new ColumnIdentifier(dims.get(i).getColumnId(), null, null);
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier = new DictionaryColumnUniqueIdentifier(table.getCarbonTableIdentifier(), columnIdentifier, columnIdentifier.getDataType(),
+ CarbonStorePath.getCarbonTablePath(table.getStorePath(), table.getCarbonTableIdentifier()));
CarbonDictionaryWriter writer =
new CarbonDictionaryWriterImpl(absoluteTableIdentifier.getStorePath(),
- absoluteTableIdentifier.getCarbonTableIdentifier(), columnIdentifier);
+ absoluteTableIdentifier.getCarbonTableIdentifier(), dictionaryColumnUniqueIdentifier);
for (String value : set[i]) {
writer.write(value);
}
@@ -327,7 +329,8 @@ public class StoreCreator {
writer.commit();
Dictionary dict = (Dictionary) dictCache.get(
new DictionaryColumnUniqueIdentifier(absoluteTableIdentifier.getCarbonTableIdentifier(),
- columnIdentifier, dims.get(i).getDataType()));
+ columnIdentifier, dims.get(i).getDataType(),
+ CarbonStorePath.getCarbonTablePath(absoluteTableIdentifier)));
CarbonDictionarySortInfoPreparator preparator =
new CarbonDictionarySortInfoPreparator();
List<String> newDistinctValues = new ArrayList<String>();
@@ -335,7 +338,7 @@ public class StoreCreator {
preparator.getDictionarySortInfo(newDistinctValues, dict, dims.get(i).getDataType());
CarbonDictionarySortIndexWriter carbonDictionaryWriter =
new CarbonDictionarySortIndexWriterImpl(
- absoluteTableIdentifier.getCarbonTableIdentifier(), columnIdentifier,
+ absoluteTableIdentifier.getCarbonTableIdentifier(), dictionaryColumnUniqueIdentifier,
absoluteTableIdentifier.getStorePath());
try {
carbonDictionaryWriter.writeSortIndex(dictionarySortInfo.getSortIndex());
[2/2] carbondata git commit: [CARBONDATA-1308] Added tableProvider to
supply carbonTable wherever needed
Posted by gv...@apache.org.
[CARBONDATA-1308] Added tableProvider to supply carbonTable wherever needed
This closes #1208
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/d3a09e27
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/d3a09e27
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/d3a09e27
Branch: refs/heads/master
Commit: d3a09e2790ec1d130feee78cdce5357a02c11628
Parents: e3f98fa
Author: dhatchayani <dh...@gmail.com>
Authored: Fri Jul 28 17:03:22 2017 +0530
Committer: Venkata Ramana G <ra...@huawei.com>
Committed: Tue Aug 22 19:48:56 2017 +0530
----------------------------------------------------------------------
.../dictionary/AbstractDictionaryCache.java | 9 ++--
.../dictionary/DictionaryCacheLoaderImpl.java | 44 ++++++++++--------
.../DictionaryColumnUniqueIdentifier.java | 12 ++++-
.../IncrementalColumnDictionaryGenerator.java | 20 ++++++--
.../CarbonDictionaryMetadataReaderImpl.java | 20 ++++----
.../core/reader/CarbonDictionaryReaderImpl.java | 23 +++++-----
.../CarbonDictionarySortIndexReaderImpl.java | 34 ++++++++------
.../executor/impl/AbstractQueryExecutor.java | 9 +++-
.../core/scan/executor/util/QueryUtil.java | 24 ++++++----
.../scan/filter/FilterExpressionProcessor.java | 19 ++++----
.../core/scan/filter/FilterProcessor.java | 3 +-
.../carbondata/core/scan/filter/FilterUtil.java | 31 +++++++++++--
.../core/scan/filter/SingleTableProvider.java | 40 ++++++++++++++++
.../core/scan/filter/TableProvider.java | 29 ++++++++++++
.../resolver/ConditionalFilterResolverImpl.java | 9 +++-
.../filter/resolver/FilterResolverIntf.java | 3 +-
.../resolver/LogicalFilterResolverImpl.java | 5 +-
.../resolver/RowLevelFilterResolverImpl.java | 4 +-
.../RowLevelRangeFilterResolverImpl.java | 3 +-
.../metadata/FilterResolverMetadata.java | 11 +++++
.../TrueConditionalResolverImpl.java | 5 +-
.../visitor/DictionaryColumnVisitor.java | 48 ++++++++++----------
.../visitor/RangeDictionaryColumnVisitor.java | 29 ++++++------
.../core/service/DictionaryService.java | 28 ++++++------
.../carbondata/core/service/PathService.java | 5 +-
.../core/service/impl/DictionaryFactory.java | 48 ++++++++++----------
.../core/service/impl/PathFactory.java | 11 ++++-
.../apache/carbondata/core/util/CarbonUtil.java | 3 +-
.../core/writer/CarbonDictionaryWriterImpl.java | 37 ++++++++-------
.../CarbonDictionarySortIndexWriterImpl.java | 27 ++++++-----
.../dictionary/AbstractDictionaryCacheTest.java | 9 +++-
.../DictionaryCacheLoaderImplTest.java | 16 +++++--
.../DictionaryColumnUniqueIdentifierTest.java | 6 +--
.../dictionary/ForwardDictionaryCacheTest.java | 7 ++-
.../reader/CarbonDictionaryReaderImplTest.java | 8 +++-
...CarbonDictionarySortIndexReaderImplTest.java | 10 ++--
.../writer/CarbonDictionaryWriterImplTest.java | 13 ++++--
...CarbonDictionarySortIndexWriterImplTest.java | 10 ++--
.../examples/GenerateDictionaryExample.scala | 4 +-
.../carbondata/hadoop/CarbonInputFormat.java | 10 +++-
.../hadoop/api/CarbonTableInputFormat.java | 14 ++++--
.../impl/DictionaryDecodeReadSupport.java | 4 +-
.../hadoop/util/CarbonInputFormatUtil.java | 6 ++-
.../hadoop/test/util/StoreCreator.java | 11 +++--
.../hive/CarbonDictionaryDecodeReadSupport.java | 4 +-
.../hive/MapredCarbonInputFormat.java | 6 ++-
.../presto/CarbondataRecordSetProvider.java | 7 ++-
.../presto/impl/CarbonTableReader.java | 11 +++--
.../presto/CarbonDictionaryDecodeSupport.scala | 5 +-
.../carbondata/spark/load/CarbonLoaderUtil.java | 3 +-
.../spark/rdd/CarbonGlobalDictionaryRDD.scala | 17 +++++--
.../spark/tasks/DictionaryWriterTask.scala | 8 ++--
.../spark/tasks/SortIndexWriterTask.scala | 9 ++--
.../spark/util/GlobalDictionaryUtil.scala | 28 ++++++++++--
.../spark/sql/CarbonDictionaryDecoder.scala | 5 +-
.../spark/util/DictionaryTestCaseUtil.scala | 4 +-
...GlobalDictionaryUtilConcurrentTestCase.scala | 10 +++-
.../spark/sql/CarbonDictionaryDecoder.scala | 10 ++--
.../spark/util/DictionaryTestCaseUtil.scala | 4 +-
.../processing/datatypes/PrimitiveDataType.java | 4 +-
.../impl/DictionaryFieldConverterImpl.java | 4 +-
.../carbondata/processing/StoreCreator.java | 9 ++--
62 files changed, 592 insertions(+), 277 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java
index 6910888..4046364 100644
--- a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java
+++ b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java
@@ -93,7 +93,7 @@ public abstract class AbstractDictionaryCache<K extends DictionaryColumnUniqueId
DictionaryService dictService = CarbonCommonFactory.getDictionaryService();
CarbonDictionaryMetadataReader columnMetadataReaderImpl = dictService
.getDictionaryMetadataReader(dictionaryColumnUniqueIdentifier.getCarbonTableIdentifier(),
- dictionaryColumnUniqueIdentifier.getColumnIdentifier(), carbonStorePath);
+ dictionaryColumnUniqueIdentifier, carbonStorePath);
CarbonDictionaryColumnMetaChunk carbonDictionaryColumnMetaChunk = null;
// read metadata file
@@ -121,7 +121,7 @@ public abstract class AbstractDictionaryCache<K extends DictionaryColumnUniqueId
CarbonDictionaryMetadataReader columnMetadataReaderImpl = dictService
.getDictionaryMetadataReader(
dictionaryColumnUniqueIdentifier.getCarbonTableIdentifier(),
- dictionaryColumnUniqueIdentifier.getColumnIdentifier(), carbonStorePath);
+ dictionaryColumnUniqueIdentifier, carbonStorePath);
CarbonDictionaryColumnMetaChunk carbonDictionaryColumnMetaChunk = null;
// read metadata file
@@ -158,7 +158,8 @@ public abstract class AbstractDictionaryCache<K extends DictionaryColumnUniqueId
DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) throws IOException {
PathService pathService = CarbonCommonFactory.getPathService();
CarbonTablePath carbonTablePath = pathService.getCarbonTablePath(carbonStorePath,
- dictionaryColumnUniqueIdentifier.getCarbonTableIdentifier());
+ dictionaryColumnUniqueIdentifier.getCarbonTableIdentifier(),
+ dictionaryColumnUniqueIdentifier);
String dictionaryFilePath = carbonTablePath.getDictionaryMetaFilePath(
dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId());
FileFactory.FileType fileType = FileFactory.getFileType(dictionaryFilePath);
@@ -280,7 +281,7 @@ public abstract class AbstractDictionaryCache<K extends DictionaryColumnUniqueId
throws IOException {
DictionaryCacheLoader dictionaryCacheLoader =
new DictionaryCacheLoaderImpl(dictionaryColumnUniqueIdentifier.getCarbonTableIdentifier(),
- carbonStorePath);
+ carbonStorePath, dictionaryColumnUniqueIdentifier);
dictionaryCacheLoader
.load(dictionaryInfo, dictionaryColumnUniqueIdentifier.getColumnIdentifier(),
dictionaryChunkStartOffset, dictionaryChunkEndOffset, loadSortIndex);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/cache/dictionary/DictionaryCacheLoaderImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/DictionaryCacheLoaderImpl.java b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/DictionaryCacheLoaderImpl.java
index 5373ca8..e6a4022 100644
--- a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/DictionaryCacheLoaderImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/DictionaryCacheLoaderImpl.java
@@ -41,6 +41,8 @@ public class DictionaryCacheLoaderImpl implements DictionaryCacheLoader {
*/
private CarbonTableIdentifier carbonTableIdentifier;
+ private DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier;
+
/**
* carbon store path
*/
@@ -51,9 +53,10 @@ public class DictionaryCacheLoaderImpl implements DictionaryCacheLoader {
* @param carbonStorePath hdfs store path
*/
public DictionaryCacheLoaderImpl(CarbonTableIdentifier carbonTableIdentifier,
- String carbonStorePath) {
+ String carbonStorePath, DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
this.carbonTableIdentifier = carbonTableIdentifier;
this.carbonStorePath = carbonStorePath;
+ this.dictionaryColumnUniqueIdentifier = dictionaryColumnUniqueIdentifier;
}
/**
@@ -74,9 +77,10 @@ public class DictionaryCacheLoaderImpl implements DictionaryCacheLoader {
long dictionaryChunkStartOffset, long dictionaryChunkEndOffset, boolean loadSortIndex)
throws IOException {
Iterator<byte[]> columnDictionaryChunkWrapper =
- load(columnIdentifier, dictionaryChunkStartOffset, dictionaryChunkEndOffset);
+ load(dictionaryColumnUniqueIdentifier, dictionaryChunkStartOffset,
+ dictionaryChunkEndOffset);
if (loadSortIndex) {
- readSortIndexFile(dictionaryInfo, columnIdentifier);
+ readSortIndexFile(dictionaryInfo, dictionaryColumnUniqueIdentifier);
}
fillDictionaryValuesAndAddToDictionaryChunks(dictionaryInfo, columnDictionaryChunkWrapper);
}
@@ -118,15 +122,15 @@ public class DictionaryCacheLoaderImpl implements DictionaryCacheLoader {
/**
* This method will load the dictionary data between a given start and end offset
*
- * @param columnIdentifier column unique identifier
+ * @param dictionaryColumnUniqueIdentifier column unique identifier
* @param startOffset start offset of dictionary file
* @param endOffset end offset of dictionary file
* @return iterator over dictionary values
* @throws IOException
*/
- private Iterator<byte[]> load(ColumnIdentifier columnIdentifier, long startOffset, long endOffset)
- throws IOException {
- CarbonDictionaryReader dictionaryReader = getDictionaryReader(columnIdentifier);
+ private Iterator<byte[]> load(DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier,
+ long startOffset, long endOffset) throws IOException {
+ CarbonDictionaryReader dictionaryReader = getDictionaryReader(dictionaryColumnUniqueIdentifier);
try {
return dictionaryReader.read(startOffset, endOffset);
} finally {
@@ -138,12 +142,13 @@ public class DictionaryCacheLoaderImpl implements DictionaryCacheLoader {
* This method will read the sort index file and load into memory
*
* @param dictionaryInfo
- * @param columnIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @throws IOException
*/
- private void readSortIndexFile(DictionaryInfo dictionaryInfo, ColumnIdentifier columnIdentifier)
- throws IOException {
- CarbonDictionarySortIndexReader sortIndexReader = getSortIndexReader(columnIdentifier);
+ private void readSortIndexFile(DictionaryInfo dictionaryInfo,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) throws IOException {
+ CarbonDictionarySortIndexReader sortIndexReader =
+ getSortIndexReader(dictionaryColumnUniqueIdentifier);
try {
dictionaryInfo.setSortOrderIndex(sortIndexReader.readSortIndex());
dictionaryInfo.setSortReverseOrderIndex(sortIndexReader.readInvertedSortIndex());
@@ -155,22 +160,25 @@ public class DictionaryCacheLoaderImpl implements DictionaryCacheLoader {
/**
* This method will create a dictionary reader instance to read the dictionary file
*
- * @param columnIdentifier unique column identifier
+ * @param dictionaryColumnUniqueIdentifier unique column identifier
* @return carbon dictionary reader instance
*/
- private CarbonDictionaryReader getDictionaryReader(ColumnIdentifier columnIdentifier) {
+ private CarbonDictionaryReader getDictionaryReader(
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
DictionaryService dictService = CarbonCommonFactory.getDictionaryService();
- return dictService
- .getDictionaryReader(carbonTableIdentifier, columnIdentifier, carbonStorePath);
+ return dictService.getDictionaryReader(carbonTableIdentifier, dictionaryColumnUniqueIdentifier,
+ carbonStorePath);
}
/**
- * @param columnIdentifier unique column identifier
+ * @param dictionaryColumnUniqueIdentifier unique column identifier
* @return sort index reader instance
*/
- private CarbonDictionarySortIndexReader getSortIndexReader(ColumnIdentifier columnIdentifier) {
+ private CarbonDictionarySortIndexReader getSortIndexReader(
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
DictionaryService dictService = CarbonCommonFactory.getDictionaryService();
return dictService
- .getDictionarySortIndexReader(carbonTableIdentifier, columnIdentifier, carbonStorePath);
+ .getDictionarySortIndexReader(carbonTableIdentifier, dictionaryColumnUniqueIdentifier,
+ carbonStorePath);
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/cache/dictionary/DictionaryColumnUniqueIdentifier.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/DictionaryColumnUniqueIdentifier.java b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/DictionaryColumnUniqueIdentifier.java
index f0c5c5a..f0182e4 100644
--- a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/DictionaryColumnUniqueIdentifier.java
+++ b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/DictionaryColumnUniqueIdentifier.java
@@ -22,6 +22,7 @@ import java.io.Serializable;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.metadata.datatype.DataType;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
/**
* dictionary column identifier which includes table identifier and column identifier
@@ -38,6 +39,8 @@ public class DictionaryColumnUniqueIdentifier implements Serializable {
*/
private ColumnIdentifier columnIdentifier;
+ private transient CarbonTablePath carbonTablePath;
+
private DataType dataType;
/**
@@ -69,9 +72,12 @@ public class DictionaryColumnUniqueIdentifier implements Serializable {
* @param dataType
*/
public DictionaryColumnUniqueIdentifier(CarbonTableIdentifier carbonTableIdentifier,
- ColumnIdentifier columnIdentifier, DataType dataType) {
+ ColumnIdentifier columnIdentifier, DataType dataType, CarbonTablePath carbonTablePath) {
this(carbonTableIdentifier, columnIdentifier);
this.dataType = dataType;
+ if (null != carbonTablePath) {
+ this.carbonTablePath = carbonTablePath;
+ }
}
public DataType getDataType() {
@@ -85,6 +91,10 @@ public class DictionaryColumnUniqueIdentifier implements Serializable {
return carbonTableIdentifier;
}
+ public CarbonTablePath getCarbonTablePath() {
+ return carbonTablePath;
+ }
+
/**
* @return columnIdentifier
*/
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGenerator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGenerator.java b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGenerator.java
index 87c575f..20802be 100644
--- a/core/src/main/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGenerator.java
+++ b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGenerator.java
@@ -42,6 +42,8 @@ import org.apache.carbondata.core.service.CarbonCommonFactory;
import org.apache.carbondata.core.service.DictionaryService;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.DataTypeUtil;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.core.writer.CarbonDictionaryWriter;
import org.apache.carbondata.core.writer.sortindex.CarbonDictionarySortIndexWriter;
import org.apache.carbondata.core.writer.sortindex.CarbonDictionarySortInfo;
@@ -113,6 +115,8 @@ public class IncrementalColumnDictionaryGenerator implements BiDictionary<Intege
// initialize params
CarbonMetadata metadata = CarbonMetadata.getInstance();
CarbonTable carbonTable = metadata.getCarbonTable(tableUniqueName);
+ CarbonTablePath carbonTablePath =
+ CarbonStorePath.getCarbonTablePath(carbonTable.getAbsoluteTableIdentifier());
CarbonTableIdentifier tableIdentifier = carbonTable.getCarbonTableIdentifier();
ColumnIdentifier columnIdentifier = dimension.getColumnIdentifier();
String storePath = carbonTable.getStorePath();
@@ -120,7 +124,7 @@ public class IncrementalColumnDictionaryGenerator implements BiDictionary<Intege
// create dictionary cache from dictionary File
DictionaryColumnUniqueIdentifier identifier =
new DictionaryColumnUniqueIdentifier(tableIdentifier, columnIdentifier,
- columnIdentifier.getDataType());
+ columnIdentifier.getDataType(), carbonTablePath);
Boolean isDictExists = CarbonUtil.isFileExistsForGivenColumn(storePath, identifier);
Dictionary dictionary = null;
long t1 = System.currentTimeMillis();
@@ -134,7 +138,7 @@ public class IncrementalColumnDictionaryGenerator implements BiDictionary<Intege
// write dictionary
CarbonDictionaryWriter dictionaryWriter = null;
dictionaryWriter = dictionaryService
- .getDictionaryWriter(tableIdentifier, columnIdentifier, storePath);
+ .getDictionaryWriter(tableIdentifier, identifier, storePath);
List<String> distinctValues = writeDictionary(dictionaryWriter, isDictExists);
long dictWriteTime = System.currentTimeMillis() - t2;
long t3 = System.currentTimeMillis();
@@ -215,14 +219,20 @@ public class IncrementalColumnDictionaryGenerator implements BiDictionary<Intege
ColumnIdentifier columnIdentifier,
String storePath) throws IOException {
CarbonDictionarySortIndexWriter carbonDictionarySortIndexWriter = null;
+ CarbonTablePath carbonTablePath = CarbonStorePath
+ .getCarbonTablePath(storePath, tableIdentifier.getDatabaseName(),
+ tableIdentifier.getTableName());
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier =
+ new DictionaryColumnUniqueIdentifier(tableIdentifier, columnIdentifier,
+ columnIdentifier.getDataType(), carbonTablePath);
try {
CarbonDictionarySortInfoPreparator preparator = new CarbonDictionarySortInfoPreparator();
CarbonDictionarySortInfo dictionarySortInfo =
preparator.getDictionarySortInfo(distinctValues, dictionary,
dimension.getDataType());
- carbonDictionarySortIndexWriter =
- dictionaryService.getDictionarySortIndexWriter(tableIdentifier, columnIdentifier,
- storePath);
+ carbonDictionarySortIndexWriter = dictionaryService
+ .getDictionarySortIndexWriter(tableIdentifier, dictionaryColumnUniqueIdentifier,
+ storePath);
carbonDictionarySortIndexWriter.writeSortIndex(dictionarySortInfo.getSortIndex());
carbonDictionarySortIndexWriter
.writeInvertedSortIndex(dictionarySortInfo.getSortIndexInverted());
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/reader/CarbonDictionaryMetadataReaderImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/reader/CarbonDictionaryMetadataReaderImpl.java b/core/src/main/java/org/apache/carbondata/core/reader/CarbonDictionaryMetadataReaderImpl.java
index 9356974..e09fda0 100644
--- a/core/src/main/java/org/apache/carbondata/core/reader/CarbonDictionaryMetadataReaderImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/reader/CarbonDictionaryMetadataReaderImpl.java
@@ -21,9 +21,9 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
-import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.service.CarbonCommonFactory;
import org.apache.carbondata.core.service.PathService;
import org.apache.carbondata.core.util.path.CarbonTablePath;
@@ -49,7 +49,7 @@ public class CarbonDictionaryMetadataReaderImpl implements CarbonDictionaryMetad
/**
* column identifier
*/
- protected ColumnIdentifier columnIdentifier;
+ protected DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier;
/**
* dictionary metadata file path
@@ -66,13 +66,14 @@ public class CarbonDictionaryMetadataReaderImpl implements CarbonDictionaryMetad
*
* @param storePath carbon dictionary meta data store path
* @param carbonTableIdentifier table identifier which will give table name and database name
- * @param columnIdentifier column unique identifier
+ * @param dictionaryColumnUniqueIdentifier column unique identifier
*/
public CarbonDictionaryMetadataReaderImpl(String storePath,
- CarbonTableIdentifier carbonTableIdentifier, ColumnIdentifier columnIdentifier) {
+ CarbonTableIdentifier carbonTableIdentifier,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
this.storePath = storePath;
this.carbonTableIdentifier = carbonTableIdentifier;
- this.columnIdentifier = columnIdentifier;
+ this.dictionaryColumnUniqueIdentifier = dictionaryColumnUniqueIdentifier;
initFileLocation();
}
@@ -169,10 +170,11 @@ public class CarbonDictionaryMetadataReaderImpl implements CarbonDictionaryMetad
*/
protected void initFileLocation() {
PathService pathService = CarbonCommonFactory.getPathService();
- CarbonTablePath carbonTablePath =
- pathService.getCarbonTablePath(this.storePath, carbonTableIdentifier);
- this.columnDictionaryMetadataFilePath =
- carbonTablePath.getDictionaryMetaFilePath(columnIdentifier.getColumnId());
+ CarbonTablePath carbonTablePath = pathService
+ .getCarbonTablePath(this.storePath, carbonTableIdentifier,
+ dictionaryColumnUniqueIdentifier);
+ this.columnDictionaryMetadataFilePath = carbonTablePath.getDictionaryMetaFilePath(
+ dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId());
}
/**
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/reader/CarbonDictionaryReaderImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/reader/CarbonDictionaryReaderImpl.java b/core/src/main/java/org/apache/carbondata/core/reader/CarbonDictionaryReaderImpl.java
index d93e3a5..0325cf6 100644
--- a/core/src/main/java/org/apache/carbondata/core/reader/CarbonDictionaryReaderImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/reader/CarbonDictionaryReaderImpl.java
@@ -24,8 +24,8 @@ import java.util.Iterator;
import java.util.List;
import org.apache.carbondata.core.cache.dictionary.ColumnDictionaryChunkIterator;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
-import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.service.CarbonCommonFactory;
import org.apache.carbondata.core.service.PathService;
import org.apache.carbondata.core.util.path.CarbonTablePath;
@@ -52,7 +52,7 @@ public class CarbonDictionaryReaderImpl implements CarbonDictionaryReader {
/**
* column name
*/
- protected ColumnIdentifier columnIdentifier;
+ protected DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier;
/**
* dictionary file path
@@ -69,13 +69,13 @@ public class CarbonDictionaryReaderImpl implements CarbonDictionaryReader {
*
* @param storePath carbon dictionary data store path
* @param carbonTableIdentifier table identifier which will give table name and database name
- * @param columnIdentifier column unique identifier
+ * @param dictionaryColumnUniqueIdentifier column unique identifier
*/
- public CarbonDictionaryReaderImpl(String storePath,
- CarbonTableIdentifier carbonTableIdentifier, ColumnIdentifier columnIdentifier) {
+ public CarbonDictionaryReaderImpl(String storePath, CarbonTableIdentifier carbonTableIdentifier,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
this.storePath = storePath;
this.carbonTableIdentifier = carbonTableIdentifier;
- this.columnIdentifier = columnIdentifier;
+ this.dictionaryColumnUniqueIdentifier = dictionaryColumnUniqueIdentifier;
initFileLocation();
}
@@ -216,10 +216,11 @@ public class CarbonDictionaryReaderImpl implements CarbonDictionaryReader {
*/
protected void initFileLocation() {
PathService pathService = CarbonCommonFactory.getPathService();
- CarbonTablePath carbonTablePath = pathService.getCarbonTablePath(
- this.storePath, carbonTableIdentifier);
- this.columnDictionaryFilePath = carbonTablePath
- .getDictionaryFilePath(columnIdentifier.getColumnId());
+ CarbonTablePath carbonTablePath = pathService
+ .getCarbonTablePath(this.storePath, carbonTableIdentifier,
+ dictionaryColumnUniqueIdentifier);
+ this.columnDictionaryFilePath = carbonTablePath.getDictionaryFilePath(
+ dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId());
}
/**
@@ -303,7 +304,7 @@ public class CarbonDictionaryReaderImpl implements CarbonDictionaryReader {
*/
protected CarbonDictionaryMetadataReader getDictionaryMetadataReader() {
return new CarbonDictionaryMetadataReaderImpl(this.storePath, carbonTableIdentifier,
- this.columnIdentifier);
+ this.dictionaryColumnUniqueIdentifier);
}
/**
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImpl.java b/core/src/main/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImpl.java
index eeb8235..8c06434 100644
--- a/core/src/main/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImpl.java
@@ -21,9 +21,9 @@ import java.util.List;
import org.apache.carbondata.common.logging.LogService;
import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
-import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.reader.CarbonDictionaryColumnMetaChunk;
import org.apache.carbondata.core.reader.CarbonDictionaryMetadataReader;
import org.apache.carbondata.core.reader.CarbonDictionaryMetadataReaderImpl;
@@ -48,7 +48,7 @@ public class CarbonDictionarySortIndexReaderImpl implements CarbonDictionarySort
/**
* column name
*/
- protected ColumnIdentifier columnIdentifier;
+ protected DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier;
/**
* store location
@@ -77,14 +77,16 @@ public class CarbonDictionarySortIndexReaderImpl implements CarbonDictionarySort
private ThriftReader dictionarySortIndexThriftReader;
/**
- * @param carbonTableIdentifier Carbon Table identifier holding the database name and table name
- * @param columnIdentifier column name
- * @param carbonStorePath carbon store path
+ * @param carbonTableIdentifier Carbon Table identifier holding the database name
+ * and table name
+ * @param dictionaryColumnUniqueIdentifier column name
+ * @param carbonStorePath carbon store path
*/
public CarbonDictionarySortIndexReaderImpl(final CarbonTableIdentifier carbonTableIdentifier,
- final ColumnIdentifier columnIdentifier, final String carbonStorePath) {
+ final DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier,
+ final String carbonStorePath) {
this.carbonTableIdentifier = carbonTableIdentifier;
- this.columnIdentifier = columnIdentifier;
+ this.dictionaryColumnUniqueIdentifier = dictionaryColumnUniqueIdentifier;
this.carbonStorePath = carbonStorePath;
}
@@ -151,21 +153,23 @@ public class CarbonDictionarySortIndexReaderImpl implements CarbonDictionarySort
protected void initPath() {
PathService pathService = CarbonCommonFactory.getPathService();
- CarbonTablePath carbonTablePath =
- pathService.getCarbonTablePath(carbonStorePath, carbonTableIdentifier);
+ CarbonTablePath carbonTablePath = pathService
+ .getCarbonTablePath(carbonStorePath, carbonTableIdentifier,
+ dictionaryColumnUniqueIdentifier);
try {
CarbonDictionaryColumnMetaChunk chunkMetaObjectForLastSegmentEntry =
getChunkMetaObjectForLastSegmentEntry();
long dictOffset = chunkMetaObjectForLastSegmentEntry.getEnd_offset();
- this.sortIndexFilePath =
- carbonTablePath.getSortIndexFilePath(columnIdentifier.getColumnId(), dictOffset);
+ this.sortIndexFilePath = carbonTablePath.getSortIndexFilePath(
+ dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId(), dictOffset);
if (!FileFactory
.isFileExist(this.sortIndexFilePath, FileFactory.getFileType(this.sortIndexFilePath))) {
- this.sortIndexFilePath =
- carbonTablePath.getSortIndexFilePath(columnIdentifier.getColumnId());
+ this.sortIndexFilePath = carbonTablePath.getSortIndexFilePath(
+ dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId());
}
} catch (IOException e) {
- this.sortIndexFilePath = carbonTablePath.getSortIndexFilePath(columnIdentifier.getColumnId());
+ this.sortIndexFilePath = carbonTablePath.getSortIndexFilePath(
+ dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId());
}
}
@@ -193,7 +197,7 @@ public class CarbonDictionarySortIndexReaderImpl implements CarbonDictionarySort
*/
protected CarbonDictionaryMetadataReader getDictionaryMetadataReader() {
return new CarbonDictionaryMetadataReaderImpl(carbonStorePath, carbonTableIdentifier,
- columnIdentifier);
+ dictionaryColumnUniqueIdentifier);
}
/**
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java b/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java
index 05d0d8d..f159744 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java
@@ -50,6 +50,7 @@ import org.apache.carbondata.core.memory.UnsafeMemoryManager;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.encoder.Encoding;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure;
import org.apache.carbondata.core.scan.executor.QueryExecutor;
@@ -58,6 +59,8 @@ import org.apache.carbondata.core.scan.executor.infos.BlockExecutionInfo;
import org.apache.carbondata.core.scan.executor.util.QueryUtil;
import org.apache.carbondata.core.scan.executor.util.RestructureUtil;
import org.apache.carbondata.core.scan.filter.FilterUtil;
+import org.apache.carbondata.core.scan.filter.SingleTableProvider;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.model.QueryDimension;
import org.apache.carbondata.core.scan.model.QueryMeasure;
import org.apache.carbondata.core.scan.model.QueryModel;
@@ -181,12 +184,16 @@ public abstract class AbstractQueryExecutor<E> implements QueryExecutor<E> {
QueryUtil.getAllFilterDimensions(queryModel.getFilterExpressionResolverTree(),
queryProperties.complexFilterDimension, queryProperties.filterMeasures);
+ CarbonTable carbonTable = queryModel.getTable();
+ TableProvider tableProvider = new SingleTableProvider(carbonTable);
+
queryStatistic = new QueryStatistic();
// dictionary column unique column id to dictionary mapping
// which will be used to get column actual data
queryProperties.columnToDictionayMapping = QueryUtil
.getDimensionDictionaryDetail(queryModel.getQueryDimension(),
- queryProperties.complexFilterDimension, queryModel.getAbsoluteTableIdentifier());
+ queryProperties.complexFilterDimension, queryModel.getAbsoluteTableIdentifier(),
+ tableProvider);
queryStatistic
.addStatistics(QueryStatisticsConstants.LOAD_DICTIONARY, System.currentTimeMillis());
queryProperties.queryStatisticsRecorder.recordStatistics(queryStatistic);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/executor/util/QueryUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/executor/util/QueryUtil.java b/core/src/main/java/org/apache/carbondata/core/scan/executor/util/QueryUtil.java
index ee04bb6..ee09793 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/executor/util/QueryUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/executor/util/QueryUtil.java
@@ -54,12 +54,15 @@ import org.apache.carbondata.core.scan.executor.infos.KeyStructureInfo;
import org.apache.carbondata.core.scan.expression.ColumnExpression;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.scan.filter.GenericQueryType;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.DimColumnResolvedFilterInfo;
import org.apache.carbondata.core.scan.model.QueryDimension;
import org.apache.carbondata.core.scan.model.QueryMeasure;
import org.apache.carbondata.core.scan.model.QueryModel;
import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.commons.lang3.ArrayUtils;
@@ -273,7 +276,8 @@ public class QueryUtil {
*/
public static Map<String, Dictionary> getDimensionDictionaryDetail(
List<QueryDimension> queryDimensions, Set<CarbonDimension> filterComplexDimensions,
- AbsoluteTableIdentifier absoluteTableIdentifier) throws IOException {
+ AbsoluteTableIdentifier absoluteTableIdentifier, TableProvider tableProvider)
+ throws IOException {
// to store dimension unique column id list, this is required as
// dimension can be present in
// query dimension, as well as some aggregation function will be applied
@@ -307,7 +311,7 @@ public class QueryUtil {
List<String> dictionaryColumnIdList =
new ArrayList<String>(dictionaryDimensionFromQuery.size());
dictionaryColumnIdList.addAll(dictionaryDimensionFromQuery);
- return getDictionaryMap(dictionaryColumnIdList, absoluteTableIdentifier);
+ return getDictionaryMap(dictionaryColumnIdList, absoluteTableIdentifier, tableProvider);
}
/**
@@ -339,11 +343,12 @@ public class QueryUtil {
* @throws IOException
*/
private static Map<String, Dictionary> getDictionaryMap(List<String> dictionaryColumnIdList,
- AbsoluteTableIdentifier absoluteTableIdentifier) throws IOException {
+ AbsoluteTableIdentifier absoluteTableIdentifier, TableProvider tableProvider)
+ throws IOException {
// this for dictionary unique identifier
List<DictionaryColumnUniqueIdentifier> dictionaryColumnUniqueIdentifiers =
getDictionaryColumnUniqueIdentifierList(dictionaryColumnIdList,
- absoluteTableIdentifier.getCarbonTableIdentifier());
+ absoluteTableIdentifier.getCarbonTableIdentifier(), tableProvider);
CacheProvider cacheProvider = CacheProvider.getInstance();
Cache<DictionaryColumnUniqueIdentifier, Dictionary> forwardDictionaryCache = cacheProvider
.createCache(CacheType.FORWARD_DICTIONARY, absoluteTableIdentifier.getStorePath());
@@ -367,9 +372,11 @@ public class QueryUtil {
* @return
*/
private static List<DictionaryColumnUniqueIdentifier> getDictionaryColumnUniqueIdentifierList(
- List<String> dictionaryColumnIdList, CarbonTableIdentifier carbonTableIdentifier) {
- CarbonTable carbonTable =
- CarbonMetadata.getInstance().getCarbonTable(carbonTableIdentifier.getTableUniqueName());
+ List<String> dictionaryColumnIdList, CarbonTableIdentifier carbonTableIdentifier,
+ TableProvider tableProvider) throws IOException {
+ CarbonTable carbonTable = tableProvider.getCarbonTable(carbonTableIdentifier);
+ CarbonTablePath carbonTablePath =
+ CarbonStorePath.getCarbonTablePath(carbonTable.getStorePath(), carbonTableIdentifier);
List<DictionaryColumnUniqueIdentifier> dictionaryColumnUniqueIdentifiers =
new ArrayList<>(dictionaryColumnIdList.size());
for (String columnId : dictionaryColumnIdList) {
@@ -380,7 +387,8 @@ public class QueryUtil {
new DictionaryColumnUniqueIdentifier(
carbonTableIdentifier,
dimension.getColumnIdentifier(),
- dimension.getDataType()
+ dimension.getDataType(),
+ carbonTablePath
)
);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterExpressionProcessor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterExpressionProcessor.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterExpressionProcessor.java
index 9bc864d..cfcf112 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterExpressionProcessor.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterExpressionProcessor.java
@@ -80,9 +80,10 @@ public class FilterExpressionProcessor implements FilterProcessor {
* @return a filter resolver tree
*/
public FilterResolverIntf getFilterResolver(Expression expressionTree,
- AbsoluteTableIdentifier tableIdentifier) throws FilterUnsupportedException, IOException {
+ AbsoluteTableIdentifier tableIdentifier, TableProvider tableProvider)
+ throws FilterUnsupportedException, IOException {
if (null != expressionTree && null != tableIdentifier) {
- return getFilterResolvertree(expressionTree, tableIdentifier);
+ return getFilterResolvertree(expressionTree, tableIdentifier, tableProvider);
}
return null;
}
@@ -315,10 +316,11 @@ public class FilterExpressionProcessor implements FilterProcessor {
* @return FilterResolverIntf type.
*/
private FilterResolverIntf getFilterResolvertree(Expression expressionTree,
- AbsoluteTableIdentifier tableIdentifier) throws FilterUnsupportedException, IOException {
+ AbsoluteTableIdentifier tableIdentifier, TableProvider tableProvider)
+ throws FilterUnsupportedException, IOException {
FilterResolverIntf filterEvaluatorTree =
createFilterResolverTree(expressionTree, tableIdentifier);
- traverseAndResolveTree(filterEvaluatorTree, tableIdentifier);
+ traverseAndResolveTree(filterEvaluatorTree, tableIdentifier, tableProvider);
return filterEvaluatorTree;
}
@@ -332,13 +334,14 @@ public class FilterExpressionProcessor implements FilterProcessor {
* @param tableIdentifier
*/
private void traverseAndResolveTree(FilterResolverIntf filterResolverTree,
- AbsoluteTableIdentifier tableIdentifier) throws FilterUnsupportedException, IOException {
+ AbsoluteTableIdentifier tableIdentifier, TableProvider tableProvider)
+ throws FilterUnsupportedException, IOException {
if (null == filterResolverTree) {
return;
}
- traverseAndResolveTree(filterResolverTree.getLeft(), tableIdentifier);
- filterResolverTree.resolve(tableIdentifier);
- traverseAndResolveTree(filterResolverTree.getRight(), tableIdentifier);
+ traverseAndResolveTree(filterResolverTree.getLeft(), tableIdentifier, tableProvider);
+ filterResolverTree.resolve(tableIdentifier, tableProvider);
+ traverseAndResolveTree(filterResolverTree.getRight(), tableIdentifier, tableProvider);
}
/**
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterProcessor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterProcessor.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterProcessor.java
index 246166d..dc392ba 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterProcessor.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterProcessor.java
@@ -41,7 +41,8 @@ public interface FilterProcessor {
* @throws FilterUnsupportedException
*/
FilterResolverIntf getFilterResolver(Expression expressionTree,
- AbsoluteTableIdentifier tableIdentifier) throws FilterUnsupportedException, IOException;
+ AbsoluteTableIdentifier tableIdentifier, TableProvider tableProvider)
+ throws FilterUnsupportedException, IOException;
/**
* This API is exposed inorder to get the required block reference node
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterUtil.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterUtil.java
index 6230771..b952b97 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/FilterUtil.java
@@ -55,6 +55,7 @@ import org.apache.carbondata.core.keygenerator.KeyGenerator;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.encoder.Encoding;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure;
import org.apache.carbondata.core.scan.expression.ColumnExpression;
@@ -96,6 +97,8 @@ import org.apache.carbondata.core.util.DataTypeConverterImpl;
import org.apache.carbondata.core.util.DataTypeUtil;
import org.apache.carbondata.core.util.comparator.Comparator;
import org.apache.carbondata.core.util.comparator.SerializableComparator;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
public final class FilterUtil {
private static final LogService LOGGER =
@@ -498,13 +501,15 @@ public final class FilterUtil {
* @throws IOException
*/
public static ColumnFilterInfo getFilterValues(AbsoluteTableIdentifier tableIdentifier,
- ColumnExpression columnExpression, List<String> evaluateResultList, boolean isIncludeFilter)
+ ColumnExpression columnExpression, List<String> evaluateResultList, boolean isIncludeFilter,
+ TableProvider tableProvider)
throws IOException {
Dictionary forwardDictionary = null;
try {
// Reading the dictionary value from cache.
forwardDictionary =
- getForwardDictionaryCache(tableIdentifier, columnExpression.getDimension());
+ getForwardDictionaryCache(tableIdentifier, columnExpression.getDimension(),
+ tableProvider);
return getFilterValues(columnExpression, evaluateResultList, forwardDictionary,
isIncludeFilter);
} finally {
@@ -566,14 +571,15 @@ public final class FilterUtil {
*/
public static ColumnFilterInfo getFilterListForAllValues(
AbsoluteTableIdentifier tableIdentifier, Expression expression,
- final ColumnExpression columnExpression, boolean isIncludeFilter)
+ final ColumnExpression columnExpression, boolean isIncludeFilter, TableProvider tableProvider)
throws IOException, FilterUnsupportedException {
Dictionary forwardDictionary = null;
List<String> evaluateResultListFinal = new ArrayList<String>(20);
DictionaryChunksWrapper dictionaryWrapper = null;
try {
forwardDictionary =
- getForwardDictionaryCache(tableIdentifier, columnExpression.getDimension());
+ getForwardDictionaryCache(tableIdentifier, columnExpression.getDimension(),
+ tableProvider);
dictionaryWrapper = forwardDictionary.getDictionaryChunks();
while (dictionaryWrapper.hasNext()) {
byte[] columnVal = dictionaryWrapper.next();
@@ -1090,9 +1096,24 @@ public final class FilterUtil {
*/
public static Dictionary getForwardDictionaryCache(AbsoluteTableIdentifier tableIdentifier,
CarbonDimension carbonDimension) throws IOException {
+ return getForwardDictionaryCache(tableIdentifier, carbonDimension, null);
+ }
+
+ /**
+ * @param tableIdentifier
+ * @param carbonDimension
+ * @param tableProvider
+ * @return
+ */
+ public static Dictionary getForwardDictionaryCache(AbsoluteTableIdentifier tableIdentifier,
+ CarbonDimension carbonDimension, TableProvider tableProvider) throws IOException {
+ CarbonTable carbonTable =
+ tableProvider.getCarbonTable(tableIdentifier.getCarbonTableIdentifier());
+ CarbonTablePath carbonTablePath =
+ CarbonStorePath.getCarbonTablePath(carbonTable.getAbsoluteTableIdentifier());
DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier =
new DictionaryColumnUniqueIdentifier(tableIdentifier.getCarbonTableIdentifier(),
- carbonDimension.getColumnIdentifier(), carbonDimension.getDataType());
+ carbonDimension.getColumnIdentifier(), carbonDimension.getDataType(), carbonTablePath);
CacheProvider cacheProvider = CacheProvider.getInstance();
Cache<DictionaryColumnUniqueIdentifier, Dictionary> forwardDictionaryCache =
cacheProvider.createCache(CacheType.FORWARD_DICTIONARY, tableIdentifier.getStorePath());
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/SingleTableProvider.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/SingleTableProvider.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/SingleTableProvider.java
new file mode 100644
index 0000000..88fc8a6
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/SingleTableProvider.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.carbondata.core.scan.filter;
+
+import java.io.IOException;
+
+import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+
+public class SingleTableProvider implements TableProvider {
+
+ private CarbonTable carbonTable;
+
+ public SingleTableProvider(CarbonTable carbonTable) {
+ this.carbonTable = carbonTable;
+ }
+
+ @Override public CarbonTable getCarbonTable(CarbonTableIdentifier carbonTableIdentifier)
+ throws IOException {
+ if (carbonTable.getCarbonTableIdentifier().equals(carbonTableIdentifier)) {
+ return carbonTable;
+ } else {
+ throw new IOException("Carbon table does not exist with identifier " + carbonTableIdentifier);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/TableProvider.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/TableProvider.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/TableProvider.java
new file mode 100644
index 0000000..d3fa15a
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/TableProvider.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.carbondata.core.scan.filter;
+
+import java.io.IOException;
+import java.io.Serializable;
+
+import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+
+public interface TableProvider extends Serializable {
+
+ CarbonTable getCarbonTable(CarbonTableIdentifier carbonTableIdentifier) throws IOException;
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/ConditionalFilterResolverImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/ConditionalFilterResolverImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/ConditionalFilterResolverImpl.java
index 8d3c366..6dd59fd 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/ConditionalFilterResolverImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/ConditionalFilterResolverImpl.java
@@ -31,6 +31,7 @@ import org.apache.carbondata.core.scan.expression.conditional.ConditionalExpress
import org.apache.carbondata.core.scan.expression.exception.FilterUnsupportedException;
import org.apache.carbondata.core.scan.expression.logical.RangeExpression;
import org.apache.carbondata.core.scan.filter.FilterUtil;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.filter.intf.FilterExecuterType;
import org.apache.carbondata.core.scan.filter.resolver.metadata.FilterResolverMetadata;
import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.DimColumnResolvedFilterInfo;
@@ -70,7 +71,8 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
*
* @throws FilterUnsupportedException
*/
- @Override public void resolve(AbsoluteTableIdentifier absoluteTableIdentifier)
+ @Override public void resolve(AbsoluteTableIdentifier absoluteTableIdentifier,
+ TableProvider tableProvider)
throws FilterUnsupportedException, IOException {
FilterResolverMetadata metadata = new FilterResolverMetadata();
metadata.setTableIdentifier(absoluteTableIdentifier);
@@ -83,6 +85,7 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
metadata.setColumnExpression(columnExpression);
metadata.setExpression(rightExp);
metadata.setIncludeFilter(isIncludeFilter);
+ metadata.setTableProvider(tableProvider);
// If imei=imei comes in filter condition then we need to
// skip processing of right expression.
// This flow has reached here assuming that this is a single
@@ -118,6 +121,7 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
metadata.setColumnExpression(columnExpression);
metadata.setExpression(leftExp);
metadata.setIncludeFilter(isIncludeFilter);
+ metadata.setTableProvider(tableProvider);
if (columnExpression.getDataType().equals(DataType.TIMESTAMP) ||
columnExpression.getDataType().equals(DataType.DATE)) {
isExpressionResolve = true;
@@ -153,6 +157,7 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
metadata.setColumnExpression(columnList.get(0));
metadata.setExpression(exp);
metadata.setIncludeFilter(isIncludeFilter);
+ metadata.setTableProvider(tableProvider);
if ((null != columnList.get(0).getDimension()) && (
!columnList.get(0).getDimension().hasEncoding(Encoding.DICTIONARY) || columnList.get(0)
.getDimension().hasEncoding(Encoding.DIRECT_DICTIONARY))
@@ -169,7 +174,7 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
== org.apache.carbondata.core.metadata.datatype.DataType.ARRAY))) {
dimColResolvedFilterInfo.setFilterValues(FilterUtil
.getFilterListForAllValues(absoluteTableIdentifier, exp, columnList.get(0),
- isIncludeFilter));
+ isIncludeFilter, tableProvider));
dimColResolvedFilterInfo.setColumnIndex(columnList.get(0).getDimension().getOrdinal());
dimColResolvedFilterInfo.setDimension(columnList.get(0).getDimension());
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/FilterResolverIntf.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/FilterResolverIntf.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/FilterResolverIntf.java
index 89a3890..a0810d5 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/FilterResolverIntf.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/FilterResolverIntf.java
@@ -25,6 +25,7 @@ import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.scan.expression.exception.FilterUnsupportedException;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.filter.intf.FilterExecuterType;
import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.DimColumnResolvedFilterInfo;
import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.MeasureColumnResolvedFilterInfo;
@@ -39,7 +40,7 @@ public interface FilterResolverIntf extends Serializable {
* @throws IOException
* @throws FilterUnsupportedException
*/
- void resolve(AbsoluteTableIdentifier absoluteTableIdentifier)
+ void resolve(AbsoluteTableIdentifier absoluteTableIdentifier, TableProvider tableProvider)
throws IOException, FilterUnsupportedException;
/**
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/LogicalFilterResolverImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/LogicalFilterResolverImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/LogicalFilterResolverImpl.java
index 368cd9a..d530155 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/LogicalFilterResolverImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/LogicalFilterResolverImpl.java
@@ -24,6 +24,7 @@ import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.scan.expression.BinaryExpression;
import org.apache.carbondata.core.scan.expression.Expression;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.filter.intf.ExpressionType;
import org.apache.carbondata.core.scan.filter.intf.FilterExecuterType;
import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.DimColumnResolvedFilterInfo;
@@ -56,8 +57,10 @@ public class LogicalFilterResolverImpl implements FilterResolverIntf {
* node for filter evaluation, so in this instance no implementation is required.
*
* @param absoluteTableIdentifier
+ * @param tableProvider
*/
- @Override public void resolve(AbsoluteTableIdentifier absoluteTableIdentifier) {
+ @Override public void resolve(AbsoluteTableIdentifier absoluteTableIdentifier,
+ TableProvider tableProvider) {
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelFilterResolverImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelFilterResolverImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelFilterResolverImpl.java
index b399fd6..33bafb1 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelFilterResolverImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelFilterResolverImpl.java
@@ -25,6 +25,7 @@ import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.scan.expression.ColumnExpression;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.scan.expression.conditional.ConditionalExpression;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.filter.intf.FilterExecuterType;
import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.DimColumnResolvedFilterInfo;
import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.MeasureColumnResolvedFilterInfo;
@@ -51,7 +52,8 @@ public class RowLevelFilterResolverImpl extends ConditionalFilterResolverImpl {
* Method which will resolve the filter expression by converting the filter member
* to its assigned dictionary values.
*/
- public void resolve(AbsoluteTableIdentifier absoluteTableIdentifier) {
+ public void resolve(AbsoluteTableIdentifier absoluteTableIdentifier,
+ TableProvider tableProvider) {
DimColumnResolvedFilterInfo dimColumnEvaluatorInfo = null;
MeasureColumnResolvedFilterInfo msrColumnEvalutorInfo = null;
int index = 0;
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelRangeFilterResolverImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelRangeFilterResolverImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelRangeFilterResolverImpl.java
index f64342f..3e27594 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelRangeFilterResolverImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelRangeFilterResolverImpl.java
@@ -40,6 +40,7 @@ import org.apache.carbondata.core.scan.expression.exception.FilterUnsupportedExc
import org.apache.carbondata.core.scan.expression.logical.BinaryLogicalExpression;
import org.apache.carbondata.core.scan.filter.ColumnFilterInfo;
import org.apache.carbondata.core.scan.filter.FilterUtil;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.filter.intf.FilterExecuterType;
import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.DimColumnResolvedFilterInfo;
import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.MeasureColumnResolvedFilterInfo;
@@ -216,7 +217,7 @@ public class RowLevelRangeFilterResolverImpl extends ConditionalFilterResolverIm
* Method which will resolve the filter expression by converting the filter
* member to its assigned dictionary values.
*/
- public void resolve(AbsoluteTableIdentifier absoluteTableIdentifier)
+ public void resolve(AbsoluteTableIdentifier absoluteTableIdentifier, TableProvider tableProvider)
throws FilterUnsupportedException {
DimColumnResolvedFilterInfo dimColumnEvaluatorInfo = null;
MeasureColumnResolvedFilterInfo msrColumnEvalutorInfo = null;
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/metadata/FilterResolverMetadata.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/metadata/FilterResolverMetadata.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/metadata/FilterResolverMetadata.java
index 1faaf86..8db80dc 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/metadata/FilterResolverMetadata.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/metadata/FilterResolverMetadata.java
@@ -19,11 +19,22 @@ package org.apache.carbondata.core.scan.filter.resolver.metadata;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.scan.expression.ColumnExpression;
import org.apache.carbondata.core.scan.expression.Expression;
+import org.apache.carbondata.core.scan.filter.TableProvider;
public class FilterResolverMetadata {
private AbsoluteTableIdentifier tableIdentifier;
private Expression expression;
private ColumnExpression columnExpression;
+
+ public TableProvider getTableProvider() {
+ return tableProvider;
+ }
+
+ public void setTableProvider(TableProvider tableProvider) {
+ this.tableProvider = tableProvider;
+ }
+
+ private TableProvider tableProvider;
private boolean isIncludeFilter;
public AbsoluteTableIdentifier getTableIdentifier() {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/TrueConditionalResolverImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/TrueConditionalResolverImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/TrueConditionalResolverImpl.java
index c9d180f..d403ef4 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/TrueConditionalResolverImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/TrueConditionalResolverImpl.java
@@ -19,6 +19,7 @@ package org.apache.carbondata.core.scan.filter.resolver.resolverinfo;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.scan.expression.Expression;
+import org.apache.carbondata.core.scan.filter.TableProvider;
import org.apache.carbondata.core.scan.filter.intf.FilterExecuterType;
import org.apache.carbondata.core.scan.filter.resolver.ConditionalFilterResolverImpl;
@@ -32,8 +33,8 @@ public class TrueConditionalResolverImpl extends ConditionalFilterResolverImpl {
super(exp, isExpressionResolve, isIncludeFilter, tableIdentifier, false);
}
- @Override
- public void resolve(AbsoluteTableIdentifier absoluteTableIdentifier) {
+ @Override public void resolve(AbsoluteTableIdentifier absoluteTableIdentifier,
+ TableProvider tableProvider) {
}
/**
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/visitor/DictionaryColumnVisitor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/visitor/DictionaryColumnVisitor.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/visitor/DictionaryColumnVisitor.java
index 2fc7503..451ce45 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/visitor/DictionaryColumnVisitor.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/visitor/DictionaryColumnVisitor.java
@@ -45,32 +45,30 @@ public class DictionaryColumnVisitor implements ResolvedFilterInfoVisitorIntf {
public void populateFilterResolvedInfo(ColumnResolvedFilterInfo visitableObj,
FilterResolverMetadata metadata) throws FilterUnsupportedException, IOException {
- if (visitableObj instanceof DimColumnResolvedFilterInfo) {
- DimColumnResolvedFilterInfo resolveDimension = (DimColumnResolvedFilterInfo) visitableObj;
- ColumnFilterInfo resolvedFilterObject = null;
- List<String> evaluateResultListFinal;
- try {
- evaluateResultListFinal = metadata.getExpression().evaluate(null).getListAsString();
- } catch (FilterIllegalMemberException e) {
- throw new FilterUnsupportedException(e);
- }
- resolvedFilterObject = FilterUtil
- .getFilterValues(metadata.getTableIdentifier(), metadata.getColumnExpression(),
- evaluateResultListFinal, metadata.isIncludeFilter());
- if (!metadata.isIncludeFilter() && null != resolvedFilterObject) {
- // Adding default surrogate key of null member inorder to not display the same while
- // displaying the report as per hive compatibility.
- // first check of surrogate key for null value is already added then
- // no need to add again otherwise result will be wrong in case of exclude filter
- // this is because two times it will flip the same bit
- if (!resolvedFilterObject.getFilterList()
- .contains(CarbonCommonConstants.MEMBER_DEFAULT_VAL_SURROGATE_KEY)) {
- resolvedFilterObject.getFilterList()
- .add(CarbonCommonConstants.MEMBER_DEFAULT_VAL_SURROGATE_KEY);
- }
- Collections.sort(resolvedFilterObject.getFilterList());
+ DimColumnResolvedFilterInfo resolveDimension = (DimColumnResolvedFilterInfo) visitableObj;
+ ColumnFilterInfo resolvedFilterObject = null;
+ List<String> evaluateResultListFinal;
+ try {
+ evaluateResultListFinal = metadata.getExpression().evaluate(null).getListAsString();
+ } catch (FilterIllegalMemberException e) {
+ throw new FilterUnsupportedException(e);
+ }
+ resolvedFilterObject = FilterUtil
+ .getFilterValues(metadata.getTableIdentifier(), metadata.getColumnExpression(),
+ evaluateResultListFinal, metadata.isIncludeFilter(), metadata.getTableProvider());
+ if (!metadata.isIncludeFilter() && null != resolvedFilterObject) {
+ // Adding default surrogate key of null member inorder to not display the same while
+ // displaying the report as per hive compatibility.
+ // first check of surrogate key for null value is already added then
+ // no need to add again otherwise result will be wrong in case of exclude filter
+ // this is because two times it will flip the same bit
+ if (!resolvedFilterObject.getFilterList()
+ .contains(CarbonCommonConstants.MEMBER_DEFAULT_VAL_SURROGATE_KEY)) {
+ resolvedFilterObject.getFilterList()
+ .add(CarbonCommonConstants.MEMBER_DEFAULT_VAL_SURROGATE_KEY);
}
- resolveDimension.setFilterValues(resolvedFilterObject);
+ Collections.sort(resolvedFilterObject.getFilterList());
}
+ resolveDimension.setFilterValues(resolvedFilterObject);
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/visitor/RangeDictionaryColumnVisitor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/visitor/RangeDictionaryColumnVisitor.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/visitor/RangeDictionaryColumnVisitor.java
index 3d16cd2..5ce818d 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/visitor/RangeDictionaryColumnVisitor.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/resolverinfo/visitor/RangeDictionaryColumnVisitor.java
@@ -44,22 +44,21 @@ public class RangeDictionaryColumnVisitor extends DictionaryColumnVisitor
*/
public void populateFilterResolvedInfo(ColumnResolvedFilterInfo visitableObj,
FilterResolverMetadata metadata) throws FilterUnsupportedException, IOException {
- if (visitableObj instanceof DimColumnResolvedFilterInfo) {
- DimColumnResolvedFilterInfo resolveDimension = (DimColumnResolvedFilterInfo) visitableObj;
- ColumnFilterInfo resolvedFilterObject = null;
- List<String> evaluateResultListFinal;
- resolvedFilterObject = FilterUtil
- .getFilterListForAllValues(metadata.getTableIdentifier(), metadata.getExpression(),
- metadata.getColumnExpression(), metadata.isIncludeFilter());
+ DimColumnResolvedFilterInfo resolveDimension = (DimColumnResolvedFilterInfo) visitableObj;
+ ColumnFilterInfo resolvedFilterObject = null;
+ List<String> evaluateResultListFinal;
+ resolvedFilterObject = FilterUtil
+ .getFilterListForAllValues(metadata.getTableIdentifier(), metadata.getExpression(),
+ metadata.getColumnExpression(), metadata.isIncludeFilter(),
+ metadata.getTableProvider());
- if (!metadata.isIncludeFilter() && null != resolvedFilterObject) {
- // Adding default surrogate key of null member inorder to not display the same while
- // displaying the report as per hive compatibility.
- resolvedFilterObject.getFilterList()
- .add(CarbonCommonConstants.MEMBER_DEFAULT_VAL_SURROGATE_KEY);
- Collections.sort(resolvedFilterObject.getFilterList());
- }
- resolveDimension.setFilterValues(resolvedFilterObject);
+ if (!metadata.isIncludeFilter() && null != resolvedFilterObject) {
+ // Adding default surrogate key of null member inorder to not display the same while
+ // displaying the report as per hive compatibility.
+ resolvedFilterObject.getFilterList()
+ .add(CarbonCommonConstants.MEMBER_DEFAULT_VAL_SURROGATE_KEY);
+ Collections.sort(resolvedFilterObject.getFilterList());
}
+ resolveDimension.setFilterValues(resolvedFilterObject);
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/service/DictionaryService.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/service/DictionaryService.java b/core/src/main/java/org/apache/carbondata/core/service/DictionaryService.java
index ef941a5..672aa83 100644
--- a/core/src/main/java/org/apache/carbondata/core/service/DictionaryService.java
+++ b/core/src/main/java/org/apache/carbondata/core/service/DictionaryService.java
@@ -16,8 +16,8 @@
*/
package org.apache.carbondata.core.service;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
-import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.reader.CarbonDictionaryMetadataReader;
import org.apache.carbondata.core.reader.CarbonDictionaryReader;
import org.apache.carbondata.core.reader.sortindex.CarbonDictionarySortIndexReader;
@@ -33,58 +33,58 @@ public interface DictionaryService {
* get dictionary writer
*
* @param carbonTableIdentifier
- * @param columnIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @param carbonStorePath
* @return
*/
CarbonDictionaryWriter getDictionaryWriter(CarbonTableIdentifier carbonTableIdentifier,
- ColumnIdentifier columnIdentifier, String carbonStorePath);
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, String carbonStorePath);
/**
* get dictionary sort index writer
*
* @param carbonTableIdentifier
- * @param columnIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @param carbonStorePath
* @return
*/
CarbonDictionarySortIndexWriter getDictionarySortIndexWriter(
- CarbonTableIdentifier carbonTableIdentifier, ColumnIdentifier columnIdentifier,
- String carbonStorePath);
+ CarbonTableIdentifier carbonTableIdentifier,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, String carbonStorePath);
/**
* get dictionary metadata reader
*
* @param carbonTableIdentifier
- * @param columnIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @param carbonStorePath
* @return
*/
CarbonDictionaryMetadataReader getDictionaryMetadataReader(
- CarbonTableIdentifier carbonTableIdentifier, ColumnIdentifier columnIdentifier,
- String carbonStorePath);
+ CarbonTableIdentifier carbonTableIdentifier,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, String carbonStorePath);
/**
* get dictionary reader
*
* @param carbonTableIdentifier
- * @param columnIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @param carbonStorePath
* @return
*/
CarbonDictionaryReader getDictionaryReader(CarbonTableIdentifier carbonTableIdentifier,
- ColumnIdentifier columnIdentifier, String carbonStorePath);
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, String carbonStorePath);
/**
* get dictionary sort index reader
*
* @param carbonTableIdentifier
- * @param columnIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @param carbonStorePath
* @return
*/
CarbonDictionarySortIndexReader getDictionarySortIndexReader(
- CarbonTableIdentifier carbonTableIdentifier, ColumnIdentifier columnIdentifier,
- String carbonStorePath);
+ CarbonTableIdentifier carbonTableIdentifier,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, String carbonStorePath);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/service/PathService.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/service/PathService.java b/core/src/main/java/org/apache/carbondata/core/service/PathService.java
index 3528d0e..9735217 100644
--- a/core/src/main/java/org/apache/carbondata/core/service/PathService.java
+++ b/core/src/main/java/org/apache/carbondata/core/service/PathService.java
@@ -16,6 +16,7 @@
*/
package org.apache.carbondata.core.service;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
import org.apache.carbondata.core.util.path.CarbonTablePath;
@@ -27,7 +28,9 @@ public interface PathService {
/**
* @param storeLocation
* @param tableIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @return store path related to tables
*/
- CarbonTablePath getCarbonTablePath(String storeLocation, CarbonTableIdentifier tableIdentifier);
+ CarbonTablePath getCarbonTablePath(String storeLocation, CarbonTableIdentifier tableIdentifier,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/service/impl/DictionaryFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/service/impl/DictionaryFactory.java b/core/src/main/java/org/apache/carbondata/core/service/impl/DictionaryFactory.java
index 8d6969c..1dc22ba 100644
--- a/core/src/main/java/org/apache/carbondata/core/service/impl/DictionaryFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/service/impl/DictionaryFactory.java
@@ -16,8 +16,8 @@
*/
package org.apache.carbondata.core.service.impl;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
-import org.apache.carbondata.core.metadata.ColumnIdentifier;
import org.apache.carbondata.core.reader.CarbonDictionaryMetadataReader;
import org.apache.carbondata.core.reader.CarbonDictionaryMetadataReaderImpl;
import org.apache.carbondata.core.reader.CarbonDictionaryReader;
@@ -41,73 +41,75 @@ public class DictionaryFactory implements DictionaryService {
* get dictionary writer
*
* @param carbonTableIdentifier
- * @param columnIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @param carbonStorePath
* @return
*/
@Override public CarbonDictionaryWriter getDictionaryWriter(
- CarbonTableIdentifier carbonTableIdentifier, ColumnIdentifier columnIdentifier,
- String carbonStorePath) {
- return new CarbonDictionaryWriterImpl(carbonStorePath, carbonTableIdentifier, columnIdentifier);
+ CarbonTableIdentifier carbonTableIdentifier,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, String carbonStorePath) {
+ return new CarbonDictionaryWriterImpl(carbonStorePath, carbonTableIdentifier,
+ dictionaryColumnUniqueIdentifier);
}
/**
* get dictionary sort index writer
*
* @param carbonTableIdentifier
- * @param columnIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @param carbonStorePath
* @return
*/
@Override public CarbonDictionarySortIndexWriter getDictionarySortIndexWriter(
- CarbonTableIdentifier carbonTableIdentifier, ColumnIdentifier columnIdentifier,
- String carbonStorePath) {
- return new CarbonDictionarySortIndexWriterImpl(carbonTableIdentifier, columnIdentifier,
- carbonStorePath);
+ CarbonTableIdentifier carbonTableIdentifier,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, String carbonStorePath) {
+ return new CarbonDictionarySortIndexWriterImpl(carbonTableIdentifier,
+ dictionaryColumnUniqueIdentifier, carbonStorePath);
}
/**
* get dictionary metadata reader
*
* @param carbonTableIdentifier
- * @param columnIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @param carbonStorePath
* @return
*/
@Override public CarbonDictionaryMetadataReader getDictionaryMetadataReader(
- CarbonTableIdentifier carbonTableIdentifier, ColumnIdentifier columnIdentifier,
- String carbonStorePath) {
+ CarbonTableIdentifier carbonTableIdentifier,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, String carbonStorePath) {
return new CarbonDictionaryMetadataReaderImpl(carbonStorePath, carbonTableIdentifier,
- columnIdentifier);
+ dictionaryColumnUniqueIdentifier);
}
/**
* get dictionary reader
*
* @param carbonTableIdentifier
- * @param columnIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @param carbonStorePath
* @return
*/
@Override public CarbonDictionaryReader getDictionaryReader(
- CarbonTableIdentifier carbonTableIdentifier, ColumnIdentifier columnIdentifier,
- String carbonStorePath) {
- return new CarbonDictionaryReaderImpl(carbonStorePath, carbonTableIdentifier, columnIdentifier);
+ CarbonTableIdentifier carbonTableIdentifier,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, String carbonStorePath) {
+ return new CarbonDictionaryReaderImpl(carbonStorePath, carbonTableIdentifier,
+ dictionaryColumnUniqueIdentifier);
}
/**
* get dictionary sort index reader
*
* @param carbonTableIdentifier
- * @param columnIdentifier
+ * @param dictionaryColumnUniqueIdentifier
* @param carbonStorePath
* @return
*/
@Override public CarbonDictionarySortIndexReader getDictionarySortIndexReader(
- CarbonTableIdentifier carbonTableIdentifier, ColumnIdentifier columnIdentifier,
- String carbonStorePath) {
- return new CarbonDictionarySortIndexReaderImpl(carbonTableIdentifier, columnIdentifier,
- carbonStorePath);
+ CarbonTableIdentifier carbonTableIdentifier,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, String carbonStorePath) {
+ return new CarbonDictionarySortIndexReaderImpl(carbonTableIdentifier,
+ dictionaryColumnUniqueIdentifier, carbonStorePath);
}
public static DictionaryService getInstance() {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/service/impl/PathFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/service/impl/PathFactory.java b/core/src/main/java/org/apache/carbondata/core/service/impl/PathFactory.java
index b3c7d61..e7e9ebe 100644
--- a/core/src/main/java/org/apache/carbondata/core/service/impl/PathFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/service/impl/PathFactory.java
@@ -16,6 +16,7 @@
*/
package org.apache.carbondata.core.service.impl;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
import org.apache.carbondata.core.service.PathService;
import org.apache.carbondata.core.util.path.CarbonStorePath;
@@ -31,10 +32,16 @@ public class PathFactory implements PathService {
/**
* @param storeLocation
* @param tableIdentifier
+ * @param
* @return store path related to tables
*/
- @Override public CarbonTablePath getCarbonTablePath(
- String storeLocation, CarbonTableIdentifier tableIdentifier) {
+ @Override public CarbonTablePath getCarbonTablePath(String storeLocation,
+ CarbonTableIdentifier tableIdentifier,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
+ if (null != dictionaryColumnUniqueIdentifier && null != dictionaryColumnUniqueIdentifier
+ .getCarbonTablePath()) {
+ return dictionaryColumnUniqueIdentifier.getCarbonTablePath();
+ }
return CarbonStorePath.getCarbonTablePath(storeLocation, tableIdentifier);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d3a09e27/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
index 3bf837b..1db1cca 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
@@ -1576,7 +1576,8 @@ public final class CarbonUtil {
DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
PathService pathService = CarbonCommonFactory.getPathService();
CarbonTablePath carbonTablePath = pathService.getCarbonTablePath(carbonStorePath,
- dictionaryColumnUniqueIdentifier.getCarbonTableIdentifier());
+ dictionaryColumnUniqueIdentifier.getCarbonTableIdentifier(),
+ dictionaryColumnUniqueIdentifier);
String dictionaryFilePath = carbonTablePath.getDictionaryFilePath(
dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId());