8784

Build # 8784

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/902bd9badf058712a9e9771cffd0e2f456d59914">902bd9bad<a href="https://github.com/apache/carbondata/commit/902bd9badf058712a9e9771cffd0e2f456d59914">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/9709ff22ac38c1322daacd1e6c51d63a3c637a8b">9709ff22a</a>">9709ff22a</a><a href="https://github.com/apache/carbondata/commit/902bd9badf058712a9e9771cffd0e2f456d59914"> into 3cd8b947c">3cd8b947c</a>

Run Details

60469 of 75063 relevant lines covered (80.56%)

1.06 hits per line

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

88.24

/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/AbstractBloomDataMapWriter.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.DataOutputStream;
import java.io.IOException;
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.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.datastore.page.ColumnPage;
import org.apache.carbondata.core.datastore.page.encoding.bool.BooleanConvert;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.encoder.Encoding;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.util.CarbonUtil;

import org.apache.hadoop.util.bloom.CarbonBloomFilter;
import org.apache.hadoop.util.bloom.Key;
import org.apache.hadoop.util.hash.Hash;

@InterfaceAudience.Internal
public abstract class AbstractBloomDataMapWriter extends DataMapWriter {
  private static final LogService LOG = LogServiceFactory.getLogService(
      BloomDataMapWriter.class.getCanonicalName());
  private int bloomFilterSize;
  private double bloomFilterFpp;
  private boolean compressBloom;
  protected int currentBlockletId;
  private List<String> currentDMFiles;
  private List<DataOutputStream> currentDataOutStreams;
  protected List<CarbonBloomFilter> indexBloomFilters;

  AbstractBloomDataMapWriter(String tablePath, String dataMapName, List<CarbonColumn> indexColumns,
      Segment segment, String shardName, SegmentProperties segmentProperties,
      int bloomFilterSize, double bloomFilterFpp, boolean compressBloom)
      throws IOException {
    super(tablePath, dataMapName, indexColumns, segment, shardName);
    this.bloomFilterSize = bloomFilterSize;
    this.bloomFilterFpp = bloomFilterFpp;
    this.compressBloom = compressBloom;
    currentDMFiles = new ArrayList<>(indexColumns.size());
    currentDataOutStreams = new ArrayList<>(indexColumns.size());
    indexBloomFilters = new ArrayList<>(indexColumns.size());
    initDataMapFile();
    resetBloomFilters();
  }

  @Override
  public void onBlockStart(String blockId) throws IOException {
  }

  @Override
  public void onBlockEnd(String blockId) throws IOException {
  }

  @Override
  public void onBlockletStart(int blockletId) {
  }

  protected void resetBloomFilters() {
    indexBloomFilters.clear();
    int[] stats = calculateBloomStats();
    for (int i = 0; i < indexColumns.size(); i++) {
      indexBloomFilters
          .add(new CarbonBloomFilter(stats[0], stats[1], Hash.MURMUR_HASH, compressBloom));
    }
  }

  /**
   * It calculates the bits size and number of hash functions to calculate bloom.
   */
  private int[] calculateBloomStats() {
    /*
     * n: how many items you expect to have in your filter
     * p: your acceptable false positive rate
     * Number of bits (m) = -n*ln(p) / (ln(2)^2)
     * Number of hashes(k) = m/n * ln(2)
     */
    double sizeinBits = -bloomFilterSize * Math.log(bloomFilterFpp) / (Math.pow(Math.log(2), 2));
    double numberOfHashes = sizeinBits / bloomFilterSize * Math.log(2);
    int[] stats = new int[2];
    stats[0] = (int) Math.ceil(sizeinBits);
    stats[1] = (int) Math.ceil(numberOfHashes);
    return stats;
  }

  @Override
  public void onBlockletEnd(int blockletId) {
    writeBloomDataMapFile();
    currentBlockletId++;
  }

  @Override
  public void onPageAdded(int blockletId, int pageId, int pageSize, ColumnPage[] pages)
      throws IOException {
    for (int rowId = 0; rowId < pageSize; rowId++) {
      // for each indexed column, add the data to index
      for (int i = 0; i < indexColumns.size(); i++) {
        Object data = pages[i].getData(rowId);
        addValue2BloomIndex(i, data);
      }
    }
  }

