| /* |
| * 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.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.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.TreeMap; |
| |
| import org.apache.carbondata.common.logging.LogServiceFactory; |
| import org.apache.carbondata.core.constants.CarbonCommonConstants; |
| import org.apache.carbondata.core.datastore.block.SegmentProperties; |
| import org.apache.carbondata.core.datastore.block.TableBlockInfo; |
| 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.filesystem.S3CarbonFile; |
| import org.apache.carbondata.core.datastore.impl.FileFactory; |
| import org.apache.carbondata.core.index.Segment; |
| import org.apache.carbondata.core.indexstore.BlockMetaInfo; |
| import org.apache.carbondata.core.indexstore.TableBlockIndexUniqueIdentifier; |
| import org.apache.carbondata.core.indexstore.TableBlockIndexUniqueIdentifierWrapper; |
| import org.apache.carbondata.core.indexstore.blockletindex.BlockletIndexFactory; |
| import org.apache.carbondata.core.indexstore.blockletindex.BlockletIndexInputSplit; |
| import org.apache.carbondata.core.indexstore.blockletindex.SegmentIndexFileStore; |
| import org.apache.carbondata.core.metadata.blocklet.DataFileFooter; |
| import org.apache.carbondata.core.metadata.blocklet.index.BlockletMinMaxIndex; |
| import org.apache.carbondata.core.metadata.schema.table.CarbonTable; |
| import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn; |
| import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension; |
| import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure; |
| import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema; |
| import org.apache.carbondata.core.scan.executor.util.QueryUtil; |
| import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf; |
| import org.apache.carbondata.core.util.path.CarbonTablePath; |
| |
| import org.apache.commons.io.FilenameUtils; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.PathFilter; |
| import org.apache.log4j.Logger; |
| |
| public class BlockletIndexUtil { |
| |
| private static final Logger LOG = |
| LogServiceFactory.getLogService(BlockletIndexUtil.class.getName()); |
| |
| public static Set<TableBlockIndexUniqueIdentifier> getSegmentUniqueIdentifiers(Segment segment) |
| throws IOException { |
| Set<TableBlockIndexUniqueIdentifier> set = new HashSet<>(); |
| set.add(new TableBlockIndexUniqueIdentifier(segment.getSegmentNo())); |
| return set; |
| } |
| |
| public static Map<String, BlockMetaInfo> getBlockMetaInfoMap( |
| TableBlockIndexUniqueIdentifierWrapper identifierWrapper, |
| SegmentIndexFileStore indexFileStore, Set<String> filesRead, |
| Map<String, BlockMetaInfo> fileNameToMetaInfoMapping, List<DataFileFooter> indexInfos) |
| throws IOException { |
| boolean isTransactionalTable = true; |
| TableBlockIndexUniqueIdentifier identifier = |
| identifierWrapper.getTableBlockIndexUniqueIdentifier(); |
| List<ColumnSchema> tableColumnList = null; |
| if (identifier.getMergeIndexFileName() != null |
| && indexFileStore.getFileData(identifier.getIndexFileName()) == null) { |
| CarbonFile indexMergeFile = FileFactory.getCarbonFile( |
| identifier.getIndexFilePath() + CarbonCommonConstants.FILE_SEPARATOR + identifier |
| .getMergeIndexFileName(), identifierWrapper.getConfiguration()); |
| if (indexMergeFile.exists() && !filesRead.contains(indexMergeFile.getPath())) { |
| indexFileStore.readAllIIndexOfSegment(new CarbonFile[] { indexMergeFile }); |
| filesRead.add(indexMergeFile.getPath()); |
| } |
| } |
| if (indexFileStore.getFileData(identifier.getIndexFileName()) == null) { |
| indexFileStore.readAllIIndexOfSegment(new CarbonFile[] { FileFactory.getCarbonFile( |
| identifier.getIndexFilePath() + CarbonCommonConstants.FILE_SEPARATOR + identifier |
| .getIndexFileName(), identifierWrapper.getConfiguration()) }); |
| } |
| Map<String, BlockMetaInfo> blockMetaInfoMap = new HashMap<>(); |
| CarbonTable carbonTable = identifierWrapper.getCarbonTable(); |
| if (carbonTable != null) { |
| isTransactionalTable = carbonTable.getTableInfo().isTransactionalTable(); |
| tableColumnList = |
| carbonTable.getTableInfo().getFactTable().getListOfColumns(); |
| } |
| DataFileFooterConverter fileFooterConverter = |
| new DataFileFooterConverter(identifierWrapper.getConfiguration()); |
| List<DataFileFooter> indexInfo = fileFooterConverter.getIndexInfo( |
| identifier.getIndexFilePath() + CarbonCommonConstants.FILE_SEPARATOR + identifier |
| .getIndexFileName(), indexFileStore.getFileData(identifier.getIndexFileName()), |
| isTransactionalTable); |
| indexInfos.addAll(indexInfo); |
| for (DataFileFooter footer : indexInfo) { |
| if ((!isTransactionalTable) && (tableColumnList.size() != 0) && |
| !isSameColumnAndDifferentDatatypeInSchema(footer.getColumnInTable(), tableColumnList)) { |
| LOG.error("Datatype of the common columns present in " + identifier.getIndexFileName() |
| + " doesn't match with the column's datatype in table schema"); |
| throw new IOException("All common columns present in the files doesn't have same datatype. " |
| + "Unsupported operation on nonTransactional table. Check logs."); |
| } |
| if ((tableColumnList != null) && (tableColumnList.size() == 0)) { |
| // Carbon reader have used dummy columnSchema. Update it with inferred schema now |
| carbonTable.getTableInfo().getFactTable().setListOfColumns(footer.getColumnInTable()); |
| CarbonTable.updateTableByTableInfo(carbonTable, carbonTable.getTableInfo()); |
| } |
| String blockPath = footer.getBlockInfo().getFilePath(); |
| if (null == blockMetaInfoMap.get(blockPath)) { |
| BlockMetaInfo blockMetaInfo = createBlockMetaInfo( |
| fileNameToMetaInfoMapping, footer.getBlockInfo()); |
| // if blockMetaInfo is null that means the file has been deleted from the file system. |
| // This can happen in case IUD scenarios where after deleting or updating the data the |
| // complete block is deleted but the entry still exists in index or merge index file |
| if (null != blockMetaInfo) { |
| blockMetaInfoMap.put(blockPath, blockMetaInfo); |
| } |
| } |
| } |
| return blockMetaInfoMap; |
| } |
| |
| /** |
| * This method will create file name to block Meta Info Mapping. This method will reduce the |
| * number of nameNode calls and using this method one namenode will fetch 1000 entries |
| * |
| * @param segmentFilePath |
| * @return |
| * @throws IOException |
| */ |
| public static Map<String, BlockMetaInfo> createCarbonDataFileBlockMetaInfoMapping( |
| String segmentFilePath, Configuration configuration) throws IOException { |
| Map<String, BlockMetaInfo> fileNameToMetaInfoMapping = new TreeMap(); |
| CarbonFile carbonFile = FileFactory.getCarbonFile(segmentFilePath, configuration); |
| if (carbonFile instanceof AbstractDFSCarbonFile && !(carbonFile instanceof S3CarbonFile)) { |
| PathFilter pathFilter = new PathFilter() { |
| @Override |
| public boolean accept(Path path) { |
| return CarbonTablePath.isCarbonDataFile(path.getName()); |
| } |
| }; |
| CarbonFile[] carbonFiles = carbonFile.locationAwareListFiles(pathFilter); |
| for (CarbonFile file : carbonFiles) { |
| String[] location = file.getLocations(); |
| long len = file.getSize(); |
| BlockMetaInfo blockMetaInfo = new BlockMetaInfo(location, len); |
| fileNameToMetaInfoMapping.put(file.getPath(), blockMetaInfo); |
| } |
| } |
| return fileNameToMetaInfoMapping; |
| } |
| |
| private static BlockMetaInfo createBlockMetaInfo( |
| Map<String, BlockMetaInfo> fileNameToMetaInfoMapping, TableBlockInfo blockInfo) |
| throws IOException { |
| String carbonDataFile = blockInfo.getFilePath(); |
| FileFactory.FileType fileType = FileFactory.getFileType(carbonDataFile); |
| switch (fileType) { |
| case S3: |
| case LOCAL: |
| // consider backward compatibility |
| // when the file size in blockInfo is not zero, use this file size in blockInfo. |
| if (blockInfo.getFileSize() != 0) { |
| return new BlockMetaInfo(new String[] { "localhost" }, blockInfo.getFileSize()); |
| } |
| // when the file size in blockInfo is zero, get the size of this file. |
| if (!FileFactory.isFileExist(carbonDataFile)) { |
| return null; |
| } |
| CarbonFile carbonFile = FileFactory.getCarbonFile(carbonDataFile); |
| return new BlockMetaInfo(new String[] { "localhost" }, carbonFile.getSize()); |
| default: |
| return fileNameToMetaInfoMapping.get(FileFactory.getFormattedPath(carbonDataFile)); |
| } |
| } |
| |
| public static Set<TableBlockIndexUniqueIdentifier> getTableBlockUniqueIdentifiers(Segment segment) |
| throws IOException { |
| Set<TableBlockIndexUniqueIdentifier> tableBlockIndexUniqueIdentifiers = new HashSet<>(); |
| Map<String, String> indexFiles = segment.getCommittedIndexFile(); |
| for (Map.Entry<String, String> indexFileEntry : indexFiles.entrySet()) { |
| String indexFile = indexFileEntry.getKey(); |
| tableBlockIndexUniqueIdentifiers.add( |
| new TableBlockIndexUniqueIdentifier(FilenameUtils.getFullPathNoEndSeparator(indexFile), |
| FilenameUtils.getName(indexFile), indexFileEntry.getValue(), segment.getSegmentNo())); |
| } |
| return tableBlockIndexUniqueIdentifiers; |
| } |
| |
| /** |
| * This method will filter out the TableBlockIndexUniqueIdentifier belongs to that distributable |
| * |
| * @param tableBlockIndexUniqueIdentifiers |
| * @param distributable |
| * @return |
| */ |
| public static TableBlockIndexUniqueIdentifier filterIdentifiersBasedOnDistributable( |
| Set<TableBlockIndexUniqueIdentifier> tableBlockIndexUniqueIdentifiers, |
| BlockletIndexInputSplit distributable) { |
| TableBlockIndexUniqueIdentifier validIdentifier = null; |
| String fileName = CarbonTablePath.DataFileUtil.getFileName(distributable.getFilePath()); |
| for (TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier : |
| tableBlockIndexUniqueIdentifiers) { |
| if (fileName.equals(tableBlockIndexUniqueIdentifier.getIndexFileName())) { |
| validIdentifier = tableBlockIndexUniqueIdentifier; |
| break; |
| } |
| } |
| return validIdentifier; |
| } |
| |
| /** |
| * This method will the index files tableBlockIndexUniqueIdentifiers of a merge index file |
| * |
| * @param identifier |
| * @param segmentIndexFileStore |
| * @return |
| * @throws IOException |
| */ |
| public static List<TableBlockIndexUniqueIdentifier> getIndexFileIdentifiersFromMergeFile( |
| TableBlockIndexUniqueIdentifier identifier, SegmentIndexFileStore segmentIndexFileStore) |
| throws IOException { |
| List<TableBlockIndexUniqueIdentifier> tableBlockIndexUniqueIdentifiers = new ArrayList<>(); |
| String mergeFilePath = |
| identifier.getIndexFilePath() + CarbonCommonConstants.FILE_SEPARATOR + identifier |
| .getIndexFileName(); |
| segmentIndexFileStore.readMergeFile(mergeFilePath); |
| List<String> indexFiles = |
| segmentIndexFileStore.getCarbonMergeFileToIndexFilesMap().get(mergeFilePath); |
| for (String indexFile : indexFiles) { |
| tableBlockIndexUniqueIdentifiers.add( |
| new TableBlockIndexUniqueIdentifier(identifier.getIndexFilePath(), indexFile, |
| identifier.getIndexFileName(), identifier.getSegmentId())); |
| } |
| return tableBlockIndexUniqueIdentifiers; |
| } |
| |
| /** |
| * Method to check if CACHE_LEVEL is set to BLOCK or BLOCKLET |
| */ |
| public static boolean isCacheLevelBlock(CarbonTable carbonTable) { |
| String cacheLevel = carbonTable.getTableInfo().getFactTable().getTableProperties() |
| .get(CarbonCommonConstants.CACHE_LEVEL); |
| if (BlockletIndexFactory.CACHE_LEVEL_BLOCKLET.equals(cacheLevel)) { |
| return false; |
| } |
| return true; |
| } |
| |
| /** |
| * This method validates whether the schema present in index and table contains the same column |
| * name but with different dataType. |
| */ |
| public static boolean isSameColumnAndDifferentDatatypeInSchema( |
| List<ColumnSchema> indexFileColumnList, List<ColumnSchema> tableColumnList) |
| throws IOException { |
| for (int i = 0; i < tableColumnList.size(); i++) { |
| for (int j = 0; j < indexFileColumnList.size(); j++) { |
| if (indexFileColumnList.get(j).getColumnName() |
| .equalsIgnoreCase(tableColumnList.get(i).getColumnName()) && !indexFileColumnList.get(j) |
| .getDataType().getName() |
| .equalsIgnoreCase(tableColumnList.get(i).getDataType().getName())) { |
| if ("varchar".equalsIgnoreCase(indexFileColumnList.get(j).getDataType().getName()) && |
| "string".equalsIgnoreCase(tableColumnList.get(i).getDataType().getName())) { |
| throw new IOException("Datatype of the Column " |
| + indexFileColumnList.get(j).getDataType().getName() |
| + " present in index file, is varchar and not same as datatype of the column " + |
| "with same name present in table, " + |
| "because carbon convert varchar of carbon to string of spark, " + |
| "please set long_string_columns for varchar column: " |
| + tableColumnList.get(i).getColumnName()); |
| } |
| LOG.error("Datatype of the Column " + indexFileColumnList.get(j).getColumnName() |
| + " present in index file, is not same as datatype of the column with same name" |
| + "present in table"); |
| return false; |
| } |
| } |
| } |
| return true; |
| } |
| |
| /** |
| * 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 to reduce the size of schema |
| ByteBuffer byteBuffer = |
| CompressorFactory.NativeSupportedCompressor.SNAPPY.getCompressor().compressByte(byteArray); |
| return byteBuffer.array(); |
| } |
| |
| /** |
| * Read column schema from binary |
| * |
| * @param schemaArray |
| * @throws IOException |
| */ |
| public static List<ColumnSchema> readColumnSchema(byte[] schemaArray) throws IOException { |
| // uncompress it. |
| schemaArray = CompressorFactory.NativeSupportedCompressor.SNAPPY.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; |
| } |
| |
| /** |
| * Method to get the min/max values for columns to be cached |
| * |
| * @param segmentProperties |
| * @param minMaxCacheColumns |
| * @param minMaxValuesForAllColumns |
| * @return |
| */ |
| public static byte[][] getMinMaxForColumnsToBeCached(SegmentProperties segmentProperties, |
| List<CarbonColumn> minMaxCacheColumns, byte[][] minMaxValuesForAllColumns) { |
| byte[][] minMaxValuesForColumnsToBeCached = minMaxValuesForAllColumns; |
| if (null != minMaxCacheColumns) { |
| minMaxValuesForColumnsToBeCached = new byte[minMaxCacheColumns.size()][]; |
| int counter = 0; |
| for (CarbonColumn column : minMaxCacheColumns) { |
| minMaxValuesForColumnsToBeCached[counter++] = |
| minMaxValuesForAllColumns[getColumnOrdinal(segmentProperties, column)]; |
| } |
| } |
| return minMaxValuesForColumnsToBeCached; |
| } |
| |
| /** |
| * Method to get the flag values for columns to be cached |
| * |
| * @param segmentProperties |
| * @param minMaxCacheColumns |
| * @param minMaxFlag |
| * @return |
| */ |
| public static boolean[] getMinMaxFlagValuesForColumnsToBeCached( |
| SegmentProperties segmentProperties, List<CarbonColumn> minMaxCacheColumns, |
| boolean[] minMaxFlag) { |
| boolean[] minMaxFlagValuesForColumnsToBeCached = minMaxFlag; |
| if (null != minMaxCacheColumns) { |
| minMaxFlagValuesForColumnsToBeCached = new boolean[minMaxCacheColumns.size()]; |
| int counter = 0; |
| for (CarbonColumn column : minMaxCacheColumns) { |
| minMaxFlagValuesForColumnsToBeCached[counter++] = |
| minMaxFlag[getColumnOrdinal(segmentProperties, column)]; |
| } |
| } |
| return minMaxFlagValuesForColumnsToBeCached; |
| } |
| |
| /** |
| * compute the column ordinal as per data is stored |
| * |
| * @param segmentProperties |
| * @param column |
| * @return |
| */ |
| public static int getColumnOrdinal(SegmentProperties segmentProperties, CarbonColumn column) { |
| if (column.isMeasure()) { |
| // as measures are stored at the end after all dimensions and complex dimensions hence add |
| // the last dimension ordinal to measure ordinal. Segment properties will store min max |
| // length in one array on the order normal dimension, complex dimension and then measure |
| return segmentProperties.getLastDimensionColOrdinal() + column.getOrdinal(); |
| } else { |
| return column.getOrdinal(); |
| } |
| } |
| |
| /** |
| * Method to check whether to serialize min/max values to executor. Returns true if |
| * filter column min/max is not cached in driver |
| * |
| * @param filterResolverTree |
| * @param minMaxCacheColumns |
| * @return |
| */ |
| public static boolean useMinMaxForBlockletPruning(FilterResolverIntf filterResolverTree, |
| List<CarbonColumn> minMaxCacheColumns) { |
| boolean serializeMinMax = false; |
| if (null != minMaxCacheColumns) { |
| Set<CarbonDimension> filterDimensions = new HashSet<>(); |
| Set<CarbonMeasure> filterMeasures = new HashSet<>(); |
| QueryUtil |
| .getAllFilterDimensionsAndMeasures(filterResolverTree, filterDimensions, filterMeasures); |
| // set flag to true if columns cached size is lesser than filter columns |
| if (minMaxCacheColumns.size() < (filterDimensions.size() + filterMeasures.size())) { |
| serializeMinMax = true; |
| } else { |
| // check if all the filter dimensions are cached |
| for (CarbonDimension filterDimension : filterDimensions) { |
| // complex dimensions are not allowed to be specified in COLUMN_META_CACHE property, so |
| // cannot validate for complex columns |
| if (filterDimension.isComplex()) { |
| continue; |
| } |
| if (!filterColumnExistsInMinMaxColumnList(minMaxCacheColumns, filterDimension)) { |
| serializeMinMax = true; |
| break; |
| } |
| } |
| // check if all the filter measures are cached only if all filter dimensions are cached |
| if (!serializeMinMax) { |
| for (CarbonMeasure filterMeasure : filterMeasures) { |
| if (!filterColumnExistsInMinMaxColumnList(minMaxCacheColumns, filterMeasure)) { |
| serializeMinMax = true; |
| break; |
| } |
| } |
| } |
| } |
| } |
| return serializeMinMax; |
| } |
| |
| /** |
| * Method to check for filter column in min/max cache columns list |
| * |
| * @param minMaxCacheColumns |
| * @param filterColumn |
| * @return |
| */ |
| private static boolean filterColumnExistsInMinMaxColumnList(List<CarbonColumn> minMaxCacheColumns, |
| CarbonColumn filterColumn) { |
| for (CarbonColumn column : minMaxCacheColumns) { |
| if (filterColumn.getColumnId().equalsIgnoreCase(column.getColumnId())) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| /** |
| * Method to update the min max flag. For CACHE_LEVEL=BLOCK, for any column if min max is not |
| * written in any of the blocklet then for that column the flag will be false for the |
| * complete block |
| * |
| * @param minMaxIndex |
| * @param minMaxFlag |
| */ |
| public static void updateMinMaxFlag(BlockletMinMaxIndex minMaxIndex, boolean[] minMaxFlag) { |
| boolean[] isMinMaxSet = minMaxIndex.getIsMinMaxSet(); |
| if (null != isMinMaxSet) { |
| for (int i = 0; i < minMaxFlag.length; i++) { |
| if (!isMinMaxSet[i]) { |
| minMaxFlag[i] = isMinMaxSet[i]; |
| } |
| } |
| } |
| } |
| |
| /** |
| * Validate whether load indexes parallel is SET or not |
| * |
| * @param carbonTable |
| * @return |
| */ |
| public static boolean loadIndexesParallel(CarbonTable carbonTable) { |
| String parentTableName = carbonTable.getParentTableName(); |
| String tableName; |
| String dbName; |
| if (!parentTableName.isEmpty()) { |
| // if the table is index table, then check the property on parent table name |
| // as index table is a child of the main table |
| tableName = parentTableName; |
| } else { |
| // if it is a normal carbon table, then check on the table name |
| tableName = carbonTable.getTableName(); |
| } |
| dbName = carbonTable.getDatabaseName(); |
| return CarbonProperties.getInstance().isIndexParallelLoadingEnabled(dbName, tableName); |
| } |
| } |
