9055

Build # 9055

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/c37bc4b97269b631db0f256c4c54eaeabc80d6c9">c37bc4b97<a href="https://github.com/apache/carbondata/commit/c37bc4b97269b631db0f256c4c54eaeabc80d6c9">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/5d7d47b388870aa1fb3d01fb526371825c0d5166">5d7d47b38</a>">5d7d47b38</a><a href="https://github.com/apache/carbondata/commit/c37bc4b97269b631db0f256c4c54eaeabc80d6c9"> into 15d38260c">15d38260c</a>

Run Details

2094 of 2094 new or added lines in 64 files covered. (100.0%)

60738 of 76327 relevant lines covered (79.58%)

1.05 hits per line

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

37.5

/core/src/main/java/org/apache/carbondata/core/datastore/page/VarLengthColumnPageBase.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.carbondata.core.datastore.page;

import java.io.IOException;

import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.constants.CarbonV3DataFormatConstants;
import org.apache.carbondata.core.datastore.ColumnType;
import org.apache.carbondata.core.datastore.TableSpec;
import org.apache.carbondata.core.datastore.page.encoding.ColumnPageEncoderMeta;
import org.apache.carbondata.core.memory.CarbonUnsafe;
import org.apache.carbondata.core.memory.MemoryBlock;
import org.apache.carbondata.core.memory.MemoryException;
import org.apache.carbondata.core.memory.UnsafeMemoryManager;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.datatype.DecimalConverterFactory;
import org.apache.carbondata.core.util.ByteUtil;
import org.apache.carbondata.core.util.ThreadLocalTaskInfo;

public abstract class VarLengthColumnPageBase extends ColumnPage {

  static final int byteBits = DataTypes.BYTE.getSizeBits();
  static final int shortBits = DataTypes.SHORT.getSizeBits();
  static final int intBits = DataTypes.INT.getSizeBits();
  static final int longBits = DataTypes.LONG.getSizeBits();
  // default size for each row, grows as needed
  static final int DEFAULT_ROW_SIZE = 8;

  static final double FACTOR = 1.25;

  final String taskId = ThreadLocalTaskInfo.getCarbonTaskInfo().getTaskId();

  // memory allocated by Unsafe
  MemoryBlock memoryBlock;

  // base address of memoryBlock
  Object baseAddress;

  // the offset of row in the unsafe memory, its size is pageSize + 1
  ColumnPage rowOffset;

  // the length of bytes added in the page
  int totalLength;

  // base offset of memoryBlock
  long baseOffset;

  // size of the allocated memory, in bytes
  int capacity;

  VarLengthColumnPageBase(ColumnPageEncoderMeta columnPageEncoderMeta, int pageSize) {
    super(columnPageEncoderMeta, pageSize);
    TableSpec.MeasureSpec spec = TableSpec.MeasureSpec.newInstance(
        columnPageEncoderMeta.getColumnSpec().getFieldName(), DataTypes.INT);
    try {
      rowOffset = ColumnPage.newPage(
          new ColumnPageEncoderMeta(spec, DataTypes.INT, columnPageEncoderMeta.getCompressorName()),
          pageSize);
    } catch (MemoryException e) {
      throw new RuntimeException(e);
    }
    totalLength = 0;
  }

  @Override
  public void setBytePage(byte[] byteData) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public void setShortPage(short[] shortData) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public void setShortIntPage(byte[] shortIntData) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public void setIntPage(int[] intData) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public void setLongPage(long[] longData) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public void setFloatPage(float[] floatData) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public void setDoublePage(double[] doubleData) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  /**
   * Create a new column page for decimal page
   */
  static ColumnPage newDecimalColumnPage(TableSpec.ColumnSpec columnSpec, byte[] lvEncodedBytes,
      String compressorName) throws MemoryException {
    DecimalConverterFactory.DecimalConverter decimalConverter =
        DecimalConverterFactory.INSTANCE.getDecimalConverter(columnSpec.getPrecision(),
            columnSpec.getScale());
    int size = decimalConverter.getSize();
    if (size < 0) {
      return getLVBytesColumnPage(columnSpec, lvEncodedBytes,
          DataTypes.createDecimalType(columnSpec.getPrecision(), columnSpec.getScale()),
          CarbonCommonConstants.INT_SIZE_IN_BYTE, compressorName);
    } else {
      // Here the size is always fixed.
      return getDecimalColumnPage(columnSpec, lvEncodedBytes, size, compressorName);
    }
  }