  protected void addValue2BloomIndex(int indexColIdx, Object value) {
    byte[] indexValue;
    // convert measure to bytes
    // convert non-dict dimensions to simple bytes without length
    // convert internal-dict dimensions to simple bytes without any encode
    if (indexColumns.get(indexColIdx).isMeasure()) {
      // NULL value of all measures are already processed in `ColumnPage.getData`
      // or `RawBytesReadSupport.readRow` with actual data type

      // Carbon stores boolean as byte. Here we convert it for `getValueAsBytes`
      if (indexColumns.get(indexColIdx).getDataType().equals(DataTypes.BOOLEAN)) {
        value = BooleanConvert.boolean2Byte((Boolean)value);
      }
      indexValue = CarbonUtil.getValueAsBytes(indexColumns.get(indexColIdx).getDataType(), value);
    } else {
      if (indexColumns.get(indexColIdx).hasEncoding(Encoding.DICTIONARY)
          || indexColumns.get(indexColIdx).hasEncoding(Encoding.DIRECT_DICTIONARY)) {
        indexValue = convertDictionaryValue(indexColIdx, value);
      } else {
        indexValue = convertNonDictionaryValue(indexColIdx, value);
      }
    }
    if (indexValue.length == 0) {
      indexValue = CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY;
    }
    indexBloomFilters.get(indexColIdx).add(new Key(indexValue));
  }

  protected abstract byte[] convertDictionaryValue(int indexColIdx, Object value);

  protected abstract byte[] convertNonDictionaryValue(int indexColIdx, Object value);

  private void initDataMapFile() throws IOException {
    if (!FileFactory.isFileExist(dataMapPath)) {
      if (!FileFactory.mkdirs(dataMapPath, FileFactory.getFileType(dataMapPath))) {
        throw new IOException("Failed to create directory " + dataMapPath);
      }
    }
    for (int indexColId = 0; indexColId < indexColumns.size(); indexColId++) {
      String dmFile = BloomIndexFileStore.getBloomIndexFile(dataMapPath,
          indexColumns.get(indexColId).getColName());
      DataOutputStream dataOutStream = null;
      try {
        FileFactory.createNewFile(dmFile, FileFactory.getFileType(dmFile));
        dataOutStream = FileFactory.getDataOutputStream(dmFile,
            FileFactory.getFileType(dmFile));
      } catch (IOException e) {
        CarbonUtil.closeStreams(dataOutStream);
        throw new IOException(e);
      }

      this.currentDMFiles.add(dmFile);
      this.currentDataOutStreams.add(dataOutStream);
    }
  }

  protected void writeBloomDataMapFile() {
    try {
      for (int indexColId = 0; indexColId < indexColumns.size(); indexColId++) {
        CarbonBloomFilter bloomFilter = indexBloomFilters.get(indexColId);
        bloomFilter.setBlockletNo(currentBlockletId);
        // only in higher version of guava-bloom-filter, it provides readFrom/writeTo interface.
        // In lower version, we use default java serializer to write bloomfilter.
        bloomFilter.write(this.currentDataOutStreams.get(indexColId));
        this.currentDataOutStreams.get(indexColId).flush();
      }
    } catch (Exception e) {
      for (DataOutputStream dataOutputStream : currentDataOutStreams) {
        CarbonUtil.closeStreams(dataOutputStream);
      }
      throw new RuntimeException(e);
    } finally {
      resetBloomFilters();
    }
  }

  @Override
  public void finish() throws IOException {
    if (!isWritingFinished()) {
      releaseResouce();
      setWritingFinished(true);
    }
  }

