3522

Build # 3522

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/820a501cadb6640fbccc3e21f058ae4e4518d621">820a501ca<a href="https://github.com/apache/carbondata/commit/820a501cadb6640fbccc3e21f058ae4e4518d621">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/32766b7cce5855347c81e1f5e3caebe99a2899e9">32766b7cc</a>">32766b7cc</a><a href="https://github.com/apache/carbondata/commit/820a501cadb6640fbccc3e21f058ae4e4518d621"> into df71291ec">df71291ec</a>

Run Details

395 of 395 new or added lines in 17 files covered. (100.0%)

64171 of 82248 relevant lines covered (78.02%)

1.03 hits per line

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

86.56

/streaming/src/main/java/org/apache/carbondata/streaming/CarbonStreamRecordWriter.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.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.List;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.compression.CompressorFactory;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.datastore.row.CarbonRow;
import org.apache.carbondata.core.metadata.blocklet.index.BlockletMinMaxIndex;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
import org.apache.carbondata.core.reader.CarbonHeaderReader;
import org.apache.carbondata.core.util.ByteUtil;
import org.apache.carbondata.core.util.CarbonMetadataUtil;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.DataTypeUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.format.FileHeader;
import org.apache.carbondata.processing.loading.BadRecordsLogger;
import org.apache.carbondata.processing.loading.BadRecordsLoggerProvider;
import org.apache.carbondata.processing.loading.CarbonDataLoadConfiguration;
import org.apache.carbondata.processing.loading.DataField;
import org.apache.carbondata.processing.loading.DataLoadProcessBuilder;
import org.apache.carbondata.processing.loading.converter.RowConverter;
import org.apache.carbondata.processing.loading.converter.impl.RowConverterImpl;
import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
import org.apache.carbondata.processing.loading.parser.RowParser;
import org.apache.carbondata.processing.loading.parser.impl.RowParserImpl;
import org.apache.carbondata.processing.store.writer.AbstractFactDataWriter;
import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
import org.apache.carbondata.streaming.segment.StreamSegment;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskID;
import org.apache.log4j.Logger;

/**
 * Stream record writer
 */
public class CarbonStreamRecordWriter extends RecordWriter<Void, Object> {

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(CarbonStreamRecordWriter.class.getName());

  // basic info
  private Configuration hadoopConf;
  private CarbonLoadModel carbonLoadModel;
  private CarbonDataLoadConfiguration configuration;
  private CarbonTable carbonTable;
  private int maxRowNums;
  private int maxCacheSize;

  // parser and converter
  private RowParser rowParser;
  private BadRecordsLogger badRecordLogger;
  private RowConverter converter;
  private CarbonRow currentRow = new CarbonRow(null);

  // encoder
  private DataField[] dataFields;
  private BitSet nullBitSet;
  private boolean[] isNoDictionaryDimensionColumn;
  private int dimensionWithComplexCount;
  private int measureCount;
  private DataType[] measureDataTypes;
  private StreamBlockletWriter output = null;
  private String compressorName;

  // data write
  private String segmentDir;
  private String fileName;
  private DataOutputStream outputStream;
  private boolean isFirstRow = true;
  private boolean hasException = false;

  // batch level stats collector
  private BlockletMinMaxIndex batchMinMaxIndex;
  private boolean isClosed = false;

  CarbonStreamRecordWriter(TaskAttemptContext job) throws IOException {
    initialize(job);
  }

  public CarbonStreamRecordWriter(TaskAttemptContext job, CarbonLoadModel carbonLoadModel)
      throws IOException {
    this.carbonLoadModel = carbonLoadModel;
    initialize(job);
  }

