8422

Build # 8422

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/8b24bedc17a98d53ec260af280af947e0319a0be">8b24bedc1">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/8b24bedc17a98d53ec260af280af947e0319a0be">8b24bedc1</a><a href="https://github.com/apache/carbondata/commit/8b24bedc17a98d53ec260af280af947e0319a0be">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/8b24bedc17a98d53ec260af280af947e0319a0be&quot;&gt;8b24bedc1&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/8b24bedc17a98d53ec260af280af947e0319a0be&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a ... (continued)

Run Details

16 of 16 new or added lines in 4 files covered. (100.0%)

58120 of 72053 relevant lines covered (80.66%)

1.07 hits per line

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

98.41

/core/src/main/java/org/apache/carbondata/core/stream/StreamPruner.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.core.stream;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.List;

import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
import org.apache.carbondata.core.reader.CarbonIndexFileReader;
import org.apache.carbondata.core.scan.filter.FilterUtil;
import org.apache.carbondata.core.scan.filter.executer.FilterExecuter;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.util.CarbonMetadataUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.format.BlockIndex;

@InterfaceAudience.Internal
public class StreamPruner {

  private CarbonTable carbonTable;
  private FilterExecuter filterExecuter;

  private int totalFileNums = 0;

  public StreamPruner(CarbonTable carbonTable) {
    this.carbonTable = carbonTable;
  }

  public void init(FilterResolverIntf filterExp) {
    if (filterExp != null) {
      // cache all columns
      List<CarbonColumn> minMaxCacheColumns = new ArrayList<>();
      for (CarbonDimension dimension : carbonTable.getDimensions()) {
        if (!dimension.isComplex()) {
          minMaxCacheColumns.add(dimension);
        }
      }
      minMaxCacheColumns.addAll(carbonTable.getMeasures());
      // prepare cardinality of all dimensions
      List<ColumnSchema> listOfColumns =
          carbonTable.getTableInfo().getFactTable().getListOfColumns();
      int[] columnCardinality = new int[listOfColumns.size()];
      for (int index = 0; index < columnCardinality.length; index++) {
        columnCardinality[index] = Integer.MAX_VALUE;
      }
      // initial filter executor
      SegmentProperties segmentProperties =
          new SegmentProperties(listOfColumns, columnCardinality);
      filterExecuter = FilterUtil.getFilterExecuterTree(
          filterExp, segmentProperties, null, minMaxCacheColumns);
    }
  }

  public List<StreamFile> prune(List<Segment> segments) throws IOException {
    if (filterExecuter == null) {
      // if filter is null, list all steam files
      return listAllStreamFiles(segments, false);
    } else {
      List<StreamFile> streamFileList = new ArrayList<>();
      for (StreamFile streamFile : listAllStreamFiles(segments, true)) {
        if (isScanRequire(streamFile)) {
          // if stream file is required to scan
          streamFileList.add(streamFile);
          streamFile.setMinMaxIndex(null);
        }
      }
      return streamFileList;
    }
  }

  private boolean isScanRequire(StreamFile streamFile) {
    // backward compatibility, old stream file without min/max index
    if (streamFile.getMinMaxIndex() == null) {
      return true;
    }
    byte[][] maxValue = streamFile.getMinMaxIndex().getMaxValues();
    byte[][] minValue = streamFile.getMinMaxIndex().getMinValues();
    BitSet bitSet = filterExecuter.isScanRequired(maxValue, minValue);
    if (!bitSet.isEmpty()) {
      return true;
    } else {
      return false;
    }
  }

  // TODO optimize and move the code to StreamSegment , but it's in the streaming module.
  private List<StreamFile> listAllStreamFiles(List<Segment> segments, boolean withMinMax)
      throws IOException {
    List<StreamFile> streamFileList = new ArrayList<>();
    for (Segment segment : segments) {
      String segmentDir = CarbonTablePath.getSegmentPath(
          carbonTable.getAbsoluteTableIdentifier().getTablePath(), segment.getSegmentNo());
      String indexFile = CarbonTablePath.getCarbonStreamIndexFilePath(segmentDir);
      FileFactory.FileType fileType = FileFactory.getFileType(indexFile);
      if (FileFactory.isFileExist(indexFile, fileType)) {
        CarbonIndexFileReader indexReader = new CarbonIndexFileReader();
        indexReader.openThriftReader(indexFile);
        try {
          while (indexReader.hasNext()) {
            BlockIndex blockIndex = indexReader.readBlockIndexInfo();
            String filePath = segmentDir + File.separator + blockIndex.getFile_name();
            long length = blockIndex.getFile_size();
            StreamFile streamFile = new StreamFile(segment.getSegmentNo(), filePath, length);
            streamFileList.add(streamFile);
            if (withMinMax) {
              if (blockIndex.getBlock_index() != null
                  && blockIndex.getBlock_index().getMin_max_index() != null) {
                streamFile.setMinMaxIndex(CarbonMetadataUtil
                    .convertExternalMinMaxIndex(blockIndex.getBlock_index().getMin_max_index()));
              }
            }
          }
        } finally {
          indexReader.closeThriftReader();
        }
      }
    }
    totalFileNums = streamFileList.size();
    return streamFileList;
  }

