apache / carbondata / 551

Build # 551

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/ce7d6282ef0297dca7707f488f3010e0ec321008">ce7d6282e<a href="https://github.com/apache/carbondata/commit/ce7d6282ef0297dca7707f488f3010e0ec321008">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/1d48b62c9b5fcd88ca57f66cd874a15e2df3010e">1d48b62c9</a>">1d48b62c9</a><a href="https://github.com/apache/carbondata/commit/ce7d6282ef0297dca7707f488f3010e0ec321008"> into d1bfb7477">d1bfb7477</a>

Run Details

274 of 274 new or added lines in 37 files covered. (100.0%)

59757 of 74233 relevant lines covered (80.5%)

1.07 hits per line

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

55.28

/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 long 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.ColumnSpec spec = TableSpec.ColumnSpec.newInstance(
        columnPageEncoderMeta.getColumnSpec().getFieldName(), DataTypes.INT, ColumnType.MEASURE);
    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;
  }

  @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 long 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.ColumnSpec spec = TableSpec.ColumnSpec.newInstance(	2✔
71	columnPageEncoderMeta.getColumnSpec().getFieldName(), DataTypes.INT, ColumnType.MEASURE);	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,	2✔
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;	2✔
231	TableSpec.ColumnSpec spec = TableSpec.ColumnSpec	2✔
232	.newInstance(columnSpec.getFieldName(), DataTypes.INT, ColumnType.MEASURE);	2✔
233	ColumnPage rowOffset = ColumnPage.newPage(	2✔
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;	2✔
239	int counter = 0;	2✔
240	// extract Length field in input and calculate total length
241	for (offset = 0; lvEncodedOffset < lvEncodedBytes.length; offset += length) {	2✔
242	length = ByteUtil.toShort(lvEncodedBytes, lvEncodedOffset);	2✔
243	rowOffset.putInt(counter, offset);	2✔
244	lvEncodedOffset += lvLength + length;	2✔
245	rowId++;	2✔
246	counter++;	2✔
247	}
248	rowOffset.putInt(counter, offset);	2✔
249
250	return getVarLengthColumnPage(columnSpec, lvEncodedBytes, dataType,	2✔
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;	2✔
260
261	VarLengthColumnPageBase page;
262	int inputDataLength = offset;	2✔
263	if (unsafe) {	2✔
264	page = new UnsafeDecimalColumnPage(	2✔
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;	2✔
274	page.rowOffset.freeMemory();	2✔
275	page.rowOffset = rowOffset;	2✔
276
277	// set data in page
278	lvEncodedOffset = 0;	2✔
279	for (int i = 0; i < numRows; i++) {	2✔
280	length = rowOffset.getInt(i + 1) - rowOffset.getInt(i);	2✔
281	page.putBytes(i, lvEncodedBytes, lvEncodedOffset + lvLength, length);	2✔
282	lvEncodedOffset += lvLength + length;	2✔
283	}
284	return page;	2✔
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	}	2✔
346
347	@Override
348	public byte getByte(int rowId) {
349	throw new UnsupportedOperationException(	×
350	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
351	}
352
353	@Override
354	public short getShort(int rowId) {
355	throw new UnsupportedOperationException(	×
356	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
357	}
358
359	@Override
360	public int getShortInt(int rowId) {
361	throw new UnsupportedOperationException(	×
362	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
363	}
364
365	@Override
366	public int getInt(int rowId) {
367	throw new UnsupportedOperationException(	×
368	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
369	}
370
371	@Override
372	public long getLong(int rowId) {
373	throw new UnsupportedOperationException(	×
374	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
375	}
376
377	@Override
378	public float getFloat(int rowId) {
379	throw new UnsupportedOperationException(	×
380	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
381	}
382
383	@Override
384	public double getDouble(int rowId) {
385	throw new UnsupportedOperationException(	×
386	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
387	}
388
389	@Override
390	public byte[] getBytePage() {
391	throw new UnsupportedOperationException(	×
392	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
393	}
394
395	@Override
396	public short[] getShortPage() {
397	throw new UnsupportedOperationException(	×
398	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
399	}
400
401	@Override
402	public byte[] getShortIntPage() {
403	throw new UnsupportedOperationException(	×
404	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
405	}
406
407	@Override
408	public int[] getIntPage() {
409	throw new UnsupportedOperationException(	×
410	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
411	}
412
413	@Override
414	public long[] getLongPage() {
415	throw new UnsupportedOperationException(	×
416	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
417	}
418
419	@Override
420	public float[] getFloatPage() {
421	throw new UnsupportedOperationException(	×
422	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
423	}
424
425	@Override
426	public double[] getDoublePage() {
427	throw new UnsupportedOperationException(	×
428	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
429	}
430
431	@Override
432	public byte[] getDecimalPage() {
433	// output LV encoded byte array
434	int offset = 0;	1✔
435	byte[] data = new byte[totalLength];	1✔
436	for (int rowId = 0; rowId < pageSize; rowId++) {	1✔
437	int length = rowOffset.getInt(rowId + 1) - rowOffset.getInt(rowId);	1✔
438	copyBytes(rowId, data, offset, length);	1✔
439	offset += length;	1✔
440	}
441	return data;	1✔
442	}
443
444	/**
445	* Copy `length` bytes from data at rowId to dest start from destOffset
446	*/
447	abstract void copyBytes(int rowId, byte[] dest, int destOffset, int length);
448
449	@Override
450	public byte[] getLVFlattenedBytePage() throws IOException {
451	// output LV encoded byte array
452	int offset = 0;	×
453	byte[] data = new byte[totalLength + ((rowOffset.getActualRowCount() - 1) * 4)];	×
454	for (int rowId = 0; rowId < rowOffset.getActualRowCount() - 1; rowId++) {	×
455	int length = rowOffset.getInt(rowId + 1) - rowOffset.getInt(rowId);	×
456	ByteUtil.setInt(data, offset, length);	×
457	copyBytes(rowId, data, offset + 4, length);	×
458	offset += 4 + length;	×
459	}
460	return data;	×
461	}
462
463	@Override public byte[] getComplexChildrenLVFlattenedBytePage() throws IOException {
464	// output LV encoded byte array
465	int offset = 0;	1✔
466	byte[] data = new byte[totalLength + ((rowOffset.getActualRowCount() - 1) * 2)];	1✔
467	for (int rowId = 0; rowId < rowOffset.getActualRowCount() - 1; rowId++) {	1✔
468	short length = (short) (rowOffset.getInt(rowId + 1) - rowOffset.getInt(rowId));	1✔
469	ByteUtil.setShort(data, offset, length);	1✔
470	copyBytes(rowId, data, offset + 2, length);	1✔
471	offset += 2 + length;	1✔
472	}
473	return data;	1✔
474	}
475
476	@Override
477	public byte[] getComplexParentFlattenedBytePage() throws IOException {
478	// output LV encoded byte array
479	int offset = 0;	2✔
480	byte[] data = new byte[totalLength];	2✔
481	for (int rowId = 0; rowId < rowOffset.getActualRowCount() - 1; rowId++) {	2✔
482	int length = (rowOffset.getInt(rowId + 1) - rowOffset.getInt(rowId));	2✔
483	copyBytes(rowId, data, offset, length);	2✔
484	offset += length;	2✔
485	}
486	return data;	2✔
487	}
488
489	@Override
490	public void convertValue(ColumnPageValueConverter codec) {
491	throw new UnsupportedOperationException(	×
492	"invalid data type: " + columnPageEncoderMeta.getStoreDataType());	×
493	}
494
495	/**
496	* reallocate memory if capacity length than current size + request size
497	*/
498	protected void ensureMemory(int requestSize) throws MemoryException {
499	if (totalLength + requestSize > capacity) {	2✔
500	int newSize = Math.max(2 * capacity, totalLength + requestSize);	2✔
501	MemoryBlock newBlock = UnsafeMemoryManager.allocateMemoryWithRetry(taskId, newSize);	2✔
502	CarbonUnsafe.getUnsafe().copyMemory(baseAddress, baseOffset,	2✔
503	newBlock.getBaseObject(), newBlock.getBaseOffset(), capacity);	2✔
504	UnsafeMemoryManager.INSTANCE.freeMemory(taskId, memoryBlock);	2✔
505	memoryBlock = newBlock;	2✔
506	baseAddress = newBlock.getBaseObject();	2✔
507	baseOffset = newBlock.getBaseOffset();	2✔
508	capacity = newSize;	2✔
509	}
510	}	2✔
511
512	/**
513	* free memory as needed
514	*/
515	public void freeMemory() {
516	if (null != rowOffset) {	2✔
517	rowOffset.freeMemory();	2✔
518	rowOffset = null;	2✔
519	}
520	}	2✔
521	}

apache / carbondata / 551

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