  /**
   * Create a new column page based on the LV (Length Value) encoded bytes
   */
  static ColumnPage newLVBytesColumnPage(TableSpec.ColumnSpec columnSpec, byte[] lvEncodedBytes,
      int lvLength, String compressorName) throws MemoryException {
    return getLVBytesColumnPage(columnSpec, lvEncodedBytes, DataTypes.BYTE_ARRAY,
        lvLength, compressorName);
  }

  /**
   * Create a new column page based on the LV (Length Value) encoded bytes
   */
  static ColumnPage newComplexLVBytesColumnPage(TableSpec.ColumnSpec columnSpec,
      byte[] lvEncodedBytes, int lvLength, String compressorName) throws MemoryException {
    return getComplexLVBytesColumnPage(columnSpec, lvEncodedBytes, DataTypes.BYTE_ARRAY,
        lvLength, compressorName);
  }

  private static ColumnPage getDecimalColumnPage(TableSpec.ColumnSpec columnSpec,
      byte[] lvEncodedBytes, int size, String compressorName) throws MemoryException {
    TableSpec.ColumnSpec spec = TableSpec.ColumnSpec
        .newInstance(columnSpec.getFieldName(), DataTypes.INT, ColumnType.MEASURE);
    ColumnPage rowOffset = ColumnPage.newPage(
        new ColumnPageEncoderMeta(spec, DataTypes.INT, compressorName),
        CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT);
    int offset;
    int rowId = 0;
    int counter = 0;
    for (offset = 0; offset < lvEncodedBytes.length; offset += size) {
      rowOffset.putInt(counter, offset);
      rowId++;
      counter++;
    }
    rowOffset.putInt(counter, offset);

    VarLengthColumnPageBase page;
    if (unsafe) {
      page = new UnsafeDecimalColumnPage(
          new ColumnPageEncoderMeta(columnSpec, columnSpec.getSchemaDataType(), compressorName),
          rowId);
    } else {
      page = new SafeDecimalColumnPage(
          new ColumnPageEncoderMeta(columnSpec, columnSpec.getSchemaDataType(), compressorName),
          rowId);
    }

    // set total length and rowOffset in page
    page.totalLength = offset;
    page.rowOffset.freeMemory();
    page.rowOffset = rowOffset;
    for (int i = 0; i < rowId; i++) {
      page.putBytes(i, lvEncodedBytes, i * size, size);
    }
    return page;
  }

  private static ColumnPage getLVBytesColumnPage(TableSpec.ColumnSpec columnSpec,
      byte[] lvEncodedBytes, DataType dataType, int lvLength, String compressorName)
      throws MemoryException {
    // extract length and data, set them to rowOffset and unsafe memory correspondingly
    int rowId = 0;
    TableSpec.ColumnSpec spec = TableSpec.ColumnSpec
        .newInstance(columnSpec.getFieldName(), DataTypes.INT, ColumnType.MEASURE);
    ColumnPage rowOffset = ColumnPage.newPage(
        new ColumnPageEncoderMeta(spec, DataTypes.INT, compressorName),
        CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT);
    int length;
    int offset;
    int lvEncodedOffset = 0;
    int counter = 0;
    // extract Length field in input and calculate total length
    for (offset = 0; lvEncodedOffset < lvEncodedBytes.length; offset += length) {
      length = ByteUtil.toInt(lvEncodedBytes, lvEncodedOffset);
      rowOffset.putInt(counter, offset);
      lvEncodedOffset += lvLength + length;
      rowId++;
      counter++;
    }
    rowOffset.putInt(counter, offset);
    return getVarLengthColumnPage(columnSpec, lvEncodedBytes, dataType,
        lvLength, rowId, rowOffset, offset, compressorName);
  }

  private static ColumnPage getComplexLVBytesColumnPage(TableSpec.ColumnSpec columnSpec,
      byte[] lvEncodedBytes, DataType dataType, int lvLength, String compressorName)
      throws MemoryException {
    // extract length and data, set them to rowOffset and unsafe memory correspondingly
    int rowId = 0;
    TableSpec.ColumnSpec spec = TableSpec.ColumnSpec
        .newInstance(columnSpec.getFieldName(), DataTypes.INT, ColumnType.MEASURE);
    ColumnPage rowOffset = ColumnPage.newPage(
        new ColumnPageEncoderMeta(spec, DataTypes.INT, compressorName),
        CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT);
    int length;
    int offset;
    int lvEncodedOffset = 0;
    int counter = 0;
    // extract Length field in input and calculate total length
    for (offset = 0; lvEncodedOffset < lvEncodedBytes.length; offset += length) {
      length = ByteUtil.toShort(lvEncodedBytes, lvEncodedOffset);
      rowOffset.putInt(counter, offset);
      lvEncodedOffset += lvLength + length;
      rowId++;
      counter++;
    }
    rowOffset.putInt(counter, offset);

    return getVarLengthColumnPage(columnSpec, lvEncodedBytes, dataType,
        lvLength, rowId, rowOffset, offset, compressorName);
  }

