apache / carbondata / 370

Build # 370

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/c974374d8a00c6083c1393808436a6d732aa072c">c974374d8<a href="https://github.com/apache/carbondata/commit/c974374d8a00c6083c1393808436a6d732aa072c">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/f2b41c296c1e83421874625f5e9036c8e68e8b1d">f2b41c296</a>">f2b41c296</a><a href="https://github.com/apache/carbondata/commit/c974374d8a00c6083c1393808436a6d732aa072c"> into a9cc43411">a9cc43411</a>

Run Details

209 of 209 new or added lines in 47 files covered. (100.0%)

58261 of 72238 relevant lines covered (80.65%)

1.07 hits per line

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

78.57

/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonFileInputFormat.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.hadoop.api;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.LinkedList;
import java.util.List;

import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.annotations.InterfaceStability;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.schema.PartitionInfo;
import org.apache.carbondata.core.metadata.schema.SchemaReader;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.TableInfo;
import org.apache.carbondata.core.readcommitter.LatestFilesReadCommittedScope;
import org.apache.carbondata.core.readcommitter.ReadCommittedScope;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.hadoop.CarbonInputSplit;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;

/**
 * InputFormat for reading carbondata files without table level metadata support,
 * schema is inferred as following steps:
 * 1. read from schema file is exists
 * 2. read from data file footer
 *
 * @param <T>
 */
@InterfaceAudience.User
@InterfaceStability.Evolving
public class CarbonFileInputFormat<T> extends CarbonInputFormat<T> implements Serializable {

  // a cache for carbon table, it will be used in task side
  private CarbonTable carbonTable;

  public CarbonTable getOrCreateCarbonTable(Configuration configuration) throws IOException {
    CarbonTable carbonTableTemp;
    if (carbonTable == null) {
      // carbon table should be created either from deserialized table info (schema saved in
      // hive metastore) or by reading schema in HDFS (schema saved in HDFS)
      TableInfo tableInfo = getTableInfo(configuration);
      CarbonTable localCarbonTable;
      if (tableInfo != null) {
        localCarbonTable = CarbonTable.buildFromTableInfo(tableInfo);
      } else {
        String schemaPath = CarbonTablePath
            .getSchemaFilePath(getAbsoluteTableIdentifier(configuration).getTablePath());
        if (!FileFactory.isFileExist(schemaPath, FileFactory.getFileType(schemaPath))) {
          TableInfo tableInfoInfer =
              SchemaReader.inferSchema(getAbsoluteTableIdentifier(configuration), true);
          localCarbonTable = CarbonTable.buildFromTableInfo(tableInfoInfer);
        } else {
          localCarbonTable =
              SchemaReader.readCarbonTableFromStore(getAbsoluteTableIdentifier(configuration));
        }
      }
      this.carbonTable = localCarbonTable;
      return localCarbonTable;
    } else {
      carbonTableTemp = this.carbonTable;
      return carbonTableTemp;
    }
  }

  /**
   * {@inheritDoc}
   * Configurations FileInputFormat.INPUT_DIR
   * are used to get table path to read.
   *
   * @param job
   * @return List<InputSplit> list of CarbonInputSplit
   * @throws IOException
   */
  @Override
  public List<InputSplit> getSplits(JobContext job) throws IOException {
    CarbonTable carbonTable = getOrCreateCarbonTable(job.getConfiguration());
    if (null == carbonTable) {
      throw new IOException("Missing/Corrupt schema file for table.");
    }
    AbsoluteTableIdentifier identifier = carbonTable.getAbsoluteTableIdentifier();

    if (getValidateSegmentsToAccess(job.getConfiguration())) {
      // get all valid segments and set them into the configuration
      // check for externalTable segment (Segment_null)
      // process and resolve the expression

      ReadCommittedScope readCommittedScope = null;
      if (carbonTable.isTransactionalTable()) {
        readCommittedScope = new LatestFilesReadCommittedScope(
            identifier.getTablePath() + "/Fact/Part0/Segment_null/", job.getConfiguration());
      } else {
        readCommittedScope = getReadCommittedScope(job.getConfiguration());
        if (readCommittedScope == null) {
          readCommittedScope = new LatestFilesReadCommittedScope(identifier.getTablePath(), job
              .getConfiguration());
        } else {
          readCommittedScope.setConfiguration(job.getConfiguration());
        }
      }
      // this will be null in case of corrupt schema file.
      PartitionInfo partitionInfo = carbonTable.getPartitionInfo(carbonTable.getTableName());
      Expression filter = getFilterPredicates(job.getConfiguration());


      // if external table Segments are found, add it to the List
      List<Segment> externalTableSegments = new ArrayList<Segment>();
      Segment seg;
      if (carbonTable.isTransactionalTable()) {
        // SDK some cases write into the Segment Path instead of Table Path i.e. inside
        // the "Fact/Part0/Segment_null". The segment in this case is named as "null".
        // The table is denoted by default as a transactional table and goes through
        // the path of CarbonFileInputFormat. The above scenario is handled in the below code.
        seg = new Segment("null", null, readCommittedScope);
        externalTableSegments.add(seg);
      } else {
        LoadMetadataDetails[] loadMetadataDetails = readCommittedScope.getSegmentList();
        for (LoadMetadataDetails load : loadMetadataDetails) {
          seg = new Segment(load.getLoadName(), null, readCommittedScope);
          externalTableSegments.add(seg);
        }
      }
      // do block filtering and get split
      List<InputSplit> splits =
          getSplits(job, filter, externalTableSegments, null, partitionInfo, null);
      if (getColumnProjection(job.getConfiguration()) == null) {
        // If the user projection is empty, use default all columns as projections.
        // All column name will be filled inside getSplits, so can update only here.
        String[]  projectionColumns = projectAllColumns(carbonTable);
        setColumnProjection(job.getConfiguration(), projectionColumns);
      }
      return splits;
    }
    return null;
  }