  private void initialize(TaskAttemptContext job) throws IOException {
    // set basic information
    hadoopConf = job.getConfiguration();
    if (carbonLoadModel == null) {
      carbonLoadModel = CarbonStreamOutputFormat.getCarbonLoadModel(hadoopConf);
      if (carbonLoadModel == null) {
        throw new IOException(
            "CarbonStreamRecordWriter require configuration: mapreduce.output.carbon.load.model");
      }
    }
    String segmentId = CarbonStreamOutputFormat.getSegmentId(hadoopConf);
    carbonLoadModel.setSegmentId(segmentId);
    carbonTable = carbonLoadModel.getCarbonDataLoadSchema().getCarbonTable();
    long taskNo = TaskID.forName(hadoopConf.get("mapred.tip.id")).getId();
    carbonLoadModel.setTaskNo("" + taskNo);
    configuration = DataLoadProcessBuilder.createConfiguration(carbonLoadModel);
    maxRowNums = hadoopConf.getInt(CarbonStreamOutputFormat.CARBON_STREAM_BLOCKLET_ROW_NUMS,
        CarbonStreamOutputFormat.CARBON_STREAM_BLOCKLET_ROW_NUMS_DEFAULT) - 1;
    maxCacheSize = hadoopConf.getInt(CarbonStreamOutputFormat.CARBON_STREAM_CACHE_SIZE,
        CarbonStreamOutputFormat.CARBON_STREAM_CACHE_SIZE_DEFAULT);

    segmentDir = CarbonTablePath.getSegmentPath(
        carbonTable.getAbsoluteTableIdentifier().getTablePath(), segmentId);
    fileName = CarbonTablePath.getCarbonDataFileName(0, taskNo, 0, 0, "0", segmentId);

    // initialize metadata
    isNoDictionaryDimensionColumn =
        CarbonDataProcessorUtil.getNoDictionaryMapping(configuration.getDataFields());
    dimensionWithComplexCount = configuration.getDimensionCount();
    measureCount = configuration.getMeasureCount();
    dataFields = configuration.getDataFields();
    measureDataTypes = new DataType[measureCount];
    for (int i = 0; i < measureCount; i++) {
      measureDataTypes[i] =
          dataFields[dimensionWithComplexCount + i].getColumn().getDataType();
    }
  }

  private void initializeAtFirstRow() throws IOException, InterruptedException {
    // initialize parser and converter
    rowParser = new RowParserImpl(dataFields, configuration);
    badRecordLogger = BadRecordsLoggerProvider.createBadRecordLogger(configuration);
    converter =
        new RowConverterImpl(configuration.getDataFields(), configuration, badRecordLogger, true);
    configuration.setCardinalityFinder(converter);
    converter.initialize();

    // initialize data writer and compressor
    String filePath = segmentDir + File.separator + fileName;
    FileFactory.FileType fileType = FileFactory.getFileType(filePath);
    CarbonFile carbonFile = FileFactory.getCarbonFile(filePath, fileType);
    if (carbonFile.exists()) {
      // if the file is existed, use the append api
      outputStream = FileFactory.getDataOutputStreamUsingAppend(filePath, fileType);
      // get the compressor from the fileheader. In legacy store,
      // the compressor name is not set and it use snappy compressor
      FileHeader header = new CarbonHeaderReader(filePath).readHeader();
      if (header.isSetCompressor_name()) {
        compressorName = header.getCompressor_name();
      } else {
        compressorName = CompressorFactory.NativeSupportedCompressor.SNAPPY.getName();
      }
    } else {
      // IF the file is not existed, use the create api
      outputStream = FileFactory.getDataOutputStream(filePath, fileType);
      compressorName = carbonTable.getTableInfo().getFactTable().getTableProperties().get(
          CarbonCommonConstants.COMPRESSOR);
      if (null == compressorName) {
        compressorName = CompressorFactory.getInstance().getCompressor().getName();
      }
      writeFileHeader();
    }

    // initialize encoder
    nullBitSet = new BitSet(dataFields.length);
    int rowBufferSize = hadoopConf.getInt(CarbonStreamOutputFormat.CARBON_ENCODER_ROW_BUFFER_SIZE,
        CarbonStreamOutputFormat.CARBON_ENCODER_ROW_BUFFER_SIZE_DEFAULT);
    output = new StreamBlockletWriter(maxCacheSize, maxRowNums, rowBufferSize,
        isNoDictionaryDimensionColumn.length, measureCount,
        measureDataTypes, compressorName);

    isFirstRow = false;
  }

