10921

Build # 10921

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/e4936965b3a6066cf856fb694cfdcff8295d3ce9">e4936965b<a href="https://github.com/apache/carbondata/commit/e4936965b3a6066cf856fb694cfdcff8295d3ce9">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/1752ec95f35f7becca4be140b59649d472b1b4e6">1752ec95f</a>">1752ec95f</a><a href="https://github.com/apache/carbondata/commit/e4936965b3a6066cf856fb694cfdcff8295d3ce9"> into 3f98c5123">3f98c5123</a>

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

62751 of 79172 relevant lines covered (79.26%)

1.05 hits per line

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

78.48

/processing/src/main/java/org/apache/carbondata/processing/datatypes/PrimitiveDataType.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.processing.datatypes;

import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.cache.Cache;
import org.apache.carbondata.core.cache.CacheProvider;
import org.apache.carbondata.core.cache.CacheType;
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.datastore.ColumnType;
import org.apache.carbondata.core.datastore.row.ComplexColumnInfo;
import org.apache.carbondata.core.devapi.BiDictionary;
import org.apache.carbondata.core.devapi.DictionaryGenerationException;
import org.apache.carbondata.core.dictionary.client.DictionaryClient;
import org.apache.carbondata.core.dictionary.generator.key.DictionaryMessage;
import org.apache.carbondata.core.dictionary.generator.key.DictionaryMessageType;
import org.apache.carbondata.core.keygenerator.KeyGenException;
import org.apache.carbondata.core.keygenerator.KeyGenerator;
import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryKeyGeneratorFactory;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
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.column.CarbonColumn;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
import org.apache.carbondata.core.util.ByteUtil;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.DataTypeUtil;
import org.apache.carbondata.processing.loading.converter.BadRecordLogHolder;
import org.apache.carbondata.processing.loading.dictionary.DictionaryServerClientDictionary;
import org.apache.carbondata.processing.loading.dictionary.DirectDictionary;
import org.apache.carbondata.processing.loading.dictionary.PreCreatedDictionary;
import org.apache.carbondata.processing.loading.exception.CarbonDataLoadingException;
import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;

import org.apache.log4j.Logger;

/**
 * Primitive DataType stateless object used in data loading
 */
public class PrimitiveDataType implements GenericDataType<Object> {

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

  /**
   * surrogate index
   */
  private int index;

  /**
   * column name
   */
  private String name;

  /**
   * column parent name
   */
  private String parentName;

  /**
   * column unique id
   */
  private String columnId;

  /**
   * key size
   */
  private int keySize;

  /**
   * array index
   */
  private int outputArrayIndex;

  /**
   * data counter
   */
  private int dataCounter;

  private BiDictionary<Integer, Object> dictionaryGenerator;

  private CarbonDimension carbonDimension;

  private boolean isDictionary;

  private String nullFormat;

  private boolean isDirectDictionary;

  private DataType dataType;

  private PrimitiveDataType(int outputArrayIndex, int dataCounter) {
    this.outputArrayIndex = outputArrayIndex;
    this.dataCounter = dataCounter;
  }

  /**
   * constructor
   *
   * @param name
   * @param parentName
   * @param columnId
   * @param isDictionary
   */
  public PrimitiveDataType(String name, DataType dataType, String parentName, String columnId,
      boolean isDictionary, String nullFormat) {
    this.name = name;
    this.parentName = parentName;
    this.columnId = columnId;
    this.isDictionary = isDictionary;
    this.nullFormat = nullFormat;
    this.dataType = dataType;
  }

