[CARBONDATA-2701] Refactor code to store minimal required info in Block and Blocklet Cache
1. Refactored code to keep only minimal information in block and blocklet cache. 2. Introduced segment properties holder at JVM level to hold the segment properties. As it is heavy object, new segment properties object will be created only when schema or cardinality is changed for a table. This closes #2454 Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/f4a58c54 Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/f4a58c54 Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/f4a58c54 Branch: refs/heads/master Commit: f4a58c54599929ca8a1e1e9ad647e54e0cfc6093 Parents: 5c483f3 Author: manishgupta88 <[email protected]> Authored: Wed Jul 4 21:00:54 2018 +0530 Committer: kunal642 <[email protected]> Committed: Mon Jul 9 17:32:26 2018 +0530 ---------------------------------------------------------------------- .../block/SegmentPropertiesAndSchemaHolder.java | 342 +++++++++++++++++++ .../core/indexstore/AbstractMemoryDMStore.java | 18 +- .../carbondata/core/indexstore/Blocklet.java | 22 +- .../indexstore/BlockletDataMapIndexStore.java | 20 +- .../core/indexstore/BlockletDetailInfo.java | 42 ++- .../core/indexstore/ExtendedBlocklet.java | 5 + .../core/indexstore/SafeMemoryDMStore.java | 20 +- .../core/indexstore/UnsafeMemoryDMStore.java | 7 +- .../indexstore/blockletindex/BlockDataMap.java | 311 +++++++++-------- .../blockletindex/BlockletDataMap.java | 69 ++-- .../blockletindex/BlockletDataMapFactory.java | 5 +- .../blockletindex/BlockletDataMapModel.java | 19 +- .../BlockletDataMapRowIndexes.java | 11 +- .../core/indexstore/row/DataMapRow.java | 11 +- .../core/indexstore/schema/SchemaGenerator.java | 19 +- .../executor/impl/AbstractQueryExecutor.java | 2 + .../core/util/BlockletDataMapUtil.java | 46 +++ .../datasources/SparkCarbonFileFormat.scala | 1 - .../spark/sql/hive/CarbonFileMetastore.scala | 3 + .../spark/sql/hive/CarbonHiveMetaStore.scala | 2 + 20 files changed, 726 insertions(+), 249 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/core/src/main/java/org/apache/carbondata/core/datastore/block/SegmentPropertiesAndSchemaHolder.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/block/SegmentPropertiesAndSchemaHolder.java b/core/src/main/java/org/apache/carbondata/core/datastore/block/SegmentPropertiesAndSchemaHolder.java new file mode 100644 index 0000000..da76bc6 --- /dev/null +++ b/core/src/main/java/org/apache/carbondata/core/datastore/block/SegmentPropertiesAndSchemaHolder.java @@ -0,0 +1,342 @@ +/* + * 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.datastore.block; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.carbondata.common.logging.LogService; +import org.apache.carbondata.common.logging.LogServiceFactory; +import org.apache.carbondata.core.constants.CarbonCommonConstants; +import org.apache.carbondata.core.indexstore.schema.CarbonRowSchema; +import org.apache.carbondata.core.indexstore.schema.SchemaGenerator; +import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier; +import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema; + +/** + * Singleton class which will help in creating the segment properties + */ +public class SegmentPropertiesAndSchemaHolder { + + /** + * Logger + */ + private static final LogService LOGGER = + LogServiceFactory.getLogService(SegmentPropertiesAndSchemaHolder.class.getName()); + /** + * SegmentPropertiesAndSchemaHolder instance + */ + private static final SegmentPropertiesAndSchemaHolder INSTANCE = + new SegmentPropertiesAndSchemaHolder(); + /** + * object level lock + */ + private static final Object lock = new Object(); + /** + * counter for maintaining the index of segmentProperties + */ + private static final AtomicInteger segmentPropertiesIndexCounter = new AtomicInteger(0); + /** + * holds segmentPropertiesWrapper to segment ID and segmentProperties Index wrapper mapping. + * Will be used while invaliding a segment and drop table + */ + private Map<SegmentPropertiesWrapper, SegmentIdAndSegmentPropertiesIndexWrapper> + segmentPropWrapperToSegmentSetMap = new ConcurrentHashMap<>(); + /** + * reverse mapping for segmentProperties index to segmentPropertiesWrapper + */ + private static Map<Integer, SegmentPropertiesWrapper> indexToSegmentPropertiesWrapperMapping = + new ConcurrentHashMap<>(); + /** + * Map to be used for table level locking while populating segmentProperties + */ + private Map<String, Object> absoluteTableIdentifierByteMap = new ConcurrentHashMap<>(); + + /** + * private constructor for singleton instance + */ + private SegmentPropertiesAndSchemaHolder() { + + } + + public static SegmentPropertiesAndSchemaHolder getInstance() { + return INSTANCE; + } + + /** + * Method to add the segment properties and avoid construction of new segment properties until + * the schema is not modified + * + * @param tableIdentifier + * @param columnsInTable + * @param columnCardinality + * @param segmentId + */ + public int addSegmentProperties(AbsoluteTableIdentifier tableIdentifier, + List<ColumnSchema> columnsInTable, int[] columnCardinality, String segmentId) { + SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper segmentPropertiesWrapper = + new SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper(tableIdentifier, + columnsInTable, columnCardinality); + SegmentIdAndSegmentPropertiesIndexWrapper segmentIdSetAndIndexWrapper = + this.segmentPropWrapperToSegmentSetMap.get(segmentPropertiesWrapper); + if (null == segmentIdSetAndIndexWrapper) { + synchronized (getOrCreateTableLock(tableIdentifier)) { + segmentIdSetAndIndexWrapper = + this.segmentPropWrapperToSegmentSetMap.get(segmentPropertiesWrapper); + if (null == segmentIdSetAndIndexWrapper) { + // create new segmentProperties + segmentPropertiesWrapper.initSegmentProperties(); + int segmentPropertiesIndex = segmentPropertiesIndexCounter.incrementAndGet(); + indexToSegmentPropertiesWrapperMapping + .put(segmentPropertiesIndex, segmentPropertiesWrapper); + LOGGER.info("Constructing new SegmentProperties for table: " + tableIdentifier + .getCarbonTableIdentifier().getTableUniqueName() + + ". Current size of segment properties" + " holder list is: " + + indexToSegmentPropertiesWrapperMapping.size()); + // populate the SegmentIdAndSegmentPropertiesIndexWrapper to maintain the set of segments + // having same SegmentPropertiesWrapper instance this will used to decide during the + // tblSegmentsProperties map clean-up for the invalid segments + segmentIdSetAndIndexWrapper = + new SegmentIdAndSegmentPropertiesIndexWrapper(segmentId, segmentPropertiesIndex); + segmentPropWrapperToSegmentSetMap + .put(segmentPropertiesWrapper, segmentIdSetAndIndexWrapper); + } + } + } else { + synchronized (getOrCreateTableLock(tableIdentifier)) { + segmentIdSetAndIndexWrapper.addSegmentId(segmentId); + } + } + return segmentIdSetAndIndexWrapper.getSegmentPropertiesIndex(); + } + + /** + * Method to create table Level lock + * + * @param absoluteTableIdentifier + * @return + */ + private Object getOrCreateTableLock(AbsoluteTableIdentifier absoluteTableIdentifier) { + Object tableLock = absoluteTableIdentifierByteMap + .get(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName()); + if (null == tableLock) { + synchronized (lock) { + tableLock = absoluteTableIdentifierByteMap + .get(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName()); + if (null == tableLock) { + tableLock = new Object(); + absoluteTableIdentifierByteMap + .put(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName(), + tableLock); + } + } + } + return tableLock; + } + + /** + * Method to get the segment properties from given index + * + * @param segmentPropertiesIndex + * @return + */ + public SegmentProperties getSegmentProperties(int segmentPropertiesIndex) { + SegmentPropertiesWrapper segmentPropertiesWrapper = + getSegmentPropertiesWrapper(segmentPropertiesIndex); + if (null != segmentPropertiesWrapper) { + return segmentPropertiesWrapper.getSegmentProperties(); + } + return null; + } + + /** + * Method to get the segment properties from given index + * + * @param segmentPropertiesWrapperIndex + * @return + */ + public SegmentPropertiesWrapper getSegmentPropertiesWrapper(int segmentPropertiesWrapperIndex) { + return indexToSegmentPropertiesWrapperMapping.get(segmentPropertiesWrapperIndex); + } + + /** + * This method will remove the segment properties from the map on drop table + * + * @param absoluteTableIdentifier + */ + public void invalidate(AbsoluteTableIdentifier absoluteTableIdentifier) { + List<SegmentPropertiesWrapper> segmentPropertiesWrappersToBeRemoved = new ArrayList<>(); + // remove segmentProperties wrapper entries from the copyOnWriteArrayList + for (Map.Entry<SegmentPropertiesWrapper, SegmentIdAndSegmentPropertiesIndexWrapper> entry : + segmentPropWrapperToSegmentSetMap.entrySet()) { + SegmentPropertiesWrapper segmentPropertiesWrapper = entry.getKey(); + if (segmentPropertiesWrapper.getTableIdentifier().getCarbonTableIdentifier() + .getTableUniqueName() + .equals(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName())) { + SegmentIdAndSegmentPropertiesIndexWrapper value = entry.getValue(); + // remove from the reverse mapping map + indexToSegmentPropertiesWrapperMapping.remove(value.getSegmentPropertiesIndex()); + segmentPropertiesWrappersToBeRemoved.add(segmentPropertiesWrapper); + } + } + // remove all the segmentPropertiesWrapper entries from map + for (SegmentPropertiesWrapper segmentPropertiesWrapper : segmentPropertiesWrappersToBeRemoved) { + segmentPropWrapperToSegmentSetMap.remove(segmentPropertiesWrapper); + } + // remove the table lock + absoluteTableIdentifierByteMap + .remove(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName()); + } + + /** + * Method to remove the given segment ID + * + * @param segmentId + * @param segmentPropertiesIndex + */ + public void invalidate(String segmentId, int segmentPropertiesIndex) { + SegmentPropertiesWrapper segmentPropertiesWrapper = + indexToSegmentPropertiesWrapperMapping.get(segmentPropertiesIndex); + if (null != segmentPropertiesWrapper) { + SegmentIdAndSegmentPropertiesIndexWrapper segmentIdAndSegmentPropertiesIndexWrapper = + segmentPropWrapperToSegmentSetMap.get(segmentPropertiesWrapper); + synchronized (segmentPropertiesWrapper.getTableIdentifier().getCarbonTableIdentifier() + .getTableUniqueName()) { + segmentIdAndSegmentPropertiesIndexWrapper.removeSegmentId(segmentId); + } + // if after removal of given SegmentId, the segmentIdSet becomes empty that means this + // segmentPropertiesWrapper is not getting used at all. In that case this object can be + // removed from all the holders + if (segmentIdAndSegmentPropertiesIndexWrapper.segmentIdSet.isEmpty()) { + indexToSegmentPropertiesWrapperMapping.remove(segmentPropertiesIndex); + segmentPropWrapperToSegmentSetMap.remove(segmentPropertiesWrapper); + } + } + } + + /** + * This class wraps tableIdentifier, columnsInTable and columnCardinality as a key to determine + * whether the SegmentProperties object can be reused. + */ + public static class SegmentPropertiesWrapper { + + private AbsoluteTableIdentifier tableIdentifier; + private List<ColumnSchema> columnsInTable; + private int[] columnCardinality; + private SegmentProperties segmentProperties; + private CarbonRowSchema[] taskSummarySchema; + + public SegmentPropertiesWrapper(AbsoluteTableIdentifier tableIdentifier, + List<ColumnSchema> columnsInTable, int[] columnCardinality) { + this.tableIdentifier = tableIdentifier; + this.columnsInTable = columnsInTable; + this.columnCardinality = columnCardinality; + } + + public void initSegmentProperties() { + segmentProperties = new SegmentProperties(columnsInTable, columnCardinality); + } + + @Override public boolean equals(Object obj) { + if (!(obj instanceof SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper)) { + return false; + } + SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper other = + (SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper) obj; + return tableIdentifier.equals(other.tableIdentifier) && columnsInTable + .equals(other.columnsInTable) && Arrays + .equals(columnCardinality, other.columnCardinality); + } + + @Override public int hashCode() { + return tableIdentifier.hashCode() + columnsInTable.hashCode() + Arrays + .hashCode(columnCardinality); + } + + public AbsoluteTableIdentifier getTableIdentifier() { + return tableIdentifier; + } + + public SegmentProperties getSegmentProperties() { + return segmentProperties; + } + + public List<ColumnSchema> getColumnsInTable() { + return columnsInTable; + } + + public int[] getColumnCardinality() { + return columnCardinality; + } + + public CarbonRowSchema[] getBlockSchema() { + return SchemaGenerator.createBlockSchema(segmentProperties); + } + + public CarbonRowSchema[] getBlocketSchema() { + return SchemaGenerator.createBlockletSchema(segmentProperties); + } + + public CarbonRowSchema[] getTaskSummarySchema() { + return taskSummarySchema; + } + + public void setTaskSummarySchema(CarbonRowSchema[] taskSummarySchema) { + this.taskSummarySchema = taskSummarySchema; + } + } + + /** + * holder for segmentId and segmentPropertiesIndex + */ + public static class SegmentIdAndSegmentPropertiesIndexWrapper { + + /** + * set holding all unique segment Id's using the same segmentProperties + */ + private Set<String> segmentIdSet; + /** + * index which maps to segmentPropertiesWrpper Index from where segmentProperties + * can be retrieved + */ + private int segmentPropertiesIndex; + + public SegmentIdAndSegmentPropertiesIndexWrapper(String segmentId, int segmentPropertiesIndex) { + segmentIdSet = new HashSet<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); + addSegmentId(segmentId); + this.segmentPropertiesIndex = segmentPropertiesIndex; + } + + public void addSegmentId(String segmentId) { + segmentIdSet.add(segmentId); + } + + public void removeSegmentId(String segmentId) { + segmentIdSet.remove(segmentId); + } + + public int getSegmentPropertiesIndex() { + return segmentPropertiesIndex; + } + } +} http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/core/src/main/java/org/apache/carbondata/core/indexstore/AbstractMemoryDMStore.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/AbstractMemoryDMStore.java b/core/src/main/java/org/apache/carbondata/core/indexstore/AbstractMemoryDMStore.java index e6bc691..5880943 100644 --- a/core/src/main/java/org/apache/carbondata/core/indexstore/AbstractMemoryDMStore.java +++ b/core/src/main/java/org/apache/carbondata/core/indexstore/AbstractMemoryDMStore.java @@ -31,33 +31,25 @@ public abstract class AbstractMemoryDMStore implements Serializable { protected boolean isMemoryFreed; - protected CarbonRowSchema[] schema; - protected final long taskId = ThreadLocalTaskInfo.getCarbonTaskInfo().getTaskId(); - public AbstractMemoryDMStore(CarbonRowSchema[] schema) { - this.schema = schema; - } - - public abstract void addIndexRow(DataMapRow indexRow) throws MemoryException; + public abstract void addIndexRow(CarbonRowSchema[] schema, DataMapRow indexRow) + throws MemoryException; - public abstract DataMapRow getDataMapRow(int index); + public abstract DataMapRow getDataMapRow(CarbonRowSchema[] schema, int index); public abstract void freeMemory(); public abstract int getMemoryUsed(); - public CarbonRowSchema[] getSchema() { - return schema; - } - public abstract int getRowCount(); public void finishWriting() throws MemoryException { // do nothing in default implementation } - public UnsafeMemoryDMStore convertToUnsafeDMStore() throws MemoryException { + public UnsafeMemoryDMStore convertToUnsafeDMStore(CarbonRowSchema[] schema) + throws MemoryException { throw new UnsupportedOperationException("Operation not allowed"); } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/core/src/main/java/org/apache/carbondata/core/indexstore/Blocklet.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/Blocklet.java b/core/src/main/java/org/apache/carbondata/core/indexstore/Blocklet.java index 777a980..c6e1681 100644 --- a/core/src/main/java/org/apache/carbondata/core/indexstore/Blocklet.java +++ b/core/src/main/java/org/apache/carbondata/core/indexstore/Blocklet.java @@ -34,11 +34,25 @@ public class Blocklet implements Writable,Serializable { /** id to identify the blocklet inside the block (it is a sequential number) */ private String blockletId; + /** + * flag to specify whether to consider blocklet Id in equals and hashcode comparison. This is + * because when CACHE_LEVEL='BLOCK' which is default value, the blocklet ID returned by + * BlockDataMap pruning will always be -1 and other datamaps will give the the correct blocklet + * ID. Therefore if we compare -1 with correct blocklet ID the comparison will become wrong and + * always false will be returned resulting in incorrect result. Default value for flag is true. + */ + private boolean compareBlockletIdForObjectMatching = true; + public Blocklet(String filePath, String blockletId) { this.filePath = filePath; this.blockletId = blockletId; } + public Blocklet(String filePath, String blockletId, boolean compareBlockletIdForObjectMatching) { + this(filePath, blockletId); + this.compareBlockletIdForObjectMatching = compareBlockletIdForObjectMatching; + } + // For serialization purpose public Blocklet() { } @@ -70,6 +84,9 @@ public class Blocklet implements Writable,Serializable { if (filePath != null ? !filePath.equals(blocklet.filePath) : blocklet.filePath != null) { return false; } + if (!compareBlockletIdForObjectMatching) { + return true; + } return blockletId != null ? blockletId.equals(blocklet.blockletId) : blocklet.blockletId == null; @@ -77,7 +94,10 @@ public class Blocklet implements Writable,Serializable { @Override public int hashCode() { int result = filePath != null ? filePath.hashCode() : 0; - result = 31 * result + (blockletId != null ? blockletId.hashCode() : 0); + result = 31 * result; + if (compareBlockletIdForObjectMatching) { + result += blockletId != null ? blockletId.hashCode() : 0; + } return result; } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/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 5e35d96..8b5eb06 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 @@ -30,6 +30,7 @@ import org.apache.carbondata.core.cache.Cache; import org.apache.carbondata.core.cache.CarbonLRUCache; import org.apache.carbondata.core.constants.CarbonCommonConstants; import org.apache.carbondata.core.datamap.dev.DataMap; +import org.apache.carbondata.core.datastore.block.SegmentPropertiesAndSchemaHolder; import org.apache.carbondata.core.indexstore.blockletindex.BlockDataMap; import org.apache.carbondata.core.indexstore.blockletindex.BlockletDataMapFactory; import org.apache.carbondata.core.indexstore.blockletindex.BlockletDataMapModel; @@ -184,6 +185,23 @@ public class BlockletDataMapIndexStore */ @Override public void invalidate( TableBlockIndexUniqueIdentifierWrapper tableSegmentUniqueIdentifierWrapper) { + BlockletDataMapIndexWrapper blockletDataMapIndexWrapper = + getIfPresent(tableSegmentUniqueIdentifierWrapper); + if (null != blockletDataMapIndexWrapper) { + // clear the segmentProperties cache + List<BlockDataMap> dataMaps = blockletDataMapIndexWrapper.getDataMaps(); + if (null != dataMaps) { + String segmentId = + tableSegmentUniqueIdentifierWrapper.getTableBlockIndexUniqueIdentifier().getSegmentId(); + for (BlockDataMap dataMap : dataMaps) { + // as segmentId will be same for all the dataMaps and segmentProperties cache is + // maintained at segment level so it need to be called only once for clearing + SegmentPropertiesAndSchemaHolder.getInstance() + .invalidate(segmentId, dataMap.getSegmentPropertiesIndex()); + break; + } + } + } lruCache.remove(tableSegmentUniqueIdentifierWrapper.getTableBlockIndexUniqueIdentifier() .getUniqueTableSegmentIdentifier()); } @@ -247,7 +265,7 @@ public class BlockletDataMapIndexStore BlockDataMap dataMap; synchronized (lock) { dataMap = (BlockDataMap) BlockletDataMapFactory.createDataMap(carbonTable); - dataMap.init(new BlockletDataMapModel( + dataMap.init(new BlockletDataMapModel(carbonTable, identifier.getIndexFilePath() + CarbonCommonConstants.FILE_SEPARATOR + identifier .getIndexFileName(), indexFileStore.getFileData(identifier.getIndexFileName()), blockMetaInfoMap, identifier.getSegmentId())); http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDetailInfo.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDetailInfo.java b/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDetailInfo.java index 8e568ca..3cd86b0 100644 --- a/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDetailInfo.java +++ b/core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDetailInfo.java @@ -26,9 +26,9 @@ import java.util.List; import org.apache.carbondata.common.logging.LogService; import org.apache.carbondata.common.logging.LogServiceFactory; -import org.apache.carbondata.core.indexstore.blockletindex.BlockDataMap; import org.apache.carbondata.core.metadata.blocklet.BlockletInfo; import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema; +import org.apache.carbondata.core.util.BlockletDataMapUtil; import org.apache.hadoop.io.Writable; @@ -171,8 +171,16 @@ public class BlockletDetailInfo implements Serializable, Writable { blockletInfo.write(out); } out.writeLong(blockFooterOffset); - out.writeInt(columnSchemaBinary.length); - out.write(columnSchemaBinary); + // convert column schema list to binary format for serializing + convertColumnSchemaToBinary(); + if (null != columnSchemaBinary) { + out.writeInt(columnSchemaBinary.length); + out.write(columnSchemaBinary); + } else { + // write -1 if columnSchemaBinary is null so that at the time of reading it can distinguish + // whether schema is written or not + out.writeInt(-1); + } out.writeInt(blockletInfoBinary.length); out.write(blockletInfoBinary); out.writeLong(blockSize); @@ -195,9 +203,12 @@ public class BlockletDetailInfo implements Serializable, Writable { } blockFooterOffset = in.readLong(); int bytesSize = in.readInt(); - byte[] schemaArray = new byte[bytesSize]; - in.readFully(schemaArray); - readColumnSchema(schemaArray); + // if byteSize is -1 that means schema binary is not written + if (bytesSize != -1) { + byte[] schemaArray = new byte[bytesSize]; + in.readFully(schemaArray); + readColumnSchema(schemaArray); + } int byteSize = in.readInt(); blockletInfoBinary = new byte[byteSize]; in.readFully(blockletInfoBinary); @@ -212,8 +223,15 @@ public class BlockletDetailInfo implements Serializable, Writable { * @throws IOException */ public void readColumnSchema(byte[] schemaArray) throws IOException { - BlockDataMap blockDataMap = new BlockDataMap(); - columnSchemas = blockDataMap.readColumnSchema(schemaArray); + if (null != columnSchemaBinary) { + columnSchemas = BlockletDataMapUtil.readColumnSchema(schemaArray); + } + } + + private void convertColumnSchemaToBinary() throws IOException { + if (null != columnSchemas) { + columnSchemaBinary = BlockletDataMapUtil.convertSchemaToBinary(columnSchemas); + } } /** @@ -260,10 +278,6 @@ public class BlockletDetailInfo implements Serializable, Writable { return columnSchemas; } - public void setColumnSchemaBinary(byte[] columnSchemaBinary) { - this.columnSchemaBinary = columnSchemaBinary; - } - public byte[] getColumnSchemaBinary() { return columnSchemaBinary; } @@ -279,4 +293,8 @@ public class BlockletDetailInfo implements Serializable, Writable { public void setLegacyStore(boolean legacyStore) { isLegacyStore = legacyStore; } + + public void setColumnSchemas(List<ColumnSchema> columnSchemas) { + this.columnSchemas = columnSchemas; + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlocklet.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlocklet.java b/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlocklet.java index 077b942..22dff8e 100644 --- a/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlocklet.java +++ b/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlocklet.java @@ -37,6 +37,11 @@ public class ExtendedBlocklet extends Blocklet { super(filePath, blockletId); } + public ExtendedBlocklet(String filePath, String blockletId, + boolean compareBlockletIdForObjectMatching) { + super(filePath, blockletId, compareBlockletIdForObjectMatching); + } + public BlockletDetailInfo getDetailInfo() { return detailInfo; } http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/core/src/main/java/org/apache/carbondata/core/indexstore/SafeMemoryDMStore.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/SafeMemoryDMStore.java b/core/src/main/java/org/apache/carbondata/core/indexstore/SafeMemoryDMStore.java index d7a1b8f..0b3d4d8 100644 --- a/core/src/main/java/org/apache/carbondata/core/indexstore/SafeMemoryDMStore.java +++ b/core/src/main/java/org/apache/carbondata/core/indexstore/SafeMemoryDMStore.java @@ -39,10 +39,6 @@ public class SafeMemoryDMStore extends AbstractMemoryDMStore { private int runningLength; - public SafeMemoryDMStore(CarbonRowSchema[] schema) { - super(schema); - } - /** * Add the index row to dataMapRows, basically to in memory. * @@ -50,13 +46,13 @@ public class SafeMemoryDMStore extends AbstractMemoryDMStore { * @return */ @Override - public void addIndexRow(DataMapRow indexRow) throws MemoryException { + public void addIndexRow(CarbonRowSchema[] schema, DataMapRow indexRow) throws MemoryException { dataMapRows.add(indexRow); runningLength += indexRow.getTotalSizeInBytes(); } @Override - public DataMapRow getDataMapRow(int index) { + public DataMapRow getDataMapRow(CarbonRowSchema[] schema, int index) { assert (index < dataMapRows.size()); return dataMapRows.get(index); } @@ -83,11 +79,13 @@ public class SafeMemoryDMStore extends AbstractMemoryDMStore { } @Override - public UnsafeMemoryDMStore convertToUnsafeDMStore() throws MemoryException { - setSchemaDataType(); - UnsafeMemoryDMStore unsafeMemoryDMStore = new UnsafeMemoryDMStore(schema); + public UnsafeMemoryDMStore convertToUnsafeDMStore(CarbonRowSchema[] schema) + throws MemoryException { + setSchemaDataType(schema); + UnsafeMemoryDMStore unsafeMemoryDMStore = new UnsafeMemoryDMStore(); for (DataMapRow dataMapRow : dataMapRows) { - unsafeMemoryDMStore.addIndexRow(dataMapRow); + dataMapRow.setSchemas(schema); + unsafeMemoryDMStore.addIndexRow(schema, dataMapRow); } unsafeMemoryDMStore.finishWriting(); return unsafeMemoryDMStore; @@ -96,7 +94,7 @@ public class SafeMemoryDMStore extends AbstractMemoryDMStore { /** * Set the dataType to the schema. Needed in case of serialization / deserialization */ - private void setSchemaDataType() { + private void setSchemaDataType(CarbonRowSchema[] schema) { for (CarbonRowSchema carbonRowSchema : schema) { carbonRowSchema.setDataType(DataTypeUtil.valueOf(carbonRowSchema.getDataType(), 0, 0)); } http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/core/src/main/java/org/apache/carbondata/core/indexstore/UnsafeMemoryDMStore.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/UnsafeMemoryDMStore.java b/core/src/main/java/org/apache/carbondata/core/indexstore/UnsafeMemoryDMStore.java index 599877c..9296c99 100644 --- a/core/src/main/java/org/apache/carbondata/core/indexstore/UnsafeMemoryDMStore.java +++ b/core/src/main/java/org/apache/carbondata/core/indexstore/UnsafeMemoryDMStore.java @@ -47,8 +47,7 @@ public class UnsafeMemoryDMStore extends AbstractMemoryDMStore { private int rowCount; - public UnsafeMemoryDMStore(CarbonRowSchema[] schema) throws MemoryException { - super(schema); + public UnsafeMemoryDMStore() throws MemoryException { this.allocatedSize = capacity; this.memoryBlock = UnsafeMemoryManager.allocateMemoryWithRetry(taskId, allocatedSize); this.pointers = new int[1000]; @@ -87,7 +86,7 @@ public class UnsafeMemoryDMStore extends AbstractMemoryDMStore { * @param indexRow * @return */ - public void addIndexRow(DataMapRow indexRow) throws MemoryException { + public void addIndexRow(CarbonRowSchema[] schema, DataMapRow indexRow) throws MemoryException { // First calculate the required memory to keep the row in unsafe int rowSize = indexRow.getTotalSizeInBytes(); // Check whether allocated memory is sufficient or not. @@ -176,7 +175,7 @@ public class UnsafeMemoryDMStore extends AbstractMemoryDMStore { } } - public DataMapRow getDataMapRow(int index) { + public DataMapRow getDataMapRow(CarbonRowSchema[] schema, int index) { assert (index < rowCount); return new UnsafeDataMapRow(schema, memoryBlock, pointers[index]); } http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockDataMap.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockDataMap.java b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockDataMap.java index 7861f15..4fe3daa 100644 --- a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockDataMap.java +++ b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockDataMap.java @@ -16,7 +16,9 @@ */ package org.apache.carbondata.core.indexstore.blockletindex; -import java.io.*; +import java.io.IOException; +import java.io.Serializable; +import java.io.UnsupportedEncodingException; import java.util.ArrayList; import java.util.BitSet; import java.util.List; @@ -27,6 +29,7 @@ import org.apache.carbondata.core.constants.CarbonCommonConstants; import org.apache.carbondata.core.datamap.dev.DataMapModel; import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap; import org.apache.carbondata.core.datastore.block.SegmentProperties; +import org.apache.carbondata.core.datastore.block.SegmentPropertiesAndSchemaHolder; import org.apache.carbondata.core.datastore.block.TableBlockInfo; import org.apache.carbondata.core.indexstore.AbstractMemoryDMStore; import org.apache.carbondata.core.indexstore.BlockMetaInfo; @@ -60,7 +63,6 @@ import org.apache.carbondata.core.util.path.CarbonTablePath; import org.apache.commons.lang3.ArrayUtils; import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.fs.Path; -import org.xerial.snappy.Snappy; /** * Datamap implementation for block. @@ -76,21 +78,27 @@ public class BlockDataMap extends CoarseGrainDataMap * for CACHE_LEVEL=BLOCK and legacy store default blocklet id will be -1 */ private static final short BLOCK_DEFAULT_BLOCKLET_ID = -1; - + /** + * store which will hold all the block or blocklet entries in one task + */ protected AbstractMemoryDMStore memoryDMStore; - + /** + * task summary holder store + */ protected AbstractMemoryDMStore taskSummaryDMStore; - - // As it is a heavy object it is not recommended to serialize this object - protected transient SegmentProperties segmentProperties; - - protected int[] columnCardinality; - - protected long blockletSchemaTime; + /** + * index of segmentProperties in the segmentProperties holder + */ + protected int segmentPropertiesIndex; /** * flag to check for store from 1.1 or any prior version */ protected boolean isLegacyStore; + /** + * flag to be used for forming the complete file path from file name. It will be true in case of + * partition table and non transactional table + */ + protected boolean isFilePathStored; @Override public void init(DataMapModel dataMapModel) throws IOException, MemoryException { long startTime = System.currentTimeMillis(); @@ -100,25 +108,30 @@ public class BlockDataMap extends CoarseGrainDataMap List<DataFileFooter> indexInfo = fileFooterConverter .getIndexInfo(blockletDataMapInfo.getFilePath(), blockletDataMapInfo.getFileData()); Path path = new Path(blockletDataMapInfo.getFilePath()); - byte[] filePath = path.getParent().toString().getBytes(CarbonCommonConstants.DEFAULT_CHARSET); + // store file path only in case of partition table, non transactional table and flat folder + // structure + byte[] filePath = null; + boolean isPartitionTable = blockletDataMapInfo.getCarbonTable().isHivePartitionTable(); + if (isPartitionTable || !blockletDataMapInfo.getCarbonTable().isTransactionalTable() + || blockletDataMapInfo.getCarbonTable().isSupportFlatFolder()) { + filePath = path.getParent().toString().getBytes(CarbonCommonConstants.DEFAULT_CHARSET); + isFilePathStored = true; + } byte[] fileName = path.getName().toString().getBytes(CarbonCommonConstants.DEFAULT_CHARSET); byte[] segmentId = blockletDataMapInfo.getSegmentId().getBytes(CarbonCommonConstants.DEFAULT_CHARSET); - byte[] schemaBinary = null; if (!indexInfo.isEmpty()) { DataFileFooter fileFooter = indexInfo.get(0); // store for 1.1 or any prior version will not have any blocklet information in file footer isLegacyStore = fileFooter.getBlockletList() == null; // init segment properties and create schema - initSegmentProperties(fileFooter); - schemaBinary = convertSchemaToBinary(fileFooter.getColumnInTable()); - createSchema(segmentProperties, blockletDataMapInfo.isAddToUnsafe()); - createSummarySchema(segmentProperties, schemaBinary, filePath, fileName, segmentId, - blockletDataMapInfo.isAddToUnsafe()); - } - // check for legacy store and load the metadata - DataMapRowImpl summaryRow = loadMetadata(blockletDataMapInfo, indexInfo); - finishWriting(filePath, fileName, segmentId, schemaBinary, summaryRow); + SegmentProperties segmentProperties = initSegmentProperties(blockletDataMapInfo, fileFooter); + createMemoryDMStore(blockletDataMapInfo.isAddToUnsafe()); + createSummarySchema(segmentProperties, blockletDataMapInfo.isAddToUnsafe()); + // check for legacy store and load the metadata + DataMapRowImpl summaryRow = loadMetadata(segmentProperties, blockletDataMapInfo, indexInfo); + finishWriting(filePath, fileName, segmentId, summaryRow); + } if (LOGGER.isDebugEnabled()) { LOGGER.debug( "Time taken to load blocklet datamap from file : " + dataMapModel.getFilePath() + " is " @@ -127,14 +140,12 @@ public class BlockDataMap extends CoarseGrainDataMap } private void finishWriting(byte[] filePath, byte[] fileName, byte[] segmentId, - byte[] schemaBinary, DataMapRowImpl summaryRow) throws MemoryException { + DataMapRowImpl summaryRow) throws MemoryException { if (memoryDMStore != null) { memoryDMStore.finishWriting(); } - // TODO: schema binary not required. Instead maintain only the segmentProperties index and - // get the cardinality and columnSchema from there if (null != taskSummaryDMStore) { - addTaskSummaryRowToUnsafeMemoryStore(summaryRow, schemaBinary, filePath, fileName, segmentId); + addTaskSummaryRowToUnsafeMemoryStore(summaryRow, filePath, fileName, segmentId); taskSummaryDMStore.finishWriting(); } } @@ -142,17 +153,19 @@ public class BlockDataMap extends CoarseGrainDataMap /** * Method to check the cache level and load metadata based on that information * + * @param segmentProperties * @param blockletDataMapInfo * @param indexInfo * @throws IOException * @throws MemoryException */ - protected DataMapRowImpl loadMetadata(BlockletDataMapModel blockletDataMapInfo, - List<DataFileFooter> indexInfo) throws IOException, MemoryException { + protected DataMapRowImpl loadMetadata(SegmentProperties segmentProperties, + BlockletDataMapModel blockletDataMapInfo, List<DataFileFooter> indexInfo) + throws IOException, MemoryException { if (isLegacyStore) { - return loadBlockInfoForOldStore(blockletDataMapInfo, indexInfo); + return loadBlockInfoForOldStore(segmentProperties, blockletDataMapInfo, indexInfo); } else { - return loadBlockMetaInfo(blockletDataMapInfo, indexInfo); + return loadBlockMetaInfo(segmentProperties, blockletDataMapInfo, indexInfo); } } @@ -162,12 +175,14 @@ public class BlockDataMap extends CoarseGrainDataMap * @param fileFooter * @throws IOException */ - private void initSegmentProperties(DataFileFooter fileFooter) throws IOException { + private SegmentProperties initSegmentProperties(BlockletDataMapModel blockletDataMapInfo, + DataFileFooter fileFooter) throws IOException { List<ColumnSchema> columnInTable = fileFooter.getColumnInTable(); - // TODO: remove blockletSchemaTime after maintaining the index of segmentProperties - blockletSchemaTime = fileFooter.getSchemaUpdatedTimeStamp(); - columnCardinality = fileFooter.getSegmentInfo().getColumnCardinality(); - segmentProperties = new SegmentProperties(columnInTable, columnCardinality); + int[] columnCardinality = fileFooter.getSegmentInfo().getColumnCardinality(); + segmentPropertiesIndex = SegmentPropertiesAndSchemaHolder.getInstance() + .addSegmentProperties(blockletDataMapInfo.getCarbonTable().getAbsoluteTableIdentifier(), + columnInTable, columnCardinality, blockletDataMapInfo.getSegmentId()); + return getSegmentProperties(); } /** @@ -179,8 +194,9 @@ public class BlockDataMap extends CoarseGrainDataMap * @throws IOException * @throws MemoryException */ - protected DataMapRowImpl loadBlockInfoForOldStore(BlockletDataMapModel blockletDataMapInfo, - List<DataFileFooter> indexInfo) throws IOException, MemoryException { + protected DataMapRowImpl loadBlockInfoForOldStore(SegmentProperties segmentProperties, + BlockletDataMapModel blockletDataMapInfo, List<DataFileFooter> indexInfo) + throws IOException, MemoryException { DataMapRowImpl summaryRow = null; for (DataFileFooter fileFooter : indexInfo) { TableBlockInfo blockInfo = fileFooter.getBlockInfo().getTableBlockInfo(); @@ -219,8 +235,9 @@ public class BlockDataMap extends CoarseGrainDataMap * @throws IOException * @throws MemoryException */ - private DataMapRowImpl loadBlockMetaInfo(BlockletDataMapModel blockletDataMapInfo, - List<DataFileFooter> indexInfo) throws IOException, MemoryException { + private DataMapRowImpl loadBlockMetaInfo(SegmentProperties segmentProperties, + BlockletDataMapModel blockletDataMapInfo, List<DataFileFooter> indexInfo) + throws IOException, MemoryException { String tempFilePath = null; DataFileFooter previousDataFileFooter = null; int footerCounter = 0; @@ -294,7 +311,8 @@ public class BlockDataMap extends CoarseGrainDataMap } byte[] blockletCount = ArrayUtils .toPrimitive(blockletCountInEachBlock.toArray(new Byte[blockletCountInEachBlock.size()])); - summaryRow.setByteArray(blockletCount, SUMMARY_BLOCKLET_COUNT); + // blocklet count index is the last index + summaryRow.setByteArray(blockletCount, getTaskSummarySchema().length - 1); return summaryRow; } @@ -314,30 +332,31 @@ public class BlockDataMap extends CoarseGrainDataMap SegmentProperties segmentProperties, String filePath, DataMapRowImpl summaryRow, BlockMetaInfo blockMetaInfo, byte[][] minValues, byte[][] maxValues) { int[] minMaxLen = segmentProperties.getColumnsValueSize(); - CarbonRowSchema[] schema = memoryDMStore.getSchema(); + CarbonRowSchema[] schema = getSchema(); + CarbonRowSchema[] taskSummarySchema = getTaskSummarySchema(); // Add one row to maintain task level min max for segment pruning if (summaryRow == null) { - summaryRow = new DataMapRowImpl(taskSummaryDMStore.getSchema()); + summaryRow = new DataMapRowImpl(taskSummarySchema); } DataMapRow row = new DataMapRowImpl(schema); int ordinal = 0; int taskMinMaxOrdinal = 0; row.setRow(addMinMax(minMaxLen, schema[ordinal], minValues), ordinal); // compute and set task level min values - addTaskMinMaxValues(summaryRow, minMaxLen, taskSummaryDMStore.getSchema(), taskMinMaxOrdinal, + addTaskMinMaxValues(summaryRow, minMaxLen, taskSummarySchema, taskMinMaxOrdinal, minValues, TASK_MIN_VALUES_INDEX, true); ordinal++; taskMinMaxOrdinal++; row.setRow(addMinMax(minMaxLen, schema[ordinal], maxValues), ordinal); // compute and set task level max values - addTaskMinMaxValues(summaryRow, minMaxLen, taskSummaryDMStore.getSchema(), taskMinMaxOrdinal, + addTaskMinMaxValues(summaryRow, minMaxLen, taskSummarySchema, taskMinMaxOrdinal, maxValues, TASK_MAX_VALUES_INDEX, false); ordinal++; // add total rows in one carbondata file row.setInt((int) fileFooter.getNumberOfRows(), ordinal++); - // add file path - // TODO: shorten file path - byte[] filePathBytes = filePath.getBytes(CarbonCommonConstants.DEFAULT_CHARSET_CLASS); + // add file name + byte[] filePathBytes = + getFileNameFromPath(filePath).getBytes(CarbonCommonConstants.DEFAULT_CHARSET_CLASS); row.setByteArray(filePathBytes, ordinal++); // add version number row.setShort(fileFooter.getVersionId().number(), ordinal++); @@ -349,27 +368,46 @@ public class BlockDataMap extends CoarseGrainDataMap setLocations(blockMetaInfo.getLocationInfo(), row, ordinal++); // store block size row.setLong(blockMetaInfo.getSize(), ordinal); - memoryDMStore.addIndexRow(row); + memoryDMStore.addIndexRow(schema, row); } catch (Exception e) { throw new RuntimeException(e); } return summaryRow; } - private void addTaskSummaryRowToUnsafeMemoryStore(DataMapRow summaryRow, byte[] schemaBinary, - byte[] filePath, byte[] fileName, byte[] segmentId) { + protected String getFileNameFromPath(String filePath) { + return CarbonTablePath.getCarbonDataFileName(filePath); + } + + protected String getFilePath() { + if (isFilePathStored) { + return getTableTaskInfo(SUMMARY_INDEX_PATH); + } + // create the segment directory path + String tablePath = SegmentPropertiesAndSchemaHolder.getInstance() + .getSegmentPropertiesWrapper(segmentPropertiesIndex).getTableIdentifier().getTablePath(); + String segmentId = getTableTaskInfo(SUMMARY_SEGMENTID); + return CarbonTablePath.getSegmentPath(tablePath, segmentId); + } + + protected String getFileNameWithFilePath(DataMapRow dataMapRow, String filePath) { + String fileName = filePath + CarbonCommonConstants.FILE_SEPARATOR + new String( + dataMapRow.getByteArray(FILE_PATH_INDEX), CarbonCommonConstants.DEFAULT_CHARSET_CLASS) + + CarbonTablePath.getCarbonDataExtension(); + return fileName; + } + + private void addTaskSummaryRowToUnsafeMemoryStore(DataMapRow summaryRow, byte[] filePath, + byte[] fileName, byte[] segmentId) { // write the task summary info to unsafe memory store if (null != summaryRow) { - // Add column schema , it is useful to generate segment properties in executor. - // So we no need to read footer again there. - if (schemaBinary != null) { - summaryRow.setByteArray(schemaBinary, SUMMARY_SCHEMA); - } - summaryRow.setByteArray(filePath, SUMMARY_INDEX_PATH); summaryRow.setByteArray(fileName, SUMMARY_INDEX_FILE_NAME); summaryRow.setByteArray(segmentId, SUMMARY_SEGMENTID); + if (null != filePath) { + summaryRow.setByteArray(filePath, SUMMARY_INDEX_PATH); + } try { - taskSummaryDMStore.addIndexRow(summaryRow); + taskSummaryDMStore.addIndexRow(getTaskSummarySchema(), summaryRow); } catch (Exception e) { throw new RuntimeException(e); } @@ -451,10 +489,8 @@ public class BlockDataMap extends CoarseGrainDataMap return updatedMinMaxValues; } - protected void createSchema(SegmentProperties segmentProperties, boolean addToUnsafe) - throws MemoryException { - CarbonRowSchema[] schema = SchemaGenerator.createBlockSchema(segmentProperties); - memoryDMStore = getMemoryDMStore(schema, addToUnsafe); + protected void createMemoryDMStore(boolean addToUnsafe) throws MemoryException { + memoryDMStore = getMemoryDMStore(addToUnsafe); } /** @@ -465,23 +501,25 @@ public class BlockDataMap extends CoarseGrainDataMap * @param segmentProperties * @throws MemoryException */ - protected void createSummarySchema(SegmentProperties segmentProperties, byte[] schemaBinary, - byte[] filePath, byte[] fileName, byte[] segmentId, boolean addToUnsafe) + protected void createSummarySchema(SegmentProperties segmentProperties, boolean addToUnsafe) throws MemoryException { // flag to check whether it is required to store blocklet count of each carbondata file as // binary in summary schema. This will be true when it is not a legacy store (>1.1 version) // and CACHE_LEVEL=BLOCK boolean storeBlockletCount = !isLegacyStore; CarbonRowSchema[] taskSummarySchema = SchemaGenerator - .createTaskSummarySchema(segmentProperties, schemaBinary, filePath, fileName, segmentId, - storeBlockletCount); - taskSummaryDMStore = getMemoryDMStore(taskSummarySchema, addToUnsafe); + .createTaskSummarySchema(segmentProperties, storeBlockletCount, isFilePathStored); + SegmentPropertiesAndSchemaHolder.getInstance() + .getSegmentPropertiesWrapper(segmentPropertiesIndex) + .setTaskSummarySchema(taskSummarySchema); + taskSummaryDMStore = getMemoryDMStore(addToUnsafe); } @Override public boolean isScanRequired(FilterResolverIntf filterExp) { FilterExecuter filterExecuter = - FilterUtil.getFilterExecuterTree(filterExp, segmentProperties, null); - DataMapRow unsafeRow = taskSummaryDMStore.getDataMapRow(taskSummaryDMStore.getRowCount() - 1); + FilterUtil.getFilterExecuterTree(filterExp, getSegmentProperties(), null); + DataMapRow unsafeRow = taskSummaryDMStore + .getDataMapRow(getTaskSummarySchema(), taskSummaryDMStore.getRowCount() - 1); boolean isScanRequired = FilterExpressionProcessor .isScanRequired(filterExecuter, getMinMaxValue(unsafeRow, TASK_MAX_VALUES_INDEX), getMinMaxValue(unsafeRow, TASK_MIN_VALUES_INDEX)); @@ -491,17 +529,20 @@ public class BlockDataMap extends CoarseGrainDataMap return false; } - private List<Blocklet> prune(FilterResolverIntf filterExp, SegmentProperties segmentProperties) { + private List<Blocklet> prune(FilterResolverIntf filterExp) { if (memoryDMStore.getRowCount() == 0) { return new ArrayList<>(); } List<Blocklet> blocklets = new ArrayList<>(); + CarbonRowSchema[] schema = getSchema(); + String filePath = getFilePath(); int numBlocklets = 0; if (filterExp == null) { numBlocklets = memoryDMStore.getRowCount(); for (int i = 0; i < numBlocklets; i++) { - DataMapRow safeRow = memoryDMStore.getDataMapRow(i).convertToSafeRow(); - blocklets.add(createBlocklet(safeRow, getBlockletId(safeRow))); + DataMapRow safeRow = memoryDMStore.getDataMapRow(schema, i).convertToSafeRow(); + blocklets.add(createBlocklet(safeRow, getFileNameWithFilePath(safeRow, filePath), + getBlockletId(safeRow))); } } else { // Remove B-tree jump logic as start and end key prepared is not @@ -509,17 +550,16 @@ public class BlockDataMap extends CoarseGrainDataMap int startIndex = 0; numBlocklets = memoryDMStore.getRowCount(); FilterExecuter filterExecuter = - FilterUtil.getFilterExecuterTree(filterExp, segmentProperties, null); + FilterUtil.getFilterExecuterTree(filterExp, getSegmentProperties(), null); while (startIndex < numBlocklets) { - DataMapRow safeRow = memoryDMStore.getDataMapRow(startIndex).convertToSafeRow(); - String filePath = new String(safeRow.getByteArray(FILE_PATH_INDEX), - CarbonCommonConstants.DEFAULT_CHARSET_CLASS); + DataMapRow safeRow = memoryDMStore.getDataMapRow(schema, startIndex).convertToSafeRow(); + String fileName = getFileNameWithFilePath(safeRow, filePath); short blockletId = getBlockletId(safeRow); boolean isValid = addBlockBasedOnMinMaxValue(filterExecuter, getMinMaxValue(safeRow, MAX_VALUES_INDEX), - getMinMaxValue(safeRow, MIN_VALUES_INDEX), filePath, blockletId); + getMinMaxValue(safeRow, MIN_VALUES_INDEX), fileName, blockletId); if (isValid) { - blocklets.add(createBlocklet(safeRow, blockletId)); + blocklets.add(createBlocklet(safeRow, fileName, blockletId)); } startIndex++; } @@ -546,7 +586,7 @@ public class BlockDataMap extends CoarseGrainDataMap // segmentProperties. // Its a temporary fix. The Interface DataMap.prune(FilterResolverIntf filterExp, // SegmentProperties segmentProperties, List<PartitionSpec> partitions) should be corrected - return prune(filterExp, this.segmentProperties); + return prune(filterExp); } private boolean validatePartitionInfo(List<PartitionSpec> partitions) { @@ -655,14 +695,17 @@ public class BlockDataMap extends CoarseGrainDataMap rowIndex++; } } - DataMapRow safeRow = memoryDMStore.getDataMapRow(rowIndex).convertToSafeRow(); - return createBlocklet(safeRow, relativeBlockletId); + DataMapRow safeRow = memoryDMStore.getDataMapRow(getSchema(), rowIndex).convertToSafeRow(); + String filePath = getFilePath(); + return createBlocklet(safeRow, getFileNameWithFilePath(safeRow, filePath), relativeBlockletId); } private byte[] getBlockletRowCountForEachBlock() { // taskSummary DM store will have only one row - return taskSummaryDMStore.getDataMapRow(taskSummaryDMStore.getRowCount() - 1) - .getByteArray(SUMMARY_BLOCKLET_COUNT); + CarbonRowSchema[] taskSummarySchema = getTaskSummarySchema(); + return taskSummaryDMStore + .getDataMapRow(taskSummarySchema, taskSummaryDMStore.getRowCount() - 1) + .getByteArray(taskSummarySchema.length - 1); } /** @@ -670,11 +713,10 @@ public class BlockDataMap extends CoarseGrainDataMap * * @return */ - public String getIndexFileName() { - DataMapRow unsafeRow = taskSummaryDMStore.getDataMapRow(0); + public String getTableTaskInfo(int index) { + DataMapRow unsafeRow = taskSummaryDMStore.getDataMapRow(getTaskSummarySchema(), 0); try { - return new String(unsafeRow.getByteArray(SUMMARY_INDEX_FILE_NAME), - CarbonCommonConstants.DEFAULT_CHARSET); + return new String(unsafeRow.getByteArray(index), CarbonCommonConstants.DEFAULT_CHARSET); } catch (UnsupportedEncodingException e) { // should never happen! throw new IllegalArgumentException("UTF8 encoding is not supported", e); @@ -694,10 +736,8 @@ public class BlockDataMap extends CoarseGrainDataMap return BLOCK_DEFAULT_BLOCKLET_ID; } - protected ExtendedBlocklet createBlocklet(DataMapRow row, short blockletId) { - ExtendedBlocklet blocklet = new ExtendedBlocklet( - new String(row.getByteArray(FILE_PATH_INDEX), CarbonCommonConstants.DEFAULT_CHARSET_CLASS), - blockletId + ""); + protected ExtendedBlocklet createBlocklet(DataMapRow row, String fileName, short blockletId) { + ExtendedBlocklet blocklet = new ExtendedBlocklet(fileName, blockletId + "", false); BlockletDetailInfo detailInfo = getBlockletDetailInfo(row, blockletId, blocklet); detailInfo.setBlockletInfoBinary(new byte[0]); blocklet.setDetailInfo(detailInfo); @@ -709,8 +749,8 @@ public class BlockDataMap extends CoarseGrainDataMap BlockletDetailInfo detailInfo = new BlockletDetailInfo(); detailInfo.setRowCount(row.getInt(ROW_COUNT_INDEX)); detailInfo.setVersionNumber(row.getShort(VERSION_INDEX)); - detailInfo.setDimLens(columnCardinality); detailInfo.setBlockletId(blockletId); + detailInfo.setDimLens(getColumnCardinality()); detailInfo.setSchemaUpdatedTimeStamp(row.getLong(SCHEMA_UPADATED_TIME_INDEX)); try { blocklet.setLocation( @@ -720,7 +760,6 @@ public class BlockDataMap extends CoarseGrainDataMap throw new RuntimeException(e); } detailInfo.setBlockFooterOffset(row.getLong(BLOCK_FOOTER_OFFSET)); - detailInfo.setColumnSchemaBinary(getColumnSchemaBinary()); detailInfo.setBlockSize(row.getLong(BLOCK_LENGTH)); detailInfo.setLegacyStore(isLegacyStore); return detailInfo; @@ -729,7 +768,7 @@ public class BlockDataMap extends CoarseGrainDataMap private String[] getFileDetails() { try { String[] fileDetails = new String[3]; - DataMapRow unsafeRow = taskSummaryDMStore.getDataMapRow(0); + DataMapRow unsafeRow = taskSummaryDMStore.getDataMapRow(getTaskSummarySchema(), 0); fileDetails[0] = new String(unsafeRow.getByteArray(SUMMARY_INDEX_PATH), CarbonCommonConstants.DEFAULT_CHARSET); fileDetails[1] = new String(unsafeRow.getByteArray(SUMMARY_INDEX_FILE_NAME), @@ -742,34 +781,10 @@ public class BlockDataMap extends CoarseGrainDataMap } } - public byte[] getColumnSchemaBinary() { - DataMapRow unsafeRow = taskSummaryDMStore.getDataMapRow(0); - return unsafeRow.getByteArray(SUMMARY_SCHEMA); - } - - /** - * Convert schema to binary - */ - private byte[] convertSchemaToBinary(List<ColumnSchema> columnSchemas) throws IOException { - ByteArrayOutputStream stream = new ByteArrayOutputStream(); - DataOutput dataOutput = new DataOutputStream(stream); - dataOutput.writeShort(columnSchemas.size()); - for (ColumnSchema columnSchema : columnSchemas) { - if (columnSchema.getColumnReferenceId() == null) { - columnSchema.setColumnReferenceId(columnSchema.getColumnUniqueId()); - } - columnSchema.write(dataOutput); - } - byte[] byteArray = stream.toByteArray(); - // Compress with snappy to reduce the size of schema - return Snappy.rawCompress(byteArray, byteArray.length); - } - @Override public void clear() { if (memoryDMStore != null) { memoryDMStore.freeMemory(); memoryDMStore = null; - segmentProperties = null; } // clear task min/max unsafe memory if (null != taskSummaryDMStore) { @@ -789,29 +804,42 @@ public class BlockDataMap extends CoarseGrainDataMap return memoryUsed; } - public SegmentProperties getSegmentProperties() { - return segmentProperties; + protected SegmentProperties getSegmentProperties() { + return SegmentPropertiesAndSchemaHolder.getInstance() + .getSegmentProperties(segmentPropertiesIndex); } - public void setSegmentProperties(SegmentProperties segmentProperties) { - this.segmentProperties = segmentProperties; + public int[] getColumnCardinality() { + return SegmentPropertiesAndSchemaHolder.getInstance() + .getSegmentPropertiesWrapper(segmentPropertiesIndex).getColumnCardinality(); } - public int[] getColumnCardinality() { - return columnCardinality; + public List<ColumnSchema> getColumnSchema() { + return SegmentPropertiesAndSchemaHolder.getInstance() + .getSegmentPropertiesWrapper(segmentPropertiesIndex).getColumnsInTable(); } - protected AbstractMemoryDMStore getMemoryDMStore(CarbonRowSchema[] schema, boolean addToUnsafe) + protected AbstractMemoryDMStore getMemoryDMStore(boolean addToUnsafe) throws MemoryException { AbstractMemoryDMStore memoryDMStore; if (addToUnsafe) { - memoryDMStore = new UnsafeMemoryDMStore(schema); + memoryDMStore = new UnsafeMemoryDMStore(); } else { - memoryDMStore = new SafeMemoryDMStore(schema); + memoryDMStore = new SafeMemoryDMStore(); } return memoryDMStore; } + protected CarbonRowSchema[] getSchema() { + return SegmentPropertiesAndSchemaHolder.getInstance() + .getSegmentPropertiesWrapper(segmentPropertiesIndex).getBlockSchema(); + } + + protected CarbonRowSchema[] getTaskSummarySchema() { + return SegmentPropertiesAndSchemaHolder.getInstance() + .getSegmentPropertiesWrapper(segmentPropertiesIndex).getTaskSummarySchema(); + } + /** * This method will ocnvert safe to unsafe memory DM store * @@ -819,40 +847,23 @@ public class BlockDataMap extends CoarseGrainDataMap */ public void convertToUnsafeDMStore() throws MemoryException { if (memoryDMStore instanceof SafeMemoryDMStore) { - UnsafeMemoryDMStore unsafeMemoryDMStore = memoryDMStore.convertToUnsafeDMStore(); + UnsafeMemoryDMStore unsafeMemoryDMStore = memoryDMStore.convertToUnsafeDMStore(getSchema()); memoryDMStore.freeMemory(); memoryDMStore = unsafeMemoryDMStore; } if (taskSummaryDMStore instanceof SafeMemoryDMStore) { - UnsafeMemoryDMStore unsafeSummaryMemoryDMStore = taskSummaryDMStore.convertToUnsafeDMStore(); + UnsafeMemoryDMStore unsafeSummaryMemoryDMStore = + taskSummaryDMStore.convertToUnsafeDMStore(getTaskSummarySchema()); taskSummaryDMStore.freeMemory(); taskSummaryDMStore = unsafeSummaryMemoryDMStore; } } - /** - * Read column schema from binary - * - * @param schemaArray - * @throws IOException - */ - public List<ColumnSchema> readColumnSchema(byte[] schemaArray) throws IOException { - // uncompress it. - schemaArray = Snappy.uncompress(schemaArray); - ByteArrayInputStream schemaStream = new ByteArrayInputStream(schemaArray); - DataInput schemaInput = new DataInputStream(schemaStream); - List<ColumnSchema> columnSchemas = new ArrayList<>(); - int size = schemaInput.readShort(); - for (int i = 0; i < size; i++) { - ColumnSchema columnSchema = new ColumnSchema(); - columnSchema.readFields(schemaInput); - columnSchemas.add(columnSchema); - } - return columnSchemas; + public void setSegmentPropertiesIndex(int segmentPropertiesIndex) { + this.segmentPropertiesIndex = segmentPropertiesIndex; } - public long getBlockletSchemaTime() { - return blockletSchemaTime; + public int getSegmentPropertiesIndex() { + return segmentPropertiesIndex; } - } http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/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 eebd288..5274b0e 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 @@ -26,6 +26,7 @@ import java.util.List; import org.apache.carbondata.core.constants.CarbonCommonConstants; import org.apache.carbondata.core.datamap.dev.DataMapModel; import org.apache.carbondata.core.datastore.block.SegmentProperties; +import org.apache.carbondata.core.datastore.block.SegmentPropertiesAndSchemaHolder; import org.apache.carbondata.core.datastore.block.TableBlockInfo; import org.apache.carbondata.core.indexstore.BlockMetaInfo; import org.apache.carbondata.core.indexstore.BlockletDetailInfo; @@ -58,26 +59,24 @@ public class BlockletDataMap extends BlockDataMap implements Serializable { * @throws IOException * @throws MemoryException */ - protected DataMapRowImpl loadMetadata(BlockletDataMapModel blockletDataMapInfo, - List<DataFileFooter> indexInfo) throws IOException, MemoryException { + protected DataMapRowImpl loadMetadata(SegmentProperties segmentProperties, + BlockletDataMapModel blockletDataMapInfo, List<DataFileFooter> indexInfo) + throws IOException, MemoryException { if (isLegacyStore) { - return loadBlockInfoForOldStore(blockletDataMapInfo, indexInfo); + return loadBlockInfoForOldStore(segmentProperties, blockletDataMapInfo, indexInfo); } else { - return loadBlockletMetaInfo(blockletDataMapInfo, indexInfo); + return loadBlockletMetaInfo(segmentProperties, blockletDataMapInfo, indexInfo); } } /** * Method to create blocklet schema * - * @param segmentProperties * @param addToUnsafe * @throws MemoryException */ - protected void createSchema(SegmentProperties segmentProperties, boolean addToUnsafe) - throws MemoryException { - CarbonRowSchema[] schema = SchemaGenerator.createBlockletSchema(segmentProperties); - memoryDMStore = getMemoryDMStore(schema, addToUnsafe); + protected void createMemoryDMStore(boolean addToUnsafe) throws MemoryException { + memoryDMStore = getMemoryDMStore(addToUnsafe); } /** @@ -88,13 +87,14 @@ public class BlockletDataMap extends BlockDataMap implements Serializable { * @param segmentProperties * @throws MemoryException */ - protected void createSummarySchema(SegmentProperties segmentProperties, byte[] schemaBinary, - byte[] filePath, byte[] fileName, byte[] segmentId, boolean addToUnsafe) + protected void createSummarySchema(SegmentProperties segmentProperties, boolean addToUnsafe) throws MemoryException { - CarbonRowSchema[] taskSummarySchema = SchemaGenerator - .createTaskSummarySchema(segmentProperties, schemaBinary, filePath, fileName, segmentId, - false); - taskSummaryDMStore = getMemoryDMStore(taskSummarySchema, addToUnsafe); + CarbonRowSchema[] taskSummarySchema = + SchemaGenerator.createTaskSummarySchema(segmentProperties, false, isFilePathStored); + SegmentPropertiesAndSchemaHolder.getInstance() + .getSegmentPropertiesWrapper(segmentPropertiesIndex) + .setTaskSummarySchema(taskSummarySchema); + taskSummaryDMStore = getMemoryDMStore(addToUnsafe); } /** @@ -105,8 +105,9 @@ public class BlockletDataMap extends BlockDataMap implements Serializable { * @throws IOException * @throws MemoryException */ - private DataMapRowImpl loadBlockletMetaInfo(BlockletDataMapModel blockletDataMapInfo, - List<DataFileFooter> indexInfo) throws IOException, MemoryException { + private DataMapRowImpl loadBlockletMetaInfo(SegmentProperties segmentProperties, + BlockletDataMapModel blockletDataMapInfo, List<DataFileFooter> indexInfo) + throws IOException, MemoryException { String tempFilePath = null; DataMapRowImpl summaryRow = null; // Relative blocklet ID is the id assigned to a blocklet within a part file @@ -142,10 +143,11 @@ public class BlockletDataMap extends BlockDataMap implements Serializable { BlockMetaInfo blockMetaInfo, int relativeBlockletId) { int[] minMaxLen = segmentProperties.getColumnsValueSize(); List<BlockletInfo> blockletList = fileFooter.getBlockletList(); - CarbonRowSchema[] schema = memoryDMStore.getSchema(); + CarbonRowSchema[] schema = getSchema(); + CarbonRowSchema[] taskSummarySchema = getTaskSummarySchema(); // Add one row to maintain task level min max for segment pruning if (!blockletList.isEmpty() && summaryRow == null) { - summaryRow = new DataMapRowImpl(taskSummaryDMStore.getSchema()); + summaryRow = new DataMapRowImpl(taskSummarySchema); } for (int index = 0; index < blockletList.size(); index++) { DataMapRow row = new DataMapRowImpl(schema); @@ -155,18 +157,19 @@ public class BlockletDataMap extends BlockDataMap implements Serializable { BlockletMinMaxIndex minMaxIndex = blockletInfo.getBlockletIndex().getMinMaxIndex(); row.setRow(addMinMax(minMaxLen, schema[ordinal], minMaxIndex.getMinValues()), ordinal); // compute and set task level min values - addTaskMinMaxValues(summaryRow, minMaxLen, taskSummaryDMStore.getSchema(), taskMinMaxOrdinal, + addTaskMinMaxValues(summaryRow, minMaxLen, taskSummarySchema, taskMinMaxOrdinal, minMaxIndex.getMinValues(), TASK_MIN_VALUES_INDEX, true); ordinal++; taskMinMaxOrdinal++; row.setRow(addMinMax(minMaxLen, schema[ordinal], minMaxIndex.getMaxValues()), ordinal); // compute and set task level max values - addTaskMinMaxValues(summaryRow, minMaxLen, taskSummaryDMStore.getSchema(), taskMinMaxOrdinal, + addTaskMinMaxValues(summaryRow, minMaxLen, taskSummarySchema, taskMinMaxOrdinal, minMaxIndex.getMaxValues(), TASK_MAX_VALUES_INDEX, false); ordinal++; row.setInt(blockletInfo.getNumberOfRows(), ordinal++); - // add file path - byte[] filePathBytes = filePath.getBytes(CarbonCommonConstants.DEFAULT_CHARSET_CLASS); + // add file name + byte[] filePathBytes = + getFileNameFromPath(filePath).getBytes(CarbonCommonConstants.DEFAULT_CHARSET_CLASS); row.setByteArray(filePathBytes, ordinal++); // add version number row.setShort(fileFooter.getVersionId().number(), ordinal++); @@ -189,7 +192,7 @@ public class BlockletDataMap extends BlockDataMap implements Serializable { row.setShort((short) blockletInfo.getNumberOfPages(), ordinal++); // for relative blocklet id i.e blocklet id that belongs to a particular carbondata file row.setShort((short) relativeBlockletId++, ordinal); - memoryDMStore.addIndexRow(row); + memoryDMStore.addIndexRow(schema, row); } catch (Exception e) { throw new RuntimeException(e); } @@ -202,20 +205,26 @@ public class BlockletDataMap extends BlockDataMap implements Serializable { super.getDetailedBlocklet(blockletId); } int absoluteBlockletId = Integer.parseInt(blockletId); - DataMapRow safeRow = memoryDMStore.getDataMapRow(absoluteBlockletId).convertToSafeRow(); + DataMapRow safeRow = + memoryDMStore.getDataMapRow(getSchema(), absoluteBlockletId).convertToSafeRow(); short relativeBlockletId = safeRow.getShort(BLOCKLET_ID_INDEX); - return createBlocklet(safeRow, relativeBlockletId); + String filePath = getFilePath(); + return createBlocklet(safeRow, getFileNameWithFilePath(safeRow, filePath), relativeBlockletId); } protected short getBlockletId(DataMapRow dataMapRow) { return dataMapRow.getShort(BLOCKLET_ID_INDEX); } - protected ExtendedBlocklet createBlocklet(DataMapRow row, short blockletId) { - ExtendedBlocklet blocklet = new ExtendedBlocklet( - new String(row.getByteArray(FILE_PATH_INDEX), CarbonCommonConstants.DEFAULT_CHARSET_CLASS), - blockletId + ""); + protected CarbonRowSchema[] getSchema() { + return SegmentPropertiesAndSchemaHolder.getInstance() + .getSegmentPropertiesWrapper(segmentPropertiesIndex).getBlocketSchema(); + } + + protected ExtendedBlocklet createBlocklet(DataMapRow row, String fileName, short blockletId) { + ExtendedBlocklet blocklet = new ExtendedBlocklet(fileName, blockletId + ""); BlockletDetailInfo detailInfo = getBlockletDetailInfo(row, blockletId, blocklet); + detailInfo.setColumnSchemas(getColumnSchema()); detailInfo.setBlockletInfoBinary(row.getByteArray(BLOCKLET_INFO_INDEX)); detailInfo.setPagesCount(row.getShort(BLOCKLET_PAGE_COUNT_INDEX)); blocklet.setDetailInfo(detailInfo); http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/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 f9bc7b8..51e6e21 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 @@ -99,7 +99,6 @@ public class BlockletDataMapFactory extends CoarseGrainDataMapFactory public static DataMap createDataMap(CarbonTable carbonTable) { boolean cacheLevelBlock = BlockletDataMapUtil.isCacheLevelBlock(carbonTable, CACHE_LEVEL_BLOCKLET); - cacheLevelBlock = false; if (cacheLevelBlock) { // case1: when CACHE_LEVEL = BLOCK return new BlockDataMap(); @@ -212,7 +211,9 @@ public class BlockletDataMapFactory extends CoarseGrainDataMapFactory BlockletDataMapIndexWrapper wrapper = cache.get(identifierWrapper); List<BlockDataMap> dataMaps = wrapper.getDataMaps(); for (DataMap dataMap : dataMaps) { - if (((BlockDataMap) dataMap).getIndexFileName().startsWith(blocklet.getFilePath())) { + if (((BlockDataMap) dataMap) + .getTableTaskInfo(BlockletDataMapRowIndexes.SUMMARY_INDEX_FILE_NAME) + .startsWith(blocklet.getFilePath())) { return ((BlockDataMap) dataMap).getDetailedBlocklet(blocklet.getBlockletId()); } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/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 index 7443d15..180c812 100644 --- 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 @@ -20,6 +20,7 @@ import java.util.Map; import org.apache.carbondata.core.datamap.dev.DataMapModel; import org.apache.carbondata.core.indexstore.BlockMetaInfo; +import org.apache.carbondata.core.metadata.schema.table.CarbonTable; /** * It is the model object to keep the information to build or initialize BlockletDataMap. @@ -30,21 +31,25 @@ public class BlockletDataMapModel extends DataMapModel { private Map<String, BlockMetaInfo> blockMetaInfoMap; + private CarbonTable carbonTable; + private String segmentId; private boolean addToUnsafe = true; - public BlockletDataMapModel(String filePath, byte[] fileData, - Map<String, BlockMetaInfo> blockMetaInfoMap, String segmentId) { + public BlockletDataMapModel(CarbonTable carbonTable, String filePath, + byte[] fileData, Map<String, BlockMetaInfo> blockMetaInfoMap, String segmentId) { super(filePath); this.fileData = fileData; this.blockMetaInfoMap = blockMetaInfoMap; this.segmentId = segmentId; + this.carbonTable = carbonTable; } - public BlockletDataMapModel(String filePath, byte[] fileData, - Map<String, BlockMetaInfo> blockMetaInfoMap, String segmentId, boolean addToUnsafe) { - this(filePath, fileData, blockMetaInfoMap, segmentId); + public BlockletDataMapModel(CarbonTable carbonTable, String filePath, + byte[] fileData, Map<String, BlockMetaInfo> blockMetaInfoMap, String segmentId, + boolean addToUnsafe) { + this(carbonTable, filePath, fileData, blockMetaInfoMap, segmentId); this.addToUnsafe = addToUnsafe; } @@ -63,4 +68,8 @@ public class BlockletDataMapModel extends DataMapModel { public boolean isAddToUnsafe() { return addToUnsafe; } + + public CarbonTable getCarbonTable() { + return carbonTable; + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapRowIndexes.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapRowIndexes.java b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapRowIndexes.java index 3826b07..7f61d77 100644 --- a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapRowIndexes.java +++ b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapRowIndexes.java @@ -52,14 +52,9 @@ public interface BlockletDataMapRowIndexes { int TASK_MAX_VALUES_INDEX = 1; - int SUMMARY_SCHEMA = 2; + int SUMMARY_INDEX_FILE_NAME = 2; - int SUMMARY_INDEX_PATH = 3; - - int SUMMARY_INDEX_FILE_NAME = 4; - - int SUMMARY_SEGMENTID = 5; - - int SUMMARY_BLOCKLET_COUNT = 6; + int SUMMARY_SEGMENTID = 3; + int SUMMARY_INDEX_PATH = 4; } http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/core/src/main/java/org/apache/carbondata/core/indexstore/row/DataMapRow.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/row/DataMapRow.java b/core/src/main/java/org/apache/carbondata/core/indexstore/row/DataMapRow.java index b8b46ef..e9ce170 100644 --- a/core/src/main/java/org/apache/carbondata/core/indexstore/row/DataMapRow.java +++ b/core/src/main/java/org/apache/carbondata/core/indexstore/row/DataMapRow.java @@ -26,7 +26,10 @@ import org.apache.carbondata.core.indexstore.schema.CarbonRowSchema; */ public abstract class DataMapRow implements Serializable { - protected CarbonRowSchema[] schemas; + /** + * This is made transient as it is temporary and should not be serialized + */ + protected transient CarbonRowSchema[] schemas; public DataMapRow(CarbonRowSchema[] schemas) { this.schemas = schemas; @@ -101,4 +104,10 @@ public abstract class DataMapRow implements Serializable { public DataMapRow convertToSafeRow() { return this; } + + public void setSchemas(CarbonRowSchema[] schemas) { + if (null == this.schemas) { + this.schemas = schemas; + } + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/core/src/main/java/org/apache/carbondata/core/indexstore/schema/SchemaGenerator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/schema/SchemaGenerator.java b/core/src/main/java/org/apache/carbondata/core/indexstore/schema/SchemaGenerator.java index 310cabf..e256fd6 100644 --- a/core/src/main/java/org/apache/carbondata/core/indexstore/schema/SchemaGenerator.java +++ b/core/src/main/java/org/apache/carbondata/core/indexstore/schema/SchemaGenerator.java @@ -100,23 +100,22 @@ public class SchemaGenerator { * @throws MemoryException */ public static CarbonRowSchema[] createTaskSummarySchema(SegmentProperties segmentProperties, - byte[] schemaBinary, byte[] filePath, byte[] fileName, byte[] segmentId, - boolean storeBlockletCount) throws MemoryException { + boolean storeBlockletCount, boolean filePathToBeStored) throws MemoryException { List<CarbonRowSchema> taskMinMaxSchemas = new ArrayList<>(); // get MinMax Schema getMinMaxSchema(segmentProperties, taskMinMaxSchemas); - // for storing column schema - taskMinMaxSchemas - .add(new CarbonRowSchema.FixedCarbonRowSchema(DataTypes.BYTE_ARRAY, schemaBinary.length)); - // for storing file path - taskMinMaxSchemas - .add(new CarbonRowSchema.FixedCarbonRowSchema(DataTypes.BYTE_ARRAY, filePath.length)); // for storing file name taskMinMaxSchemas - .add(new CarbonRowSchema.FixedCarbonRowSchema(DataTypes.BYTE_ARRAY, fileName.length)); + .add(new CarbonRowSchema.VariableCarbonRowSchema(DataTypes.BYTE_ARRAY)); // for storing segmentid taskMinMaxSchemas - .add(new CarbonRowSchema.FixedCarbonRowSchema(DataTypes.BYTE_ARRAY, segmentId.length)); + .add(new CarbonRowSchema.VariableCarbonRowSchema(DataTypes.BYTE_ARRAY)); + // store path only in case of partition table or non transactional table + if (filePathToBeStored) { + // for storing file path + taskMinMaxSchemas + .add(new CarbonRowSchema.VariableCarbonRowSchema(DataTypes.BYTE_ARRAY)); + } // flag to check whether it is required to store blocklet count of each carbondata file as // binary in summary schema. This will be true when it is not a legacy store (>1.1 version) // and CACHE_LEVEL=BLOCK http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/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 0f11bb0..7e116d9 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 @@ -255,6 +255,8 @@ public abstract class AbstractQueryExecutor<E> implements QueryExecutor<E> { BlockletInfo blockletInfo, short blockletId) { TableBlockInfo info = blockInfo.copy(); BlockletDetailInfo detailInfo = info.getDetailInfo(); + // set column schema details + detailInfo.setColumnSchemas(fileFooter.getColumnInTable()); detailInfo.setRowCount(blockletInfo.getNumberOfRows()); byte[][] maxValues = blockletInfo.getBlockletIndex().getMinMaxIndex().getMaxValues(); byte[][] minValues = blockletInfo.getBlockletIndex().getMinMaxIndex().getMinValues(); http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/core/src/main/java/org/apache/carbondata/core/util/BlockletDataMapUtil.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/util/BlockletDataMapUtil.java b/core/src/main/java/org/apache/carbondata/core/util/BlockletDataMapUtil.java index d3857f0..9f4c6c5 100644 --- a/core/src/main/java/org/apache/carbondata/core/util/BlockletDataMapUtil.java +++ b/core/src/main/java/org/apache/carbondata/core/util/BlockletDataMapUtil.java @@ -17,6 +17,12 @@ package org.apache.carbondata.core.util; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInput; +import java.io.DataInputStream; +import java.io.DataOutput; +import java.io.DataOutputStream; import java.io.IOException; import java.math.BigDecimal; import java.nio.ByteBuffer; @@ -31,6 +37,7 @@ import java.util.TreeMap; import org.apache.carbondata.core.constants.CarbonCommonConstants; import org.apache.carbondata.core.datamap.Segment; import org.apache.carbondata.core.datastore.block.SegmentProperties; +import org.apache.carbondata.core.datastore.compression.CompressorFactory; import org.apache.carbondata.core.datastore.filesystem.AbstractDFSCarbonFile; import org.apache.carbondata.core.datastore.filesystem.CarbonFile; import org.apache.carbondata.core.datastore.impl.FileFactory; @@ -321,4 +328,43 @@ public class BlockletDataMapUtil { } return updatedValues; } + + /** + * Convert schema to binary + */ + public static byte[] convertSchemaToBinary(List<ColumnSchema> columnSchemas) throws IOException { + ByteArrayOutputStream stream = new ByteArrayOutputStream(); + DataOutput dataOutput = new DataOutputStream(stream); + dataOutput.writeShort(columnSchemas.size()); + for (ColumnSchema columnSchema : columnSchemas) { + if (columnSchema.getColumnReferenceId() == null) { + columnSchema.setColumnReferenceId(columnSchema.getColumnUniqueId()); + } + columnSchema.write(dataOutput); + } + byte[] byteArray = stream.toByteArray(); + // Compress with snappy to reduce the size of schema + return CompressorFactory.getInstance().getCompressor().compressByte(byteArray); + } + + /** + * Read column schema from binary + * + * @param schemaArray + * @throws IOException + */ + public static List<ColumnSchema> readColumnSchema(byte[] schemaArray) throws IOException { + // uncompress it. + schemaArray = CompressorFactory.getInstance().getCompressor().unCompressByte(schemaArray); + ByteArrayInputStream schemaStream = new ByteArrayInputStream(schemaArray); + DataInput schemaInput = new DataInputStream(schemaStream); + List<ColumnSchema> columnSchemas = new ArrayList<>(); + int size = schemaInput.readShort(); + for (int i = 0; i < size; i++) { + ColumnSchema columnSchema = new ColumnSchema(); + columnSchema.readFields(schemaInput); + columnSchemas.add(columnSchema); + } + return columnSchemas; + } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonFileFormat.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonFileFormat.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonFileFormat.scala index 697eec5..b0ad549 100644 --- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonFileFormat.scala +++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonFileFormat.scala @@ -260,7 +260,6 @@ class SparkCarbonFileFormat extends FileFormat val prunedBlocklets = dataMapExprWrapper.prune(segments, null) val detailInfo = prunedBlocklets.get(0).getDetailInfo - detailInfo.readColumnSchema(detailInfo.getColumnSchemaBinary) split.setDetailInfo(detailInfo) val carbonReader = if (readVector) { http://git-wip-us.apache.org/repos/asf/carbondata/blob/f4a58c54/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala index 1f8e359..1670d8a 100644 --- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala +++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala @@ -37,6 +37,7 @@ import org.apache.carbondata.common.logging.LogServiceFactory import org.apache.carbondata.core.cache.dictionary.ManageDictionaryAndBTree import org.apache.carbondata.core.constants.CarbonCommonConstants import org.apache.carbondata.core.datamap.DataMapStoreManager +import org.apache.carbondata.core.datastore.block.SegmentPropertiesAndSchemaHolder import org.apache.carbondata.core.datastore.impl.FileFactory import org.apache.carbondata.core.fileoperations.FileWriteOperation import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonMetadata, CarbonTableIdentifier} @@ -490,6 +491,7 @@ class CarbonFileMetastore extends CarbonMetaStore { val tableIdentifier = TableIdentifier(tableName, Option(dbName)) sparkSession.sessionState.catalog.refreshTable(tableIdentifier) DataMapStoreManager.getInstance().clearDataMaps(absoluteTableIdentifier) + SegmentPropertiesAndSchemaHolder.getInstance().invalidate(absoluteTableIdentifier) } else { if (!isTransactionalCarbonTable(absoluteTableIdentifier)) { removeTableFromMetadata(dbName, tableName) @@ -498,6 +500,7 @@ class CarbonFileMetastore extends CarbonMetaStore { val tableIdentifier = TableIdentifier(tableName, Option(dbName)) sparkSession.sessionState.catalog.refreshTable(tableIdentifier) DataMapStoreManager.getInstance().clearDataMaps(absoluteTableIdentifier) + SegmentPropertiesAndSchemaHolder.getInstance().invalidate(absoluteTableIdentifier) } } }
