content: /*
   * 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.index.bloom;
  
  import java.io.ByteArrayInputStream;
  import java.io.DataInputStream;
  import java.io.DataOutputStream;
  import java.io.File;
  import java.io.IOException;
  import java.nio.charset.Charset;
  import java.util.ArrayList;
  import java.util.List;
  
  import org.apache.carbondata.common.annotations.InterfaceAudience;
  import org.apache.carbondata.common.logging.LogServiceFactory;
  import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
  import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
  import org.apache.carbondata.core.datastore.impl.FileFactory;
  import org.apache.carbondata.core.util.CarbonUtil;
  
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.util.bloom.CarbonBloomFilter;
  import org.apache.log4j.Logger;
  
  /**
   * This class works for merging and loading bloom index
   */
  @InterfaceAudience.Internal
  public class BloomIndexFileStore {
  
    private static final Logger LOGGER =
            LogServiceFactory.getLogService(BloomIndexFileStore.class.getName());
  
    // suffix of original generated file
    public static final String BLOOM_INDEX_SUFFIX = ".bloomindex";
    // suffix of merged bloom index file
    public static final String MERGE_BLOOM_INDEX_SUFFIX = ".bloomindexmerge";
    // directory to store merged bloom index files
    public static final String MERGE_BLOOM_INDEX_SHARD_NAME = "mergeShard";
    /**
     * flag file for merging
     * if flag file exists, query won't use mergeShard
     * if flag file not exists and mergeShard generated, query will use mergeShard
     */
    public static final String MERGE_INPROGRESS_FILE = "mergeShard.inprogress";
  
    public static void mergeBloomIndexFile(String dmSegmentPathString, List<String> indexCols) {
  
      // Step 1. check current folders
  
      // get all shard paths of old store
      CarbonFile segmentPath = FileFactory.getCarbonFile(dmSegmentPathString);
      CarbonFile[] shardPaths = segmentPath.listFiles(new CarbonFileFilter() {
        @Override
        public boolean accept(CarbonFile file) {
          return file.isDirectory() && !file.getName().equals(MERGE_BLOOM_INDEX_SHARD_NAME);
        }
      });
  
      String mergeShardPath = dmSegmentPathString + File.separator + MERGE_BLOOM_INDEX_SHARD_NAME;
      String mergeInprogressFile = dmSegmentPathString + File.separator + MERGE_INPROGRESS_FILE;
  
      // Step 2. prepare for fail-safe merging
  
      try {
        // delete mergeShard folder if exists
        if (FileFactory.isFileExist(mergeShardPath)) {
          FileFactory.deleteFile(mergeShardPath);
        }
        // create flag file before creating mergeShard folder
        if (!FileFactory.isFileExist(mergeInprogressFile)) {
          FileFactory.createNewFile(mergeInprogressFile);
        }
        // create mergeShard output folder
        if (!FileFactory.mkdirs(mergeShardPath)) {
          throw new RuntimeException("Failed to create directory " + mergeShardPath);
        }
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
  
      // Step 3. merge index files
      // Query won't use mergeShard until MERGE_INPROGRESS_FILE is deleted
  
      // for each index column, merge the bloomindex files from all shards into one
      for (String indexCol: indexCols) {
        String mergeIndexFile = getMergeBloomIndexFile(mergeShardPath, indexCol);
        DataInputStream dataInputStream = null;
        DataOutputStream dataOutputStream = null;
        try {
          FileFactory.createNewFile(mergeIndexFile);
          dataOutputStream = FileFactory.getDataOutputStream(
              mergeIndexFile);
          for (CarbonFile shardPath : shardPaths) {
            String bloomIndexFile = getBloomIndexFile(shardPath.getCanonicalPath(), indexCol);
            dataInputStream = FileFactory.getDataInputStream(bloomIndexFile);
            byte[] fileData = new byte[(int) FileFactory.getCarbonFile(bloomIndexFile).getSize()];
            dataInputStream.readFully(fileData);
            byte[] shardName = shardPath.getName().getBytes(Charset.forName("UTF-8"));
            dataOutputStream.writeInt(shardName.length);
            dataOutputStream.write(shardName);
            dataOutputStream.writeInt(fileData.length);
            dataOutputStream.write(fileData);
            CarbonUtil.closeStream(dataInputStream);
          }
        } catch (IOException e) {
          LOGGER.error("Error occurs while merge bloom index file of column: " + indexCol, e);
          // if any column failed, delete merge shard for this segment and exit
          FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(mergeShardPath));
          FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(mergeInprogressFile));
          throw new RuntimeException(
              "Error occurs while merge bloom index file of column: " + indexCol, e);
        } finally {
          CarbonUtil.closeStreams(dataInputStream, dataOutputStream);
        }
      }
  
      // Step 4. delete flag file and mergeShard can be used
      try {
        FileFactory.deleteFile(mergeInprogressFile);
      } catch (IOException e) {
        LOGGER.error("Error occurs while deleting file " + mergeInprogressFile, e);
        throw new RuntimeException("Error occurs while deleting file " + mergeInprogressFile);
      }
      // remove old store
      for (CarbonFile shardpath: shardPaths) {
        FileFactory.deleteAllCarbonFilesOfDir(shardpath);
      }
    }
  
    /**
     * load bloom filter from bloom index file
     */
    public static List<CarbonBloomFilter> loadBloomFilterFromFile(
            String shardPath, String colName) {
      if (shardPath.endsWith(MERGE_BLOOM_INDEX_SHARD_NAME)) {
        return loadMergeBloomIndex(shardPath, colName);
      } else {
        return loadBloomIndex(shardPath, colName);
      }
    }
  
    /**
     * load bloom filter of {@code colName} from {@code shardPath}
     */
    public static List<CarbonBloomFilter> loadBloomIndex(
            String shardPath, String colName) {
      DataInputStream dataInStream = null;
      List<CarbonBloomFilter> bloomFilters = new ArrayList<>();
      try {
        String indexFile = getBloomIndexFile(shardPath, colName);
        dataInStream = FileFactory.getDataInputStream(indexFile);
        while (dataInStream.available() > 0) {
          CarbonBloomFilter bloomFilter = new CarbonBloomFilter();
          bloomFilter.readFields(dataInStream);
          bloomFilter.setShardName(new Path(shardPath).getName());
          bloomFilters.add(bloomFilter);
        }
        LOGGER.info(String.format("Read %d bloom indices from %s", bloomFilters.size(), indexFile));
  
        return bloomFilters;
      } catch (IOException e) {
        LOGGER.error("Error occurs while reading bloom index", e);
        throw new RuntimeException("Error occurs while reading bloom index", e);
      } finally {
        CarbonUtil.closeStreams(dataInStream);
      }
    }
  
    /**
     * load bloom filter of {@code colName} from {@code mergeShardPath}
     */
    public static List<CarbonBloomFilter> loadMergeBloomIndex(
            String mergeShardPath, String colName) {
      String mergeIndexFile = getMergeBloomIndexFile(mergeShardPath, colName);
      DataInputStream mergeIndexInStream = null;
      List<CarbonBloomFilter> bloomFilters = new ArrayList<>();
      try {
        mergeIndexInStream = FileFactory.getDataInputStream(mergeIndexFile);
        while (mergeIndexInStream.available() > 0) {
          // read shard name
          int shardNameByteLength = mergeIndexInStream.readInt();
          byte[] shardNameBytes = new byte[shardNameByteLength];
          mergeIndexInStream.readFully(shardNameBytes);
          String shardName = new String(shardNameBytes, Charset.forName("UTF-8"));
          // read bloom index file data
          int indexFileByteLength = mergeIndexInStream.readInt();
          byte[] indexFileBytes = new byte[indexFileByteLength];
          mergeIndexInStream.readFully(indexFileBytes);
          // warp byte array as input stream to get bloom filters
          ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(indexFileBytes);
          DataInputStream indexDataInStream = new DataInputStream(byteArrayInputStream);
          while (indexDataInStream.available() > 0) {
            CarbonBloomFilter bloomFilter = new CarbonBloomFilter();
            bloomFilter.readFields(indexDataInStream);
            bloomFilter.setShardName(shardName);
            bloomFilters.add(bloomFilter);
          }
        }
        LOGGER.info(
            String.format("Read %d bloom indices from %s", bloomFilters.size(), mergeIndexFile));
        return bloomFilters;
      } catch (IOException e) {
        LOGGER.error("Error occurs while reading merge bloom index", e);
        throw new RuntimeException("Error occurs while reading merge bloom index", e);
      } finally {
        CarbonUtil.closeStreams(mergeIndexInStream);
      }
    }
  
    /**
     * get bloom index file
     */
    public static String getBloomIndexFile(String shardPath, String colName) {
      return shardPath.concat(File.separator).concat(colName).concat(BLOOM_INDEX_SUFFIX);
    }
  
    /**
     * get merge bloom index file
     */
    public static String getMergeBloomIndexFile(String mergeShardPath, String colName) {
      return mergeShardPath.concat(File.separator).concat(colName).concat(MERGE_BLOOM_INDEX_SUFFIX);
    }
  }
  