You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ra...@apache.org on 2017/11/13 22:11:30 UTC
[04/49] carbondata git commit: [CARBONDATA-1617] Merging carbonindex
files within segment
[CARBONDATA-1617] Merging carbonindex files within segment
Merge the carbonindex files after data load, so that we can reduce the IO calls to namenode and improves the read performance for first query
This closes #1436
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/0586146a
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/0586146a
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/0586146a
Branch: refs/heads/fgdatamap
Commit: 0586146a8bd953db63e1d99608ba8a77a9f5a899
Parents: b491609
Author: ravipesala <ra...@gmail.com>
Authored: Wed Oct 25 11:13:22 2017 +0530
Committer: Jacky Li <ja...@qq.com>
Committed: Wed Nov 1 21:14:34 2017 +0530
----------------------------------------------------------------------
.../core/constants/CarbonCommonConstants.java | 8 +
.../carbondata/core/datamap/TableDataMap.java | 6 +-
.../carbondata/core/datamap/dev/DataMap.java | 2 +-
.../core/datamap/dev/DataMapFactory.java | 4 +-
.../core/datamap/dev/DataMapModel.java | 34 ++++
.../indexstore/BlockletDataMapIndexStore.java | 64 +++++--
.../blockletindex/BlockletDataMap.java | 13 +-
.../blockletindex/BlockletDataMapFactory.java | 47 ++---
.../blockletindex/BlockletDataMapModel.java | 33 ++++
.../blockletindex/SegmentIndexFileStore.java | 189 +++++++++++++++++++
.../core/reader/CarbonIndexFileReader.java | 10 +
.../carbondata/core/reader/ThriftReader.java | 9 +
.../util/AbstractDataFileFooterConverter.java | 8 +-
.../apache/carbondata/core/util/CarbonUtil.java | 4 +-
.../core/util/path/CarbonTablePath.java | 49 +++--
.../core/writer/CarbonIndexFileMergeWriter.java | 116 ++++++++++++
.../main/thrift/carbondata_index_merge.thrift | 32 ++++
.../hadoop/api/CarbonTableInputFormat.java | 14 +-
.../sdv/generated/MergeIndexTestCase.scala | 122 ++++++++++++
.../cluster/sdv/suite/SDVSuites.scala | 3 +-
.../CarbonIndexFileMergeTestCase.scala | 143 ++++++++++++++
...ompactionSupportGlobalSortFunctionTest.scala | 6 +-
...mpactionSupportGlobalSortParameterTest.scala | 8 +-
.../dataload/TestBatchSortDataLoad.scala | 8 +-
.../dataload/TestGlobalSortDataLoad.scala | 7 +-
.../testsuite/datamap/DataMapWriterSuite.scala | 2 +-
.../spark/rdd/CarbonMergeFilesRDD.scala | 82 ++++++++
.../apache/carbondata/spark/rdd/Compactor.scala | 6 +-
.../carbondata/spark/util/CommonUtil.scala | 20 +-
.../spark/sql/catalyst/CarbonDDLSqlParser.scala | 3 +-
.../spark/rdd/CarbonDataRDDFactory.scala | 23 ++-
.../sql/execution/strategy/DDLStrategy.scala | 3 +-
.../processing/merger/CarbonDataMergerUtil.java | 37 ++--
.../processing/merger/CompactionType.java | 1 +
34 files changed, 996 insertions(+), 120 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
index 4ee3001..3fed18f 100644
--- a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
+++ b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
@@ -1376,6 +1376,14 @@ public final class CarbonCommonConstants {
public static final String BITSET_PIPE_LINE_DEFAULT = "true";
+ /**
+ * It is internal configuration and used only for test purpose.
+ * It will merge the carbon index files with in the segment to single segment.
+ */
+ public static final String CARBON_MERGE_INDEX_IN_SEGMENT = "carbon.merge.index.in.segment";
+
+ public static final String CARBON_MERGE_INDEX_IN_SEGMENT_DEFAULT = "true";
+
private CarbonCommonConstants() {
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java b/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
index 3e5e9e4..1cf1def 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
@@ -118,7 +118,11 @@ public final class TableDataMap implements EventListener {
public List<ExtendedBlocklet> prune(DataMapDistributable distributable,
FilterResolverIntf filterExp) throws IOException {
List<ExtendedBlocklet> detailedBlocklets = new ArrayList<>();
- List<Blocklet> blocklets = dataMapFactory.getDataMap(distributable).prune(filterExp);
+ List<Blocklet> blocklets = new ArrayList<>();
+ List<DataMap> dataMaps = dataMapFactory.getDataMaps(distributable);
+ for (DataMap dataMap : dataMaps) {
+ blocklets.addAll(dataMap.prune(filterExp));
+ }
for (Blocklet blocklet: blocklets) {
ExtendedBlocklet detailedBlocklet =
blockletDetailsFetcher.getExtendedBlocklet(blocklet, distributable.getSegmentId());
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMap.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMap.java b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMap.java
index f6ea885..ada23ad 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMap.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMap.java
@@ -31,7 +31,7 @@ public interface DataMap {
/**
* It is called to load the data map to memory or to initialize it.
*/
- void init(String filePath) throws MemoryException, IOException;
+ void init(DataMapModel dataMapModel) throws MemoryException, IOException;
/**
* Prune the datamap with filter expression. It returns the list of
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapFactory.java b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapFactory.java
index 62cf813..cf0519b 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapFactory.java
@@ -45,9 +45,9 @@ public interface DataMapFactory {
List<DataMap> getDataMaps(String segmentId) throws IOException;
/**
- * Get datamap for distributable object.
+ * Get datamaps for distributable object.
*/
- DataMap getDataMap(DataMapDistributable distributable);
+ List<DataMap> getDataMaps(DataMapDistributable distributable) throws IOException;
/**
* Get all distributable objects of a segmentid
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapModel.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapModel.java b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapModel.java
new file mode 100644
index 0000000..76bbeee
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapModel.java
@@ -0,0 +1,34 @@
+/*
+ * 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.datamap.dev;
+
+/**
+ * Information required to build datamap
+ */
+public class DataMapModel {
+
+ private String filePath;
+
+ public DataMapModel(String filePath) {
+ this.filePath = filePath;
+ }
+
+ public String getFilePath() {
+ return filePath;
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDataMapIndexStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDataMapIndexStore.java b/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDataMapIndexStore.java
index 9d4af7b..1ea4806 100644
--- a/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDataMapIndexStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDataMapIndexStore.java
@@ -18,6 +18,7 @@ package org.apache.carbondata.core.indexstore;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -27,7 +28,10 @@ import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.cache.Cache;
import org.apache.carbondata.core.cache.CarbonLRUCache;
import org.apache.carbondata.core.indexstore.blockletindex.BlockletDataMap;
+import org.apache.carbondata.core.indexstore.blockletindex.BlockletDataMapModel;
+import org.apache.carbondata.core.indexstore.blockletindex.SegmentIndexFileStore;
import org.apache.carbondata.core.memory.MemoryException;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
/**
* Class to handle loading, unloading,clearing,storing of the table
@@ -66,13 +70,19 @@ public class BlockletDataMapIndexStore
segmentLockMap = new ConcurrentHashMap<String, Object>();
}
- @Override public BlockletDataMap get(TableBlockIndexUniqueIdentifier tableSegmentUniqueIdentifier)
+ @Override
+ public BlockletDataMap get(TableBlockIndexUniqueIdentifier identifier)
throws IOException {
- String lruCacheKey = tableSegmentUniqueIdentifier.getUniqueTableSegmentIdentifier();
+ String lruCacheKey = identifier.getUniqueTableSegmentIdentifier();
BlockletDataMap dataMap = (BlockletDataMap) lruCache.get(lruCacheKey);
if (dataMap == null) {
try {
- dataMap = loadAndGetDataMap(tableSegmentUniqueIdentifier);
+ SegmentIndexFileStore indexFileStore = new SegmentIndexFileStore();
+ indexFileStore.readAllIIndexOfSegment(
+ CarbonTablePath.getSegmentPath(
+ identifier.getAbsoluteTableIdentifier().getTablePath(),
+ identifier.getSegmentId()));
+ dataMap = loadAndGetDataMap(identifier, indexFileStore);
} catch (MemoryException e) {
LOGGER.error("memory exception when loading datamap: " + e.getMessage());
throw new RuntimeException(e.getMessage(), e);
@@ -81,18 +91,41 @@ public class BlockletDataMapIndexStore
return dataMap;
}
- @Override public List<BlockletDataMap> getAll(
+ @Override
+ public List<BlockletDataMap> getAll(
List<TableBlockIndexUniqueIdentifier> tableSegmentUniqueIdentifiers) throws IOException {
List<BlockletDataMap> blockletDataMaps = new ArrayList<>(tableSegmentUniqueIdentifiers.size());
+ List<TableBlockIndexUniqueIdentifier> missedIdentifiers = new ArrayList<>();
+ // Get the datamaps for each indexfile from cache.
try {
for (TableBlockIndexUniqueIdentifier identifier : tableSegmentUniqueIdentifiers) {
- blockletDataMaps.add(get(identifier));
+ BlockletDataMap ifPresent = getIfPresent(identifier);
+ if (ifPresent != null) {
+ blockletDataMaps.add(ifPresent);
+ } else {
+ missedIdentifiers.add(identifier);
+ }
+ }
+ if (missedIdentifiers.size() > 0) {
+ Map<String, SegmentIndexFileStore> segmentIndexFileStoreMap = new HashMap<>();
+ for (TableBlockIndexUniqueIdentifier identifier: missedIdentifiers) {
+ SegmentIndexFileStore indexFileStore =
+ segmentIndexFileStoreMap.get(identifier.getSegmentId());
+ if (indexFileStore == null) {
+ String segmentPath = CarbonTablePath
+ .getSegmentPath(identifier.getAbsoluteTableIdentifier().getTablePath(),
+ identifier.getSegmentId());
+ indexFileStore = new SegmentIndexFileStore();
+ indexFileStore.readAllIIndexOfSegment(segmentPath);
+ segmentIndexFileStoreMap.put(identifier.getSegmentId(), indexFileStore);
+ }
+ blockletDataMaps.add(loadAndGetDataMap(identifier, indexFileStore));
+ }
}
} catch (Throwable e) {
for (BlockletDataMap dataMap : blockletDataMaps) {
dataMap.clear();
}
- e.printStackTrace();
throw new IOException("Problem in loading segment blocks.", e);
}
return blockletDataMaps;
@@ -104,7 +137,8 @@ public class BlockletDataMapIndexStore
* @param tableSegmentUniqueIdentifier
* @return
*/
- @Override public BlockletDataMap getIfPresent(
+ @Override
+ public BlockletDataMap getIfPresent(
TableBlockIndexUniqueIdentifier tableSegmentUniqueIdentifier) {
BlockletDataMap dataMap = (BlockletDataMap) lruCache
.get(tableSegmentUniqueIdentifier.getUniqueTableSegmentIdentifier());
@@ -116,7 +150,8 @@ public class BlockletDataMapIndexStore
*
* @param tableSegmentUniqueIdentifier
*/
- @Override public void invalidate(TableBlockIndexUniqueIdentifier tableSegmentUniqueIdentifier) {
+ @Override
+ public void invalidate(TableBlockIndexUniqueIdentifier tableSegmentUniqueIdentifier) {
lruCache.remove(tableSegmentUniqueIdentifier.getUniqueTableSegmentIdentifier());
}
@@ -130,10 +165,11 @@ public class BlockletDataMapIndexStore
* @throws IOException
*/
private BlockletDataMap loadAndGetDataMap(
- TableBlockIndexUniqueIdentifier tableSegmentUniqueIdentifier)
+ TableBlockIndexUniqueIdentifier identifier,
+ SegmentIndexFileStore indexFileStore)
throws IOException, MemoryException {
String uniqueTableSegmentIdentifier =
- tableSegmentUniqueIdentifier.getUniqueTableSegmentIdentifier();
+ identifier.getUniqueTableSegmentIdentifier();
Object lock = segmentLockMap.get(uniqueTableSegmentIdentifier);
if (lock == null) {
lock = addAndGetSegmentLock(uniqueTableSegmentIdentifier);
@@ -141,8 +177,9 @@ public class BlockletDataMapIndexStore
BlockletDataMap dataMap = null;
synchronized (lock) {
dataMap = new BlockletDataMap();
- dataMap.init(tableSegmentUniqueIdentifier.getFilePath());
- lruCache.put(tableSegmentUniqueIdentifier.getUniqueTableSegmentIdentifier(), dataMap,
+ dataMap.init(new BlockletDataMapModel(identifier.getFilePath(),
+ indexFileStore.getFileData(identifier.getCarbonIndexFileName())));
+ lruCache.put(identifier.getUniqueTableSegmentIdentifier(), dataMap,
dataMap.getMemorySize());
}
return dataMap;
@@ -170,7 +207,8 @@ public class BlockletDataMapIndexStore
*
* @param tableSegmentUniqueIdentifiers
*/
- @Override public void clearAccessCount(
+ @Override
+ public void clearAccessCount(
List<TableBlockIndexUniqueIdentifier> tableSegmentUniqueIdentifiers) {
for (TableBlockIndexUniqueIdentifier segmentUniqueIdentifier : tableSegmentUniqueIdentifiers) {
BlockletDataMap cacheable =
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMap.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMap.java b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMap.java
index 0d7bb71..3e083cc 100644
--- a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMap.java
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMap.java
@@ -35,6 +35,7 @@ import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.cache.Cacheable;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datamap.dev.DataMap;
+import org.apache.carbondata.core.datamap.dev.DataMapModel;
import org.apache.carbondata.core.datastore.IndexKey;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.datastore.block.TableBlockInfo;
@@ -96,10 +97,13 @@ public class BlockletDataMap implements DataMap, Cacheable {
private int[] columnCardinality;
@Override
- public void init(String filePath) throws IOException, MemoryException {
+ public void init(DataMapModel dataMapModel) throws IOException, MemoryException {
long startTime = System.currentTimeMillis();
+ assert (dataMapModel instanceof BlockletDataMapModel);
+ BlockletDataMapModel blockletDataMapInfo = (BlockletDataMapModel) dataMapModel;
DataFileFooterConverter fileFooterConverter = new DataFileFooterConverter();
- List<DataFileFooter> indexInfo = fileFooterConverter.getIndexInfo(filePath);
+ List<DataFileFooter> indexInfo = fileFooterConverter
+ .getIndexInfo(blockletDataMapInfo.getFilePath(), blockletDataMapInfo.getFileData());
for (DataFileFooter fileFooter : indexInfo) {
List<ColumnSchema> columnInTable = fileFooter.getColumnInTable();
if (segmentProperties == null) {
@@ -119,8 +123,9 @@ public class BlockletDataMap implements DataMap, Cacheable {
if (unsafeMemoryDMStore != null) {
unsafeMemoryDMStore.finishWriting();
}
- LOGGER.info("Time taken to load blocklet datamap from file : " + filePath + "is " +
- (System.currentTimeMillis() - startTime));
+ LOGGER.info(
+ "Time taken to load blocklet datamap from file : " + dataMapModel.getFilePath() + "is " + (
+ System.currentTimeMillis() - startTime));
}
private void loadToUnsafe(DataFileFooter fileFooter, SegmentProperties segmentProperties,
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapFactory.java b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapFactory.java
index 5edc5b7..a1bbba5 100644
--- a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapFactory.java
@@ -31,7 +31,6 @@ import org.apache.carbondata.core.datamap.dev.DataMap;
import org.apache.carbondata.core.datamap.dev.DataMapFactory;
import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
-import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.events.ChangeEvent;
import org.apache.carbondata.core.indexstore.Blocklet;
@@ -78,15 +77,16 @@ public class BlockletDataMapFactory implements DataMapFactory, BlockletDetailsFe
}
private List<TableBlockIndexUniqueIdentifier> getTableBlockIndexUniqueIdentifiers(
- String segmentId) {
+ String segmentId) throws IOException {
List<TableBlockIndexUniqueIdentifier> tableBlockIndexUniqueIdentifiers =
segmentMap.get(segmentId);
if (tableBlockIndexUniqueIdentifiers == null) {
tableBlockIndexUniqueIdentifiers = new ArrayList<>();
- CarbonFile[] listFiles = getCarbonIndexFiles(segmentId);
- for (int i = 0; i < listFiles.length; i++) {
+ String path = CarbonTablePath.getSegmentPath(identifier.getTablePath(), segmentId);
+ List<String> indexFiles = new SegmentIndexFileStore().getIndexFilesFromSegment(path);
+ for (int i = 0; i < indexFiles.size(); i++) {
tableBlockIndexUniqueIdentifiers.add(
- new TableBlockIndexUniqueIdentifier(identifier, segmentId, listFiles[i].getName()));
+ new TableBlockIndexUniqueIdentifier(identifier, segmentId, indexFiles.get(i)));
}
segmentMap.put(segmentId, tableBlockIndexUniqueIdentifiers);
}
@@ -141,19 +141,10 @@ public class BlockletDataMapFactory implements DataMapFactory, BlockletDetailsFe
throw new IOException("Blocklet with blockid " + blocklet.getPath() + " not found ");
}
- private CarbonFile[] getCarbonIndexFiles(String segmentId) {
- String path = identifier.getTablePath() + "/Fact/Part0/Segment_" + segmentId;
- CarbonFile carbonFile = FileFactory.getCarbonFile(path);
- return carbonFile.listFiles(new CarbonFileFilter() {
- @Override public boolean accept(CarbonFile file) {
- return file.getName().endsWith(".carbonindex");
- }
- });
- }
@Override
public List<DataMapDistributable> toDistributable(String segmentId) {
- CarbonFile[] carbonIndexFiles = getCarbonIndexFiles(segmentId);
+ CarbonFile[] carbonIndexFiles = SegmentIndexFileStore.getCarbonIndexFiles(segmentId);
List<DataMapDistributable> distributables = new ArrayList<>();
for (int i = 0; i < carbonIndexFiles.length; i++) {
Path path = new Path(carbonIndexFiles[i].getPath());
@@ -195,18 +186,30 @@ public class BlockletDataMapFactory implements DataMapFactory, BlockletDetailsFe
}
@Override
- public DataMap getDataMap(DataMapDistributable distributable) {
+ public List<DataMap> getDataMaps(DataMapDistributable distributable) throws IOException {
BlockletDataMapDistributable mapDistributable = (BlockletDataMapDistributable) distributable;
- TableBlockIndexUniqueIdentifier uniqueIdentifier =
- new TableBlockIndexUniqueIdentifier(identifier, distributable.getSegmentId(),
- mapDistributable.getFilePath());
- DataMap dataMap;
+ List<TableBlockIndexUniqueIdentifier> identifiers = new ArrayList<>();
+ if (mapDistributable.getFilePath().endsWith(CarbonTablePath.INDEX_FILE_EXT)) {
+ identifiers.add(new TableBlockIndexUniqueIdentifier(identifier, distributable.getSegmentId(),
+ mapDistributable.getFilePath()));
+ } else if (mapDistributable.getFilePath().endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT)) {
+ SegmentIndexFileStore fileStore = new SegmentIndexFileStore();
+ List<String> indexFiles = fileStore.getIndexFilesFromMergeFile(
+ CarbonTablePath.getSegmentPath(identifier.getTablePath(), mapDistributable.getSegmentId())
+ + "/" + mapDistributable.getFilePath());
+ for (String indexFile : indexFiles) {
+ identifiers.add(
+ new TableBlockIndexUniqueIdentifier(identifier, distributable.getSegmentId(),
+ indexFile));
+ }
+ }
+ List<DataMap> dataMaps;
try {
- dataMap = cache.get(uniqueIdentifier);
+ dataMaps = cache.getAll(identifiers);
} catch (IOException e) {
throw new RuntimeException(e);
}
- return dataMap;
+ return dataMaps;
}
@Override
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapModel.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapModel.java b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapModel.java
new file mode 100644
index 0000000..7ffa32d
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapModel.java
@@ -0,0 +1,33 @@
+/*
+ * 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.indexstore.blockletindex;
+
+import org.apache.carbondata.core.datamap.dev.DataMapModel;
+
+public class BlockletDataMapModel extends DataMapModel {
+
+ private byte[] fileData;
+
+ public BlockletDataMapModel(String filePath, byte[] fileData) {
+ super(filePath);
+ this.fileData = fileData;
+ }
+
+ public byte[] getFileData() {
+ return fileData;
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/SegmentIndexFileStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/SegmentIndexFileStore.java b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/SegmentIndexFileStore.java
new file mode 100644
index 0000000..22d3d43
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/SegmentIndexFileStore.java
@@ -0,0 +1,189 @@
+/*
+ * 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.indexstore.blockletindex;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.*;
+
+import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
+import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
+import org.apache.carbondata.core.datastore.impl.FileFactory;
+import org.apache.carbondata.core.reader.ThriftReader;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
+import org.apache.carbondata.format.MergedBlockIndex;
+import org.apache.carbondata.format.MergedBlockIndexHeader;
+
+import org.apache.thrift.TBase;
+
+/**
+ * This class manages reading of index files with in the segment. The files it read can be
+ * carbonindex or carbonindexmerge files.
+ */
+public class SegmentIndexFileStore {
+
+ /**
+ * Stores the indexfile name and related binary file data in it.
+ */
+ private Map<String, byte[]> carbonIndexMap;
+
+ public SegmentIndexFileStore() throws IOException {
+ carbonIndexMap = new HashMap<>();
+ }
+
+ /**
+ * Read all index files and keep the cache in it.
+ *
+ * @param segmentPath
+ * @throws IOException
+ */
+ public void readAllIIndexOfSegment(String segmentPath) throws IOException {
+ CarbonFile[] carbonIndexFiles = getCarbonIndexFiles(segmentPath);
+ for (int i = 0; i < carbonIndexFiles.length; i++) {
+ if (carbonIndexFiles[i].getName().endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT)) {
+ readMergeFile(carbonIndexFiles[i].getCanonicalPath());
+ } else if (carbonIndexFiles[i].getName().endsWith(CarbonTablePath.INDEX_FILE_EXT)) {
+ readIndexFile(carbonIndexFiles[i]);
+ }
+ }
+ }
+
+ /**
+ * Read all index file names of the segment
+ *
+ * @param segmentPath
+ * @return
+ * @throws IOException
+ */
+ public List<String> getIndexFilesFromSegment(String segmentPath) throws IOException {
+ CarbonFile[] carbonIndexFiles = getCarbonIndexFiles(segmentPath);
+ Set<String> indexFiles = new HashSet<>();
+ for (int i = 0; i < carbonIndexFiles.length; i++) {
+ if (carbonIndexFiles[i].getName().endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT)) {
+ indexFiles.addAll(getIndexFilesFromMergeFile(carbonIndexFiles[i].getCanonicalPath()));
+ } else if (carbonIndexFiles[i].getName().endsWith(CarbonTablePath.INDEX_FILE_EXT)) {
+ indexFiles.add(carbonIndexFiles[i].getName());
+ }
+ }
+ return new ArrayList<>(indexFiles);
+ }
+
+ /**
+ * List all the index files inside merge file.
+ * @param mergeFile
+ * @return
+ * @throws IOException
+ */
+ public List<String> getIndexFilesFromMergeFile(String mergeFile) throws IOException {
+ List<String> indexFiles = new ArrayList<>();
+ ThriftReader thriftReader = new ThriftReader(mergeFile);
+ thriftReader.open();
+ MergedBlockIndexHeader indexHeader = readMergeBlockIndexHeader(thriftReader);
+ List<String> file_names = indexHeader.getFile_names();
+ indexFiles.addAll(file_names);
+ thriftReader.close();
+ return indexFiles;
+ }
+
+ /**
+ * Read carbonindexmerge file and update the map
+ *
+ * @param mergeFilePath
+ * @throws IOException
+ */
+ private void readMergeFile(String mergeFilePath) throws IOException {
+ ThriftReader thriftReader = new ThriftReader(mergeFilePath);
+ thriftReader.open();
+ MergedBlockIndexHeader indexHeader = readMergeBlockIndexHeader(thriftReader);
+ MergedBlockIndex mergedBlockIndex = readMergeBlockIndex(thriftReader);
+ List<String> file_names = indexHeader.getFile_names();
+ List<ByteBuffer> fileData = mergedBlockIndex.getFileData();
+ assert (file_names.size() == fileData.size());
+ for (int i = 0; i < file_names.size(); i++) {
+ carbonIndexMap.put(file_names.get(i), fileData.get(i).array());
+ }
+ thriftReader.close();
+ }
+
+ /**
+ * Read carbonindex file and convert to stream and add to map
+ *
+ * @param indexFile
+ * @throws IOException
+ */
+ private void readIndexFile(CarbonFile indexFile) throws IOException {
+ String indexFilePath = indexFile.getCanonicalPath();
+ DataInputStream dataInputStream =
+ FileFactory.getDataInputStream(indexFilePath, FileFactory.getFileType(indexFilePath));
+ byte[] bytes = new byte[(int) indexFile.getSize()];
+ dataInputStream.readFully(bytes);
+ carbonIndexMap.put(indexFile.getName(), bytes);
+ }
+
+ private MergedBlockIndexHeader readMergeBlockIndexHeader(ThriftReader thriftReader)
+ throws IOException {
+ return (MergedBlockIndexHeader) thriftReader.read(new ThriftReader.TBaseCreator() {
+ @Override public TBase create() {
+ return new MergedBlockIndexHeader();
+ }
+ });
+ }
+
+ private MergedBlockIndex readMergeBlockIndex(ThriftReader thriftReader) throws IOException {
+ return (MergedBlockIndex) thriftReader.read(new ThriftReader.TBaseCreator() {
+ @Override public TBase create() {
+ return new MergedBlockIndex();
+ }
+ });
+ }
+
+ /**
+ * Get the carbonindex file content
+ *
+ * @param fileName
+ * @return
+ */
+ public byte[] getFileData(String fileName) {
+ return carbonIndexMap.get(fileName);
+ }
+
+ /**
+ * List all the index files of the segment.
+ *
+ * @param segmentPath
+ * @return
+ */
+ public static CarbonFile[] getCarbonIndexFiles(String segmentPath) {
+ CarbonFile carbonFile = FileFactory.getCarbonFile(segmentPath);
+ return carbonFile.listFiles(new CarbonFileFilter() {
+ @Override public boolean accept(CarbonFile file) {
+ return file.getName().endsWith(CarbonTablePath.INDEX_FILE_EXT) || file.getName()
+ .endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT);
+ }
+ });
+ }
+
+ /**
+ * Return the map that contain index file name and content of the file.
+ *
+ * @return
+ */
+ public Map<String, byte[]> getCarbonIndexMap() {
+ return carbonIndexMap;
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/reader/CarbonIndexFileReader.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/reader/CarbonIndexFileReader.java b/core/src/main/java/org/apache/carbondata/core/reader/CarbonIndexFileReader.java
index dc46242..4617a12 100644
--- a/core/src/main/java/org/apache/carbondata/core/reader/CarbonIndexFileReader.java
+++ b/core/src/main/java/org/apache/carbondata/core/reader/CarbonIndexFileReader.java
@@ -80,6 +80,16 @@ public class CarbonIndexFileReader {
}
/**
+ * Open the thrift reader
+ *
+ * @param fileData
+ * @throws IOException
+ */
+ public void openThriftReader(byte[] fileData) throws IOException {
+ thriftReader = new ThriftReader(fileData);
+ }
+
+ /**
* check if any more object is present
*
* @return true if any more object can be read
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/reader/ThriftReader.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/reader/ThriftReader.java b/core/src/main/java/org/apache/carbondata/core/reader/ThriftReader.java
index 3e2aab4..221a285 100644
--- a/core/src/main/java/org/apache/carbondata/core/reader/ThriftReader.java
+++ b/core/src/main/java/org/apache/carbondata/core/reader/ThriftReader.java
@@ -17,6 +17,7 @@
package org.apache.carbondata.core.reader;
+import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
@@ -70,6 +71,14 @@ public class ThriftReader {
}
/**
+ * Constructor.
+ */
+ public ThriftReader(byte[] fileData) {
+ dataInputStream = new DataInputStream(new ByteArrayInputStream(fileData));
+ binaryIn = new TCompactProtocol(new TIOStreamTransport(dataInputStream));
+ }
+
+ /**
* Opens the fileName for reading.
*/
public void open() throws IOException {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/util/AbstractDataFileFooterConverter.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/AbstractDataFileFooterConverter.java b/core/src/main/java/org/apache/carbondata/core/util/AbstractDataFileFooterConverter.java
index 15f85d6..cba9931 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/AbstractDataFileFooterConverter.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/AbstractDataFileFooterConverter.java
@@ -126,13 +126,17 @@ public abstract class AbstractDataFileFooterConverter {
* @return list of index info
* @throws IOException problem while reading the index file
*/
- public List<DataFileFooter> getIndexInfo(String filePath) throws IOException {
+ public List<DataFileFooter> getIndexInfo(String filePath, byte[] fileData) throws IOException {
CarbonIndexFileReader indexReader = new CarbonIndexFileReader();
List<DataFileFooter> dataFileFooters = new ArrayList<DataFileFooter>();
String parentPath = filePath.substring(0, filePath.lastIndexOf("/"));
try {
// open the reader
- indexReader.openThriftReader(filePath);
+ if (fileData != null) {
+ indexReader.openThriftReader(fileData);
+ } else {
+ indexReader.openThriftReader(filePath);
+ }
// get the index header
org.apache.carbondata.format.IndexHeader readIndexHeader = indexReader.readIndexHeader();
List<ColumnSchema> columnSchemaList = new ArrayList<ColumnSchema>();
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/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 ef3e71d..77789c5 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
@@ -389,7 +389,9 @@ public final class CarbonUtil {
public static void deleteFiles(File[] intermediateFiles) throws IOException {
for (int i = 0; i < intermediateFiles.length; i++) {
- if (!intermediateFiles[i].delete()) {
+ // ignore deleting for index file since it is inside merged file.
+ if (!intermediateFiles[i].delete() && !intermediateFiles[i].getName()
+ .endsWith(CarbonTablePath.INDEX_FILE_EXT)) {
throw new IOException("Problem while deleting intermediate file");
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java b/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
index 7be9c76..02a000a 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
@@ -33,26 +33,28 @@ import org.apache.hadoop.fs.Path;
*/
public class CarbonTablePath extends Path {
- protected static final String METADATA_DIR = "Metadata";
- protected static final String DICTIONARY_EXT = ".dict";
- protected static final String DICTIONARY_META_EXT = ".dictmeta";
- protected static final String SORT_INDEX_EXT = ".sortindex";
- protected static final String SCHEMA_FILE = "schema";
- protected static final String TABLE_STATUS_FILE = "tablestatus";
- protected static final String TABLE_UPDATE_STATUS_FILE = "tableupdatestatus";
- protected static final String FACT_DIR = "Fact";
- protected static final String SEGMENT_PREFIX = "Segment_";
- protected static final String PARTITION_PREFIX = "Part";
- protected static final String CARBON_DATA_EXT = ".carbondata";
- protected static final String CARBON_DELTE_DELTA_EXT = ".deletedelta";
- protected static final String CARBON_UPDATE_DELTA_EXT = ".updatedelta";
- protected static final String DATA_PART_PREFIX = "part-";
- protected static final String BATCH_PREFIX = "_batchno";
- protected static final String INDEX_FILE_EXT = ".carbonindex";
- protected static final String DELETE_DELTA_FILE_EXT = ".deletedelta";
-
- protected String tablePath;
- protected CarbonTableIdentifier carbonTableIdentifier;
+ private static final String METADATA_DIR = "Metadata";
+ private static final String DICTIONARY_EXT = ".dict";
+ private static final String DICTIONARY_META_EXT = ".dictmeta";
+ private static final String SORT_INDEX_EXT = ".sortindex";
+ private static final String SCHEMA_FILE = "schema";
+ private static final String TABLE_STATUS_FILE = "tablestatus";
+ private static final String TABLE_UPDATE_STATUS_FILE = "tableupdatestatus";
+ private static final String FACT_DIR = "Fact";
+ private static final String SEGMENT_PREFIX = "Segment_";
+ private static final String PARTITION_PREFIX = "Part";
+ private static final String CARBON_DATA_EXT = ".carbondata";
+ private static final String CARBON_DELTE_DELTA_EXT = ".deletedelta";
+ private static final String CARBON_UPDATE_DELTA_EXT = ".updatedelta";
+ private static final String DATA_PART_PREFIX = "part-";
+ private static final String BATCH_PREFIX = "_batchno";
+ private static final String DELETE_DELTA_FILE_EXT = ".deletedelta";
+
+ public static final String INDEX_FILE_EXT = ".carbonindex";
+ public static final String MERGE_INDEX_FILE_EXT = ".carbonindexmerge";
+
+ private String tablePath;
+ private CarbonTableIdentifier carbonTableIdentifier;
/**
* structure CarbonTablePath object to manage table paths
@@ -746,4 +748,11 @@ public class CarbonTablePath extends Path {
return DataFileUtil.getTaskNo(actualBlockName) + "-" + DataFileUtil.getBucketNo(actualBlockName)
+ "-" + DataFileUtil.getTimeStampFromFileName(actualBlockName) + INDEX_FILE_EXT;
}
+
+ /**
+ * Get the segment path from table path and segmentid
+ */
+ public static String getSegmentPath(String tablePath, String segmentId) {
+ return tablePath + "/Fact/Part0/Segment_" + segmentId;
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/core/src/main/java/org/apache/carbondata/core/writer/CarbonIndexFileMergeWriter.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/writer/CarbonIndexFileMergeWriter.java b/core/src/main/java/org/apache/carbondata/core/writer/CarbonIndexFileMergeWriter.java
new file mode 100644
index 0000000..8d9bddc
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/writer/CarbonIndexFileMergeWriter.java
@@ -0,0 +1,116 @@
+/*
+ * 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.writer;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
+import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
+import org.apache.carbondata.core.datastore.impl.FileFactory;
+import org.apache.carbondata.core.indexstore.blockletindex.SegmentIndexFileStore;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
+import org.apache.carbondata.format.MergedBlockIndex;
+import org.apache.carbondata.format.MergedBlockIndexHeader;
+
+public class CarbonIndexFileMergeWriter {
+
+ /**
+ * thrift writer object
+ */
+ private ThriftWriter thriftWriter;
+
+ /**
+ * Merge all the carbonindex files of segment to a merged file
+ * @param segmentPath
+ * @throws IOException
+ */
+ public void mergeCarbonIndexFilesOfSegment(String segmentPath) throws IOException {
+ CarbonFile carbonFile = FileFactory.getCarbonFile(segmentPath);
+ CarbonFile[] indexFiles = carbonFile.listFiles(new CarbonFileFilter() {
+ @Override public boolean accept(CarbonFile file) {
+ return file.getName().endsWith(CarbonTablePath.INDEX_FILE_EXT);
+ }
+ });
+ if (indexFiles.length > 0) {
+ SegmentIndexFileStore fileStore = new SegmentIndexFileStore();
+ fileStore.readAllIIndexOfSegment(segmentPath);
+ openThriftWriter(segmentPath + "/" +
+ System.currentTimeMillis() + CarbonTablePath.MERGE_INDEX_FILE_EXT);
+ Map<String, byte[]> indexMap = fileStore.getCarbonIndexMap();
+ MergedBlockIndexHeader indexHeader = new MergedBlockIndexHeader();
+ MergedBlockIndex mergedBlockIndex = new MergedBlockIndex();
+ List<String> fileNames = new ArrayList<>(indexMap.size());
+ List<ByteBuffer> data = new ArrayList<>(indexMap.size());
+ for (Map.Entry<String, byte[]> entry : indexMap.entrySet()) {
+ fileNames.add(entry.getKey());
+ data.add(ByteBuffer.wrap(entry.getValue()));
+ }
+ indexHeader.setFile_names(fileNames);
+ mergedBlockIndex.setFileData(data);
+ writeMergedBlockIndexHeader(indexHeader);
+ writeMergedBlockIndex(mergedBlockIndex);
+ close();
+ for (CarbonFile indexFile : indexFiles) {
+ indexFile.delete();
+ }
+ }
+
+ }
+
+ /**
+ * It writes thrift object to file
+ *
+ * @throws IOException
+ */
+ private void writeMergedBlockIndexHeader(MergedBlockIndexHeader indexObject) throws IOException {
+ thriftWriter.write(indexObject);
+ }
+
+ /**
+ * It writes thrift object to file
+ *
+ * @throws IOException
+ */
+ private void writeMergedBlockIndex(MergedBlockIndex indexObject) throws IOException {
+ thriftWriter.write(indexObject);
+ }
+
+ /**
+ * Below method will be used to open the thrift writer
+ *
+ * @param filePath file path where data need to be written
+ * @throws IOException throws io exception in case of any failure
+ */
+ private void openThriftWriter(String filePath) throws IOException {
+ // create thrift writer instance
+ thriftWriter = new ThriftWriter(filePath, false);
+ // open the file stream
+ thriftWriter.open();
+ }
+
+ /**
+ * Below method will be used to close the thrift object
+ */
+ private void close() throws IOException {
+ thriftWriter.close();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/format/src/main/thrift/carbondata_index_merge.thrift
----------------------------------------------------------------------
diff --git a/format/src/main/thrift/carbondata_index_merge.thrift b/format/src/main/thrift/carbondata_index_merge.thrift
new file mode 100644
index 0000000..468d90d
--- /dev/null
+++ b/format/src/main/thrift/carbondata_index_merge.thrift
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+
+/**
+ * File format description for CarbonData merged index file.
+ */
+namespace java org.apache.carbondata.format
+
+struct MergedBlockIndexHeader{
+ 1: required list<string> file_names; // list of carbon index file names
+}
+
+/**
+ * Merged carbon index file information
+ */
+struct MergedBlockIndex {
+ 1: required list<binary> fileData; // content of the indexed files
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/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 314ffd5..9fbeb8a 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
@@ -19,6 +19,7 @@ package org.apache.carbondata.hadoop.api;
import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.util.ArrayList;
@@ -576,7 +577,10 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
// for partition table, the task id of carbaondata file name is the partition id.
// if this partition is not required, here will skip it.
if (matchedPartitions == null || matchedPartitions.get(partitionIndex)) {
- resultFilterredBlocks.add(convertToCarbonInputSplit(blocklet));
+ CarbonInputSplit inputSplit = convertToCarbonInputSplit(blocklet);
+ if (inputSplit != null) {
+ resultFilterredBlocks.add(inputSplit);
+ }
}
}
}
@@ -588,7 +592,13 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
private CarbonInputSplit convertToCarbonInputSplit(ExtendedBlocklet blocklet)
throws IOException {
- blocklet.updateLocations();
+ try {
+ blocklet.updateLocations();
+ } catch (FileNotFoundException e) {
+ // In case of clean files there is a chance of carbondata file is deleted but index file
+ // exist inside merged file. So just return null.
+ return null;
+ }
org.apache.carbondata.hadoop.CarbonInputSplit split =
org.apache.carbondata.hadoop.CarbonInputSplit.from(blocklet.getSegmentId(),
new FileSplit(new Path(blocklet.getPath()), 0, blocklet.getLength(),
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/MergeIndexTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/MergeIndexTestCase.scala b/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/MergeIndexTestCase.scala
new file mode 100644
index 0000000..01146ee
--- /dev/null
+++ b/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/MergeIndexTestCase.scala
@@ -0,0 +1,122 @@
+
+/*
+ * 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.cluster.sdv.generated
+
+import org.apache.spark.sql.CarbonEnv
+import org.apache.spark.sql.common.util._
+import org.scalatest.BeforeAndAfterAll
+
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.datastore.filesystem.{CarbonFile, CarbonFileFilter}
+import org.apache.carbondata.core.datastore.impl.FileFactory
+import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
+import org.apache.carbondata.core.util.CarbonProperties
+import org.apache.carbondata.core.util.path.CarbonTablePath
+
+/**
+ * Test Class for AlterTableTestCase to verify all scenerios
+ */
+
+class MergeIndexTestCase extends QueryTest with BeforeAndAfterAll {
+
+
+ override protected def afterAll(): Unit = {
+ sql("DROP TABLE IF EXISTS nonindexmerge")
+ sql("DROP TABLE IF EXISTS indexmerge")
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "true")
+ }
+
+ test("Verify correctness of index merge sdv") {
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "false")
+ sql(s"""drop table if exists carbon_automation_nonmerge""").collect
+
+ sql(s"""create table carbon_automation_nonmerge (imei string,deviceInformationId int,MAC string,deviceColor string,device_backColor string,modelId string,marketName string,AMSize string,ROMSize string,CUPAudit string,CPIClocked string,series string,productionDate timestamp,bomCode string,internalModels string, deliveryTime string, channelsId string, channelsName string , deliveryAreaId string, deliveryCountry string, deliveryProvince string, deliveryCity string,deliveryDistrict string, deliveryStreet string, oxSingleNumber string, ActiveCheckTime string, ActiveAreaId string, ActiveCountry string, ActiveProvince string, Activecity string, ActiveDistrict string, ActiveStreet string, ActiveOperatorId string, Active_releaseId string, Active_EMUIVersion string, Active_operaSysVersion string, Active_BacVerNumber string, Active_BacFlashVer string, Active_webUIVersion string, Active_webUITypeCarrVer string,Active_webTypeDataVerNumber string, Active_operatorsVersion string, Active_phoneP
ADPartitionedVersions string, Latest_YEAR int, Latest_MONTH int, Latest_DAY int, Latest_HOUR string, Latest_areaId string, Latest_country string, Latest_province string, Latest_city string, Latest_district string, Latest_street string, Latest_releaseId string, Latest_EMUIVersion string, Latest_operaSysVersion string, Latest_BacVerNumber string, Latest_BacFlashVer string, Latest_webUIVersion string, Latest_webUITypeCarrVer string, Latest_webTypeDataVerNumber string, Latest_operatorsVersion string, Latest_phonePADPartitionedVersions string, Latest_operatorId string, gamePointDescription string,gamePointId double,contractNumber double) STORED BY 'org.apache.carbondata.format' TBLPROPERTIES ('DICTIONARY_INCLUDE'='deviceInformationId,Latest_YEAR,Latest_MONTH,Latest_DAY')""").collect
+
+ sql(s"""LOAD DATA INPATH '$resourcesPath/Data/VmaLL100' INTO TABLE carbon_automation_nonmerge OPTIONS('DELIMITER'=',','QUOTECHAR'='"','FILEHEADER'='imei,deviceInformationId,MAC,deviceColor,device_backColor,modelId,marketName,AMSize,ROMSize,CUPAudit,CPIClocked,series,productionDate,bomCode,internalModels,deliveryTime,channelsId,channelsName,deliveryAreaId,deliveryCountry,deliveryProvince,deliveryCity,deliveryDistrict,deliveryStreet,oxSingleNumber,contractNumber,ActiveCheckTime,ActiveAreaId,ActiveCountry,ActiveProvince,Activecity,ActiveDistrict,ActiveStreet,ActiveOperatorId,Active_releaseId,Active_EMUIVersion,Active_operaSysVersion,Active_BacVerNumber,Active_BacFlashVer,Active_webUIVersion,Active_webUITypeCarrVer,Active_webTypeDataVerNumber,Active_operatorsVersion,Active_phonePADPartitionedVersions,Latest_YEAR,Latest_MONTH,Latest_DAY,Latest_HOUR,Latest_areaId,Latest_country,Latest_province,Latest_city,Latest_district,Latest_street,Latest_releaseId,Latest_EMUIVersion,Latest_operaSy
sVersion,Latest_BacVerNumber,Latest_BacFlashVer,Latest_webUIVersion,Latest_webUITypeCarrVer,Latest_webTypeDataVerNumber,Latest_operatorsVersion,Latest_phonePADPartitionedVersions,Latest_operatorId,gamePointId,gamePointDescription')""").collect
+ assert(getIndexFileCount("default", "carbon_automation_nonmerge", "0") == 2)
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "true")
+ sql("DROP TABLE IF EXISTS carbon_automation_merge")
+ sql(s"""create table carbon_automation_merge (imei string,deviceInformationId int,MAC string,deviceColor string,device_backColor string,modelId string,marketName string,AMSize string,ROMSize string,CUPAudit string,CPIClocked string,series string,productionDate timestamp,bomCode string,internalModels string, deliveryTime string, channelsId string, channelsName string , deliveryAreaId string, deliveryCountry string, deliveryProvince string, deliveryCity string,deliveryDistrict string, deliveryStreet string, oxSingleNumber string, ActiveCheckTime string, ActiveAreaId string, ActiveCountry string, ActiveProvince string, Activecity string, ActiveDistrict string, ActiveStreet string, ActiveOperatorId string, Active_releaseId string, Active_EMUIVersion string, Active_operaSysVersion string, Active_BacVerNumber string, Active_BacFlashVer string, Active_webUIVersion string, Active_webUITypeCarrVer string,Active_webTypeDataVerNumber string, Active_operatorsVersion string, Active_phonePADP
artitionedVersions string, Latest_YEAR int, Latest_MONTH int, Latest_DAY int, Latest_HOUR string, Latest_areaId string, Latest_country string, Latest_province string, Latest_city string, Latest_district string, Latest_street string, Latest_releaseId string, Latest_EMUIVersion string, Latest_operaSysVersion string, Latest_BacVerNumber string, Latest_BacFlashVer string, Latest_webUIVersion string, Latest_webUITypeCarrVer string, Latest_webTypeDataVerNumber string, Latest_operatorsVersion string, Latest_phonePADPartitionedVersions string, Latest_operatorId string, gamePointDescription string,gamePointId double,contractNumber double) STORED BY 'org.apache.carbondata.format' TBLPROPERTIES ('DICTIONARY_INCLUDE'='deviceInformationId,Latest_YEAR,Latest_MONTH,Latest_DAY')""").collect
+
+ sql(s"""LOAD DATA INPATH '$resourcesPath/Data/VmaLL100' INTO TABLE carbon_automation_merge OPTIONS('DELIMITER'=',','QUOTECHAR'='"','FILEHEADER'='imei,deviceInformationId,MAC,deviceColor,device_backColor,modelId,marketName,AMSize,ROMSize,CUPAudit,CPIClocked,series,productionDate,bomCode,internalModels,deliveryTime,channelsId,channelsName,deliveryAreaId,deliveryCountry,deliveryProvince,deliveryCity,deliveryDistrict,deliveryStreet,oxSingleNumber,contractNumber,ActiveCheckTime,ActiveAreaId,ActiveCountry,ActiveProvince,Activecity,ActiveDistrict,ActiveStreet,ActiveOperatorId,Active_releaseId,Active_EMUIVersion,Active_operaSysVersion,Active_BacVerNumber,Active_BacFlashVer,Active_webUIVersion,Active_webUITypeCarrVer,Active_webTypeDataVerNumber,Active_operatorsVersion,Active_phonePADPartitionedVersions,Latest_YEAR,Latest_MONTH,Latest_DAY,Latest_HOUR,Latest_areaId,Latest_country,Latest_province,Latest_city,Latest_district,Latest_street,Latest_releaseId,Latest_EMUIVersion,Latest_operaSysVe
rsion,Latest_BacVerNumber,Latest_BacFlashVer,Latest_webUIVersion,Latest_webUITypeCarrVer,Latest_webTypeDataVerNumber,Latest_operatorsVersion,Latest_phonePADPartitionedVersions,Latest_operatorId,gamePointId,gamePointDescription')""").collect
+
+ assert(getIndexFileCount("default", "carbon_automation_merge", "0") == 0)
+ checkAnswer(sql("""Select count(*) from carbon_automation_nonmerge"""),
+ sql("""Select count(*) from carbon_automation_merge"""))
+ }
+
+ test("Verify command of index merge sdv") {
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "false")
+ sql(s"""drop table if exists carbon_automation_nonmerge""").collect
+
+ sql(s"""create table carbon_automation_nonmerge (imei string,deviceInformationId int,MAC string,deviceColor string,device_backColor string,modelId string,marketName string,AMSize string,ROMSize string,CUPAudit string,CPIClocked string,series string,productionDate timestamp,bomCode string,internalModels string, deliveryTime string, channelsId string, channelsName string , deliveryAreaId string, deliveryCountry string, deliveryProvince string, deliveryCity string,deliveryDistrict string, deliveryStreet string, oxSingleNumber string, ActiveCheckTime string, ActiveAreaId string, ActiveCountry string, ActiveProvince string, Activecity string, ActiveDistrict string, ActiveStreet string, ActiveOperatorId string, Active_releaseId string, Active_EMUIVersion string, Active_operaSysVersion string, Active_BacVerNumber string, Active_BacFlashVer string, Active_webUIVersion string, Active_webUITypeCarrVer string,Active_webTypeDataVerNumber string, Active_operatorsVersion string, Active_phoneP
ADPartitionedVersions string, Latest_YEAR int, Latest_MONTH int, Latest_DAY int, Latest_HOUR string, Latest_areaId string, Latest_country string, Latest_province string, Latest_city string, Latest_district string, Latest_street string, Latest_releaseId string, Latest_EMUIVersion string, Latest_operaSysVersion string, Latest_BacVerNumber string, Latest_BacFlashVer string, Latest_webUIVersion string, Latest_webUITypeCarrVer string, Latest_webTypeDataVerNumber string, Latest_operatorsVersion string, Latest_phonePADPartitionedVersions string, Latest_operatorId string, gamePointDescription string,gamePointId double,contractNumber double) STORED BY 'org.apache.carbondata.format' TBLPROPERTIES ('DICTIONARY_INCLUDE'='deviceInformationId,Latest_YEAR,Latest_MONTH,Latest_DAY')""").collect
+
+ sql(s"""LOAD DATA INPATH '$resourcesPath/Data/VmaLL100' INTO TABLE carbon_automation_nonmerge OPTIONS('DELIMITER'=',','QUOTECHAR'='"','FILEHEADER'='imei,deviceInformationId,MAC,deviceColor,device_backColor,modelId,marketName,AMSize,ROMSize,CUPAudit,CPIClocked,series,productionDate,bomCode,internalModels,deliveryTime,channelsId,channelsName,deliveryAreaId,deliveryCountry,deliveryProvince,deliveryCity,deliveryDistrict,deliveryStreet,oxSingleNumber,contractNumber,ActiveCheckTime,ActiveAreaId,ActiveCountry,ActiveProvince,Activecity,ActiveDistrict,ActiveStreet,ActiveOperatorId,Active_releaseId,Active_EMUIVersion,Active_operaSysVersion,Active_BacVerNumber,Active_BacFlashVer,Active_webUIVersion,Active_webUITypeCarrVer,Active_webTypeDataVerNumber,Active_operatorsVersion,Active_phonePADPartitionedVersions,Latest_YEAR,Latest_MONTH,Latest_DAY,Latest_HOUR,Latest_areaId,Latest_country,Latest_province,Latest_city,Latest_district,Latest_street,Latest_releaseId,Latest_EMUIVersion,Latest_operaSy
sVersion,Latest_BacVerNumber,Latest_BacFlashVer,Latest_webUIVersion,Latest_webUITypeCarrVer,Latest_webTypeDataVerNumber,Latest_operatorsVersion,Latest_phonePADPartitionedVersions,Latest_operatorId,gamePointId,gamePointDescription')""").collect
+ sql(s"""LOAD DATA INPATH '$resourcesPath/Data/VmaLL100' INTO TABLE carbon_automation_nonmerge OPTIONS('DELIMITER'=',','QUOTECHAR'='"','FILEHEADER'='imei,deviceInformationId,MAC,deviceColor,device_backColor,modelId,marketName,AMSize,ROMSize,CUPAudit,CPIClocked,series,productionDate,bomCode,internalModels,deliveryTime,channelsId,channelsName,deliveryAreaId,deliveryCountry,deliveryProvince,deliveryCity,deliveryDistrict,deliveryStreet,oxSingleNumber,contractNumber,ActiveCheckTime,ActiveAreaId,ActiveCountry,ActiveProvince,Activecity,ActiveDistrict,ActiveStreet,ActiveOperatorId,Active_releaseId,Active_EMUIVersion,Active_operaSysVersion,Active_BacVerNumber,Active_BacFlashVer,Active_webUIVersion,Active_webUITypeCarrVer,Active_webTypeDataVerNumber,Active_operatorsVersion,Active_phonePADPartitionedVersions,Latest_YEAR,Latest_MONTH,Latest_DAY,Latest_HOUR,Latest_areaId,Latest_country,Latest_province,Latest_city,Latest_district,Latest_street,Latest_releaseId,Latest_EMUIVersion,Latest_operaSy
sVersion,Latest_BacVerNumber,Latest_BacFlashVer,Latest_webUIVersion,Latest_webUITypeCarrVer,Latest_webTypeDataVerNumber,Latest_operatorsVersion,Latest_phonePADPartitionedVersions,Latest_operatorId,gamePointId,gamePointDescription')""").collect
+ val rows = sql("""Select count(*) from carbon_automation_nonmerge""").collect()
+ assert(getIndexFileCount("default", "carbon_automation_nonmerge", "0") == 2)
+ assert(getIndexFileCount("default", "carbon_automation_nonmerge", "1") == 2)
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "true")
+ sql("ALTER TABLE carbon_automation_nonmerge COMPACT 'SEGMENT_INDEX_COMPACTION'").collect()
+ assert(getIndexFileCount("default", "carbon_automation_nonmerge", "0") == 0)
+ assert(getIndexFileCount("default", "carbon_automation_nonmerge", "1") == 0)
+ checkAnswer(sql("""Select count(*) from carbon_automation_nonmerge"""), rows)
+ }
+
+ test("Verify index index merge with compaction sdv") {
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "false")
+ sql(s"""drop table if exists carbon_automation_nonmerge""").collect
+
+ sql(s"""create table carbon_automation_nonmerge (imei string,deviceInformationId int,MAC string,deviceColor string,device_backColor string,modelId string,marketName string,AMSize string,ROMSize string,CUPAudit string,CPIClocked string,series string,productionDate timestamp,bomCode string,internalModels string, deliveryTime string, channelsId string, channelsName string , deliveryAreaId string, deliveryCountry string, deliveryProvince string, deliveryCity string,deliveryDistrict string, deliveryStreet string, oxSingleNumber string, ActiveCheckTime string, ActiveAreaId string, ActiveCountry string, ActiveProvince string, Activecity string, ActiveDistrict string, ActiveStreet string, ActiveOperatorId string, Active_releaseId string, Active_EMUIVersion string, Active_operaSysVersion string, Active_BacVerNumber string, Active_BacFlashVer string, Active_webUIVersion string, Active_webUITypeCarrVer string,Active_webTypeDataVerNumber string, Active_operatorsVersion string, Active_phoneP
ADPartitionedVersions string, Latest_YEAR int, Latest_MONTH int, Latest_DAY int, Latest_HOUR string, Latest_areaId string, Latest_country string, Latest_province string, Latest_city string, Latest_district string, Latest_street string, Latest_releaseId string, Latest_EMUIVersion string, Latest_operaSysVersion string, Latest_BacVerNumber string, Latest_BacFlashVer string, Latest_webUIVersion string, Latest_webUITypeCarrVer string, Latest_webTypeDataVerNumber string, Latest_operatorsVersion string, Latest_phonePADPartitionedVersions string, Latest_operatorId string, gamePointDescription string,gamePointId double,contractNumber double) STORED BY 'org.apache.carbondata.format' TBLPROPERTIES ('DICTIONARY_INCLUDE'='deviceInformationId,Latest_YEAR,Latest_MONTH,Latest_DAY')""").collect
+
+ sql(s"""LOAD DATA INPATH '$resourcesPath/Data/VmaLL100' INTO TABLE carbon_automation_nonmerge OPTIONS('DELIMITER'=',','QUOTECHAR'='"','FILEHEADER'='imei,deviceInformationId,MAC,deviceColor,device_backColor,modelId,marketName,AMSize,ROMSize,CUPAudit,CPIClocked,series,productionDate,bomCode,internalModels,deliveryTime,channelsId,channelsName,deliveryAreaId,deliveryCountry,deliveryProvince,deliveryCity,deliveryDistrict,deliveryStreet,oxSingleNumber,contractNumber,ActiveCheckTime,ActiveAreaId,ActiveCountry,ActiveProvince,Activecity,ActiveDistrict,ActiveStreet,ActiveOperatorId,Active_releaseId,Active_EMUIVersion,Active_operaSysVersion,Active_BacVerNumber,Active_BacFlashVer,Active_webUIVersion,Active_webUITypeCarrVer,Active_webTypeDataVerNumber,Active_operatorsVersion,Active_phonePADPartitionedVersions,Latest_YEAR,Latest_MONTH,Latest_DAY,Latest_HOUR,Latest_areaId,Latest_country,Latest_province,Latest_city,Latest_district,Latest_street,Latest_releaseId,Latest_EMUIVersion,Latest_operaSy
sVersion,Latest_BacVerNumber,Latest_BacFlashVer,Latest_webUIVersion,Latest_webUITypeCarrVer,Latest_webTypeDataVerNumber,Latest_operatorsVersion,Latest_phonePADPartitionedVersions,Latest_operatorId,gamePointId,gamePointDescription')""").collect
+ sql(s"""LOAD DATA INPATH '$resourcesPath/Data/VmaLL100' INTO TABLE carbon_automation_nonmerge OPTIONS('DELIMITER'=',','QUOTECHAR'='"','FILEHEADER'='imei,deviceInformationId,MAC,deviceColor,device_backColor,modelId,marketName,AMSize,ROMSize,CUPAudit,CPIClocked,series,productionDate,bomCode,internalModels,deliveryTime,channelsId,channelsName,deliveryAreaId,deliveryCountry,deliveryProvince,deliveryCity,deliveryDistrict,deliveryStreet,oxSingleNumber,contractNumber,ActiveCheckTime,ActiveAreaId,ActiveCountry,ActiveProvince,Activecity,ActiveDistrict,ActiveStreet,ActiveOperatorId,Active_releaseId,Active_EMUIVersion,Active_operaSysVersion,Active_BacVerNumber,Active_BacFlashVer,Active_webUIVersion,Active_webUITypeCarrVer,Active_webTypeDataVerNumber,Active_operatorsVersion,Active_phonePADPartitionedVersions,Latest_YEAR,Latest_MONTH,Latest_DAY,Latest_HOUR,Latest_areaId,Latest_country,Latest_province,Latest_city,Latest_district,Latest_street,Latest_releaseId,Latest_EMUIVersion,Latest_operaSy
sVersion,Latest_BacVerNumber,Latest_BacFlashVer,Latest_webUIVersion,Latest_webUITypeCarrVer,Latest_webTypeDataVerNumber,Latest_operatorsVersion,Latest_phonePADPartitionedVersions,Latest_operatorId,gamePointId,gamePointDescription')""").collect
+ sql(s"""LOAD DATA INPATH '$resourcesPath/Data/VmaLL100' INTO TABLE carbon_automation_nonmerge OPTIONS('DELIMITER'=',','QUOTECHAR'='"','FILEHEADER'='imei,deviceInformationId,MAC,deviceColor,device_backColor,modelId,marketName,AMSize,ROMSize,CUPAudit,CPIClocked,series,productionDate,bomCode,internalModels,deliveryTime,channelsId,channelsName,deliveryAreaId,deliveryCountry,deliveryProvince,deliveryCity,deliveryDistrict,deliveryStreet,oxSingleNumber,contractNumber,ActiveCheckTime,ActiveAreaId,ActiveCountry,ActiveProvince,Activecity,ActiveDistrict,ActiveStreet,ActiveOperatorId,Active_releaseId,Active_EMUIVersion,Active_operaSysVersion,Active_BacVerNumber,Active_BacFlashVer,Active_webUIVersion,Active_webUITypeCarrVer,Active_webTypeDataVerNumber,Active_operatorsVersion,Active_phonePADPartitionedVersions,Latest_YEAR,Latest_MONTH,Latest_DAY,Latest_HOUR,Latest_areaId,Latest_country,Latest_province,Latest_city,Latest_district,Latest_street,Latest_releaseId,Latest_EMUIVersion,Latest_operaSy
sVersion,Latest_BacVerNumber,Latest_BacFlashVer,Latest_webUIVersion,Latest_webUITypeCarrVer,Latest_webTypeDataVerNumber,Latest_operatorsVersion,Latest_phonePADPartitionedVersions,Latest_operatorId,gamePointId,gamePointDescription')""").collect
+ val rows = sql("""Select count(*) from carbon_automation_nonmerge""").collect()
+ assert(getIndexFileCount("default", "carbon_automation_nonmerge", "0") == 2)
+ assert(getIndexFileCount("default", "carbon_automation_nonmerge", "1") == 2)
+ assert(getIndexFileCount("default", "carbon_automation_nonmerge", "1") == 2)
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "true")
+ sql("ALTER TABLE carbon_automation_nonmerge COMPACT 'minor'").collect()
+ assert(getIndexFileCount("default", "carbon_automation_nonmerge", "0.1") == 0)
+ checkAnswer(sql("""Select count(*) from carbon_automation_nonmerge"""), rows)
+ }
+
+ private def getIndexFileCount(dbName: String, tableName: String, segment: String): Int = {
+ val identifier = AbsoluteTableIdentifier.from(CarbonEnv.getInstance(sqlContext.sparkSession).storePath, dbName, tableName)
+ val path = CarbonTablePath
+ .getSegmentPath(identifier.getTablePath, segment)
+ val carbonFiles = FileFactory.getCarbonFile(path).listFiles(new CarbonFileFilter {
+ override def accept(file: CarbonFile): Boolean = file.getName.endsWith(CarbonTablePath
+ .INDEX_FILE_EXT)
+ })
+ if (carbonFiles != null) {
+ carbonFiles.length
+ } else {
+ 0
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/suite/SDVSuites.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/suite/SDVSuites.scala b/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/suite/SDVSuites.scala
index 049a460..2e00b0c 100644
--- a/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/suite/SDVSuites.scala
+++ b/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/suite/SDVSuites.scala
@@ -78,7 +78,8 @@ class SDVSuites1 extends Suites with BeforeAndAfterAll {
new PartitionTestCase ::
new QueriesBasicTestCase ::
new BatchSortLoad3TestCase ::
- new GlobalSortTestCase :: Nil
+ new GlobalSortTestCase ::
+ new MergeIndexTestCase :: Nil
override val nestedSuites = suites.toIndexedSeq
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CarbonIndexFileMergeTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CarbonIndexFileMergeTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CarbonIndexFileMergeTestCase.scala
new file mode 100644
index 0000000..110557c
--- /dev/null
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CarbonIndexFileMergeTestCase.scala
@@ -0,0 +1,143 @@
+/*
+ * 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.spark.testsuite.datacompaction
+
+import org.apache.spark.sql.test.util.QueryTest
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.datastore.filesystem.{CarbonFile, CarbonFileFilter}
+import org.apache.carbondata.core.datastore.impl.FileFactory
+import org.apache.carbondata.core.metadata.CarbonMetadata
+import org.apache.carbondata.core.util.CarbonProperties
+import org.apache.carbondata.core.util.path.CarbonTablePath
+
+class CarbonIndexFileMergeTestCase
+ extends QueryTest with BeforeAndAfterEach with BeforeAndAfterAll {
+ val file2 = resourcesPath + "/compaction/fil2.csv"
+
+ override protected def beforeAll(): Unit = {
+ val n = 150000
+ CompactionSupportGlobalSortBigFileTest.createFile(file2, n * 4, n)
+ }
+
+ override protected def afterAll(): Unit = {
+ CompactionSupportGlobalSortBigFileTest.deleteFile(file2)
+ sql("DROP TABLE IF EXISTS nonindexmerge")
+ sql("DROP TABLE IF EXISTS indexmerge")
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "true")
+ }
+
+ test("Verify correctness of index merge") {
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "false")
+ sql("DROP TABLE IF EXISTS nonindexmerge")
+ sql(
+ """
+ | CREATE TABLE nonindexmerge(id INT, name STRING, city STRING, age INT)
+ | STORED BY 'org.apache.carbondata.format'
+ | TBLPROPERTIES('SORT_COLUMNS'='city,name', 'SORT_SCOPE'='GLOBAL_SORT')
+ """.stripMargin)
+ sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE nonindexmerge OPTIONS('header'='false', " +
+ s"'GLOBAL_SORT_PARTITIONS'='100')")
+ assert(getIndexFileCount("default_nonindexmerge", "0") == 100)
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "true")
+ sql("DROP TABLE IF EXISTS indexmerge")
+ sql(
+ """
+ | CREATE TABLE indexmerge(id INT, name STRING, city STRING, age INT)
+ | STORED BY 'org.apache.carbondata.format'
+ | TBLPROPERTIES('SORT_COLUMNS'='city,name', 'SORT_SCOPE'='GLOBAL_SORT')
+ """.stripMargin)
+ sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE indexmerge OPTIONS('header'='false', " +
+ s"'GLOBAL_SORT_PARTITIONS'='100')")
+ assert(getIndexFileCount("default_indexmerge", "0") == 0)
+ checkAnswer(sql("""Select count(*) from nonindexmerge"""),
+ sql("""Select count(*) from indexmerge"""))
+ }
+
+ test("Verify command of index merge") {
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "false")
+ sql("DROP TABLE IF EXISTS nonindexmerge")
+ sql(
+ """
+ | CREATE TABLE nonindexmerge(id INT, name STRING, city STRING, age INT)
+ | STORED BY 'org.apache.carbondata.format'
+ | TBLPROPERTIES('SORT_COLUMNS'='city,name', 'SORT_SCOPE'='GLOBAL_SORT')
+ """.stripMargin)
+ sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE nonindexmerge OPTIONS('header'='false', " +
+ s"'GLOBAL_SORT_PARTITIONS'='100')")
+ sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE nonindexmerge OPTIONS('header'='false', " +
+ s"'GLOBAL_SORT_PARTITIONS'='100')")
+ val rows = sql("""Select count(*) from nonindexmerge""").collect()
+ assert(getIndexFileCount("default_nonindexmerge", "0") == 100)
+ assert(getIndexFileCount("default_nonindexmerge", "1") == 100)
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "true")
+ sql("ALTER TABLE nonindexmerge COMPACT 'SEGMENT_INDEX_COMPACTION'").collect()
+ assert(getIndexFileCount("default_nonindexmerge", "0") == 0)
+ assert(getIndexFileCount("default_nonindexmerge", "1") == 0)
+ checkAnswer(sql("""Select count(*) from nonindexmerge"""), rows)
+ }
+
+ test("Verify index index merge with compaction") {
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "false")
+ sql("DROP TABLE IF EXISTS nonindexmerge")
+ sql(
+ """
+ | CREATE TABLE nonindexmerge(id INT, name STRING, city STRING, age INT)
+ | STORED BY 'org.apache.carbondata.format'
+ | TBLPROPERTIES('SORT_COLUMNS'='city,name', 'SORT_SCOPE'='GLOBAL_SORT')
+ """.stripMargin)
+ sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE nonindexmerge OPTIONS('header'='false', " +
+ s"'GLOBAL_SORT_PARTITIONS'='100')")
+ sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE nonindexmerge OPTIONS('header'='false', " +
+ s"'GLOBAL_SORT_PARTITIONS'='100')")
+ sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE nonindexmerge OPTIONS('header'='false', " +
+ s"'GLOBAL_SORT_PARTITIONS'='100')")
+ val rows = sql("""Select count(*) from nonindexmerge""").collect()
+ assert(getIndexFileCount("default_nonindexmerge", "0") == 100)
+ assert(getIndexFileCount("default_nonindexmerge", "1") == 100)
+ assert(getIndexFileCount("default_nonindexmerge", "1") == 100)
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "true")
+ sql("ALTER TABLE nonindexmerge COMPACT 'minor'").collect()
+ assert(getIndexFileCount("default_nonindexmerge", "0.1") == 0)
+ checkAnswer(sql("""Select count(*) from nonindexmerge"""), rows)
+ }
+
+ private def getIndexFileCount(tableName: String, segment: String): Int = {
+ val table = CarbonMetadata.getInstance().getCarbonTable(tableName)
+ val path = CarbonTablePath
+ .getSegmentPath(table.getAbsoluteTableIdentifier.getTablePath, segment)
+ val carbonFiles = FileFactory.getCarbonFile(path).listFiles(new CarbonFileFilter {
+ override def accept(file: CarbonFile): Boolean = file.getName.endsWith(CarbonTablePath
+ .INDEX_FILE_EXT)
+ })
+ if (carbonFiles != null) {
+ carbonFiles.length
+ } else {
+ 0
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CompactionSupportGlobalSortFunctionTest.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CompactionSupportGlobalSortFunctionTest.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CompactionSupportGlobalSortFunctionTest.scala
index 6f8648d..c7b39ad 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CompactionSupportGlobalSortFunctionTest.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CompactionSupportGlobalSortFunctionTest.scala
@@ -24,6 +24,7 @@ import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.indexstore.blockletindex.SegmentIndexFileStore
import org.apache.carbondata.core.util.CarbonProperties
class CompactionSupportGlobalSortFunctionTest extends QueryTest with BeforeAndAfterEach with BeforeAndAfterAll {
@@ -527,9 +528,6 @@ class CompactionSupportGlobalSortFunctionTest extends QueryTest with BeforeAndAf
private def getIndexFileCount(tableName: String, segmentNo: String = "0"): Int = {
val store = storeLocation + "/default/" + tableName + "/Fact/Part0/Segment_" + segmentNo
- val list = new File(store).list(new FilenameFilter {
- override def accept(dir: File, name: String) = name.endsWith(".carbonindex")
- })
- list.size
+ new SegmentIndexFileStore().getIndexFilesFromSegment(store).size()
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CompactionSupportGlobalSortParameterTest.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CompactionSupportGlobalSortParameterTest.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CompactionSupportGlobalSortParameterTest.scala
index 1511b51..78c835a 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CompactionSupportGlobalSortParameterTest.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CompactionSupportGlobalSortParameterTest.scala
@@ -25,6 +25,9 @@ import org.apache.spark.sql.Row
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+import org.apache.carbondata.core.indexstore.blockletindex.SegmentIndexFileStore
+import org.apache.carbondata.core.util.path.CarbonTablePath
+
class CompactionSupportGlobalSortParameterTest extends QueryTest with BeforeAndAfterEach with BeforeAndAfterAll {
val filePath: String = s"$resourcesPath/globalsort"
val file1: String = resourcesPath + "/globalsort/sample1.csv"
@@ -526,9 +529,6 @@ class CompactionSupportGlobalSortParameterTest extends QueryTest with BeforeAndA
private def getIndexFileCount(tableName: String, segmentNo: String = "0"): Int = {
val store = storeLocation + "/default/" + tableName + "/Fact/Part0/Segment_" + segmentNo
- val list = new File(store).list(new FilenameFilter {
- override def accept(dir: File, name: String) = name.endsWith(".carbonindex")
- })
- list.size
+ new SegmentIndexFileStore().getIndexFilesFromSegment(store).size()
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestBatchSortDataLoad.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestBatchSortDataLoad.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestBatchSortDataLoad.scala
index 4c6c56d..cb3da40 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestBatchSortDataLoad.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestBatchSortDataLoad.scala
@@ -21,10 +21,13 @@ import java.io.{BufferedWriter, File, FileWriter, FilenameFilter}
import org.apache.spark.sql.Row
import org.scalatest.BeforeAndAfterAll
+
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.spark.sql.test.util.QueryTest
+import org.apache.carbondata.core.indexstore.blockletindex.SegmentIndexFileStore
+
class TestBatchSortDataLoad extends QueryTest with BeforeAndAfterAll {
var filePath: String = _
@@ -184,10 +187,7 @@ class TestBatchSortDataLoad extends QueryTest with BeforeAndAfterAll {
def getIndexfileCount(tableName: String, segmentNo: String = "0"): Int = {
val store = storeLocation +"/default/"+ tableName + "/Fact/Part0/Segment_"+segmentNo
- val list = new File(store).list(new FilenameFilter {
- override def accept(dir: File, name: String) = name.endsWith(".carbonindex")
- })
- list.size
+ new SegmentIndexFileStore().getIndexFilesFromSegment(store).size()
}
override def afterAll {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestGlobalSortDataLoad.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestGlobalSortDataLoad.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestGlobalSortDataLoad.scala
index 35a56ea..5d0c055 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestGlobalSortDataLoad.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestGlobalSortDataLoad.scala
@@ -27,6 +27,8 @@ import org.apache.spark.sql.test.TestQueryExecutor.projectPath
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+import org.apache.carbondata.core.indexstore.blockletindex.SegmentIndexFileStore
+
class TestGlobalSortDataLoad extends QueryTest with BeforeAndAfterEach with BeforeAndAfterAll {
var filePath: String = s"$resourcesPath/globalsort"
@@ -325,9 +327,6 @@ class TestGlobalSortDataLoad extends QueryTest with BeforeAndAfterEach with Befo
private def getIndexFileCount(tableName: String, segmentNo: String = "0"): Int = {
val store = storeLocation + "/default/" + tableName + "/Fact/Part0/Segment_" + segmentNo
- val list = new File(store).list(new FilenameFilter {
- override def accept(dir: File, name: String) = name.endsWith(".carbonindex")
- })
- list.size
+ new SegmentIndexFileStore().getIndexFilesFromSegment(store).size()
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0586146a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala
index ba6ad31..ff900ce 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala
@@ -45,7 +45,7 @@ class C2DataMapFactory() extends DataMapFactory {
override def clear(): Unit = {}
- override def getDataMap(distributable: DataMapDistributable): DataMap = ???
+ override def getDataMaps(distributable: DataMapDistributable): java.util.List[DataMap] = ???
override def getDataMaps(segmentId: String): util.List[DataMap] = ???