3765

Build # 3765

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/adbabdd64c047f3d040bf3241115fabea2b146e6">adbabdd64<a href="https://github.com/apache/carbondata/commit/adbabdd64c047f3d040bf3241115fabea2b146e6">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/9ae532457b137e8d60996e2c389e04ddcee4d70b">9ae532457</a>">9ae532457</a><a href="https://github.com/apache/carbondata/commit/adbabdd64c047f3d040bf3241115fabea2b146e6"> into afef23270">afef23270</a>

Run Details

64022 of 82808 relevant lines covered (77.31%)

1.01 hits per line

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

66.98

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

    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, 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) {
      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, 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, e);
      } 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	String mergeShardPath = dmSegmentPathString + File.separator + MERGE_BLOOM_INDEX_SHARD_NAME;	1✔
77	String mergeInprogressFile = dmSegmentPathString + File.separator + MERGE_INPROGRESS_FILE;	1✔
78
79	// Step 2. prepare for fail-safe merging
80
81	try {
82	// delete mergeShard folder if exists
83	if (FileFactory.isFileExist(mergeShardPath)) {	1✔
84	FileFactory.deleteFile(mergeShardPath, FileFactory.getFileType(mergeShardPath));	×
85	}
86	// create flag file before creating mergeShard folder
87	if (!FileFactory.isFileExist(mergeInprogressFile)) {	1✔
88	FileFactory.createNewFile(	1✔
89	mergeInprogressFile, FileFactory.getFileType(mergeInprogressFile));	1✔
90	}
91	// create mergeShard output folder
92	if (!FileFactory.mkdirs(mergeShardPath, FileFactory.getFileType(mergeShardPath))) {	1✔
93	throw new RuntimeException("Failed to create directory " + mergeShardPath);	×
94	}
95	} catch (IOException e) {	×
96	throw new RuntimeException(e);	×
97	}	1✔
98
99	// Step 3. merge index files
100	// Query won't use mergeShard until MERGE_INPROGRESS_FILE is deleted
101
102	// for each index column, merge the bloomindex files from all shards into one
103	for (String indexCol: indexCols) {	1✔
104	String mergeIndexFile = getMergeBloomIndexFile(mergeShardPath, indexCol);	1✔
105	DataInputStream dataInputStream = null;	1✔
106	DataOutputStream dataOutputStream = null;	1✔
107	try {
108	FileFactory.createNewFile(mergeIndexFile, FileFactory.getFileType(mergeIndexFile));	1✔
109	dataOutputStream = FileFactory.getDataOutputStream(	1✔
110	mergeIndexFile, FileFactory.getFileType(mergeIndexFile));	1✔
111	for (CarbonFile shardPath : shardPaths) {	1✔
112	String bloomIndexFile = getBloomIndexFile(shardPath.getCanonicalPath(), indexCol);	1✔
113	dataInputStream = FileFactory.getDataInputStream(	1✔
114	bloomIndexFile, FileFactory.getFileType(bloomIndexFile));	1✔
115	byte[] fileData = new byte[(int) FileFactory.getCarbonFile(bloomIndexFile).getSize()];	1✔
116	dataInputStream.readFully(fileData);	1✔
117	byte[] shardName = shardPath.getName().getBytes(Charset.forName("UTF-8"));	1✔
118	dataOutputStream.writeInt(shardName.length);	1✔
119	dataOutputStream.write(shardName);	1✔
120	dataOutputStream.writeInt(fileData.length);	1✔
121	dataOutputStream.write(fileData);	1✔
122	CarbonUtil.closeStream(dataInputStream);	1✔
123	}
124	} catch (IOException e) {	×
125	LOGGER.error("Error occurs while merge bloom index file of column: " + indexCol, e);	×
126	// if any column failed, delete merge shard for this segment and exit
127	FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(mergeShardPath));	×
128	FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(mergeInprogressFile));	×
129	throw new RuntimeException(	×
130	"Error occurs while merge bloom index file of column: " + indexCol, e);
131	} finally {
132	CarbonUtil.closeStreams(dataInputStream, dataOutputStream);	1✔
133	}	1✔
134	}	1✔
135
136	// Step 4. delete flag file and mergeShard can be used
137	try {
138	FileFactory.deleteFile(mergeInprogressFile, FileFactory.getFileType(mergeInprogressFile));	1✔
139	} catch (IOException e) {	×
140	LOGGER.error("Error occurs while deleting file " + mergeInprogressFile, e);	×
141	throw new RuntimeException("Error occurs while deleting file " + mergeInprogressFile);	×
142	}	1✔
143	// remove old store
144	for (CarbonFile shardpath: shardPaths) {	1✔
145	FileFactory.deleteAllCarbonFilesOfDir(shardpath);	1✔
146	}
147	}	1✔
148
149	/**
150	* load bloom filter from bloom index file
151	*/
152	public static List<CarbonBloomFilter> loadBloomFilterFromFile(
153	String shardPath, String colName) {
154	if (shardPath.endsWith(MERGE_BLOOM_INDEX_SHARD_NAME)) {	1✔
155	return loadMergeBloomIndex(shardPath, colName);	1✔
156	} else {
157	return loadBloomIndex(shardPath, colName);	×
158	}
159	}
160
161	/**
162	* load bloom filter of {@code colName} from {@code shardPath}
163	*/
164	public static List<CarbonBloomFilter> loadBloomIndex(
165	String shardPath, String colName) {
166	DataInputStream dataInStream = null;	×
167	List<CarbonBloomFilter> bloomFilters = new ArrayList<>();	×
168	try {
169	String indexFile = getBloomIndexFile(shardPath, colName);	×
170	dataInStream = FileFactory.getDataInputStream(indexFile, FileFactory.getFileType(indexFile));	×
171	while (dataInStream.available() > 0) {	×
172	CarbonBloomFilter bloomFilter = new CarbonBloomFilter();	×
173	bloomFilter.readFields(dataInStream);	×
174	bloomFilter.setShardName(new Path(shardPath).getName());	×
175	bloomFilters.add(bloomFilter);	×
176	}	×
177	LOGGER.info(String.format("Read %d bloom indices from %s", bloomFilters.size(), indexFile));	×
178
179	return bloomFilters;	×
180	} catch (IOException e) {	×
181	LOGGER.error("Error occurs while reading bloom index", e);	×
182	throw new RuntimeException("Error occurs while reading bloom index", e);	×
183	} finally {
184	CarbonUtil.closeStreams(dataInStream);	×
185	}	×
186	}
187
188
189	/**
190	* load bloom filter of {@code colName} from {@code mergeShardPath}
191	*/
192	public static List<CarbonBloomFilter> loadMergeBloomIndex(
193	String mergeShardPath, String colName) {
194	String mergeIndexFile = getMergeBloomIndexFile(mergeShardPath, colName);	1✔
195	DataInputStream mergeIndexInStream = null;	1✔
196	List<CarbonBloomFilter> bloomFilters = new ArrayList<>();	1✔
197	try {
198	mergeIndexInStream = FileFactory.getDataInputStream(	1✔
199	mergeIndexFile, FileFactory.getFileType(mergeIndexFile));	1✔
200	while (mergeIndexInStream.available() > 0) {	1✔
201	// read shard name
202	int shardNameByteLength = mergeIndexInStream.readInt();	1✔
203	byte[] shardNameBytes = new byte[shardNameByteLength];	1✔
204	mergeIndexInStream.readFully(shardNameBytes);	1✔
205	String shardName = new String(shardNameBytes, Charset.forName("UTF-8"));	1✔
206	// read bloom index file data
207	int indexFileByteLength = mergeIndexInStream.readInt();	1✔
208	byte[] indexFileBytes = new byte[indexFileByteLength];	1✔
209	mergeIndexInStream.readFully(indexFileBytes);	1✔
210	// warp byte array as input stream to get bloom filters
211	ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(indexFileBytes);	1✔
212	DataInputStream indexDataInStream = new DataInputStream(byteArrayInputStream);	1✔
213	while (indexDataInStream.available() > 0) {	1✔
214	CarbonBloomFilter bloomFilter = new CarbonBloomFilter();	1✔
215	bloomFilter.readFields(indexDataInStream);	1✔
216	bloomFilter.setShardName(shardName);	1✔
217	bloomFilters.add(bloomFilter);	1✔
218	}	1✔
219	}	1✔
220	LOGGER.info(	1✔
221	String.format("Read %d bloom indices from %s", bloomFilters.size(), mergeIndexFile));	1✔
222	return bloomFilters;	1✔
223	} catch (IOException e) {	×
224	LOGGER.error("Error occurs while reading merge bloom index", e);	×
225	throw new RuntimeException("Error occurs while reading merge bloom index", e);	×
226	} finally {
227	CarbonUtil.closeStreams(mergeIndexInStream);	1✔
228	}	×
229	}
230
231	/**
232	* get bloom index file
233	*/
234	public static String getBloomIndexFile(String shardPath, String colName) {
235	return shardPath.concat(File.separator).concat(colName).concat(BLOOM_INDEX_SUFFIX);	1✔
236	}
237
238	/**
239	* get merge bloom index file
240	*/
241	public static String getMergeBloomIndexFile(String mergeShardPath, String colName) {
242	return mergeShardPath.concat(File.separator).concat(colName).concat(MERGE_BLOOM_INDEX_SUFFIX);	1✔
243	}
244	}

apache / carbondata / 3765

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