  protected void releaseResouce() {
    for (int indexColId = 0; indexColId < indexColumns.size(); indexColId++) {
      CarbonUtil.closeStreams(
          currentDataOutStreams.get(indexColId));
    }
  }
}

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
18	package org.apache.carbondata.datamap.bloom;
19
20	import java.io.DataOutputStream;
21	import java.io.IOException;
22	import java.util.ArrayList;
23	import java.util.List;
24
25	import org.apache.carbondata.common.annotations.InterfaceAudience;
26	import org.apache.carbondata.common.logging.LogService;
27	import org.apache.carbondata.common.logging.LogServiceFactory;
28	import org.apache.carbondata.core.constants.CarbonCommonConstants;
29	import org.apache.carbondata.core.datamap.Segment;
30	import org.apache.carbondata.core.datamap.dev.DataMapWriter;
31	import org.apache.carbondata.core.datastore.block.SegmentProperties;
32	import org.apache.carbondata.core.datastore.impl.FileFactory;
33	import org.apache.carbondata.core.datastore.page.ColumnPage;
34	import org.apache.carbondata.core.datastore.page.encoding.bool.BooleanConvert;
35	import org.apache.carbondata.core.metadata.datatype.DataTypes;
36	import org.apache.carbondata.core.metadata.encoder.Encoding;
37	import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
38	import org.apache.carbondata.core.util.CarbonUtil;
39
40	import org.apache.hadoop.util.bloom.CarbonBloomFilter;
41	import org.apache.hadoop.util.bloom.Key;
42	import org.apache.hadoop.util.hash.Hash;
43
44	@InterfaceAudience.Internal
45	public abstract class AbstractBloomDataMapWriter extends DataMapWriter {
46	private static final LogService LOG = LogServiceFactory.getLogService(	1✔
47	BloomDataMapWriter.class.getCanonicalName());	1✔
48	private int bloomFilterSize;
49	private double bloomFilterFpp;
50	private boolean compressBloom;
51	protected int currentBlockletId;
52	private List<String> currentDMFiles;
53	private List<DataOutputStream> currentDataOutStreams;
54	protected List<CarbonBloomFilter> indexBloomFilters;
55
56	AbstractBloomDataMapWriter(String tablePath, String dataMapName, List<CarbonColumn> indexColumns,
57	Segment segment, String shardName, SegmentProperties segmentProperties,
58	int bloomFilterSize, double bloomFilterFpp, boolean compressBloom)
59	throws IOException {
60	super(tablePath, dataMapName, indexColumns, segment, shardName);	1✔
61	this.bloomFilterSize = bloomFilterSize;	1✔
62	this.bloomFilterFpp = bloomFilterFpp;	1✔
63	this.compressBloom = compressBloom;	1✔
64	currentDMFiles = new ArrayList<>(indexColumns.size());	1✔
65	currentDataOutStreams = new ArrayList<>(indexColumns.size());	1✔
66	indexBloomFilters = new ArrayList<>(indexColumns.size());	1✔
67	initDataMapFile();	1✔
68	resetBloomFilters();	1✔
69	}	1✔
70
71	@Override
72	public void onBlockStart(String blockId) throws IOException {
73	}	1✔
74
75	@Override
76	public void onBlockEnd(String blockId) throws IOException {
77	}	1✔
78
79	@Override
80	public void onBlockletStart(int blockletId) {
81	}	1✔
82
83	protected void resetBloomFilters() {
84	indexBloomFilters.clear();	1✔
85	int[] stats = calculateBloomStats();	1✔
86	for (int i = 0; i < indexColumns.size(); i++) {	1✔
87	indexBloomFilters	1✔
88	.add(new CarbonBloomFilter(stats[0], stats[1], Hash.MURMUR_HASH, compressBloom));	1✔
89	}
90	}	1✔
91
92	/**
93	* It calculates the bits size and number of hash functions to calculate bloom.
94	*/
95	private int[] calculateBloomStats() {
96	/*
97	* n: how many items you expect to have in your filter
98	* p: your acceptable false positive rate
99	* Number of bits (m) = -n*ln(p) / (ln(2)^2)
100	* Number of hashes(k) = m/n * ln(2)
101	*/
102	double sizeinBits = -bloomFilterSize * Math.log(bloomFilterFpp) / (Math.pow(Math.log(2), 2));	1✔
103	double numberOfHashes = sizeinBits / bloomFilterSize * Math.log(2);	1✔
104	int[] stats = new int[2];	1✔
105	stats[0] = (int) Math.ceil(sizeinBits);	1✔
106	stats[1] = (int) Math.ceil(numberOfHashes);	1✔
107	return stats;	1✔
108	}
109
110	@Override
111	public void onBlockletEnd(int blockletId) {
112	writeBloomDataMapFile();	1✔
113	currentBlockletId++;	1✔
114	}	1✔
115
116	@Override
117	public void onPageAdded(int blockletId, int pageId, int pageSize, ColumnPage[] pages)
118	throws IOException {
119	for (int rowId = 0; rowId < pageSize; rowId++) {	1✔
120	// for each indexed column, add the data to index
121	for (int i = 0; i < indexColumns.size(); i++) {	1✔
122	Object data = pages[i].getData(rowId);	1✔
123	addValue2BloomIndex(i, data);	1✔
124	}
125	}
126	}	1✔
127
128	protected void addValue2BloomIndex(int indexColIdx, Object value) {
129	byte[] indexValue;
130	// convert measure to bytes
131	// convert non-dict dimensions to simple bytes without length
132	// convert internal-dict dimensions to simple bytes without any encode
133	if (indexColumns.get(indexColIdx).isMeasure()) {	1✔
134	// NULL value of all measures are already processed in `ColumnPage.getData`
135	// or `RawBytesReadSupport.readRow` with actual data type
136
137	// Carbon stores boolean as byte. Here we convert it for `getValueAsBytes`
138	if (indexColumns.get(indexColIdx).getDataType().equals(DataTypes.BOOLEAN)) {	1✔
139	value = BooleanConvert.boolean2Byte((Boolean)value);	1✔
140	}
141	indexValue = CarbonUtil.getValueAsBytes(indexColumns.get(indexColIdx).getDataType(), value);	1✔
142	} else {
143	if (indexColumns.get(indexColIdx).hasEncoding(Encoding.DICTIONARY)	1✔
144	\|\| indexColumns.get(indexColIdx).hasEncoding(Encoding.DIRECT_DICTIONARY)) {	1✔
145	indexValue = convertDictionaryValue(indexColIdx, value);	1✔
146	} else {
147	indexValue = convertNonDictionaryValue(indexColIdx, value);	1✔
148	}
149	}
150	if (indexValue.length == 0) {	1✔
151	indexValue = CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY;	×
152	}
153	indexBloomFilters.get(indexColIdx).add(new Key(indexValue));	1✔
154	}	1✔
155
156	protected abstract byte[] convertDictionaryValue(int indexColIdx, Object value);
157
158	protected abstract byte[] convertNonDictionaryValue(int indexColIdx, Object value);
159
160	private void initDataMapFile() throws IOException {
161	if (!FileFactory.isFileExist(dataMapPath)) {	1✔
162	if (!FileFactory.mkdirs(dataMapPath, FileFactory.getFileType(dataMapPath))) {	1✔
163	throw new IOException("Failed to create directory " + dataMapPath);	×
164	}
165	}
166	for (int indexColId = 0; indexColId < indexColumns.size(); indexColId++) {	1✔
167	String dmFile = BloomIndexFileStore.getBloomIndexFile(dataMapPath,	1✔
168	indexColumns.get(indexColId).getColName());	1✔
169	DataOutputStream dataOutStream = null;	1✔
170	try {
171	FileFactory.createNewFile(dmFile, FileFactory.getFileType(dmFile));	1✔
172	dataOutStream = FileFactory.getDataOutputStream(dmFile,	1✔
173	FileFactory.getFileType(dmFile));	1✔
174	} catch (IOException e) {	×
175	CarbonUtil.closeStreams(dataOutStream);	×
176	throw new IOException(e);	×
177	}	1✔
178
179	this.currentDMFiles.add(dmFile);	1✔
180	this.currentDataOutStreams.add(dataOutStream);	1✔
181	}
182	}	1✔
183
184	protected void writeBloomDataMapFile() {
185	try {
186	for (int indexColId = 0; indexColId < indexColumns.size(); indexColId++) {	1✔
187	CarbonBloomFilter bloomFilter = indexBloomFilters.get(indexColId);	1✔
188	bloomFilter.setBlockletNo(currentBlockletId);	1✔
189	// only in higher version of guava-bloom-filter, it provides readFrom/writeTo interface.
190	// In lower version, we use default java serializer to write bloomfilter.
191	bloomFilter.write(this.currentDataOutStreams.get(indexColId));	1✔
192	this.currentDataOutStreams.get(indexColId).flush();	1✔
193	}
194	} catch (Exception e) {	×
195	for (DataOutputStream dataOutputStream : currentDataOutStreams) {	×
196	CarbonUtil.closeStreams(dataOutputStream);	×
197	}	×
198	throw new RuntimeException(e);	×
199	} finally {
200	resetBloomFilters();	1✔
201	}	1✔
202	}	1✔
203
204	@Override
205	public void finish() throws IOException {
206	if (!isWritingFinished()) {	1✔
207	releaseResouce();	1✔
208	setWritingFinished(true);	1✔
209	}
210	}	1✔
211
212	protected void releaseResouce() {
213	for (int indexColId = 0; indexColId < indexColumns.size(); indexColId++) {	1✔
214	CarbonUtil.closeStreams(	1✔
215	currentDataOutStreams.get(indexColId));	1✔
216	}
217	}	1✔
218	}

apache / carbondata / 8784

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