8341

Build # 8341

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/eeee74e90f5cb36d17c2325071d90407b3052c80">eeee74e90<a href="https://github.com/apache/carbondata/commit/eeee74e90f5cb36d17c2325071d90407b3052c80">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/81dd2b5c55fbb22779655aba9c5ea4fcb3a96c26">81dd2b5c5</a>">81dd2b5c5</a><a href="https://github.com/apache/carbondata/commit/eeee74e90f5cb36d17c2325071d90407b3052c80"> into 50248f51b">50248f51b</a>

Run Details

588 of 588 new or added lines in 63 files covered. (100.0%)

58240 of 72133 relevant lines covered (80.74%)

1.07 hits per line

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

74.58

/streaming/src/main/java/org/apache/carbondata/streaming/CarbonStreamInputFormat.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.streaming;

import java.io.IOException;
import java.lang.reflect.Constructor;

import org.apache.carbondata.core.cache.Cache;
import org.apache.carbondata.core.cache.dictionary.Dictionary;
import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.encoder.Encoding;
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.scan.complextypes.ArrayQueryType;
import org.apache.carbondata.core.scan.complextypes.PrimitiveQueryType;
import org.apache.carbondata.core.scan.complextypes.StructQueryType;
import org.apache.carbondata.core.scan.filter.GenericQueryType;
import org.apache.carbondata.core.scan.model.QueryModel;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.hadoop.InputMetricsStats;
import org.apache.carbondata.streaming.CarbonStreamUtils;

import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

/**
 * Stream input format
 */
public class CarbonStreamInputFormat extends FileInputFormat<Void, Object> {

  public static final String READ_BUFFER_SIZE = "carbon.stream.read.buffer.size";
  public static final String READ_BUFFER_SIZE_DEFAULT = "65536";
  public static final String STREAM_RECORD_READER_INSTANCE =
      "org.apache.carbondata.stream.CarbonStreamRecordReader";
  // return raw row for handoff
  private boolean useRawRow = false;

  public void setUseRawRow(boolean useRawRow) {
    this.useRawRow = useRawRow;
  }

  public void setInputMetricsStats(InputMetricsStats inputMetricsStats) {
    this.inputMetricsStats = inputMetricsStats;
  }

  public void setIsVectorReader(boolean vectorReader) {
    isVectorReader = vectorReader;
  }

  public void setModel(QueryModel model) {
    this.model = model;
  }

  // InputMetricsStats
  private InputMetricsStats inputMetricsStats;
  // vector reader
  private boolean isVectorReader;
  private QueryModel model;

  @Override
  public RecordReader<Void, Object> createRecordReader(InputSplit split, TaskAttemptContext context)
      throws IOException, InterruptedException {
    try {
      Constructor cons = CarbonStreamUtils
          .getConstructorWithReflection(STREAM_RECORD_READER_INSTANCE, boolean.class,
              InputMetricsStats.class, QueryModel.class, boolean.class);
      return (RecordReader) CarbonStreamUtils
          .getInstanceWithReflection(cons, isVectorReader, inputMetricsStats, model, useRawRow);

    } catch (Exception e) {
      throw new IOException(e);
    }
  }

  public static GenericQueryType[] getComplexDimensions(CarbonTable carbontable,
      CarbonColumn[] carbonColumns, Cache<DictionaryColumnUniqueIdentifier, Dictionary> cache)
      throws IOException {
    GenericQueryType[] queryTypes = new GenericQueryType[carbonColumns.length];
    for (int i = 0; i < carbonColumns.length; i++) {
      if (carbonColumns[i].isComplex()) {
        if (DataTypes.isArrayType(carbonColumns[i].getDataType())) {
          queryTypes[i] = new ArrayQueryType(carbonColumns[i].getColName(),
              carbonColumns[i].getColName(), i);
        } else if (DataTypes.isStructType(carbonColumns[i].getDataType())) {
          queryTypes[i] = new StructQueryType(carbonColumns[i].getColName(),
              carbonColumns[i].getColName(), i);
        } else {
          throw new UnsupportedOperationException(
              carbonColumns[i].getDataType().getName() + " is not supported");
        }

        fillChildren(carbontable, queryTypes[i], (CarbonDimension) carbonColumns[i], i, cache);
      }
    }

    return queryTypes;
  }