  /**
   * Constructor
   * @param carbonColumn
   * @param parentName
   * @param columnId
   * @param carbonDimension
   * @param absoluteTableIdentifier
   * @param client
   * @param useOnePass
   * @param localCache
   * @param nullFormat
   */
  public PrimitiveDataType(CarbonColumn carbonColumn, String parentName, String columnId,
      CarbonDimension carbonDimension, AbsoluteTableIdentifier absoluteTableIdentifier,
      DictionaryClient client, Boolean useOnePass, Map<Object, Integer> localCache,
      String nullFormat) {
    this.name = carbonColumn.getColName();
    this.parentName = parentName;
    this.columnId = columnId;
    this.carbonDimension = carbonDimension;
    this.isDictionary = isDictionaryDimension(carbonDimension);
    this.nullFormat = nullFormat;
    this.dataType = carbonColumn.getDataType();

    DictionaryColumnUniqueIdentifier identifier =
        new DictionaryColumnUniqueIdentifier(absoluteTableIdentifier,
            carbonDimension.getColumnIdentifier(), carbonDimension.getDataType());
    try {
      if (carbonDimension.hasEncoding(Encoding.DIRECT_DICTIONARY)
          || carbonColumn.getDataType() == DataTypes.DATE) {
        dictionaryGenerator = new DirectDictionary(DirectDictionaryKeyGeneratorFactory
            .getDirectDictionaryGenerator(carbonDimension.getDataType(),
                getDateFormat(carbonDimension)));
        isDirectDictionary = true;
      } else if (carbonDimension.hasEncoding(Encoding.DICTIONARY)) {
        CacheProvider cacheProvider = CacheProvider.getInstance();
        Cache<DictionaryColumnUniqueIdentifier, Dictionary> cache =
            cacheProvider.createCache(CacheType.REVERSE_DICTIONARY);
        Dictionary dictionary = null;
        if (useOnePass) {
          if (CarbonUtil.isFileExistsForGivenColumn(identifier)) {
            dictionary = cache.get(identifier);
          }
          DictionaryMessage dictionaryMessage = new DictionaryMessage();
          dictionaryMessage.setColumnName(carbonDimension.getColName());
          // for table initialization
          dictionaryMessage
              .setTableUniqueId(absoluteTableIdentifier.getCarbonTableIdentifier().getTableId());
          dictionaryMessage.setData("0");
          // for generate dictionary
          dictionaryMessage.setType(DictionaryMessageType.DICT_GENERATION);
          dictionaryGenerator = new DictionaryServerClientDictionary(dictionary, client,
              dictionaryMessage, localCache);
        } else {
          dictionary = cache.get(identifier);
          dictionaryGenerator = new PreCreatedDictionary(dictionary);
        }
      }
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  /**
   * get dateformat
   * @param carbonDimension
   * @return
   */
  private String getDateFormat(CarbonDimension carbonDimension) {
    String format;
    String dateFormat = null;
    if (this.carbonDimension.getDataType() == DataTypes.DATE) {
      dateFormat = carbonDimension.getDateFormat();
    }
    if (dateFormat != null && !dateFormat.trim().isEmpty()) {
      format = dateFormat;
    } else {
      format = CarbonUtil.getFormatFromProperty(dataType);
    }
    return format;
  }

  private boolean isDictionaryDimension(CarbonDimension carbonDimension) {
    if (carbonDimension.hasEncoding(Encoding.DICTIONARY)) {
      return true;
    } else {
      return false;
    }
  }

  /*
   * primitive column will not have any child column
   */
  @Override
  public void addChildren(GenericDataType children) {

  }

  /*
   * get column name
   */
  @Override
  public String getName() {
    return name;
  }

  /*
   * get column parent name
   */
  @Override
  public String getParentName() {
    return parentName;
  }

  /*
   * get column unique id
   */
  @Override
  public String getColumnNames() {
    return columnId;
  }

  /*
   * primitive column will not have any children
   */
  @Override
  public void getAllPrimitiveChildren(List<GenericDataType> primitiveChild) {

  }

  /*
   * get surrogate index
   */
  @Override
  public int getSurrogateIndex() {
    return index;
  }

  /*
   * set surrogate index
   */
  @Override public void setSurrogateIndex(int surrIndex) {
    if (this.carbonDimension != null && !this.carbonDimension.hasEncoding(Encoding.DICTIONARY)) {
      index = 0;
    } else if (this.carbonDimension == null && isDictionary == false) {
      index = 0;
    } else {
      index = surrIndex;
    }
  }

  @Override public boolean getIsColumnDictionary() {
    return isDictionary;
  }

  @Override public void writeByteArray(Object input, DataOutputStream dataOutputStream,
      BadRecordLogHolder logHolder) throws IOException, DictionaryGenerationException {
    String parsedValue =
        input == null ? null : DataTypeUtil.parseValue(input.toString(), carbonDimension);
    String message = logHolder.getColumnMessageMap().get(carbonDimension.getColName());
    if (this.isDictionary) {
      Integer surrogateKey;
      if (null == parsedValue) {
        surrogateKey = CarbonCommonConstants.MEMBER_DEFAULT_VAL_SURROGATE_KEY;
        if (null == message) {
          message = CarbonDataProcessorUtil
              .prepareFailureReason(carbonDimension.getColName(), carbonDimension.getDataType());
          logHolder.getColumnMessageMap().put(carbonDimension.getColName(), message);
          logHolder.setReason(message);
        }
      } else {
        if (dictionaryGenerator instanceof DirectDictionary && input instanceof Long) {
          surrogateKey = ((DirectDictionary) dictionaryGenerator).generateKey((long) input);
        } else {
          surrogateKey = dictionaryGenerator.getOrGenerateKey(parsedValue);
        }
        if (surrogateKey == CarbonCommonConstants.INVALID_SURROGATE_KEY) {
          surrogateKey = CarbonCommonConstants.MEMBER_DEFAULT_VAL_SURROGATE_KEY;
          message = CarbonDataProcessorUtil
              .prepareFailureReason(carbonDimension.getColName(), carbonDimension.getDataType());
          logHolder.getColumnMessageMap().put(carbonDimension.getColName(), message);
          logHolder.setReason(message);
        }
      }
      dataOutputStream.writeInt(surrogateKey);
    } else {
      // Transform into ByteArray for No Dictionary.
      // TODO have to refactor and place all the cases present in NonDictionaryFieldConverterImpl
      if (null == parsedValue && this.carbonDimension.getDataType() != DataTypes.STRING) {
        updateNullValue(dataOutputStream, logHolder);
      } else if (null == parsedValue || parsedValue.equals(nullFormat)) {
        updateNullValue(dataOutputStream, logHolder);
      } else {
        String dateFormat = null;
        if (this.carbonDimension.getDataType() == DataTypes.DATE) {
          dateFormat = carbonDimension.getDateFormat();
        } else if (this.carbonDimension.getDataType() == DataTypes.TIMESTAMP) {
          dateFormat = carbonDimension.getTimestampFormat();
        }
        try {
          if (!this.carbonDimension.getUseActualData()) {
            byte[] value = null;
            if (isDirectDictionary) {
              int surrogateKey;
              if (!(input instanceof Long) && !(input instanceof Integer)) {
                SimpleDateFormat parser = new SimpleDateFormat(getDateFormat(carbonDimension));
                parser.parse(parsedValue);
              }
              // If the input is a long value then this means that logical type was provided by
              // the user using AvroCarbonWriter. In this case directly generate surrogate key
              // using dictionaryGenerator.
              if (dictionaryGenerator instanceof DirectDictionary && input instanceof Long) {
                surrogateKey = ((DirectDictionary) dictionaryGenerator).generateKey((long) input);
              } else if (dictionaryGenerator instanceof DirectDictionary
                  && input instanceof Integer) {
                // In case of file format, for complex type date or time type, input data comes as a
                // Integer object, so just assign the surrogate key with the input object value
                surrogateKey = (int) input;
              } else {
                surrogateKey = dictionaryGenerator.getOrGenerateKey(parsedValue);
              }
              if (surrogateKey == CarbonCommonConstants.INVALID_SURROGATE_KEY) {
                value = new byte[0];
              } else {
                value = ByteUtil.toXorBytes(surrogateKey);
              }
            } else {
              // If the input is a long value then this means that logical type was provided by
              // the user using AvroCarbonWriter. In this case directly generate Bytes from value.
              if (this.carbonDimension.getDataType().equals(DataTypes.DATE)
                  || this.carbonDimension.getDataType().equals(DataTypes.TIMESTAMP)
                  && input instanceof Long) {
                if (dictionaryGenerator != null) {
                  value = ByteUtil.toXorBytes(((DirectDictionary) dictionaryGenerator)
                      .generateKey((long) input));
                } else {
                  value = ByteUtil.toXorBytes(Long.parseLong(parsedValue));
                }
              } else {
                value = DataTypeUtil.getBytesBasedOnDataTypeForNoDictionaryColumn(parsedValue,
                    this.carbonDimension.getDataType(), dateFormat);
              }
              if (this.carbonDimension.getDataType() == DataTypes.STRING
                  && value.length > CarbonCommonConstants.MAX_CHARS_PER_COLUMN_DEFAULT) {
                throw new CarbonDataLoadingException("Dataload failed, String size cannot exceed "
                    + CarbonCommonConstants.MAX_CHARS_PER_COLUMN_DEFAULT + " bytes");
              }
            }
            updateValueToByteStream(dataOutputStream, value);
          } else {
            Object value;
            if (dictionaryGenerator instanceof DirectDictionary
                && input instanceof Long) {
              value = ByteUtil.toXorBytes(
                  ((DirectDictionary) dictionaryGenerator).generateKey((long) input));
            } else {
              value = DataTypeUtil.getDataDataTypeForNoDictionaryColumn(parsedValue,
                  this.carbonDimension.getDataType(), dateFormat);
            }
            if (this.carbonDimension.getDataType() == DataTypes.STRING
                && value.toString().length() > CarbonCommonConstants.MAX_CHARS_PER_COLUMN_DEFAULT) {
              throw new CarbonDataLoadingException("Dataload failed, String size cannot exceed "
                  + CarbonCommonConstants.MAX_CHARS_PER_COLUMN_DEFAULT + " bytes");
            }
            if (parsedValue.length() > 0) {
              updateValueToByteStream(dataOutputStream,
                  parsedValue.getBytes(Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
            } else {
              updateNullValue(dataOutputStream, logHolder);
            }
          }
        } catch (NumberFormatException e) {
          // Update logHolder for bad record and put null in dataOutputStream.
          updateNullValue(dataOutputStream, logHolder);
        } catch (CarbonDataLoadingException e) {
          throw e;
        } catch (ParseException ex) {
          updateNullValue(dataOutputStream, logHolder);
        } catch (Throwable ex) {
          // TODO have to implemented the Bad Records LogHolder.
          // Same like NonDictionaryFieldConverterImpl.
          throw ex;
        }
      }
    }
  }

  private void updateValueToByteStream(DataOutputStream dataOutputStream, byte[] value)
      throws IOException {
    dataOutputStream.writeShort(value.length);
    dataOutputStream.write(value);
  }

  private void updateNullValue(DataOutputStream dataOutputStream, BadRecordLogHolder logHolder)
      throws IOException {
    if (this.carbonDimension.getDataType() == DataTypes.STRING) {
      dataOutputStream.writeShort(CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY.length);
      dataOutputStream.write(CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY);
    } else {
      dataOutputStream.writeShort(CarbonCommonConstants.EMPTY_BYTE_ARRAY.length);
      dataOutputStream.write(CarbonCommonConstants.EMPTY_BYTE_ARRAY);
    }
    String message = logHolder.getColumnMessageMap().get(carbonDimension.getColName());
    if (null == message) {
      message = CarbonDataProcessorUtil
          .prepareFailureReason(carbonDimension.getColName(), carbonDimension.getDataType());
      logHolder.getColumnMessageMap().put(carbonDimension.getColName(), message);
      logHolder.setReason(message);
    }
  }

  @Override public void fillCardinality(List<Integer> dimCardWithComplex) {
    if (!this.carbonDimension.hasEncoding(Encoding.DICTIONARY)) {
      return;
    }
    dimCardWithComplex.add(dictionaryGenerator.size());
  }

  @Override
  public void parseComplexValue(ByteBuffer byteArrayInput, DataOutputStream dataOutputStream,
      KeyGenerator[] generator)
      throws IOException, KeyGenException {
    if (!this.isDictionary) {
      int sizeOfData = byteArrayInput.getShort();
      dataOutputStream.writeShort(sizeOfData);
      byte[] bb = new byte[sizeOfData];
      byteArrayInput.get(bb, 0, sizeOfData);
      dataOutputStream.write(bb);
    } else {
      int data = byteArrayInput.getInt();
      byte[] v = generator[index].generateKey(new int[] { data });
      dataOutputStream.write(v);
    }
  }

  /*
   * get all columns count
   */
  @Override
  public int getColsCount() {
    return 1;
  }

  /*
   * set outputarray
   */
  @Override
  public void setOutputArrayIndex(int outputArrayIndex) {
    this.outputArrayIndex = outputArrayIndex;
  }

  /*
   * get output array
   */
  @Override
  public int getMaxOutputArrayIndex() {
    return outputArrayIndex;
  }

  /*
   * split column and return metadata and primitive column
   */
  @Override public void getColumnarDataForComplexType(List<ArrayList<byte[]>> columnsArray,
      ByteBuffer inputArray) {
    if (!isDictionary) {
      byte[] key = new byte[inputArray.getShort()];
      inputArray.get(key);
      columnsArray.get(outputArrayIndex).add(key);
    } else {
      byte[] key = new byte[keySize];
      inputArray.get(key);
      columnsArray.get(outputArrayIndex).add(key);
    }
    dataCounter++;
  }

  /*
   * return datacounter
   */
  @Override
  public int getDataCounter() {
    return this.dataCounter;
  }

  /**
   * set key size
   * @param keySize
   */
  public void setKeySize(int keySize) {
    this.keySize = keySize;
  }

  /*
   * fill agg key block
   */
  @Override
  public void fillAggKeyBlock(List<Boolean> aggKeyBlockWithComplex, boolean[] aggKeyBlock) {
    aggKeyBlockWithComplex.add(aggKeyBlock[index]);
  }

  /*
   * fill block key size
   */
  @Override
  public void fillBlockKeySize(List<Integer> blockKeySizeWithComplex, int[] primitiveBlockKeySize) {
    blockKeySizeWithComplex.add(primitiveBlockKeySize[index]);
    this.keySize = primitiveBlockKeySize[index];
  }

  /*
   * fill cardinality
   */
  @Override
  public void fillCardinalityAfterDataLoad(List<Integer> dimCardWithComplex,
      int[] maxSurrogateKeyArray) {
    dimCardWithComplex.add(maxSurrogateKeyArray[index]);
  }

  @Override
  public GenericDataType<Object> deepCopy() {
    PrimitiveDataType dataType = new PrimitiveDataType(this.outputArrayIndex, 0);
    dataType.carbonDimension = this.carbonDimension;
    dataType.isDictionary = this.isDictionary;
    dataType.parentName = this.parentName;
    dataType.columnId = this.columnId;
    dataType.dictionaryGenerator = this.dictionaryGenerator;
    dataType.nullFormat = this.nullFormat;
    dataType.setKeySize(this.keySize);
    dataType.setSurrogateIndex(this.index);
    dataType.name = this.name;
    dataType.dataType = this.dataType;
    return dataType;
  }

  @Override
  public void getComplexColumnInfo(List<ComplexColumnInfo> columnInfoList) {
    columnInfoList.add(
        new ComplexColumnInfo(ColumnType.COMPLEX_PRIMITIVE, dataType,
            name, !isDictionary));
  }

}

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.processing.datatypes;
19
20	import java.io.DataOutputStream;
21	import java.io.IOException;
22	import java.nio.ByteBuffer;
23	import java.nio.charset.Charset;
24	import java.text.ParseException;
25	import java.text.SimpleDateFormat;
26	import java.util.ArrayList;
27	import java.util.List;
28	import java.util.Map;
29
30	import org.apache.carbondata.common.logging.LogServiceFactory;
31	import org.apache.carbondata.core.cache.Cache;
32	import org.apache.carbondata.core.cache.CacheProvider;
33	import org.apache.carbondata.core.cache.CacheType;
34	import org.apache.carbondata.core.cache.dictionary.Dictionary;
35	import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
36	import org.apache.carbondata.core.constants.CarbonCommonConstants;
37	import org.apache.carbondata.core.datastore.ColumnType;
38	import org.apache.carbondata.core.datastore.row.ComplexColumnInfo;
39	import org.apache.carbondata.core.devapi.BiDictionary;
40	import org.apache.carbondata.core.devapi.DictionaryGenerationException;
41	import org.apache.carbondata.core.dictionary.client.DictionaryClient;
42	import org.apache.carbondata.core.dictionary.generator.key.DictionaryMessage;
43	import org.apache.carbondata.core.dictionary.generator.key.DictionaryMessageType;
44	import org.apache.carbondata.core.keygenerator.KeyGenException;
45	import org.apache.carbondata.core.keygenerator.KeyGenerator;
46	import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryKeyGeneratorFactory;
47	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
48	import org.apache.carbondata.core.metadata.datatype.DataType;
49	import org.apache.carbondata.core.metadata.datatype.DataTypes;
50	import org.apache.carbondata.core.metadata.encoder.Encoding;
51	import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
52	import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
53	import org.apache.carbondata.core.util.ByteUtil;
54	import org.apache.carbondata.core.util.CarbonUtil;
55	import org.apache.carbondata.core.util.DataTypeUtil;
56	import org.apache.carbondata.processing.loading.converter.BadRecordLogHolder;
57	import org.apache.carbondata.processing.loading.dictionary.DictionaryServerClientDictionary;
58	import org.apache.carbondata.processing.loading.dictionary.DirectDictionary;
59	import org.apache.carbondata.processing.loading.dictionary.PreCreatedDictionary;
60	import org.apache.carbondata.processing.loading.exception.CarbonDataLoadingException;
61	import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
62
63	import org.apache.log4j.Logger;
64
65	/**
66	* Primitive DataType stateless object used in data loading
67	*/
68	public class PrimitiveDataType implements GenericDataType<Object> {
69
70	private static final Logger LOGGER =	1✔
71	LogServiceFactory.getLogService(PrimitiveDataType.class.getName());	1✔
72
73	/**
74	* surrogate index
75	*/
76	private int index;
77
78	/**
79	* column name
80	*/
81	private String name;
82
83	/**
84	* column parent name
85	*/
86	private String parentName;
87
88	/**
89	* column unique id
90	*/
91	private String columnId;
92
93	/**
94	* key size
95	*/
96	private int keySize;
97
98	/**
99	* array index
100	*/
101	private int outputArrayIndex;
102
103	/**
104	* data counter
105	*/
106	private int dataCounter;
107
108	private BiDictionary<Integer, Object> dictionaryGenerator;
109
110	private CarbonDimension carbonDimension;
111
112	private boolean isDictionary;
113
114	private String nullFormat;
115
116	private boolean isDirectDictionary;
117
118	private DataType dataType;
119
120	private PrimitiveDataType(int outputArrayIndex, int dataCounter) {	1✔
121	this.outputArrayIndex = outputArrayIndex;	1✔
122	this.dataCounter = dataCounter;	1✔
123	}	1✔
124
125	/**
126	* constructor
127	*
128	* @param name
129	* @param parentName
130	* @param columnId
131	* @param isDictionary
132	*/
133	public PrimitiveDataType(String name, DataType dataType, String parentName, String columnId,
134	boolean isDictionary, String nullFormat) {	1✔
135	this.name = name;	1✔
136	this.parentName = parentName;	1✔
137	this.columnId = columnId;	1✔
138	this.isDictionary = isDictionary;	1✔
139	this.nullFormat = nullFormat;	1✔
140	this.dataType = dataType;	1✔
141	}	1✔
142
143	/**
144	* Constructor
145	* @param carbonColumn
146	* @param parentName
147	* @param columnId
148	* @param carbonDimension
149	* @param absoluteTableIdentifier
150	* @param client
151	* @param useOnePass
152	* @param localCache
153	* @param nullFormat
154	*/
155	public PrimitiveDataType(CarbonColumn carbonColumn, String parentName, String columnId,
156	CarbonDimension carbonDimension, AbsoluteTableIdentifier absoluteTableIdentifier,
157	DictionaryClient client, Boolean useOnePass, Map<Object, Integer> localCache,
158	String nullFormat) {	1✔
159	this.name = carbonColumn.getColName();	1✔
160	this.parentName = parentName;	1✔
161	this.columnId = columnId;	1✔
162	this.carbonDimension = carbonDimension;	1✔
163	this.isDictionary = isDictionaryDimension(carbonDimension);	1✔
164	this.nullFormat = nullFormat;	1✔
165	this.dataType = carbonColumn.getDataType();	1✔
166
167	DictionaryColumnUniqueIdentifier identifier =	1✔
168	new DictionaryColumnUniqueIdentifier(absoluteTableIdentifier,
169	carbonDimension.getColumnIdentifier(), carbonDimension.getDataType());	1✔
170	try {
171	if (carbonDimension.hasEncoding(Encoding.DIRECT_DICTIONARY)	1✔
172	\|\| carbonColumn.getDataType() == DataTypes.DATE) {	1✔
173	dictionaryGenerator = new DirectDictionary(DirectDictionaryKeyGeneratorFactory	1✔
174	.getDirectDictionaryGenerator(carbonDimension.getDataType(),	1✔
175	getDateFormat(carbonDimension)));	1✔
176	isDirectDictionary = true;	1✔
177	} else if (carbonDimension.hasEncoding(Encoding.DICTIONARY)) {	1✔
178	CacheProvider cacheProvider = CacheProvider.getInstance();	1✔
179	Cache<DictionaryColumnUniqueIdentifier, Dictionary> cache =	1✔
180	cacheProvider.createCache(CacheType.REVERSE_DICTIONARY);	1✔
181	Dictionary dictionary = null;	1✔
182	if (useOnePass) {	1✔
183	if (CarbonUtil.isFileExistsForGivenColumn(identifier)) {	×
184	dictionary = cache.get(identifier);	×
185	}
186	DictionaryMessage dictionaryMessage = new DictionaryMessage();	×
187	dictionaryMessage.setColumnName(carbonDimension.getColName());	×
188	// for table initialization
189	dictionaryMessage	×
190	.setTableUniqueId(absoluteTableIdentifier.getCarbonTableIdentifier().getTableId());	×
191	dictionaryMessage.setData("0");	×
192	// for generate dictionary
193	dictionaryMessage.setType(DictionaryMessageType.DICT_GENERATION);	×
194	dictionaryGenerator = new DictionaryServerClientDictionary(dictionary, client,	×
195	dictionaryMessage, localCache);
196	} else {	×
197	dictionary = cache.get(identifier);	1✔
198	dictionaryGenerator = new PreCreatedDictionary(dictionary);	1✔
199	}
200	}
201	} catch (IOException e) {	×
202	throw new RuntimeException(e);	×
203	}	1✔
204	}	1✔
205
206	/**
207	* get dateformat
208	* @param carbonDimension
209	* @return
210	*/
211	private String getDateFormat(CarbonDimension carbonDimension) {
212	String format;
213	String dateFormat = null;	1✔
214	if (this.carbonDimension.getDataType() == DataTypes.DATE) {	1✔
215	dateFormat = carbonDimension.getDateFormat();	1✔
216	}
217	if (dateFormat != null && !dateFormat.trim().isEmpty()) {	1✔
218	format = dateFormat;	1✔
219	} else {
220	format = CarbonUtil.getFormatFromProperty(dataType);	1✔
221	}
222	return format;	1✔
223	}
224
225	private boolean isDictionaryDimension(CarbonDimension carbonDimension) {
226	if (carbonDimension.hasEncoding(Encoding.DICTIONARY)) {	1✔
227	return true;	1✔
228	} else {
229	return false;	1✔
230	}
231	}
232
233	/*
234	* primitive column will not have any child column
235	*/
236	@Override
237	public void addChildren(GenericDataType children) {
238
239	}	1✔
240
241	/*
242	* get column name
243	*/
244	@Override
245	public String getName() {
246	return name;	×
247	}
248
249	/*
250	* get column parent name
251	*/
252	@Override
253	public String getParentName() {
254	return parentName;	1✔
255	}
256
257	/*
258	* get column unique id
259	*/
260	@Override
261	public String getColumnNames() {
262	return columnId;	×
263	}
264
265	/*
266	* primitive column will not have any children
267	*/
268	@Override
269	public void getAllPrimitiveChildren(List<GenericDataType> primitiveChild) {
270
271	}	×
272
273	/*
274	* get surrogate index
275	*/
276	@Override
277	public int getSurrogateIndex() {
278	return index;	1✔
279	}
280
281	/*
282	* set surrogate index
283	*/
284	@Override public void setSurrogateIndex(int surrIndex) {
285	if (this.carbonDimension != null && !this.carbonDimension.hasEncoding(Encoding.DICTIONARY)) {	1✔
286	index = 0;	×
287	} else if (this.carbonDimension == null && isDictionary == false) {	1✔
288	index = 0;	1✔
289	} else {
290	index = surrIndex;	1✔
291	}
292	}	1✔
293
294	@Override public boolean getIsColumnDictionary() {
295	return isDictionary;	1✔
296	}
297
298	@Override public void writeByteArray(Object input, DataOutputStream dataOutputStream,
299	BadRecordLogHolder logHolder) throws IOException, DictionaryGenerationException {
300	String parsedValue =	1✔
301	input == null ? null : DataTypeUtil.parseValue(input.toString(), carbonDimension);	1✔
302	String message = logHolder.getColumnMessageMap().get(carbonDimension.getColName());	1✔
303	if (this.isDictionary) {	1✔
304	Integer surrogateKey;
305	if (null == parsedValue) {	1✔
306	surrogateKey = CarbonCommonConstants.MEMBER_DEFAULT_VAL_SURROGATE_KEY;	1✔
307	if (null == message) {	1✔
308	message = CarbonDataProcessorUtil	1✔
309	.prepareFailureReason(carbonDimension.getColName(), carbonDimension.getDataType());	1✔
310	logHolder.getColumnMessageMap().put(carbonDimension.getColName(), message);	1✔
311	logHolder.setReason(message);	1✔
312	}
313	} else {
314	if (dictionaryGenerator instanceof DirectDictionary && input instanceof Long) {	1✔
315	surrogateKey = ((DirectDictionary) dictionaryGenerator).generateKey((long) input);	×
316	} else {
317	surrogateKey = dictionaryGenerator.getOrGenerateKey(parsedValue);	1✔
318	}
319	if (surrogateKey == CarbonCommonConstants.INVALID_SURROGATE_KEY) {	1✔
320	surrogateKey = CarbonCommonConstants.MEMBER_DEFAULT_VAL_SURROGATE_KEY;	×
321	message = CarbonDataProcessorUtil	×
322	.prepareFailureReason(carbonDimension.getColName(), carbonDimension.getDataType());	×
323	logHolder.getColumnMessageMap().put(carbonDimension.getColName(), message);	×
324	logHolder.setReason(message);	×
325	}
326	}
327	dataOutputStream.writeInt(surrogateKey);	1✔
328	} else {	1✔
329	// Transform into ByteArray for No Dictionary.
330	// TODO have to refactor and place all the cases present in NonDictionaryFieldConverterImpl
331	if (null == parsedValue && this.carbonDimension.getDataType() != DataTypes.STRING) {	1✔
332	updateNullValue(dataOutputStream, logHolder);	1✔
333	} else if (null == parsedValue \|\| parsedValue.equals(nullFormat)) {	1✔
334	updateNullValue(dataOutputStream, logHolder);	1✔
335	} else {
336	String dateFormat = null;	1✔
337	if (this.carbonDimension.getDataType() == DataTypes.DATE) {	1✔
338	dateFormat = carbonDimension.getDateFormat();	1✔
339	} else if (this.carbonDimension.getDataType() == DataTypes.TIMESTAMP) {	1✔
340	dateFormat = carbonDimension.getTimestampFormat();	1✔
341	}
342	try {
343	if (!this.carbonDimension.getUseActualData()) {	1✔
344	byte[] value = null;	1✔
345	if (isDirectDictionary) {	1✔
346	int surrogateKey;
347	if (!(input instanceof Long) && !(input instanceof Integer)) {	1✔
348	SimpleDateFormat parser = new SimpleDateFormat(getDateFormat(carbonDimension));	1✔
349	parser.parse(parsedValue);	1✔
350	}
351	// If the input is a long value then this means that logical type was provided by
352	// the user using AvroCarbonWriter. In this case directly generate surrogate key
353	// using dictionaryGenerator.
354	if (dictionaryGenerator instanceof DirectDictionary && input instanceof Long) {	1✔
355	surrogateKey = ((DirectDictionary) dictionaryGenerator).generateKey((long) input);	1✔
356	} else if (dictionaryGenerator instanceof DirectDictionary	1✔
357	&& input instanceof Integer) {
358	// In case of file format, for complex type date or time type, input data comes as a
359	// Integer object, so just assign the surrogate key with the input object value
360	surrogateKey = (int) input;	×
361	} else {
362	surrogateKey = dictionaryGenerator.getOrGenerateKey(parsedValue);	1✔
363	}
364	if (surrogateKey == CarbonCommonConstants.INVALID_SURROGATE_KEY) {	1✔
365	value = new byte[0];	×
366	} else {
367	value = ByteUtil.toXorBytes(surrogateKey);	1✔
368	}
369	} else {	1✔
370	// If the input is a long value then this means that logical type was provided by
371	// the user using AvroCarbonWriter. In this case directly generate Bytes from value.
372	if (this.carbonDimension.getDataType().equals(DataTypes.DATE)	1✔
373	\|\| this.carbonDimension.getDataType().equals(DataTypes.TIMESTAMP)	1✔
374	&& input instanceof Long) {
375	if (dictionaryGenerator != null) {	1✔
376	value = ByteUtil.toXorBytes(((DirectDictionary) dictionaryGenerator)	×
377	.generateKey((long) input));	×
378	} else {
379	value = ByteUtil.toXorBytes(Long.parseLong(parsedValue));	1✔
380	}
381	} else {
382	value = DataTypeUtil.getBytesBasedOnDataTypeForNoDictionaryColumn(parsedValue,	1✔
383	this.carbonDimension.getDataType(), dateFormat);	1✔
384	}
385	if (this.carbonDimension.getDataType() == DataTypes.STRING	1✔
386	&& value.length > CarbonCommonConstants.MAX_CHARS_PER_COLUMN_DEFAULT) {
387	throw new CarbonDataLoadingException("Dataload failed, String size cannot exceed "	×
388	+ CarbonCommonConstants.MAX_CHARS_PER_COLUMN_DEFAULT + " bytes");
389	}
390	}
391	updateValueToByteStream(dataOutputStream, value);	1✔
392	} else {	1✔
393	Object value;
394	if (dictionaryGenerator instanceof DirectDictionary	×
395	&& input instanceof Long) {
396	value = ByteUtil.toXorBytes(	×
397	((DirectDictionary) dictionaryGenerator).generateKey((long) input));	×
398	} else {
399	value = DataTypeUtil.getDataDataTypeForNoDictionaryColumn(parsedValue,	×
400	this.carbonDimension.getDataType(), dateFormat);	×
401	}
402	if (this.carbonDimension.getDataType() == DataTypes.STRING	×
403	&& value.toString().length() > CarbonCommonConstants.MAX_CHARS_PER_COLUMN_DEFAULT) {	×
404	throw new CarbonDataLoadingException("Dataload failed, String size cannot exceed "	×
405	+ CarbonCommonConstants.MAX_CHARS_PER_COLUMN_DEFAULT + " bytes");
406	}
407	if (parsedValue.length() > 0) {	×
408	updateValueToByteStream(dataOutputStream,	×
409	parsedValue.getBytes(Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));	×
410	} else {
411	updateNullValue(dataOutputStream, logHolder);	×
412	}
413	}
414	} catch (NumberFormatException e) {	1✔
415	// Update logHolder for bad record and put null in dataOutputStream.
416	updateNullValue(dataOutputStream, logHolder);	1✔
417	} catch (CarbonDataLoadingException e) {	×
418	throw e;	×
419	} catch (ParseException ex) {	1✔
420	updateNullValue(dataOutputStream, logHolder);	1✔
421	} catch (Throwable ex) {	×
422	// TODO have to implemented the Bad Records LogHolder.
423	// Same like NonDictionaryFieldConverterImpl.
424	throw ex;	×
425	}	1✔
426	}
427	}
428	}	1✔
429
430	private void updateValueToByteStream(DataOutputStream dataOutputStream, byte[] value)
431	throws IOException {
432	dataOutputStream.writeShort(value.length);	1✔
433	dataOutputStream.write(value);	1✔
434	}	1✔
435
436	private void updateNullValue(DataOutputStream dataOutputStream, BadRecordLogHolder logHolder)
437	throws IOException {
438	if (this.carbonDimension.getDataType() == DataTypes.STRING) {	1✔
439	dataOutputStream.writeShort(CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY.length);	1✔
440	dataOutputStream.write(CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY);	1✔
441	} else {
442	dataOutputStream.writeShort(CarbonCommonConstants.EMPTY_BYTE_ARRAY.length);	1✔
443	dataOutputStream.write(CarbonCommonConstants.EMPTY_BYTE_ARRAY);	1✔
444	}
445	String message = logHolder.getColumnMessageMap().get(carbonDimension.getColName());	1✔
446	if (null == message) {	1✔
447	message = CarbonDataProcessorUtil	1✔
448	.prepareFailureReason(carbonDimension.getColName(), carbonDimension.getDataType());	1✔
449	logHolder.getColumnMessageMap().put(carbonDimension.getColName(), message);	1✔
450	logHolder.setReason(message);	1✔
451	}
452	}	1✔
453
454	@Override public void fillCardinality(List<Integer> dimCardWithComplex) {
455	if (!this.carbonDimension.hasEncoding(Encoding.DICTIONARY)) {	1✔
456	return;	×
457	}
458	dimCardWithComplex.add(dictionaryGenerator.size());	1✔
459	}	1✔
460
461	@Override
462	public void parseComplexValue(ByteBuffer byteArrayInput, DataOutputStream dataOutputStream,
463	KeyGenerator[] generator)
464	throws IOException, KeyGenException {
465	if (!this.isDictionary) {	1✔
466	int sizeOfData = byteArrayInput.getShort();	1✔
467	dataOutputStream.writeShort(sizeOfData);	1✔
468	byte[] bb = new byte[sizeOfData];	1✔
469	byteArrayInput.get(bb, 0, sizeOfData);	1✔
470	dataOutputStream.write(bb);	1✔
471	} else {	1✔
472	int data = byteArrayInput.getInt();	1✔
473	byte[] v = generator[index].generateKey(new int[] { data });	1✔
474	dataOutputStream.write(v);	1✔
475	}
476	}	1✔
477
478	/*
479	* get all columns count
480	*/
481	@Override
482	public int getColsCount() {
483	return 1;	1✔
484	}
485
486	/*
487	* set outputarray
488	*/
489	@Override
490	public void setOutputArrayIndex(int outputArrayIndex) {
491	this.outputArrayIndex = outputArrayIndex;	1✔
492	}	1✔
493
494	/*
495	* get output array
496	*/
497	@Override
498	public int getMaxOutputArrayIndex() {
499	return outputArrayIndex;	1✔
500	}
501
502	/*
503	* split column and return metadata and primitive column
504	*/
505	@Override public void getColumnarDataForComplexType(List<ArrayList<byte[]>> columnsArray,
506	ByteBuffer inputArray) {
507	if (!isDictionary) {	1✔
508	byte[] key = new byte[inputArray.getShort()];	1✔
509	inputArray.get(key);	1✔
510	columnsArray.get(outputArrayIndex).add(key);	1✔
511	} else {	1✔
512	byte[] key = new byte[keySize];	1✔
513	inputArray.get(key);	1✔
514	columnsArray.get(outputArrayIndex).add(key);	1✔
515	}
516	dataCounter++;	1✔
517	}	1✔
518
519	/*
520	* return datacounter
521	*/
522	@Override
523	public int getDataCounter() {
524	return this.dataCounter;	1✔
525	}
526
527	/**
528	* set key size
529	* @param keySize
530	*/
531	public void setKeySize(int keySize) {
532	this.keySize = keySize;	1✔
533	}	1✔
534
535	/*
536	* fill agg key block
537	*/
538	@Override
539	public void fillAggKeyBlock(List<Boolean> aggKeyBlockWithComplex, boolean[] aggKeyBlock) {
540	aggKeyBlockWithComplex.add(aggKeyBlock[index]);	×
541	}	×
542
543	/*
544	* fill block key size
545	*/
546	@Override
547	public void fillBlockKeySize(List<Integer> blockKeySizeWithComplex, int[] primitiveBlockKeySize) {
548	blockKeySizeWithComplex.add(primitiveBlockKeySize[index]);	1✔
549	this.keySize = primitiveBlockKeySize[index];	1✔
550	}	1✔
551
552	/*
553	* fill cardinality
554	*/
555	@Override
556	public void fillCardinalityAfterDataLoad(List<Integer> dimCardWithComplex,
557	int[] maxSurrogateKeyArray) {
558	dimCardWithComplex.add(maxSurrogateKeyArray[index]);	×
559	}	×
560
561	@Override
562	public GenericDataType<Object> deepCopy() {
563	PrimitiveDataType dataType = new PrimitiveDataType(this.outputArrayIndex, 0);	1✔
564	dataType.carbonDimension = this.carbonDimension;	1✔
565	dataType.isDictionary = this.isDictionary;	1✔
566	dataType.parentName = this.parentName;	1✔
567	dataType.columnId = this.columnId;	1✔
568	dataType.dictionaryGenerator = this.dictionaryGenerator;	1✔
569	dataType.nullFormat = this.nullFormat;	1✔
570	dataType.setKeySize(this.keySize);	1✔
571	dataType.setSurrogateIndex(this.index);	1✔
572	dataType.name = this.name;	1✔
573	dataType.dataType = this.dataType;	1✔
574	return dataType;	1✔
575	}
576
577	@Override
578	public void getComplexColumnInfo(List<ComplexColumnInfo> columnInfoList) {
579	columnInfoList.add(	1✔
580	new ComplexColumnInfo(ColumnType.COMPLEX_PRIMITIVE, dataType,
581	name, !isDictionary));
582	}	1✔
583
584	}

apache / carbondata / 10921

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