index/bloom/src/main/java/org/apache/carbondata/index/bloom/BloomIndexFileStore.java - carbondata - Git at Google

 /*
  * 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);
   }
 }
	/*
	* 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);
	}
	}