  private static VarLengthColumnPageBase getVarLengthColumnPage(TableSpec.ColumnSpec columnSpec,
      byte[] lvEncodedBytes, DataType dataType, int lvLength, int rowId, ColumnPage rowOffset,
      int offset, String compressorName) throws MemoryException {
    int lvEncodedOffset;
    int length;
    int numRows = rowId;

    VarLengthColumnPageBase page;
    int inputDataLength = offset;
    if (unsafe) {
      page = new UnsafeDecimalColumnPage(
          new ColumnPageEncoderMeta(columnSpec, dataType, compressorName), numRows,
          inputDataLength);
    } else {
      page = new SafeDecimalColumnPage(
          new ColumnPageEncoderMeta(columnSpec, dataType, compressorName), numRows);
    }

    // set total length and rowOffset in page
    page.totalLength = offset;
    page.rowOffset.freeMemory();
    page.rowOffset = rowOffset;

    // set data in page
    lvEncodedOffset = 0;
    for (int i = 0; i < numRows; i++) {
      length = rowOffset.getInt(i + 1) - rowOffset.getInt(i);
      page.putBytes(i, lvEncodedBytes, lvEncodedOffset + lvLength, length);
      lvEncodedOffset += lvLength + length;
    }
    return page;
  }

  @Override
  public void putByte(int rowId, byte value) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public void putShort(int rowId, short value) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public void putShortInt(int rowId, int value) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public void putInt(int rowId, int value) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public void putLong(int rowId, long value) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public void putDouble(int rowId, double value) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public void putFloat(int rowId, float value) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  abstract void putBytesAtRow(int rowId, byte[] bytes);

  @Override
  public void putBytes(int rowId, byte[] bytes) {
    // rowId * 4 represents the length of L in LV
    if (bytes.length > (Integer.MAX_VALUE - totalLength - rowId * 4)) {
      // since we later store a column page in a byte array, so its maximum size is 2GB
      throw new RuntimeException("Carbondata only support maximum 2GB size for one column page,"
          + " exceed this limit at rowId " + rowId);
    }
    if (rowId == 0) {
      rowOffset.putInt(0, 0);
    }
    rowOffset.putInt(rowId + 1, rowOffset.getInt(rowId) + bytes.length);
    putBytesAtRow(rowId, bytes);
    totalLength += bytes.length;
    pageSize = rowId + 1;
  }