  public int getTotalFileNums() {
    return totalFileNums;
  }
}

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.core.stream;
19
20	import java.io.File;
21	import java.io.IOException;
22	import java.util.ArrayList;
23	import java.util.BitSet;
24	import java.util.List;
25
26	import org.apache.carbondata.common.annotations.InterfaceAudience;
27	import org.apache.carbondata.core.datamap.Segment;
28	import org.apache.carbondata.core.datastore.block.SegmentProperties;
29	import org.apache.carbondata.core.datastore.impl.FileFactory;
30	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
31	import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
32	import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
33	import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
34	import org.apache.carbondata.core.reader.CarbonIndexFileReader;
35	import org.apache.carbondata.core.scan.filter.FilterUtil;
36	import org.apache.carbondata.core.scan.filter.executer.FilterExecuter;
37	import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
38	import org.apache.carbondata.core.util.CarbonMetadataUtil;
39	import org.apache.carbondata.core.util.path.CarbonTablePath;
40	import org.apache.carbondata.format.BlockIndex;
41
42	@InterfaceAudience.Internal
43	public class StreamPruner {
44
45	private CarbonTable carbonTable;
46	private FilterExecuter filterExecuter;
47
48	private int totalFileNums = 0;	1✔
49
50	public StreamPruner(CarbonTable carbonTable) {	1✔
51	this.carbonTable = carbonTable;	1✔
52	}	1✔
53
54	public void init(FilterResolverIntf filterExp) {
55	if (filterExp != null) {	1✔
56	// cache all columns
57	List<CarbonColumn> minMaxCacheColumns = new ArrayList<>();	1✔
58	for (CarbonDimension dimension : carbonTable.getDimensions()) {	1✔
59	if (!dimension.isComplex()) {	1✔
60	minMaxCacheColumns.add(dimension);	1✔
61	}
62	}	1✔
63	minMaxCacheColumns.addAll(carbonTable.getMeasures());	1✔
64	// prepare cardinality of all dimensions
65	List<ColumnSchema> listOfColumns =	1✔
66	carbonTable.getTableInfo().getFactTable().getListOfColumns();	1✔
67	int[] columnCardinality = new int[listOfColumns.size()];	1✔
68	for (int index = 0; index < columnCardinality.length; index++) {	1✔
69	columnCardinality[index] = Integer.MAX_VALUE;	1✔
70	}
71	// initial filter executor
72	SegmentProperties segmentProperties =	1✔
73	new SegmentProperties(listOfColumns, columnCardinality);
74	filterExecuter = FilterUtil.getFilterExecuterTree(	1✔
75	filterExp, segmentProperties, null, minMaxCacheColumns);
76	}
77	}	1✔
78
79	public List<StreamFile> prune(List<Segment> segments) throws IOException {
80	if (filterExecuter == null) {	1✔
81	// if filter is null, list all steam files
82	return listAllStreamFiles(segments, false);	1✔
83	} else {
84	List<StreamFile> streamFileList = new ArrayList<>();	1✔
85	for (StreamFile streamFile : listAllStreamFiles(segments, true)) {	1✔
86	if (isScanRequire(streamFile)) {	1✔
87	// if stream file is required to scan
88	streamFileList.add(streamFile);	1✔
89	streamFile.setMinMaxIndex(null);	1✔
90	}
91	}	1✔
92	return streamFileList;	1✔
93	}
94	}
95
96	private boolean isScanRequire(StreamFile streamFile) {
97	// backward compatibility, old stream file without min/max index
98	if (streamFile.getMinMaxIndex() == null) {	1✔
99	return true;	×
100	}
101	byte[][] maxValue = streamFile.getMinMaxIndex().getMaxValues();	1✔
102	byte[][] minValue = streamFile.getMinMaxIndex().getMinValues();	1✔
103	BitSet bitSet = filterExecuter.isScanRequired(maxValue, minValue);	1✔
104	if (!bitSet.isEmpty()) {	1✔
105	return true;	1✔
106	} else {
107	return false;	1✔
108	}
109	}
110
111	// TODO optimize and move the code to StreamSegment , but it's in the streaming module.
112	private List<StreamFile> listAllStreamFiles(List<Segment> segments, boolean withMinMax)
113	throws IOException {
114	List<StreamFile> streamFileList = new ArrayList<>();	1✔
115	for (Segment segment : segments) {	1✔
116	String segmentDir = CarbonTablePath.getSegmentPath(	1✔
117	carbonTable.getAbsoluteTableIdentifier().getTablePath(), segment.getSegmentNo());	1✔
118	String indexFile = CarbonTablePath.getCarbonStreamIndexFilePath(segmentDir);	1✔
119	FileFactory.FileType fileType = FileFactory.getFileType(indexFile);	1✔
120	if (FileFactory.isFileExist(indexFile, fileType)) {	1✔
121	CarbonIndexFileReader indexReader = new CarbonIndexFileReader();	1✔
122	indexReader.openThriftReader(indexFile);	1✔
123	try {
124	while (indexReader.hasNext()) {	1✔
125	BlockIndex blockIndex = indexReader.readBlockIndexInfo();	1✔
126	String filePath = segmentDir + File.separator + blockIndex.getFile_name();	1✔
127	long length = blockIndex.getFile_size();	1✔
128	StreamFile streamFile = new StreamFile(segment.getSegmentNo(), filePath, length);	1✔
129	streamFileList.add(streamFile);	1✔
130	if (withMinMax) {	1✔
131	if (blockIndex.getBlock_index() != null	1✔
132	&& blockIndex.getBlock_index().getMin_max_index() != null) {	1✔
133	streamFile.setMinMaxIndex(CarbonMetadataUtil	1✔
134	.convertExternalMinMaxIndex(blockIndex.getBlock_index().getMin_max_index()));	1✔
135	}
136	}
137	}	1✔
138	} finally {
139	indexReader.closeThriftReader();	1✔
140	}	1✔
141	}
142	}	1✔
143	totalFileNums = streamFileList.size();	1✔
144	return streamFileList;	1✔
145	}
146
147	public int getTotalFileNums() {
148	return totalFileNums;	1✔
149	}
150	}

apache / carbondata / 8422

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