8459

Build # 8459

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/4c1e348a6790dcf318f0895d3849070ed71653fa">4c1e348a6<a href="https://github.com/apache/carbondata/commit/4c1e348a6790dcf318f0895d3849070ed71653fa">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/3debb970430b5443ee113aa04b434e6ba8ac3be8">3debb9704</a>">3debb9704</a><a href="https://github.com/apache/carbondata/commit/4c1e348a6790dcf318f0895d3849070ed71653fa"> into 73a5885a4">73a5885a4</a>

Run Details

13 of 13 new or added lines in 5 files covered. (100.0%)

53480 of 72168 relevant lines covered (74.1%)

0.88 hits per line

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

0.0

/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;	×
49
50	public StreamPruner(CarbonTable carbonTable) {	×
51	this.carbonTable = carbonTable;	×
52	}	×
53
54	public void init(FilterResolverIntf filterExp) {
55	if (filterExp != null) {	×
56	// cache all columns
57	List<CarbonColumn> minMaxCacheColumns = new ArrayList<>();	×
58	for (CarbonDimension dimension : carbonTable.getDimensions()) {	×
59	if (!dimension.isComplex()) {	×
60	minMaxCacheColumns.add(dimension);	×
61	}
62	}	×
63	minMaxCacheColumns.addAll(carbonTable.getMeasures());	×
64	// prepare cardinality of all dimensions
65	List<ColumnSchema> listOfColumns =	×
66	carbonTable.getTableInfo().getFactTable().getListOfColumns();	×
67	int[] columnCardinality = new int[listOfColumns.size()];	×
68	for (int index = 0; index < columnCardinality.length; index++) {	×
69	columnCardinality[index] = Integer.MAX_VALUE;	×
70	}
71	// initial filter executor
72	SegmentProperties segmentProperties =	×
73	new SegmentProperties(listOfColumns, columnCardinality);
74	filterExecuter = FilterUtil.getFilterExecuterTree(	×
75	filterExp, segmentProperties, null, minMaxCacheColumns);
76	}
77	}	×
78
79	public List<StreamFile> prune(List<Segment> segments) throws IOException {
80	if (filterExecuter == null) {	×
81	// if filter is null, list all steam files
82	return listAllStreamFiles(segments, false);	×
83	} else {
84	List<StreamFile> streamFileList = new ArrayList<>();	×
85	for (StreamFile streamFile : listAllStreamFiles(segments, true)) {	×
86	if (isScanRequire(streamFile)) {	×
87	// if stream file is required to scan
88	streamFileList.add(streamFile);	×
89	streamFile.setMinMaxIndex(null);	×
90	}
91	}	×
92	return streamFileList;	×
93	}
94	}
95
96	private boolean isScanRequire(StreamFile streamFile) {
97	// backward compatibility, old stream file without min/max index
98	if (streamFile.getMinMaxIndex() == null) {	×
99	return true;	×
100	}
101	byte[][] maxValue = streamFile.getMinMaxIndex().getMaxValues();	×
102	byte[][] minValue = streamFile.getMinMaxIndex().getMinValues();	×
103	BitSet bitSet = filterExecuter.isScanRequired(maxValue, minValue);	×
104	if (!bitSet.isEmpty()) {	×
105	return true;	×
106	} else {
107	return false;	×
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<>();	×
115	for (Segment segment : segments) {	×
116	String segmentDir = CarbonTablePath.getSegmentPath(	×
117	carbonTable.getAbsoluteTableIdentifier().getTablePath(), segment.getSegmentNo());	×
118	String indexFile = CarbonTablePath.getCarbonStreamIndexFilePath(segmentDir);	×
119	FileFactory.FileType fileType = FileFactory.getFileType(indexFile);	×
120	if (FileFactory.isFileExist(indexFile, fileType)) {	×
121	CarbonIndexFileReader indexReader = new CarbonIndexFileReader();	×
122	indexReader.openThriftReader(indexFile);	×
123	try {
124	while (indexReader.hasNext()) {	×
125	BlockIndex blockIndex = indexReader.readBlockIndexInfo();	×
126	String filePath = segmentDir + File.separator + blockIndex.getFile_name();	×
127	long length = blockIndex.getFile_size();	×
128	StreamFile streamFile = new StreamFile(segment.getSegmentNo(), filePath, length);	×
129	streamFileList.add(streamFile);	×
130	if (withMinMax) {	×
131	if (blockIndex.getBlock_index() != null	×
132	&& blockIndex.getBlock_index().getMin_max_index() != null) {	×
133	streamFile.setMinMaxIndex(CarbonMetadataUtil	×
134	.convertExternalMinMaxIndex(blockIndex.getBlock_index().getMin_max_index()));	×
135	}
136	}
137	}	×
138	} finally {
139	indexReader.closeThriftReader();	×
140	}	×
141	}
142	}	×
143	totalFileNums = streamFileList.size();	×
144	return streamFileList;	×
145	}
146
147	public int getTotalFileNums() {
148	return totalFileNums;	×
149	}
150	}

apache / carbondata / 8459

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