  @Override public void write(Void key, Object value) throws IOException, InterruptedException {
    if (isFirstRow) {
      initializeAtFirstRow();
    }
    // null bit set
    nullBitSet.clear();
    Object[] rowData = (Object[]) value;
    currentRow.setRawData(rowData);
    // parse and convert row
    currentRow.setData(rowParser.parseRow(rowData));
    CarbonRow updatedCarbonRow = converter.convert(currentRow);
    if (updatedCarbonRow == null) {
      output.skipRow();
      currentRow.clearData();
    } else {
      for (int i = 0; i < dataFields.length; i++) {
        if (null == currentRow.getObject(i)) {
          nullBitSet.set(i);
        }
      }
      output.nextRow();
      byte[] b = nullBitSet.toByteArray();
      output.writeShort(b.length);
      if (b.length > 0) {
        output.writeBytes(b);
      }
      int dimCount = 0;
      Object columnValue;
      // primitive type dimension
      for (; dimCount < isNoDictionaryDimensionColumn.length; dimCount++) {
        columnValue = currentRow.getObject(dimCount);
        if (null != columnValue) {
          if (isNoDictionaryDimensionColumn[dimCount]) {
            byte[] col = (byte[]) columnValue;
            output.writeShort(col.length);
            output.writeBytes(col);
            output.dimStatsCollectors[dimCount].update(col);
          } else {
            output.writeInt((int) columnValue);
            output.dimStatsCollectors[dimCount].update(ByteUtil.toBytes((int) columnValue));
          }
        } else {
          output.dimStatsCollectors[dimCount].updateNull(0);
        }
      }
      // complex type dimension
      for (; dimCount < dimensionWithComplexCount; dimCount++) {
        columnValue = currentRow.getObject(dimCount);
        if (null != columnValue) {
          byte[] col = (byte[]) columnValue;
          output.writeShort(col.length);
          output.writeBytes(col);
        }
      }
      // measure
      DataType dataType;
      for (int msrCount = 0; msrCount < measureCount; msrCount++) {
        columnValue = currentRow.getObject(dimCount + msrCount);
        if (null != columnValue) {
          dataType = measureDataTypes[msrCount];
          if (dataType == DataTypes.BOOLEAN) {
            output.writeBoolean((boolean) columnValue);
            output.msrStatsCollectors[msrCount].update((byte) ((boolean) columnValue ? 1 : 0));
          } else if (dataType == DataTypes.SHORT) {
            output.writeShort((short) columnValue);
            output.msrStatsCollectors[msrCount].update((short) columnValue);
          } else if (dataType == DataTypes.INT) {
            output.writeInt((int) columnValue);
            output.msrStatsCollectors[msrCount].update((int) columnValue);
          } else if (dataType == DataTypes.LONG) {
            output.writeLong((long) columnValue);
            output.msrStatsCollectors[msrCount].update((long) columnValue);
          } else if (dataType == DataTypes.DOUBLE) {
            output.writeDouble((double) columnValue);
            output.msrStatsCollectors[msrCount].update((double) columnValue);
          } else if (DataTypes.isDecimal(dataType)) {
            BigDecimal val = (BigDecimal) columnValue;
            byte[] bigDecimalInBytes = DataTypeUtil.bigDecimalToByte(val);
            output.writeShort(bigDecimalInBytes.length);
            output.writeBytes(bigDecimalInBytes);
            output.msrStatsCollectors[msrCount].update((BigDecimal) columnValue);
          } else {
            String msg =
                "unsupported data type:" + dataFields[dimCount + msrCount].getColumn().getDataType()
                .getName();
            LOGGER.error(msg);
            throw new IOException(msg);
          }
        } else {
          output.msrStatsCollectors[msrCount].updateNull(0);
        }
      }
    }

    if (output.isFull()) {
      appendBlockletToDataFile();
    }
  }