  /**
   * {@inheritDoc}
   * Configurations FileInputFormat.INPUT_DIR, CarbonTableInputFormat.INPUT_SEGMENT_NUMBERS
   * are used to get table path to read.
   *
   * @return
   * @throws IOException
   */
  private List<InputSplit> getSplits(JobContext job, Expression expression,
      List<Segment> validSegments, BitSet matchedPartitions, PartitionInfo partitionInfo,
      List<Integer> oldPartitionIdList) throws IOException {

    numSegments = validSegments.size();
    List<InputSplit> result = new LinkedList<InputSplit>();

    // for each segment fetch blocks matching filter in Driver BTree
    List<CarbonInputSplit> dataBlocksOfSegment =
        getDataBlocksOfSegment(job, carbonTable, expression, matchedPartitions,
            validSegments, partitionInfo, oldPartitionIdList);
    numBlocks = dataBlocksOfSegment.size();
    result.addAll(dataBlocksOfSegment);
    return result;
  }
}

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.hadoop.api;
19
20	import java.io.IOException;
21	import java.io.Serializable;
22	import java.util.ArrayList;
23	import java.util.BitSet;
24	import java.util.LinkedList;
25	import java.util.List;
26
27	import org.apache.carbondata.common.annotations.InterfaceAudience;
28	import org.apache.carbondata.common.annotations.InterfaceStability;
29	import org.apache.carbondata.core.datamap.Segment;
30	import org.apache.carbondata.core.datastore.impl.FileFactory;
31	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
32	import org.apache.carbondata.core.metadata.schema.PartitionInfo;
33	import org.apache.carbondata.core.metadata.schema.SchemaReader;
34	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
35	import org.apache.carbondata.core.metadata.schema.table.TableInfo;
36	import org.apache.carbondata.core.readcommitter.LatestFilesReadCommittedScope;
37	import org.apache.carbondata.core.readcommitter.ReadCommittedScope;
38	import org.apache.carbondata.core.scan.expression.Expression;
39	import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
40	import org.apache.carbondata.core.util.path.CarbonTablePath;
41	import org.apache.carbondata.hadoop.CarbonInputSplit;
42
43	import org.apache.hadoop.conf.Configuration;
44	import org.apache.hadoop.mapreduce.InputSplit;
45	import org.apache.hadoop.mapreduce.JobContext;
46
47	/**
48	* InputFormat for reading carbondata files without table level metadata support,
49	* schema is inferred as following steps:
50	* 1. read from schema file is exists
51	* 2. read from data file footer
52	*
53	* @param <T>
54	*/
55	@InterfaceAudience.User
56	@InterfaceStability.Evolving
57	public class CarbonFileInputFormat<T> extends CarbonInputFormat<T> implements Serializable {	1✔
58
59	// a cache for carbon table, it will be used in task side
60	private CarbonTable carbonTable;
61
62	public CarbonTable getOrCreateCarbonTable(Configuration configuration) throws IOException {
63	CarbonTable carbonTableTemp;
64	if (carbonTable == null) {	1✔
65	// carbon table should be created either from deserialized table info (schema saved in
66	// hive metastore) or by reading schema in HDFS (schema saved in HDFS)
67	TableInfo tableInfo = getTableInfo(configuration);	1✔
68	CarbonTable localCarbonTable;
69	if (tableInfo != null) {	1✔
70	localCarbonTable = CarbonTable.buildFromTableInfo(tableInfo);	1✔
71	} else {
72	String schemaPath = CarbonTablePath	×
73	.getSchemaFilePath(getAbsoluteTableIdentifier(configuration).getTablePath());	×
74	if (!FileFactory.isFileExist(schemaPath, FileFactory.getFileType(schemaPath))) {	×
75	TableInfo tableInfoInfer =	×
76	SchemaReader.inferSchema(getAbsoluteTableIdentifier(configuration), true);	×
77	localCarbonTable = CarbonTable.buildFromTableInfo(tableInfoInfer);	×
78	} else {	×
79	localCarbonTable =	×
80	SchemaReader.readCarbonTableFromStore(getAbsoluteTableIdentifier(configuration));	×
81	}
82	}
83	this.carbonTable = localCarbonTable;	1✔
84	return localCarbonTable;	1✔
85	} else {
86	carbonTableTemp = this.carbonTable;	1✔
87	return carbonTableTemp;	1✔
88	}
89	}
90
91	/**
92	* {@inheritDoc}
93	* Configurations FileInputFormat.INPUT_DIR
94	* are used to get table path to read.
95	*
96	* @param job
97	* @return List<InputSplit> list of CarbonInputSplit
98	* @throws IOException
99	*/
100	@Override
101	public List<InputSplit> getSplits(JobContext job) throws IOException {
102	CarbonTable carbonTable = getOrCreateCarbonTable(job.getConfiguration());	1✔
103	if (null == carbonTable) {	1✔
104	throw new IOException("Missing/Corrupt schema file for table.");	×
105	}
106	AbsoluteTableIdentifier identifier = carbonTable.getAbsoluteTableIdentifier();	1✔
107
108	if (getValidateSegmentsToAccess(job.getConfiguration())) {	1✔
109	// get all valid segments and set them into the configuration
110	// check for externalTable segment (Segment_null)
111	// process and resolve the expression
112
113	ReadCommittedScope readCommittedScope = null;	1✔
114	if (carbonTable.isTransactionalTable()) {	1✔
115	readCommittedScope = new LatestFilesReadCommittedScope(	1✔
116	identifier.getTablePath() + "/Fact/Part0/Segment_null/", job.getConfiguration());	1✔
117	} else {
118	readCommittedScope = getReadCommittedScope(job.getConfiguration());	1✔
119	if (readCommittedScope == null) {	1✔
120	readCommittedScope = new LatestFilesReadCommittedScope(identifier.getTablePath(), job	1✔
121	.getConfiguration());	1✔
122	} else {
123	readCommittedScope.setConfiguration(job.getConfiguration());	×
124	}
125	}
126	// this will be null in case of corrupt schema file.
127	PartitionInfo partitionInfo = carbonTable.getPartitionInfo(carbonTable.getTableName());	1✔
128	Expression filter = getFilterPredicates(job.getConfiguration());	1✔
129
130
131	// if external table Segments are found, add it to the List
132	List<Segment> externalTableSegments = new ArrayList<Segment>();	1✔
133	Segment seg;
134	if (carbonTable.isTransactionalTable()) {	1✔
135	// SDK some cases write into the Segment Path instead of Table Path i.e. inside
136	// the "Fact/Part0/Segment_null". The segment in this case is named as "null".
137	// The table is denoted by default as a transactional table and goes through
138	// the path of CarbonFileInputFormat. The above scenario is handled in the below code.
139	seg = new Segment("null", null, readCommittedScope);	1✔
140	externalTableSegments.add(seg);	1✔
141	} else {
142	LoadMetadataDetails[] loadMetadataDetails = readCommittedScope.getSegmentList();	1✔
143	for (LoadMetadataDetails load : loadMetadataDetails) {	1✔
144	seg = new Segment(load.getLoadName(), null, readCommittedScope);	1✔
145	externalTableSegments.add(seg);	1✔
146	}
147	}
148	// do block filtering and get split
149	List<InputSplit> splits =	1✔
150	getSplits(job, filter, externalTableSegments, null, partitionInfo, null);	1✔
151	if (getColumnProjection(job.getConfiguration()) == null) {	1✔
152	// If the user projection is empty, use default all columns as projections.
153	// All column name will be filled inside getSplits, so can update only here.
154	String[] projectionColumns = projectAllColumns(carbonTable);	1✔
155	setColumnProjection(job.getConfiguration(), projectionColumns);	1✔
156	}
157	return splits;	1✔
158	}
159	return null;	×
160	}
161
162	/**
163	* {@inheritDoc}
164	* Configurations FileInputFormat.INPUT_DIR, CarbonTableInputFormat.INPUT_SEGMENT_NUMBERS
165	* are used to get table path to read.
166	*
167	* @return
168	* @throws IOException
169	*/
170	private List<InputSplit> getSplits(JobContext job, Expression expression,
171	List<Segment> validSegments, BitSet matchedPartitions, PartitionInfo partitionInfo,
172	List<Integer> oldPartitionIdList) throws IOException {
173
174	numSegments = validSegments.size();	1✔
175	List<InputSplit> result = new LinkedList<InputSplit>();	1✔
176
177	// for each segment fetch blocks matching filter in Driver BTree
178	List<CarbonInputSplit> dataBlocksOfSegment =	1✔
179	getDataBlocksOfSegment(job, carbonTable, expression, matchedPartitions,	1✔
180	validSegments, partitionInfo, oldPartitionIdList);
181	numBlocks = dataBlocksOfSegment.size();	1✔
182	result.addAll(dataBlocksOfSegment);	1✔
183	return result;	1✔
184	}
185	}

apache / carbondata / 370

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