  private static void fillChildren(CarbonTable carbontable, GenericQueryType parentQueryType,
      CarbonDimension dimension, int parentBlockIndex,
      Cache<DictionaryColumnUniqueIdentifier, Dictionary> cache) throws IOException {
    for (int i = 0; i < dimension.getNumberOfChild(); i++) {
      CarbonDimension child = dimension.getListOfChildDimensions().get(i);
      DataType dataType = child.getDataType();
      GenericQueryType queryType = null;
      if (DataTypes.isArrayType(dataType)) {
        queryType =
            new ArrayQueryType(child.getColName(), dimension.getColName(), ++parentBlockIndex);

      } else if (DataTypes.isStructType(dataType)) {
        queryType =
            new StructQueryType(child.getColName(), dimension.getColName(), ++parentBlockIndex);
        parentQueryType.addChildren(queryType);
      } else {
        boolean isDirectDictionary =
            CarbonUtil.hasEncoding(child.getEncoder(), Encoding.DIRECT_DICTIONARY);
        boolean isDictionary =
            CarbonUtil.hasEncoding(child.getEncoder(), Encoding.DICTIONARY);
        Dictionary dictionary = null;
        if (isDictionary) {
          String dictionaryPath = carbontable.getTableInfo().getFactTable().getTableProperties()
              .get(CarbonCommonConstants.DICTIONARY_PATH);
          DictionaryColumnUniqueIdentifier dictionarIdentifier =
              new DictionaryColumnUniqueIdentifier(carbontable.getAbsoluteTableIdentifier(),
                  child.getColumnIdentifier(), child.getDataType(), dictionaryPath);
          dictionary = cache.get(dictionarIdentifier);
        }
        queryType =
            new PrimitiveQueryType(child.getColName(), dimension.getColName(), ++parentBlockIndex,
                child.getDataType(), 4, dictionary,
                isDirectDictionary);
      }
      parentQueryType.addChildren(queryType);
      if (child.getNumberOfChild() > 0) {
        fillChildren(carbontable, queryType, child, parentBlockIndex, cache);
      }
    }
  }
}

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.streaming;
19
20	import java.io.IOException;
21	import java.lang.reflect.Constructor;
22
23	import org.apache.carbondata.core.cache.Cache;
24	import org.apache.carbondata.core.cache.dictionary.Dictionary;
25	import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
26	import org.apache.carbondata.core.constants.CarbonCommonConstants;
27	import org.apache.carbondata.core.metadata.datatype.DataType;
28	import org.apache.carbondata.core.metadata.datatype.DataTypes;
29	import org.apache.carbondata.core.metadata.encoder.Encoding;
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.scan.complextypes.ArrayQueryType;
34	import org.apache.carbondata.core.scan.complextypes.PrimitiveQueryType;
35	import org.apache.carbondata.core.scan.complextypes.StructQueryType;
36	import org.apache.carbondata.core.scan.filter.GenericQueryType;
37	import org.apache.carbondata.core.scan.model.QueryModel;
38	import org.apache.carbondata.core.util.CarbonUtil;
39	import org.apache.carbondata.hadoop.InputMetricsStats;
40	import org.apache.carbondata.streaming.CarbonStreamUtils;
41
42	import org.apache.hadoop.mapreduce.InputSplit;
43	import org.apache.hadoop.mapreduce.RecordReader;
44	import org.apache.hadoop.mapreduce.TaskAttemptContext;
45	import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
46
47	/**
48	* Stream input format
49	*/
50	public class CarbonStreamInputFormat extends FileInputFormat<Void, Object> {	1✔
51
52	public static final String READ_BUFFER_SIZE = "carbon.stream.read.buffer.size";
53	public static final String READ_BUFFER_SIZE_DEFAULT = "65536";
54	public static final String STREAM_RECORD_READER_INSTANCE =
55	"org.apache.carbondata.stream.CarbonStreamRecordReader";
56	// return raw row for handoff
57	private boolean useRawRow = false;	1✔
58
59	public void setUseRawRow(boolean useRawRow) {
60	this.useRawRow = useRawRow;	1✔
61	}	1✔
62
63	public void setInputMetricsStats(InputMetricsStats inputMetricsStats) {
64	this.inputMetricsStats = inputMetricsStats;	1✔
65	}	1✔
66
67	public void setIsVectorReader(boolean vectorReader) {
68	isVectorReader = vectorReader;	1✔
69	}	1✔
70
71	public void setModel(QueryModel model) {
72	this.model = model;	1✔
73	}	1✔
74
75	// InputMetricsStats
76	private InputMetricsStats inputMetricsStats;
77	// vector reader
78	private boolean isVectorReader;
79	private QueryModel model;
80
81	@Override
82	public RecordReader<Void, Object> createRecordReader(InputSplit split, TaskAttemptContext context)
83	throws IOException, InterruptedException {
84	try {
85	Constructor cons = CarbonStreamUtils	1✔
86	.getConstructorWithReflection(STREAM_RECORD_READER_INSTANCE, boolean.class,	1✔
87	InputMetricsStats.class, QueryModel.class, boolean.class);
88	return (RecordReader) CarbonStreamUtils	1✔
89	.getInstanceWithReflection(cons, isVectorReader, inputMetricsStats, model, useRawRow);	1✔
90
91	} catch (Exception e) {	×
92	throw new IOException(e);	×
93	}
94	}
95
96	public static GenericQueryType[] getComplexDimensions(CarbonTable carbontable,
97	CarbonColumn[] carbonColumns, Cache<DictionaryColumnUniqueIdentifier, Dictionary> cache)
98	throws IOException {
99	GenericQueryType[] queryTypes = new GenericQueryType[carbonColumns.length];	1✔
100	for (int i = 0; i < carbonColumns.length; i++) {	1✔
101	if (carbonColumns[i].isComplex()) {	1✔
102	if (DataTypes.isArrayType(carbonColumns[i].getDataType())) {	1✔
103	queryTypes[i] = new ArrayQueryType(carbonColumns[i].getColName(),	×
104	carbonColumns[i].getColName(), i);	×
105	} else if (DataTypes.isStructType(carbonColumns[i].getDataType())) {	1✔
106	queryTypes[i] = new StructQueryType(carbonColumns[i].getColName(),	1✔
107	carbonColumns[i].getColName(), i);	1✔
108	} else {
109	throw new UnsupportedOperationException(	×
110	carbonColumns[i].getDataType().getName() + " is not supported");	×
111	}
112
113	fillChildren(carbontable, queryTypes[i], (CarbonDimension) carbonColumns[i], i, cache);	1✔
114	}
115	}
116
117	return queryTypes;	1✔
118	}
119
120	private static void fillChildren(CarbonTable carbontable, GenericQueryType parentQueryType,
121	CarbonDimension dimension, int parentBlockIndex,
122	Cache<DictionaryColumnUniqueIdentifier, Dictionary> cache) throws IOException {
123	for (int i = 0; i < dimension.getNumberOfChild(); i++) {	1✔
124	CarbonDimension child = dimension.getListOfChildDimensions().get(i);	1✔
125	DataType dataType = child.getDataType();	1✔
126	GenericQueryType queryType = null;	1✔
127	if (DataTypes.isArrayType(dataType)) {	1✔
128	queryType =	1✔
129	new ArrayQueryType(child.getColName(), dimension.getColName(), ++parentBlockIndex);	1✔
130
131	} else if (DataTypes.isStructType(dataType)) {	1✔
132	queryType =	×
133	new StructQueryType(child.getColName(), dimension.getColName(), ++parentBlockIndex);	×
134	parentQueryType.addChildren(queryType);	×
135	} else {
136	boolean isDirectDictionary =	1✔
137	CarbonUtil.hasEncoding(child.getEncoder(), Encoding.DIRECT_DICTIONARY);	1✔
138	boolean isDictionary =	1✔
139	CarbonUtil.hasEncoding(child.getEncoder(), Encoding.DICTIONARY);	1✔
140	Dictionary dictionary = null;	1✔
141	if (isDictionary) {	1✔
142	String dictionaryPath = carbontable.getTableInfo().getFactTable().getTableProperties()	×
143	.get(CarbonCommonConstants.DICTIONARY_PATH);	×
144	DictionaryColumnUniqueIdentifier dictionarIdentifier =	×
145	new DictionaryColumnUniqueIdentifier(carbontable.getAbsoluteTableIdentifier(),	×
146	child.getColumnIdentifier(), child.getDataType(), dictionaryPath);	×
147	dictionary = cache.get(dictionarIdentifier);	×
148	}
149	queryType =	1✔
150	new PrimitiveQueryType(child.getColName(), dimension.getColName(), ++parentBlockIndex,	1✔
151	child.getDataType(), 4, dictionary,	1✔
152	isDirectDictionary);
153	}
154	parentQueryType.addChildren(queryType);	1✔
155	if (child.getNumberOfChild() > 0) {	1✔
156	fillChildren(carbontable, queryType, child, parentBlockIndex, cache);	1✔
157	}
158	}
159	}	1✔
160	}

apache / carbondata / 8341

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