  private void writeFileHeader() throws IOException {
    List<ColumnSchema> wrapperColumnSchemaList = CarbonUtil
        .getColumnSchemaList(carbonTable.getDimensionByTableName(carbonTable.getTableName()),
            carbonTable.getMeasureByTableName(carbonTable.getTableName()));
    int[] dimLensWithComplex = new int[wrapperColumnSchemaList.size()];
    for (int i = 0; i < dimLensWithComplex.length; i++) {
      dimLensWithComplex[i] = Integer.MAX_VALUE;
    }
    int[] dictionaryColumnCardinality =
        CarbonUtil.getFormattedCardinality(dimLensWithComplex, wrapperColumnSchemaList);
    List<Integer> cardinality = new ArrayList<>();
    List<org.apache.carbondata.format.ColumnSchema> columnSchemaList = AbstractFactDataWriter
        .getColumnSchemaListAndCardinality(cardinality, dictionaryColumnCardinality,
            wrapperColumnSchemaList);
    FileHeader fileHeader =
        CarbonMetadataUtil.getFileHeader(true, columnSchemaList, System.currentTimeMillis());
    fileHeader.setIs_footer_present(false);
    fileHeader.setIs_splitable(true);
    fileHeader.setSync_marker(CarbonStreamOutputFormat.CARBON_SYNC_MARKER);
    fileHeader.setCompressor_name(compressorName);
    outputStream.write(CarbonUtil.getByteArray(fileHeader));
  }

  /**
   * write a blocklet to file
   */
  private void appendBlockletToDataFile() throws IOException {
    if (output.getRowIndex() == -1) {
      return;
    }
    output.apppendBlocklet(outputStream);
    outputStream.flush();
    if (!isClosed) {
      batchMinMaxIndex = StreamSegment.mergeBlockletMinMax(
          batchMinMaxIndex, output.generateBlockletMinMax(), measureDataTypes);
    }
    // reset data
    output.reset();
  }

  public BlockletMinMaxIndex getBatchMinMaxIndex() {
    if (output == null) {
      return StreamSegment.mergeBlockletMinMax(
          batchMinMaxIndex, null, measureDataTypes);
    }
    return StreamSegment.mergeBlockletMinMax(
        batchMinMaxIndex, output.generateBlockletMinMax(), measureDataTypes);
  }

  public DataType[] getMeasureDataTypes() {
    return measureDataTypes;
  }

  @Override
  public void close(TaskAttemptContext context) throws IOException, InterruptedException {
    try {
      isClosed = true;
      // append remain buffer data
      if (!hasException && !isFirstRow) {
        appendBlockletToDataFile();
        converter.finish();
      }
    } finally {
      // close resource
      CarbonUtil.closeStreams(outputStream);
      if (output != null) {
        output.close();
      }
      if (badRecordLogger != null) {
        badRecordLogger.closeStreams();
      }
    }
  }

  public String getSegmentDir() {
    return segmentDir;
  }

  public String getFileName() {
    return fileName;
  }

