[GitHub] carbondata pull request #2244: [CARBONDATA-2310] Refactored code to improve ...

[dhatchayani <gi...@git.apache.org>]

Github user dhatchayani commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/2244#discussion_r185435135
  
    --- Diff: core/src/main/java/org/apache/carbondata/core/util/BlockletDataMapUtil.java ---
    @@ -0,0 +1,198 @@
    +/*
    + * 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.IOException;
    +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.LogService;
    +import org.apache.carbondata.common.logging.LogServiceFactory;
    +import org.apache.carbondata.core.constants.CarbonCommonConstants;
    +import org.apache.carbondata.core.datamap.Segment;
    +import org.apache.carbondata.core.datastore.filesystem.AbstractDFSCarbonFile;
    +import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
    +import org.apache.carbondata.core.datastore.impl.FileFactory;
    +import org.apache.carbondata.core.indexstore.BlockMetaInfo;
    +import org.apache.carbondata.core.indexstore.TableBlockIndexUniqueIdentifier;
    +import org.apache.carbondata.core.indexstore.blockletindex.BlockletDataMapDistributable;
    +import org.apache.carbondata.core.indexstore.blockletindex.SegmentIndexFileStore;
    +import org.apache.carbondata.core.metadata.blocklet.DataFileFooter;
    +import org.apache.carbondata.core.util.path.CarbonTablePath;
    +
    +import org.apache.hadoop.fs.LocatedFileStatus;
    +import org.apache.hadoop.fs.Path;
    +import org.apache.hadoop.fs.PathFilter;
    +import org.apache.hadoop.fs.RemoteIterator;
    +
    +
    +public class BlockletDataMapUtil {
    +
    +  private static final LogService LOGGER =
    +      LogServiceFactory.getLogService(BlockletDataMapUtil.class.getName());
    +
    +  public static Map<String, BlockMetaInfo> getBlockMetaInfoMap(
    +      TableBlockIndexUniqueIdentifier identifier, SegmentIndexFileStore indexFileStore,
    +      Set<String> filesRead, Map<String, BlockMetaInfo> fileNameToMetaInfoMapping)
    +      throws IOException {
    +    if (identifier.getMergeIndexFileName() != null
    +        && indexFileStore.getFileData(identifier.getIndexFileName()) == null) {
    +      CarbonFile indexMergeFile = FileFactory.getCarbonFile(
    +          identifier.getIndexFilePath() + CarbonCommonConstants.FILE_SEPARATOR + identifier
    +              .getMergeIndexFileName());
    +      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()) });
    +    }
    +    DataFileFooterConverter fileFooterConverter = new DataFileFooterConverter();
    +    Map<String, BlockMetaInfo> blockMetaInfoMap = new HashMap<>();
    +    List<DataFileFooter> indexInfo = fileFooterConverter.getIndexInfo(
    +        identifier.getIndexFilePath() + CarbonCommonConstants.FILE_SEPARATOR + identifier
    +            .getIndexFileName(), indexFileStore.getFileData(identifier.getIndexFileName()));
    +    for (DataFileFooter footer : indexInfo) {
    +      String blockPath = footer.getBlockInfo().getTableBlockInfo().getFilePath();
    +      if (FileFactory.isFileExist(blockPath)) {
    --- End diff --
    
    This check is required in case of update/delete operation. 


---