apache / carbondata / 830

Build # 830

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/3f4ceffbf1992941ed16c8ea7ad1da9c70d588de">3f4ceffbf<a href="https://github.com/apache/carbondata/commit/3f4ceffbf1992941ed16c8ea7ad1da9c70d588de">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/5f3c1ebf6c7c73bf264b312a8de89d7d95665a51">5f3c1ebf6</a>">5f3c1ebf6</a><a href="https://github.com/apache/carbondata/commit/3f4ceffbf1992941ed16c8ea7ad1da9c70d588de"> into c01636163">c01636163</a>

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

60487 of 75063 relevant lines covered (80.58%)

1.06 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.92

/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomIndexFileStore.java

/*
 * 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.datamap.bloom;

import java.io.*;
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.LogService;
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;

/**
 * This class works for merging and loading bloom index
 */
@InterfaceAudience.Internal
public class BloomIndexFileStore {

  private static final LogService 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) {
    // get all shard paths of old store
    CarbonFile segmentPath = FileFactory.getCarbonFile(dmSegmentPathString,
            FileFactory.getFileType(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;
    try {
      // delete mergeShard folder if exists
      if (FileFactory.isFileExist(mergeShardPath)) {
        FileFactory.deleteFile(mergeShardPath, FileFactory.getFileType(mergeShardPath));
      }
      // create flag file before creating mergeShard folder
      if (!FileFactory.isFileExist(mergeInprogressFile)) {
        FileFactory.createNewFile(
            mergeInprogressFile, FileFactory.getFileType(mergeInprogressFile));
      }
      // create mergeShard output folder
      if (!FileFactory.mkdirs(mergeShardPath, FileFactory.getFileType(mergeShardPath))) {
        throw new RuntimeException("Failed to create directory " + mergeShardPath);
      }
    } catch (IOException e) {
      LOGGER.error(e, "Error occurs while create directory " + mergeShardPath);
      throw new RuntimeException("Error occurs while create directory " + mergeShardPath);
    }

    // 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, FileFactory.getFileType(mergeIndexFile));
        dataOutputStream = FileFactory.getDataOutputStream(
            mergeIndexFile, FileFactory.getFileType(mergeIndexFile));
        for (CarbonFile shardPath : shardPaths) {
          String bloomIndexFile = getBloomIndexFile(shardPath.getCanonicalPath(), indexCol);
          dataInputStream = FileFactory.getDataInputStream(
              bloomIndexFile, FileFactory.getFileType(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(e, "Error occurs while merge bloom index file of column: " + indexCol);
        // delete merge shard of bloom index for this segment when failed
        FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(mergeShardPath));
        throw new RuntimeException(
            "Error occurs while merge bloom index file of column: " + indexCol);
      } finally {
        CarbonUtil.closeStreams(dataInputStream, dataOutputStream);
      }
    }
    // delete flag file and mergeShard can be used
    try {
      FileFactory.deleteFile(mergeInprogressFile, FileFactory.getFileType(mergeInprogressFile));
    } catch (IOException e) {
      LOGGER.error(e, "Error occurs while deleting file " + mergeInprogressFile);
      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, FileFactory.getFileType(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(e, "Error occurs while reading bloom index");
      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, FileFactory.getFileType(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(e, "Error occurs while reading merge bloom index");
      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);
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17	package org.apache.carbondata.datamap.bloom;
18
19	import java.io.*;
20	import java.nio.charset.Charset;
21	import java.util.ArrayList;
22	import java.util.List;
23
24	import org.apache.carbondata.common.annotations.InterfaceAudience;
25	import org.apache.carbondata.common.logging.LogService;
26	import org.apache.carbondata.common.logging.LogServiceFactory;
27	import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
28	import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
29	import org.apache.carbondata.core.datastore.impl.FileFactory;
30	import org.apache.carbondata.core.util.CarbonUtil;
31
32	import org.apache.hadoop.fs.Path;
33	import org.apache.hadoop.util.bloom.CarbonBloomFilter;
34
35	/**
36	* This class works for merging and loading bloom index
37	*/
38	@InterfaceAudience.Internal
39	public class BloomIndexFileStore {	×
40
41	private static final LogService LOGGER =	1✔
42	LogServiceFactory.getLogService(BloomIndexFileStore.class.getName());	1✔
43
44	// suffix of original generated file
45	public static final String BLOOM_INDEX_SUFFIX = ".bloomindex";
46	// suffix of merged bloom index file
47	public static final String MERGE_BLOOM_INDEX_SUFFIX = ".bloomindexmerge";
48	// directory to store merged bloom index files
49	public static final String MERGE_BLOOM_INDEX_SHARD_NAME = "mergeShard";
50	/**
51	* flag file for merging
52	* if flag file exists, query won't use mergeShard
53	* if flag file not exists and mergeShard generated, query will use mergeShard
54	*/
55	public static final String MERGE_INPROGRESS_FILE = "mergeShard.inprogress";
56
57
58	public static void mergeBloomIndexFile(String dmSegmentPathString, List<String> indexCols) {
59	// get all shard paths of old store
60	CarbonFile segmentPath = FileFactory.getCarbonFile(dmSegmentPathString,	1✔
61	FileFactory.getFileType(dmSegmentPathString));	1✔
62	CarbonFile[] shardPaths = segmentPath.listFiles(new CarbonFileFilter() {	1✔
63	@Override
64	public boolean accept(CarbonFile file) {
65	return file.isDirectory() && !file.getName().equals(MERGE_BLOOM_INDEX_SHARD_NAME);	1✔
66	}
67	});
68
69	String mergeShardPath = dmSegmentPathString + File.separator + MERGE_BLOOM_INDEX_SHARD_NAME;	1✔
70	String mergeInprogressFile = dmSegmentPathString + File.separator + MERGE_INPROGRESS_FILE;	1✔
71	try {
72	// delete mergeShard folder if exists
73	if (FileFactory.isFileExist(mergeShardPath)) {	1✔
74	FileFactory.deleteFile(mergeShardPath, FileFactory.getFileType(mergeShardPath));	1✔
75	}
76	// create flag file before creating mergeShard folder
77	if (!FileFactory.isFileExist(mergeInprogressFile)) {	1✔
78	FileFactory.createNewFile(	1✔
79	mergeInprogressFile, FileFactory.getFileType(mergeInprogressFile));	1✔
80	}
81	// create mergeShard output folder
82	if (!FileFactory.mkdirs(mergeShardPath, FileFactory.getFileType(mergeShardPath))) {	1✔
83	throw new RuntimeException("Failed to create directory " + mergeShardPath);	×
84	}
85	} catch (IOException e) {	×
86	LOGGER.error(e, "Error occurs while create directory " + mergeShardPath);	×
87	throw new RuntimeException("Error occurs while create directory " + mergeShardPath);	×
88	}	1✔
89
90	// for each index column, merge the bloomindex files from all shards into one
91	for (String indexCol: indexCols) {	1✔
92	String mergeIndexFile = getMergeBloomIndexFile(mergeShardPath, indexCol);	1✔
93	DataInputStream dataInputStream = null;	1✔
94	DataOutputStream dataOutputStream = null;	1✔
95	try {
96	FileFactory.createNewFile(mergeIndexFile, FileFactory.getFileType(mergeIndexFile));	1✔
97	dataOutputStream = FileFactory.getDataOutputStream(	1✔
98	mergeIndexFile, FileFactory.getFileType(mergeIndexFile));	1✔
99	for (CarbonFile shardPath : shardPaths) {	1✔
100	String bloomIndexFile = getBloomIndexFile(shardPath.getCanonicalPath(), indexCol);	1✔
101	dataInputStream = FileFactory.getDataInputStream(	1✔
102	bloomIndexFile, FileFactory.getFileType(bloomIndexFile));	1✔
103	byte[] fileData = new byte[(int) FileFactory.getCarbonFile(bloomIndexFile).getSize()];	1✔
104	dataInputStream.readFully(fileData);	1✔
105	byte[] shardName = shardPath.getName().getBytes(Charset.forName("UTF-8"));	1✔
106	dataOutputStream.writeInt(shardName.length);	1✔
107	dataOutputStream.write(shardName);	1✔
108	dataOutputStream.writeInt(fileData.length);	1✔
109	dataOutputStream.write(fileData);	1✔
110	CarbonUtil.closeStream(dataInputStream);	1✔
111	}
112	} catch (IOException e) {	×
113	LOGGER.error(e, "Error occurs while merge bloom index file of column: " + indexCol);	×
114	// delete merge shard of bloom index for this segment when failed
115	FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(mergeShardPath));	×
116	throw new RuntimeException(	×
117	"Error occurs while merge bloom index file of column: " + indexCol);
118	} finally {
119	CarbonUtil.closeStreams(dataInputStream, dataOutputStream);	1✔
120	}	1✔
121	}	1✔
122	// delete flag file and mergeShard can be used
123	try {
124	FileFactory.deleteFile(mergeInprogressFile, FileFactory.getFileType(mergeInprogressFile));	1✔
125	} catch (IOException e) {	×
126	LOGGER.error(e, "Error occurs while deleting file " + mergeInprogressFile);	×
127	throw new RuntimeException("Error occurs while deleting file " + mergeInprogressFile);	×
128	}	1✔
129	// remove old store
130	for (CarbonFile shardpath: shardPaths) {	1✔
131	FileFactory.deleteAllCarbonFilesOfDir(shardpath);	1✔
132	}
133	}	1✔
134
135	/**
136	* load bloom filter from bloom index file
137	*/
138	public static List<CarbonBloomFilter> loadBloomFilterFromFile(
139	String shardPath, String colName) {
140	if (shardPath.endsWith(MERGE_BLOOM_INDEX_SHARD_NAME)) {	1✔
141	return loadMergeBloomIndex(shardPath, colName);	1✔
142	} else {
143	return loadBloomIndex(shardPath, colName);	×
144	}
145	}
146
147	/**
148	* load bloom filter of {@code colName} from {@code shardPath}
149	*/
150	public static List<CarbonBloomFilter> loadBloomIndex(
151	String shardPath, String colName) {
152	DataInputStream dataInStream = null;	×
153	List<CarbonBloomFilter> bloomFilters = new ArrayList<>();	×
154	try {
155	String indexFile = getBloomIndexFile(shardPath, colName);	×
156	dataInStream = FileFactory.getDataInputStream(indexFile, FileFactory.getFileType(indexFile));	×
157	while (dataInStream.available() > 0) {	×
158	CarbonBloomFilter bloomFilter = new CarbonBloomFilter();	×
159	bloomFilter.readFields(dataInStream);	×
160	bloomFilter.setShardName(new Path(shardPath).getName());	×
161	bloomFilters.add(bloomFilter);	×
162	}	×
163	LOGGER.info(String.format("Read %d bloom indices from %s", bloomFilters.size(), indexFile));	×
164
165	return bloomFilters;	×
166	} catch (IOException e) {	×
167	LOGGER.error(e, "Error occurs while reading bloom index");	×
168	throw new RuntimeException("Error occurs while reading bloom index", e);	×
169	} finally {
170	CarbonUtil.closeStreams(dataInStream);	×
171	}	×
172	}
173
174
175	/**
176	* load bloom filter of {@code colName} from {@code mergeShardPath}
177	*/
178	public static List<CarbonBloomFilter> loadMergeBloomIndex(
179	String mergeShardPath, String colName) {
180	String mergeIndexFile = getMergeBloomIndexFile(mergeShardPath, colName);	1✔
181	DataInputStream mergeIndexInStream = null;	1✔
182	List<CarbonBloomFilter> bloomFilters = new ArrayList<>();	1✔
183	try {
184	mergeIndexInStream = FileFactory.getDataInputStream(	1✔
185	mergeIndexFile, FileFactory.getFileType(mergeIndexFile));	1✔
186	while (mergeIndexInStream.available() > 0) {	1✔
187	// read shard name
188	int shardNameByteLength = mergeIndexInStream.readInt();	1✔
189	byte[] shardNameBytes = new byte[shardNameByteLength];	1✔
190	mergeIndexInStream.readFully(shardNameBytes);	1✔
191	String shardName = new String(shardNameBytes, Charset.forName("UTF-8"));	1✔
192	// read bloom index file data
193	int indexFileByteLength = mergeIndexInStream.readInt();	1✔
194	byte[] indexFileBytes = new byte[indexFileByteLength];	1✔
195	mergeIndexInStream.readFully(indexFileBytes);	1✔
196	// warp byte array as input stream to get bloom filters
197	ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(indexFileBytes);	1✔
198	DataInputStream indexDataInStream = new DataInputStream(byteArrayInputStream);	1✔
199	while (indexDataInStream.available() > 0) {	1✔
200	CarbonBloomFilter bloomFilter = new CarbonBloomFilter();	1✔
201	bloomFilter.readFields(indexDataInStream);	1✔
202	bloomFilter.setShardName(shardName);	1✔
203	bloomFilters.add(bloomFilter);	1✔
204	}	1✔
205	}	1✔
206	LOGGER.info(	1✔
207	String.format("Read %d bloom indices from %s", bloomFilters.size(), mergeIndexFile));	1✔
208	return bloomFilters;	1✔
209	} catch (IOException e) {	×
210	LOGGER.error(e, "Error occurs while reading merge bloom index");	×
211	throw new RuntimeException("Error occurs while reading merge bloom index", e);	×
212	} finally {
213	CarbonUtil.closeStreams(mergeIndexInStream);	1✔
214	}	×
215	}
216
217	/**
218	* get bloom index file
219	*/
220	public static String getBloomIndexFile(String shardPath, String colName) {
221	return shardPath.concat(File.separator).concat(colName).concat(BLOOM_INDEX_SUFFIX);	1✔
222	}
223
224	/**
225	* get merge bloom index file
226	*/
227	public static String getMergeBloomIndexFile(String mergeShardPath, String colName) {
228	return mergeShardPath.concat(File.separator).concat(colName).concat(MERGE_BLOOM_INDEX_SUFFIX);	1✔
229	}
230	}

apache / carbondata / 830

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous