9258

Build # 9258

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/e3a149d9eb03c6938422c4e0bdbae3e62f491305">e3a149d9e<a href="https://github.com/apache/carbondata/commit/e3a149d9eb03c6938422c4e0bdbae3e62f491305">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/bcab5ac630e39a7dadee09d5b9157642d061b5e1">bcab5ac63</a>">bcab5ac63</a><a href="https://github.com/apache/carbondata/commit/e3a149d9eb03c6938422c4e0bdbae3e62f491305"> into 278d17178">278d17178</a>

Run Details

18 of 18 new or added lines in 6 files covered. (100.0%)

60522 of 75015 relevant lines covered (80.68%)

1.07 hits per line

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

66.67

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

    // check whether need to merge
    String mergeShardPath = dmSegmentPathString + File.separator + MERGE_BLOOM_INDEX_SHARD_NAME;
    String mergeInprogressFile = dmSegmentPathString + File.separator + MERGE_INPROGRESS_FILE;
    try {
      if (shardPaths.length == 0 || FileFactory.isFileExist(mergeShardPath)) {
        LOGGER.info("No shard data to merge or already merged for path " + mergeShardPath);
        return;
      }
    } catch (IOException e) {
      throw new RuntimeException(e.getMessage(), e);
    }

    // Step 2. prepare for fail-safe merging

    try {
      // 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 for mergeShard: " + mergeShardPath);
      }
    } catch (IOException e) {
      throw new RuntimeException(e.getMessage(), 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, 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("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);
      } finally {
        CarbonUtil.closeStreams(dataInputStream, dataOutputStream);
      }
    }

    // Step 4. delete flag file and mergeShard can be used
    try {
      FileFactory.deleteFile(mergeInprogressFile, FileFactory.getFileType(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, 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("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, 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("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);
  }
}

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.ByteArrayInputStream;
20	import java.io.DataInputStream;
21	import java.io.DataOutputStream;
22	import java.io.File;
23	import java.io.IOException;
24	import java.nio.charset.Charset;
25	import java.util.ArrayList;
26	import java.util.List;
27
28	import org.apache.carbondata.common.annotations.InterfaceAudience;
29	import org.apache.carbondata.common.logging.LogServiceFactory;
30	import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
31	import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
32	import org.apache.carbondata.core.datastore.impl.FileFactory;
33	import org.apache.carbondata.core.util.CarbonUtil;
34
35	import org.apache.hadoop.fs.Path;
36	import org.apache.hadoop.util.bloom.CarbonBloomFilter;
37	import org.apache.log4j.Logger;
38
39	/**
40	* This class works for merging and loading bloom index
41	*/
42	@InterfaceAudience.Internal
43	public class BloomIndexFileStore {	×
44
45	private static final Logger LOGGER =	1✔
46	LogServiceFactory.getLogService(BloomIndexFileStore.class.getName());	1✔
47
48	// suffix of original generated file
49	public static final String BLOOM_INDEX_SUFFIX = ".bloomindex";
50	// suffix of merged bloom index file
51	public static final String MERGE_BLOOM_INDEX_SUFFIX = ".bloomindexmerge";
52	// directory to store merged bloom index files
53	public static final String MERGE_BLOOM_INDEX_SHARD_NAME = "mergeShard";
54	/**
55	* flag file for merging
56	* if flag file exists, query won't use mergeShard
57	* if flag file not exists and mergeShard generated, query will use mergeShard
58	*/
59	public static final String MERGE_INPROGRESS_FILE = "mergeShard.inprogress";
60
61
62	public static void mergeBloomIndexFile(String dmSegmentPathString, List<String> indexCols) {
63
64	// Step 1. check current folders
65
66	// get all shard paths of old store
67	CarbonFile segmentPath = FileFactory.getCarbonFile(dmSegmentPathString,	1✔
68	FileFactory.getFileType(dmSegmentPathString));	1✔
69	CarbonFile[] shardPaths = segmentPath.listFiles(new CarbonFileFilter() {	1✔
70	@Override
71	public boolean accept(CarbonFile file) {
72	return file.isDirectory() && !file.getName().equals(MERGE_BLOOM_INDEX_SHARD_NAME);	1✔
73	}
74	});
75
76	// check whether need to merge
77	String mergeShardPath = dmSegmentPathString + File.separator + MERGE_BLOOM_INDEX_SHARD_NAME;	1✔
78	String mergeInprogressFile = dmSegmentPathString + File.separator + MERGE_INPROGRESS_FILE;	1✔
79	try {
80	if (shardPaths.length == 0 \|\| FileFactory.isFileExist(mergeShardPath)) {	1✔
81	LOGGER.info("No shard data to merge or already merged for path " + mergeShardPath);	×
82	return;	×
83	}
84	} catch (IOException e) {	×
85	throw new RuntimeException(e.getMessage(), e);	×
86	}	1✔
87
88	// Step 2. prepare for fail-safe merging
89
90	try {
91	// create flag file before creating mergeShard folder
92	if (!FileFactory.isFileExist(mergeInprogressFile)) {	1✔
93	FileFactory.createNewFile(	1✔
94	mergeInprogressFile, FileFactory.getFileType(mergeInprogressFile));	1✔
95	}
96	// create mergeShard output folder
97	if (!FileFactory.mkdirs(mergeShardPath, FileFactory.getFileType(mergeShardPath))) {	1✔
98	throw new RuntimeException("Failed to create directory for mergeShard: " + mergeShardPath);	×
99	}
100	} catch (IOException e) {	×
101	throw new RuntimeException(e.getMessage(), e);	×
102	}	1✔
103
104	// Step 3. merge index files
105	// Query won't use mergeShard until MERGE_INPROGRESS_FILE is deleted
106
107	// for each index column, merge the bloomindex files from all shards into one
108	for (String indexCol: indexCols) {	1✔
109	String mergeIndexFile = getMergeBloomIndexFile(mergeShardPath, indexCol);	1✔
110	DataInputStream dataInputStream = null;	1✔
111	DataOutputStream dataOutputStream = null;	1✔
112	try {
113	FileFactory.createNewFile(mergeIndexFile, FileFactory.getFileType(mergeIndexFile));	1✔
114	dataOutputStream = FileFactory.getDataOutputStream(	1✔
115	mergeIndexFile, FileFactory.getFileType(mergeIndexFile));	1✔
116	for (CarbonFile shardPath : shardPaths) {	1✔
117	String bloomIndexFile = getBloomIndexFile(shardPath.getCanonicalPath(), indexCol);	1✔
118	dataInputStream = FileFactory.getDataInputStream(	1✔
119	bloomIndexFile, FileFactory.getFileType(bloomIndexFile));	1✔
120	byte[] fileData = new byte[(int) FileFactory.getCarbonFile(bloomIndexFile).getSize()];	1✔
121	dataInputStream.readFully(fileData);	1✔
122	byte[] shardName = shardPath.getName().getBytes(Charset.forName("UTF-8"));	1✔
123	dataOutputStream.writeInt(shardName.length);	1✔
124	dataOutputStream.write(shardName);	1✔
125	dataOutputStream.writeInt(fileData.length);	1✔
126	dataOutputStream.write(fileData);	1✔
127	CarbonUtil.closeStream(dataInputStream);	1✔
128	}
129	} catch (IOException e) {	×
130	LOGGER.error("Error occurs while merge bloom index file of column: " + indexCol, e);	×
131	// if any column failed, delete merge shard for this segment and exit
132	FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(mergeShardPath));	×
133	FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(mergeInprogressFile));	×
134	throw new RuntimeException(	×
135	"Error occurs while merge bloom index file of column: " + indexCol);
136	} finally {
137	CarbonUtil.closeStreams(dataInputStream, dataOutputStream);	1✔
138	}	1✔
139	}	1✔
140
141	// Step 4. delete flag file and mergeShard can be used
142	try {
143	FileFactory.deleteFile(mergeInprogressFile, FileFactory.getFileType(mergeInprogressFile));	1✔
144	} catch (IOException e) {	×
145	LOGGER.error("Error occurs while deleting file " + mergeInprogressFile, e);	×
146	throw new RuntimeException("Error occurs while deleting file " + mergeInprogressFile);	×
147	}	1✔
148	// remove old store
149	for (CarbonFile shardpath: shardPaths) {	1✔
150	FileFactory.deleteAllCarbonFilesOfDir(shardpath);	1✔
151	}
152	}	1✔
153
154	/**
155	* load bloom filter from bloom index file
156	*/
157	public static List<CarbonBloomFilter> loadBloomFilterFromFile(
158	String shardPath, String colName) {
159	if (shardPath.endsWith(MERGE_BLOOM_INDEX_SHARD_NAME)) {	1✔
160	return loadMergeBloomIndex(shardPath, colName);	1✔
161	} else {
162	return loadBloomIndex(shardPath, colName);	×
163	}
164	}
165
166	/**
167	* load bloom filter of {@code colName} from {@code shardPath}
168	*/
169	public static List<CarbonBloomFilter> loadBloomIndex(
170	String shardPath, String colName) {
171	DataInputStream dataInStream = null;	×
172	List<CarbonBloomFilter> bloomFilters = new ArrayList<>();	×
173	try {
174	String indexFile = getBloomIndexFile(shardPath, colName);	×
175	dataInStream = FileFactory.getDataInputStream(indexFile, FileFactory.getFileType(indexFile));	×
176	while (dataInStream.available() > 0) {	×
177	CarbonBloomFilter bloomFilter = new CarbonBloomFilter();	×
178	bloomFilter.readFields(dataInStream);	×
179	bloomFilter.setShardName(new Path(shardPath).getName());	×
180	bloomFilters.add(bloomFilter);	×
181	}	×
182	LOGGER.info(String.format("Read %d bloom indices from %s", bloomFilters.size(), indexFile));	×
183
184	return bloomFilters;	×
185	} catch (IOException e) {	×
186	LOGGER.error("Error occurs while reading bloom index", e);	×
187	throw new RuntimeException("Error occurs while reading bloom index", e);	×
188	} finally {
189	CarbonUtil.closeStreams(dataInStream);	×
190	}
191	}
192
193
194	/**
195	* load bloom filter of {@code colName} from {@code mergeShardPath}
196	*/
197	public static List<CarbonBloomFilter> loadMergeBloomIndex(
198	String mergeShardPath, String colName) {
199	String mergeIndexFile = getMergeBloomIndexFile(mergeShardPath, colName);	1✔
200	DataInputStream mergeIndexInStream = null;	1✔
201	List<CarbonBloomFilter> bloomFilters = new ArrayList<>();	1✔
202	try {
203	mergeIndexInStream = FileFactory.getDataInputStream(	1✔
204	mergeIndexFile, FileFactory.getFileType(mergeIndexFile));	1✔
205	while (mergeIndexInStream.available() > 0) {	1✔
206	// read shard name
207	int shardNameByteLength = mergeIndexInStream.readInt();	1✔
208	byte[] shardNameBytes = new byte[shardNameByteLength];	1✔
209	mergeIndexInStream.readFully(shardNameBytes);	1✔
210	String shardName = new String(shardNameBytes, Charset.forName("UTF-8"));	1✔
211	// read bloom index file data
212	int indexFileByteLength = mergeIndexInStream.readInt();	1✔
213	byte[] indexFileBytes = new byte[indexFileByteLength];	1✔
214	mergeIndexInStream.readFully(indexFileBytes);	1✔
215	// warp byte array as input stream to get bloom filters
216	ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(indexFileBytes);	1✔
217	DataInputStream indexDataInStream = new DataInputStream(byteArrayInputStream);	1✔
218	while (indexDataInStream.available() > 0) {	1✔
219	CarbonBloomFilter bloomFilter = new CarbonBloomFilter();	1✔
220	bloomFilter.readFields(indexDataInStream);	1✔
221	bloomFilter.setShardName(shardName);	1✔
222	bloomFilters.add(bloomFilter);	1✔
223	}	1✔
224	}	1✔
225	LOGGER.info(	1✔
226	String.format("Read %d bloom indices from %s", bloomFilters.size(), mergeIndexFile));	1✔
227	return bloomFilters;	1✔
228	} catch (IOException e) {	×
229	LOGGER.error("Error occurs while reading merge bloom index", e);	×
230	throw new RuntimeException("Error occurs while reading merge bloom index", e);	×
231	} finally {
232	CarbonUtil.closeStreams(mergeIndexInStream);	1✔
233	}
234	}
235
236	/**
237	* get bloom index file
238	*/
239	public static String getBloomIndexFile(String shardPath, String colName) {
240	return shardPath.concat(File.separator).concat(colName).concat(BLOOM_INDEX_SUFFIX);	1✔
241	}
242
243	/**
244	* get merge bloom index file
245	*/
246	public static String getMergeBloomIndexFile(String mergeShardPath, String colName) {
247	return mergeShardPath.concat(File.separator).concat(colName).concat(MERGE_BLOOM_INDEX_SUFFIX);	1✔
248	}
249	}

apache / carbondata / 9258

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