  public void setHasException(boolean hasException) {
    this.hasException = hasException;
  }
}

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.DataOutputStream;
21	import java.io.File;
22	import java.io.IOException;
23	import java.math.BigDecimal;
24	import java.util.ArrayList;
25	import java.util.BitSet;
26	import java.util.List;
27
28	import org.apache.carbondata.common.logging.LogServiceFactory;
29	import org.apache.carbondata.core.constants.CarbonCommonConstants;
30	import org.apache.carbondata.core.datastore.compression.CompressorFactory;
31	import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
32	import org.apache.carbondata.core.datastore.impl.FileFactory;
33	import org.apache.carbondata.core.datastore.row.CarbonRow;
34	import org.apache.carbondata.core.metadata.blocklet.index.BlockletMinMaxIndex;
35	import org.apache.carbondata.core.metadata.datatype.DataType;
36	import org.apache.carbondata.core.metadata.datatype.DataTypes;
37	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
38	import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
39	import org.apache.carbondata.core.reader.CarbonHeaderReader;
40	import org.apache.carbondata.core.util.ByteUtil;
41	import org.apache.carbondata.core.util.CarbonMetadataUtil;
42	import org.apache.carbondata.core.util.CarbonUtil;
43	import org.apache.carbondata.core.util.DataTypeUtil;
44	import org.apache.carbondata.core.util.path.CarbonTablePath;
45	import org.apache.carbondata.format.FileHeader;
46	import org.apache.carbondata.processing.loading.BadRecordsLogger;
47	import org.apache.carbondata.processing.loading.BadRecordsLoggerProvider;
48	import org.apache.carbondata.processing.loading.CarbonDataLoadConfiguration;
49	import org.apache.carbondata.processing.loading.DataField;
50	import org.apache.carbondata.processing.loading.DataLoadProcessBuilder;
51	import org.apache.carbondata.processing.loading.converter.RowConverter;
52	import org.apache.carbondata.processing.loading.converter.impl.RowConverterImpl;
53	import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
54	import org.apache.carbondata.processing.loading.parser.RowParser;
55	import org.apache.carbondata.processing.loading.parser.impl.RowParserImpl;
56	import org.apache.carbondata.processing.store.writer.AbstractFactDataWriter;
57	import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
58	import org.apache.carbondata.streaming.segment.StreamSegment;
59
60	import org.apache.hadoop.conf.Configuration;
61	import org.apache.hadoop.mapreduce.RecordWriter;
62	import org.apache.hadoop.mapreduce.TaskAttemptContext;
63	import org.apache.hadoop.mapreduce.TaskID;
64	import org.apache.log4j.Logger;
65
66	/**
67	* Stream record writer
68	*/
69	public class CarbonStreamRecordWriter extends RecordWriter<Void, Object> {
70
71	private static final Logger LOGGER =	1✔
72	LogServiceFactory.getLogService(CarbonStreamRecordWriter.class.getName());	1✔
73
74	// basic info
75	private Configuration hadoopConf;
76	private CarbonLoadModel carbonLoadModel;
77	private CarbonDataLoadConfiguration configuration;
78	private CarbonTable carbonTable;
79	private int maxRowNums;
80	private int maxCacheSize;
81
82	// parser and converter
83	private RowParser rowParser;
84	private BadRecordsLogger badRecordLogger;
85	private RowConverter converter;
86	private CarbonRow currentRow = new CarbonRow(null);	1✔
87
88	// encoder
89	private DataField[] dataFields;
90	private BitSet nullBitSet;
91	private boolean[] isNoDictionaryDimensionColumn;
92	private int dimensionWithComplexCount;
93	private int measureCount;
94	private DataType[] measureDataTypes;
95	private StreamBlockletWriter output = null;	1✔
96	private String compressorName;
97
98	// data write
99	private String segmentDir;
100	private String fileName;
101	private DataOutputStream outputStream;
102	private boolean isFirstRow = true;	1✔
103	private boolean hasException = false;	1✔
104
105	// batch level stats collector
106	private BlockletMinMaxIndex batchMinMaxIndex;
107	private boolean isClosed = false;	1✔
108
109	CarbonStreamRecordWriter(TaskAttemptContext job) throws IOException {	×
110	initialize(job);	×
111	}	×
112
113	public CarbonStreamRecordWriter(TaskAttemptContext job, CarbonLoadModel carbonLoadModel)
114	throws IOException {	1✔
115	this.carbonLoadModel = carbonLoadModel;	1✔
116	initialize(job);	1✔
117	}	1✔
118
119	private void initialize(TaskAttemptContext job) throws IOException {
120	// set basic information
121	hadoopConf = job.getConfiguration();	1✔
122	if (carbonLoadModel == null) {	1✔
123	carbonLoadModel = CarbonStreamOutputFormat.getCarbonLoadModel(hadoopConf);	×
124	if (carbonLoadModel == null) {	×
125	throw new IOException(	×
126	"CarbonStreamRecordWriter require configuration: mapreduce.output.carbon.load.model");
127	}
128	}
129	String segmentId = CarbonStreamOutputFormat.getSegmentId(hadoopConf);	1✔
130	carbonLoadModel.setSegmentId(segmentId);	1✔
131	carbonTable = carbonLoadModel.getCarbonDataLoadSchema().getCarbonTable();	1✔
132	long taskNo = TaskID.forName(hadoopConf.get("mapred.tip.id")).getId();	1✔
133	carbonLoadModel.setTaskNo("" + taskNo);	1✔
134	configuration = DataLoadProcessBuilder.createConfiguration(carbonLoadModel);	1✔
135	maxRowNums = hadoopConf.getInt(CarbonStreamOutputFormat.CARBON_STREAM_BLOCKLET_ROW_NUMS,	1✔
136	CarbonStreamOutputFormat.CARBON_STREAM_BLOCKLET_ROW_NUMS_DEFAULT) - 1;
137	maxCacheSize = hadoopConf.getInt(CarbonStreamOutputFormat.CARBON_STREAM_CACHE_SIZE,	1✔
138	CarbonStreamOutputFormat.CARBON_STREAM_CACHE_SIZE_DEFAULT);
139
140	segmentDir = CarbonTablePath.getSegmentPath(	1✔
141	carbonTable.getAbsoluteTableIdentifier().getTablePath(), segmentId);	1✔
142	fileName = CarbonTablePath.getCarbonDataFileName(0, taskNo, 0, 0, "0", segmentId);	1✔
143
144	// initialize metadata
145	isNoDictionaryDimensionColumn =	1✔
146	CarbonDataProcessorUtil.getNoDictionaryMapping(configuration.getDataFields());	1✔
147	dimensionWithComplexCount = configuration.getDimensionCount();	1✔
148	measureCount = configuration.getMeasureCount();	1✔
149	dataFields = configuration.getDataFields();	1✔
150	measureDataTypes = new DataType[measureCount];	1✔
151	for (int i = 0; i < measureCount; i++) {	1✔
152	measureDataTypes[i] =	1✔
153	dataFields[dimensionWithComplexCount + i].getColumn().getDataType();	1✔
154	}
155	}	1✔
156
157	private void initializeAtFirstRow() throws IOException, InterruptedException {
158	// initialize parser and converter
159	rowParser = new RowParserImpl(dataFields, configuration);	1✔
160	badRecordLogger = BadRecordsLoggerProvider.createBadRecordLogger(configuration);	1✔
161	converter =	1✔
162	new RowConverterImpl(configuration.getDataFields(), configuration, badRecordLogger, true);	1✔
163	configuration.setCardinalityFinder(converter);	1✔
164	converter.initialize();	1✔
165
166	// initialize data writer and compressor
167	String filePath = segmentDir + File.separator + fileName;	1✔
168	FileFactory.FileType fileType = FileFactory.getFileType(filePath);	1✔
169	CarbonFile carbonFile = FileFactory.getCarbonFile(filePath, fileType);	1✔
170	if (carbonFile.exists()) {	1✔
171	// if the file is existed, use the append api
172	outputStream = FileFactory.getDataOutputStreamUsingAppend(filePath, fileType);	1✔
173	// get the compressor from the fileheader. In legacy store,
174	// the compressor name is not set and it use snappy compressor
175	FileHeader header = new CarbonHeaderReader(filePath).readHeader();	1✔
176	if (header.isSetCompressor_name()) {	1✔
177	compressorName = header.getCompressor_name();	1✔
178	} else {
179	compressorName = CompressorFactory.NativeSupportedCompressor.SNAPPY.getName();	×
180	}
181	} else {	1✔
182	// IF the file is not existed, use the create api
183	outputStream = FileFactory.getDataOutputStream(filePath, fileType);	1✔
184	compressorName = carbonTable.getTableInfo().getFactTable().getTableProperties().get(	1✔
185	CarbonCommonConstants.COMPRESSOR);
186	if (null == compressorName) {	1✔
187	compressorName = CompressorFactory.getInstance().getCompressor().getName();	1✔
188	}
189	writeFileHeader();	1✔
190	}
191
192	// initialize encoder
193	nullBitSet = new BitSet(dataFields.length);	1✔
194	int rowBufferSize = hadoopConf.getInt(CarbonStreamOutputFormat.CARBON_ENCODER_ROW_BUFFER_SIZE,	1✔
195	CarbonStreamOutputFormat.CARBON_ENCODER_ROW_BUFFER_SIZE_DEFAULT);
196	output = new StreamBlockletWriter(maxCacheSize, maxRowNums, rowBufferSize,	1✔
197	isNoDictionaryDimensionColumn.length, measureCount,
198	measureDataTypes, compressorName);
199
200	isFirstRow = false;	1✔
201	}	1✔
202
203	@Override public void write(Void key, Object value) throws IOException, InterruptedException {
204	if (isFirstRow) {	1✔
205	initializeAtFirstRow();	1✔
206	}
207	// null bit set
208	nullBitSet.clear();	1✔
209	Object[] rowData = (Object[]) value;	1✔
210	currentRow.setRawData(rowData);	1✔
211	// parse and convert row
212	currentRow.setData(rowParser.parseRow(rowData));	1✔
213	CarbonRow updatedCarbonRow = converter.convert(currentRow);	1✔
214	if (updatedCarbonRow == null) {	1✔
215	output.skipRow();	1✔
216	currentRow.clearData();	1✔
217	} else {
218	for (int i = 0; i < dataFields.length; i++) {	1✔
219	if (null == currentRow.getObject(i)) {	1✔
220	nullBitSet.set(i);	1✔
221	}
222	}
223	output.nextRow();	1✔
224	byte[] b = nullBitSet.toByteArray();	1✔
225	output.writeShort(b.length);	1✔
226	if (b.length > 0) {	1✔
227	output.writeBytes(b);	1✔
228	}
229	int dimCount = 0;	1✔
230	Object columnValue;
231	// primitive type dimension
232	for (; dimCount < isNoDictionaryDimensionColumn.length; dimCount++) {	1✔
233	columnValue = currentRow.getObject(dimCount);	1✔
234	if (null != columnValue) {	1✔
235	if (isNoDictionaryDimensionColumn[dimCount]) {	1✔
236	byte[] col = (byte[]) columnValue;	1✔
237	output.writeShort(col.length);	1✔
238	output.writeBytes(col);	1✔
239	output.dimStatsCollectors[dimCount].update(col);	1✔
240	} else {	1✔
241	output.writeInt((int) columnValue);	1✔
242	output.dimStatsCollectors[dimCount].update(ByteUtil.toBytes((int) columnValue));	1✔
243	}
244	} else {
245	output.dimStatsCollectors[dimCount].updateNull(0);	×
246	}
247	}
248	// complex type dimension
249	for (; dimCount < dimensionWithComplexCount; dimCount++) {	1✔
250	columnValue = currentRow.getObject(dimCount);	1✔
251	if (null != columnValue) {	1✔
252	byte[] col = (byte[]) columnValue;	1✔
253	output.writeShort(col.length);	1✔
254	output.writeBytes(col);	1✔
255	}
256	}
257	// measure
258	DataType dataType;
259	for (int msrCount = 0; msrCount < measureCount; msrCount++) {	1✔
260	columnValue = currentRow.getObject(dimCount + msrCount);	1✔
261	if (null != columnValue) {	1✔
262	dataType = measureDataTypes[msrCount];	1✔
263	if (dataType == DataTypes.BOOLEAN) {	1✔
264	output.writeBoolean((boolean) columnValue);	×
265	output.msrStatsCollectors[msrCount].update((byte) ((boolean) columnValue ? 1 : 0));	×
266	} else if (dataType == DataTypes.SHORT) {	1✔
267	output.writeShort((short) columnValue);	×
268	output.msrStatsCollectors[msrCount].update((short) columnValue);	×
269	} else if (dataType == DataTypes.INT) {	1✔
270	output.writeInt((int) columnValue);	1✔
271	output.msrStatsCollectors[msrCount].update((int) columnValue);	1✔
272	} else if (dataType == DataTypes.LONG) {	1✔
273	output.writeLong((long) columnValue);	×
274	output.msrStatsCollectors[msrCount].update((long) columnValue);	×
275	} else if (dataType == DataTypes.DOUBLE) {	1✔
276	output.writeDouble((double) columnValue);	1✔
277	output.msrStatsCollectors[msrCount].update((double) columnValue);	1✔
278	} else if (DataTypes.isDecimal(dataType)) {	1✔
279	BigDecimal val = (BigDecimal) columnValue;	1✔
280	byte[] bigDecimalInBytes = DataTypeUtil.bigDecimalToByte(val);	1✔
281	output.writeShort(bigDecimalInBytes.length);	1✔
282	output.writeBytes(bigDecimalInBytes);	1✔
283	output.msrStatsCollectors[msrCount].update((BigDecimal) columnValue);	1✔
284	} else {	1✔
285	String msg =	×
286	"unsupported data type:" + dataFields[dimCount + msrCount].getColumn().getDataType()	×
287	.getName();	×
288	LOGGER.error(msg);	×
289	throw new IOException(msg);	×
290	}
291	} else {
292	output.msrStatsCollectors[msrCount].updateNull(0);	1✔
293	}
294	}
295	}
296
297	if (output.isFull()) {	1✔
298	appendBlockletToDataFile();	×
299	}
300	}	1✔
301
302	private void writeFileHeader() throws IOException {
303	List<ColumnSchema> wrapperColumnSchemaList = CarbonUtil	1✔
304	.getColumnSchemaList(carbonTable.getDimensionByTableName(carbonTable.getTableName()),	1✔
305	carbonTable.getMeasureByTableName(carbonTable.getTableName()));	1✔
306	int[] dimLensWithComplex = new int[wrapperColumnSchemaList.size()];	1✔
307	for (int i = 0; i < dimLensWithComplex.length; i++) {	1✔
308	dimLensWithComplex[i] = Integer.MAX_VALUE;	1✔
309	}
310	int[] dictionaryColumnCardinality =	1✔
311	CarbonUtil.getFormattedCardinality(dimLensWithComplex, wrapperColumnSchemaList);	1✔
312	List<Integer> cardinality = new ArrayList<>();	1✔
313	List<org.apache.carbondata.format.ColumnSchema> columnSchemaList = AbstractFactDataWriter	1✔
314	.getColumnSchemaListAndCardinality(cardinality, dictionaryColumnCardinality,	1✔
315	wrapperColumnSchemaList);
316	FileHeader fileHeader =	1✔
317	CarbonMetadataUtil.getFileHeader(true, columnSchemaList, System.currentTimeMillis());	1✔
318	fileHeader.setIs_footer_present(false);	1✔
319	fileHeader.setIs_splitable(true);	1✔
320	fileHeader.setSync_marker(CarbonStreamOutputFormat.CARBON_SYNC_MARKER);	1✔
321	fileHeader.setCompressor_name(compressorName);	1✔
322	outputStream.write(CarbonUtil.getByteArray(fileHeader));	1✔
323	}	1✔
324
325	/**
326	* write a blocklet to file
327	*/
328	private void appendBlockletToDataFile() throws IOException {
329	if (output.getRowIndex() == -1) {	1✔
330	return;	×
331	}
332	output.apppendBlocklet(outputStream);	1✔
333	outputStream.flush();	1✔
334	if (!isClosed) {	1✔
335	batchMinMaxIndex = StreamSegment.mergeBlockletMinMax(	×
336	batchMinMaxIndex, output.generateBlockletMinMax(), measureDataTypes);	×
337	}
338	// reset data
339	output.reset();	1✔
340	}	1✔
341
342	public BlockletMinMaxIndex getBatchMinMaxIndex() {
343	if (output == null) {	1✔
344	return StreamSegment.mergeBlockletMinMax(	1✔
345	batchMinMaxIndex, null, measureDataTypes);
346	}
347	return StreamSegment.mergeBlockletMinMax(	1✔
348	batchMinMaxIndex, output.generateBlockletMinMax(), measureDataTypes);	1✔
349	}
350
351	public DataType[] getMeasureDataTypes() {
352	return measureDataTypes;	×
353	}
354
355	@Override
356	public void close(TaskAttemptContext context) throws IOException, InterruptedException {
357	try {
358	isClosed = true;	1✔
359	// append remain buffer data
360	if (!hasException && !isFirstRow) {	1✔
361	appendBlockletToDataFile();	1✔
362	converter.finish();	1✔
363	}
364	} finally {
365	// close resource
366	CarbonUtil.closeStreams(outputStream);	1✔
367	if (output != null) {	1✔
368	output.close();	1✔
369	}
370	if (badRecordLogger != null) {	1✔
371	badRecordLogger.closeStreams();	1✔
372	}
373	}	×
374	}	1✔
375
376	public String getSegmentDir() {
377	return segmentDir;	1✔
378	}
379
380	public String getFileName() {
381	return fileName;	1✔
382	}
383
384	public void setHasException(boolean hasException) {
385	this.hasException = hasException;	1✔
386	}	1✔
387	}

apache / carbondata / 3522

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