  @Override
  public byte getByte(int rowId) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public short getShort(int rowId) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public int getShortInt(int rowId) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public int getInt(int rowId) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public long getLong(int rowId) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public float getFloat(int rowId) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public double getDouble(int rowId) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public byte[] getBytePage() {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public short[] getShortPage() {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public byte[] getShortIntPage() {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public int[] getIntPage() {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public long[] getLongPage() {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public float[] getFloatPage() {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public double[] getDoublePage() {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  @Override
  public byte[] getDecimalPage() {
    // output LV encoded byte array
    int offset = 0;
    byte[] data = new byte[totalLength];
    for (int rowId = 0; rowId < pageSize; rowId++) {
      int length = rowOffset.getInt(rowId + 1) - rowOffset.getInt(rowId);
      copyBytes(rowId, data, offset, length);
      offset += length;
    }
    return data;
  }

  /**
   * Copy `length` bytes from data at rowId to dest start from destOffset
   */
  abstract void copyBytes(int rowId, byte[] dest, int destOffset, int length);

  @Override
  public byte[] getLVFlattenedBytePage() throws IOException {
    // output LV encoded byte array
    int offset = 0;
    byte[] data = new byte[totalLength + ((rowOffset.getActualRowCount() - 1) * 4)];
    for (int rowId = 0; rowId < rowOffset.getActualRowCount() - 1; rowId++) {
      int length = rowOffset.getInt(rowId + 1) - rowOffset.getInt(rowId);
      ByteUtil.setInt(data, offset, length);
      copyBytes(rowId, data, offset + 4, length);
      offset += 4 + length;
    }
    return data;
  }

  @Override public byte[] getComplexChildrenLVFlattenedBytePage() throws IOException {
    // output LV encoded byte array
    int offset = 0;
    byte[] data = new byte[totalLength + ((rowOffset.getActualRowCount() - 1) * 2)];
    for (int rowId = 0; rowId < rowOffset.getActualRowCount() - 1; rowId++) {
      short length = (short) (rowOffset.getInt(rowId + 1) - rowOffset.getInt(rowId));
      ByteUtil.setShort(data, offset, length);
      copyBytes(rowId, data, offset + 2, length);
      offset += 2 + length;
    }
    return data;
  }

  @Override
  public byte[] getComplexParentFlattenedBytePage() throws IOException {
    // output LV encoded byte array
    int offset = 0;
    byte[] data = new byte[totalLength];
    for (int rowId = 0; rowId < rowOffset.getActualRowCount() - 1; rowId++) {
      int length =  (rowOffset.getInt(rowId + 1) - rowOffset.getInt(rowId));
      copyBytes(rowId, data, offset, length);
      offset += length;
    }
    return data;
  }

  @Override
  public void convertValue(ColumnPageValueConverter codec) {
    throw new UnsupportedOperationException(
        "invalid data type: " + columnPageEncoderMeta.getStoreDataType());
  }

  /**
   * reallocate memory if capacity length than current size + request size
   */
  protected void ensureMemory(int requestSize) throws MemoryException {
    if (totalLength + requestSize > capacity) {
      int newSize = Math.max(2 * capacity, totalLength + requestSize);
      MemoryBlock newBlock = UnsafeMemoryManager.allocateMemoryWithRetry(taskId, newSize);
      CarbonUnsafe.getUnsafe().copyMemory(baseAddress, baseOffset,
          newBlock.getBaseObject(), newBlock.getBaseOffset(), capacity);
      UnsafeMemoryManager.INSTANCE.freeMemory(taskId, memoryBlock);
      memoryBlock = newBlock;
      baseAddress = newBlock.getBaseObject();
      baseOffset = newBlock.getBaseOffset();
      capacity = newSize;
    }
  }

  /**
   * free memory as needed
   */
  public void freeMemory() {
    if (null != rowOffset) {
      rowOffset.freeMemory();
      rowOffset = null;
    }
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.core.datastore.page;
19
20	import java.io.IOException;
21
22	import org.apache.carbondata.core.constants.CarbonCommonConstants;
23	import org.apache.carbondata.core.constants.CarbonV3DataFormatConstants;
24	import org.apache.carbondata.core.datastore.ColumnType;
25	import org.apache.carbondata.core.datastore.TableSpec;
26	import org.apache.carbondata.core.datastore.page.encoding.ColumnPageEncoderMeta;
27	import org.apache.carbondata.core.memory.CarbonUnsafe;
28	import org.apache.carbondata.core.memory.MemoryBlock;
29	import org.apache.carbondata.core.memory.MemoryException;
30	import org.apache.carbondata.core.memory.UnsafeMemoryManager;
31	import org.apache.carbondata.core.metadata.datatype.DataType;
32	import org.apache.carbondata.core.metadata.datatype.DataTypes;
33	import org.apache.carbondata.core.metadata.datatype.DecimalConverterFactory;
34	import org.apache.carbondata.core.util.ByteUtil;
35	import org.apache.carbondata.core.util.ThreadLocalTaskInfo;
36
37	public abstract class VarLengthColumnPageBase extends ColumnPage {
38
39	static final int byteBits = DataTypes.BYTE.getSizeBits();	2✔
40	static final int shortBits = DataTypes.SHORT.getSizeBits();	2✔
41	static final int intBits = DataTypes.INT.getSizeBits();	2✔
42	static final int longBits = DataTypes.LONG.getSizeBits();	2✔
43	// default size for each row, grows as needed
44	static final int DEFAULT_ROW_SIZE = 8;
45
46	static final double FACTOR = 1.25;
47
48	final String taskId = ThreadLocalTaskInfo.getCarbonTaskInfo().getTaskId();	2✔
49
50	// memory allocated by Unsafe
51	MemoryBlock memoryBlock;
52
53	// base address of memoryBlock
54	Object baseAddress;
55
56	// the offset of row in the unsafe memory, its size is pageSize + 1
57	ColumnPage rowOffset;
58
59	// the length of bytes added in the page
60	int totalLength;
61
62	// base offset of memoryBlock
63	long baseOffset;
64
65	// size of the allocated memory, in bytes
66	int capacity;
67
68	VarLengthColumnPageBase(ColumnPageEncoderMeta columnPageEncoderMeta, int pageSize) {
69	super(columnPageEncoderMeta, pageSize);	2✔
70	TableSpec.MeasureSpec spec = TableSpec.MeasureSpec.newInstance(	2✔
71	columnPageEncoderMeta.getColumnSpec().getFieldName(), DataTypes.INT);	2✔
72	try {
73	rowOffset = ColumnPage.newPage(	2✔
74	new ColumnPageEncoderMeta(spec, DataTypes.INT, columnPageEncoderMeta.getCompressorName()),	2✔
75	pageSize);
76	} catch (MemoryException e) {	×
77	throw new RuntimeException(e);	×
78	}	2✔
79	totalLength = 0;	2✔
80	}	2✔
81
82	@Override
83	public void setBytePage(byte[] byteData) {
84	throw new UnsupportedOperationException(	×
85	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
86	}
87
88	@Override
89	public void setShortPage(short[] shortData) {
90	throw new UnsupportedOperationException(	×
91	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
92	}
93
94	@Override
95	public void setShortIntPage(byte[] shortIntData) {
96	throw new UnsupportedOperationException(	×
97	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
98	}
99
100	@Override
101	public void setIntPage(int[] intData) {
102	throw new UnsupportedOperationException(	×
103	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
104	}
105
106	@Override
107	public void setLongPage(long[] longData) {
108	throw new UnsupportedOperationException(	×
109	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
110	}
111
112	@Override
113	public void setFloatPage(float[] floatData) {
114	throw new UnsupportedOperationException(	×
115	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
116	}
117
118	@Override
119	public void setDoublePage(double[] doubleData) {
120	throw new UnsupportedOperationException(	×
121	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
122	}
123
124	/**
125	* Create a new column page for decimal page
126	*/
127	static ColumnPage newDecimalColumnPage(TableSpec.ColumnSpec columnSpec, byte[] lvEncodedBytes,
128	String compressorName) throws MemoryException {
129	DecimalConverterFactory.DecimalConverter decimalConverter =	1✔
130	DecimalConverterFactory.INSTANCE.getDecimalConverter(columnSpec.getPrecision(),	1✔
131	columnSpec.getScale());	1✔
132	int size = decimalConverter.getSize();	1✔
133	if (size < 0) {	1✔
134	return getLVBytesColumnPage(columnSpec, lvEncodedBytes,	×
135	DataTypes.createDecimalType(columnSpec.getPrecision(), columnSpec.getScale()),	×
136	CarbonCommonConstants.INT_SIZE_IN_BYTE, compressorName);
137	} else {
138	// Here the size is always fixed.
139	return getDecimalColumnPage(columnSpec, lvEncodedBytes, size, compressorName);	1✔
140	}
141	}
142
143	/**
144	* Create a new column page based on the LV (Length Value) encoded bytes
145	*/
146	static ColumnPage newLVBytesColumnPage(TableSpec.ColumnSpec columnSpec, byte[] lvEncodedBytes,
147	int lvLength, String compressorName) throws MemoryException {
148	return getLVBytesColumnPage(columnSpec, lvEncodedBytes, DataTypes.BYTE_ARRAY,	×
149	lvLength, compressorName);
150	}
151
152	/**
153	* Create a new column page based on the LV (Length Value) encoded bytes
154	*/
155	static ColumnPage newComplexLVBytesColumnPage(TableSpec.ColumnSpec columnSpec,
156	byte[] lvEncodedBytes, int lvLength, String compressorName) throws MemoryException {
157	return getComplexLVBytesColumnPage(columnSpec, lvEncodedBytes, DataTypes.BYTE_ARRAY,	×
158	lvLength, compressorName);
159	}
160
161	private static ColumnPage getDecimalColumnPage(TableSpec.ColumnSpec columnSpec,
162	byte[] lvEncodedBytes, int size, String compressorName) throws MemoryException {
163	TableSpec.ColumnSpec spec = TableSpec.ColumnSpec	1✔
164	.newInstance(columnSpec.getFieldName(), DataTypes.INT, ColumnType.MEASURE);	1✔
165	ColumnPage rowOffset = ColumnPage.newPage(	1✔
166	new ColumnPageEncoderMeta(spec, DataTypes.INT, compressorName),
167	CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT);
168	int offset;
169	int rowId = 0;	1✔
170	int counter = 0;	1✔
171	for (offset = 0; offset < lvEncodedBytes.length; offset += size) {	1✔
172	rowOffset.putInt(counter, offset);	1✔
173	rowId++;	1✔
174	counter++;	1✔
175	}
176	rowOffset.putInt(counter, offset);	1✔
177
178	VarLengthColumnPageBase page;
179	if (unsafe) {	1✔
180	page = new UnsafeDecimalColumnPage(	1✔
181	new ColumnPageEncoderMeta(columnSpec, columnSpec.getSchemaDataType(), compressorName),	1✔
182	rowId);
183	} else {
184	page = new SafeDecimalColumnPage(	×
185	new ColumnPageEncoderMeta(columnSpec, columnSpec.getSchemaDataType(), compressorName),	×
186	rowId);
187	}
188
189	// set total length and rowOffset in page
190	page.totalLength = offset;	1✔
191	page.rowOffset.freeMemory();	1✔
192	page.rowOffset = rowOffset;	1✔
193	for (int i = 0; i < rowId; i++) {	1✔
194	page.putBytes(i, lvEncodedBytes, i * size, size);	1✔
195	}
196	return page;	1✔
197	}
198
199	private static ColumnPage getLVBytesColumnPage(TableSpec.ColumnSpec columnSpec,
200	byte[] lvEncodedBytes, DataType dataType, int lvLength, String compressorName)
201	throws MemoryException {
202	// extract length and data, set them to rowOffset and unsafe memory correspondingly
203	int rowId = 0;	×
204	TableSpec.ColumnSpec spec = TableSpec.ColumnSpec	×
205	.newInstance(columnSpec.getFieldName(), DataTypes.INT, ColumnType.MEASURE);	×
206	ColumnPage rowOffset = ColumnPage.newPage(	×
207	new ColumnPageEncoderMeta(spec, DataTypes.INT, compressorName),
208	CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT);
209	int length;
210	int offset;
211	int lvEncodedOffset = 0;	×
212	int counter = 0;	×
213	// extract Length field in input and calculate total length
214	for (offset = 0; lvEncodedOffset < lvEncodedBytes.length; offset += length) {	×
215	length = ByteUtil.toInt(lvEncodedBytes, lvEncodedOffset);	×
216	rowOffset.putInt(counter, offset);	×
217	lvEncodedOffset += lvLength + length;	×
218	rowId++;	×
219	counter++;	×
220	}
221	rowOffset.putInt(counter, offset);	×
222	return getVarLengthColumnPage(columnSpec, lvEncodedBytes, dataType,	×
223	lvLength, rowId, rowOffset, offset, compressorName);
224	}
225
226	private static ColumnPage getComplexLVBytesColumnPage(TableSpec.ColumnSpec columnSpec,
227	byte[] lvEncodedBytes, DataType dataType, int lvLength, String compressorName)
228	throws MemoryException {
229	// extract length and data, set them to rowOffset and unsafe memory correspondingly
230	int rowId = 0;	×
231	TableSpec.ColumnSpec spec = TableSpec.ColumnSpec	×
232	.newInstance(columnSpec.getFieldName(), DataTypes.INT, ColumnType.MEASURE);	×
233	ColumnPage rowOffset = ColumnPage.newPage(	×
234	new ColumnPageEncoderMeta(spec, DataTypes.INT, compressorName),
235	CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT);
236	int length;
237	int offset;
238	int lvEncodedOffset = 0;	×
239	int counter = 0;	×
240	// extract Length field in input and calculate total length
241	for (offset = 0; lvEncodedOffset < lvEncodedBytes.length; offset += length) {	×
242	length = ByteUtil.toShort(lvEncodedBytes, lvEncodedOffset);	×
243	rowOffset.putInt(counter, offset);	×
244	lvEncodedOffset += lvLength + length;	×
245	rowId++;	×
246	counter++;	×
247	}
248	rowOffset.putInt(counter, offset);	×
249
250	return getVarLengthColumnPage(columnSpec, lvEncodedBytes, dataType,	×
251	lvLength, rowId, rowOffset, offset, compressorName);
252	}
253
254	private static VarLengthColumnPageBase getVarLengthColumnPage(TableSpec.ColumnSpec columnSpec,
255	byte[] lvEncodedBytes, DataType dataType, int lvLength, int rowId, ColumnPage rowOffset,
256	int offset, String compressorName) throws MemoryException {
257	int lvEncodedOffset;
258	int length;
259	int numRows = rowId;	×
260
261	VarLengthColumnPageBase page;
262	int inputDataLength = offset;	×
263	if (unsafe) {	×
264	page = new UnsafeDecimalColumnPage(	×
265	new ColumnPageEncoderMeta(columnSpec, dataType, compressorName), numRows,
266	inputDataLength);
267	} else {
268	page = new SafeDecimalColumnPage(	×
269	new ColumnPageEncoderMeta(columnSpec, dataType, compressorName), numRows);
270	}
271
272	// set total length and rowOffset in page
273	page.totalLength = offset;	×
274	page.rowOffset.freeMemory();	×
275	page.rowOffset = rowOffset;	×
276
277	// set data in page
278	lvEncodedOffset = 0;	×
279	for (int i = 0; i < numRows; i++) {	×
280	length = rowOffset.getInt(i + 1) - rowOffset.getInt(i);	×
281	page.putBytes(i, lvEncodedBytes, lvEncodedOffset + lvLength, length);	×
282	lvEncodedOffset += lvLength + length;	×
283	}
284	return page;	×
285	}
286
287	@Override
288	public void putByte(int rowId, byte value) {
289	throw new UnsupportedOperationException(	×
290	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
291	}
292
293	@Override
294	public void putShort(int rowId, short value) {
295	throw new UnsupportedOperationException(	×
296	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
297	}
298
299	@Override
300	public void putShortInt(int rowId, int value) {
301	throw new UnsupportedOperationException(	×
302	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
303	}
304
305	@Override
306	public void putInt(int rowId, int value) {
307	throw new UnsupportedOperationException(	×
308	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
309	}
310
311	@Override
312	public void putLong(int rowId, long value) {
313	throw new UnsupportedOperationException(	×
314	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
315	}
316
317	@Override
318	public void putDouble(int rowId, double value) {
319	throw new UnsupportedOperationException(	×
320	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
321	}
322
323	@Override
324	public void putFloat(int rowId, float value) {
325	throw new UnsupportedOperationException(	×
326	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
327	}
328
329	abstract void putBytesAtRow(int rowId, byte[] bytes);
330
331	@Override
332	public void putBytes(int rowId, byte[] bytes) {
333	// rowId * 4 represents the length of L in LV
334	if (bytes.length > (Integer.MAX_VALUE - totalLength - rowId * 4)) {	2✔
335	// since we later store a column page in a byte array, so its maximum size is 2GB
336	throw new RuntimeException("Carbondata only support maximum 2GB size for one column page,"	×
337	+ " exceed this limit at rowId " + rowId);
338	}
339	if (rowId == 0) {	2✔
340	rowOffset.putInt(0, 0);	2✔
341	}
342	rowOffset.putInt(rowId + 1, rowOffset.getInt(rowId) + bytes.length);	2✔
343	putBytesAtRow(rowId, bytes);	2✔
344	totalLength += bytes.length;	2✔
345	pageSize = rowId + 1;	2✔
346	}	2✔
347
348	@Override
349	public byte getByte(int rowId) {
350	throw new UnsupportedOperationException(	×
351	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
352	}
353
354	@Override
355	public short getShort(int rowId) {
356	throw new UnsupportedOperationException(	×
357	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
358	}
359
360	@Override
361	public int getShortInt(int rowId) {
362	throw new UnsupportedOperationException(	×
363	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
364	}
365
366	@Override
367	public int getInt(int rowId) {
368	throw new UnsupportedOperationException(	×
369	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
370	}
371
372	@Override
373	public long getLong(int rowId) {
374	throw new UnsupportedOperationException(	×
375	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
376	}
377
378	@Override
379	public float getFloat(int rowId) {
380	throw new UnsupportedOperationException(	×
381	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
382	}
383
384	@Override
385	public double getDouble(int rowId) {
386	throw new UnsupportedOperationException(	×
387	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
388	}
389
390	@Override
391	public byte[] getBytePage() {
392	throw new UnsupportedOperationException(	×
393	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
394	}
395
396	@Override
397	public short[] getShortPage() {
398	throw new UnsupportedOperationException(	×
399	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
400	}
401
402	@Override
403	public byte[] getShortIntPage() {
404	throw new UnsupportedOperationException(	×
405	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
406	}
407
408	@Override
409	public int[] getIntPage() {
410	throw new UnsupportedOperationException(	×
411	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
412	}
413
414	@Override
415	public long[] getLongPage() {
416	throw new UnsupportedOperationException(	×
417	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
418	}
419
420	@Override
421	public float[] getFloatPage() {
422	throw new UnsupportedOperationException(	×
423	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
424	}
425
426	@Override
427	public double[] getDoublePage() {
428	throw new UnsupportedOperationException(	×
429	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
430	}
431
432	@Override
433	public byte[] getDecimalPage() {
434	// output LV encoded byte array
435	int offset = 0;	1✔
436	byte[] data = new byte[totalLength];	1✔
437	for (int rowId = 0; rowId < pageSize; rowId++) {	1✔
438	int length = rowOffset.getInt(rowId + 1) - rowOffset.getInt(rowId);	1✔
439	copyBytes(rowId, data, offset, length);	1✔
440	offset += length;	1✔
441	}
442	return data;	1✔
443	}
444
445	/**
446	* Copy `length` bytes from data at rowId to dest start from destOffset
447	*/
448	abstract void copyBytes(int rowId, byte[] dest, int destOffset, int length);
449
450	@Override
451	public byte[] getLVFlattenedBytePage() throws IOException {
452	// output LV encoded byte array
453	int offset = 0;	×
454	byte[] data = new byte[totalLength + ((rowOffset.getActualRowCount() - 1) * 4)];	×
455	for (int rowId = 0; rowId < rowOffset.getActualRowCount() - 1; rowId++) {	×
456	int length = rowOffset.getInt(rowId + 1) - rowOffset.getInt(rowId);	×
457	ByteUtil.setInt(data, offset, length);	×
458	copyBytes(rowId, data, offset + 4, length);	×
459	offset += 4 + length;	×
460	}
461	return data;	×
462	}
463
464	@Override public byte[] getComplexChildrenLVFlattenedBytePage() throws IOException {
465	// output LV encoded byte array
466	int offset = 0;	×
467	byte[] data = new byte[totalLength + ((rowOffset.getActualRowCount() - 1) * 2)];	×
468	for (int rowId = 0; rowId < rowOffset.getActualRowCount() - 1; rowId++) {	×
469	short length = (short) (rowOffset.getInt(rowId + 1) - rowOffset.getInt(rowId));	×
470	ByteUtil.setShort(data, offset, length);	×
471	copyBytes(rowId, data, offset + 2, length);	×
472	offset += 2 + length;	×
473	}
474	return data;	×
475	}
476
477	@Override
478	public byte[] getComplexParentFlattenedBytePage() throws IOException {
479	// output LV encoded byte array
480	int offset = 0;	1✔
481	byte[] data = new byte[totalLength];	1✔
482	for (int rowId = 0; rowId < rowOffset.getActualRowCount() - 1; rowId++) {	1✔
483	int length = (rowOffset.getInt(rowId + 1) - rowOffset.getInt(rowId));	1✔
484	copyBytes(rowId, data, offset, length);	1✔
485	offset += length;	1✔
486	}
487	return data;	1✔
488	}
489
490	@Override
491	public void convertValue(ColumnPageValueConverter codec) {
492	throw new UnsupportedOperationException(	×
493	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
494	}
495
496	/**
497	* reallocate memory if capacity length than current size + request size
498	*/
499	protected void ensureMemory(int requestSize) throws MemoryException {
500	if (totalLength + requestSize > capacity) {	2✔
501	int newSize = Math.max(2 * capacity, totalLength + requestSize);	1✔
502	MemoryBlock newBlock = UnsafeMemoryManager.allocateMemoryWithRetry(taskId, newSize);	1✔
503	CarbonUnsafe.getUnsafe().copyMemory(baseAddress, baseOffset,	1✔
504	newBlock.getBaseObject(), newBlock.getBaseOffset(), capacity);	1✔
505	UnsafeMemoryManager.INSTANCE.freeMemory(taskId, memoryBlock);	1✔
506	memoryBlock = newBlock;	1✔
507	baseAddress = newBlock.getBaseObject();	1✔
508	baseOffset = newBlock.getBaseOffset();	1✔
509	capacity = newSize;	1✔
510	}
511	}	2✔
512
513	/**
514	* free memory as needed
515	*/
516	public void freeMemory() {
517	if (null != rowOffset) {	2✔
518	rowOffset.freeMemory();	2✔
519	rowOffset = null;	2✔
520	}
521	}	2✔
522	}

apache / carbondata / 9055

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