apache / carbondata / 685

Build # 685

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/bd70e3b1ac946e91cbf20b538023c521bb9e09c8">bd70e3b1a<a href="https://github.com/apache/carbondata/commit/bd70e3b1ac946e91cbf20b538023c521bb9e09c8">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/c32d1580dbc68bb1471f1934ac772ab216635bd1">c32d1580d</a>">c32d1580d</a><a href="https://github.com/apache/carbondata/commit/bd70e3b1ac946e91cbf20b538023c521bb9e09c8"> into 2ab2254be">2ab2254be</a>

Run Details

29 of 29 new or added lines in 7 files covered. (100.0%)

59941 of 74401 relevant lines covered (80.56%)

1.07 hits per line

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

76.47

/core/src/main/java/org/apache/carbondata/core/indexstore/UnsafeMemoryDMStore.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.indexstore;

import org.apache.carbondata.core.indexstore.row.DataMapRow;
import org.apache.carbondata.core.indexstore.row.UnsafeDataMapRow;
import org.apache.carbondata.core.indexstore.schema.CarbonRowSchema;
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 static org.apache.carbondata.core.memory.CarbonUnsafe.BYTE_ARRAY_OFFSET;
import static org.apache.carbondata.core.memory.CarbonUnsafe.getUnsafe;

/**
 * Store the data map row @{@link DataMapRow} data to unsafe.
 */
public class UnsafeMemoryDMStore extends AbstractMemoryDMStore {

  private static final long serialVersionUID = -5344592407101055335L;

  private transient MemoryBlock memoryBlock;

  private static int capacity = 8 * 1024;

  private int allocatedSize;

  private int runningLength;

  private int[] pointers;

  private int rowCount;

  public UnsafeMemoryDMStore() throws MemoryException {
    this.allocatedSize = capacity;
    this.memoryBlock = UnsafeMemoryManager.allocateMemoryWithRetry(taskId, allocatedSize, true);
    this.pointers = new int[1000];
  }

  /**
   * Check memory is sufficient or not, if not sufficient allocate more memory and copy old data to
   * new one.
   *
   * @param rowSize
   */
  private void ensureSize(int rowSize) throws MemoryException {
    if (runningLength + rowSize >= allocatedSize) {
      increaseMemory(runningLength + rowSize);
    }
    if (this.pointers.length <= rowCount + 1) {
      int[] newPointer = new int[pointers.length + 1000];
      System.arraycopy(pointers, 0, newPointer, 0, pointers.length);
      this.pointers = newPointer;
    }
  }

  private void increaseMemory(int requiredMemory) throws MemoryException {
    MemoryBlock newMemoryBlock =
        UnsafeMemoryManager.allocateMemoryWithRetry(taskId, allocatedSize + requiredMemory, true);
    getUnsafe().copyMemory(this.memoryBlock.getBaseObject(), this.memoryBlock.getBaseOffset(),
        newMemoryBlock.getBaseObject(), newMemoryBlock.getBaseOffset(), runningLength);
    UnsafeMemoryManager.INSTANCE.freeMemory(taskId, this.memoryBlock, true);
    allocatedSize = allocatedSize + requiredMemory;
    this.memoryBlock = newMemoryBlock;
  }

  /**
   * Add the index row to unsafe.
   *
   * @param indexRow
   * @return
   */
  public void addIndexRow(CarbonRowSchema[] schema, DataMapRow indexRow) throws MemoryException {
    // First calculate the required memory to keep the row in unsafe
    int rowSize = indexRow.getTotalSizeInBytes();
    // Check whether allocated memory is sufficient or not.
    ensureSize(rowSize);
    int pointer = runningLength;

    for (int i = 0; i < schema.length; i++) {
      addToUnsafe(schema[i], indexRow, i);
    }
    pointers[rowCount++] = pointer;
  }

  private void addToUnsafe(CarbonRowSchema schema, DataMapRow row, int index) {
    switch (schema.getSchemaType()) {
      case FIXED:
        DataType dataType = schema.getDataType();
        if (dataType == DataTypes.BYTE) {
          getUnsafe()
              .putByte(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,
                  row.getByte(index));
          runningLength += row.getSizeInBytes(index);
        } else if (dataType == DataTypes.BOOLEAN) {
          getUnsafe()
              .putBoolean(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,
                  row.getBoolean(index));
          runningLength += row.getSizeInBytes(index);
        } else if (dataType == DataTypes.SHORT) {
          getUnsafe()
              .putShort(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,
                  row.getShort(index));
          runningLength += row.getSizeInBytes(index);
        } else if (dataType == DataTypes.INT) {
          getUnsafe()
              .putInt(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,
                  row.getInt(index));
          runningLength += row.getSizeInBytes(index);
        } else if (dataType == DataTypes.LONG) {
          getUnsafe()
              .putLong(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,
                  row.getLong(index));
          runningLength += row.getSizeInBytes(index);
        } else if (dataType == DataTypes.FLOAT) {
          getUnsafe()
              .putFloat(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,
                  row.getFloat(index));
          runningLength += row.getSizeInBytes(index);
        } else if (dataType == DataTypes.DOUBLE) {
          getUnsafe()
              .putDouble(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,
                  row.getDouble(index));
          runningLength += row.getSizeInBytes(index);
        } else if (dataType == DataTypes.BYTE_ARRAY) {
          byte[] data = row.getByteArray(index);
          getUnsafe().copyMemory(data, BYTE_ARRAY_OFFSET, memoryBlock.getBaseObject(),
              memoryBlock.getBaseOffset() + runningLength, data.length);
          runningLength += row.getSizeInBytes(index);
        } else {
          throw new UnsupportedOperationException(
              "unsupported data type for unsafe storage: " + schema.getDataType());
        }
        break;
      case VARIABLE_SHORT:
        byte[] data = row.getByteArray(index);
        getUnsafe().putShort(memoryBlock.getBaseObject(),
            memoryBlock.getBaseOffset() + runningLength, (short) data.length);
        runningLength += 2;
        getUnsafe().copyMemory(data, BYTE_ARRAY_OFFSET, memoryBlock.getBaseObject(),
            memoryBlock.getBaseOffset() + runningLength, data.length);
        runningLength += data.length;
        break;
      case VARIABLE_INT:
        byte[] data2 = row.getByteArray(index);
        getUnsafe().putInt(memoryBlock.getBaseObject(),
            memoryBlock.getBaseOffset() + runningLength, data2.length);
        runningLength += 4;
        getUnsafe().copyMemory(data2, BYTE_ARRAY_OFFSET, memoryBlock.getBaseObject(),
            memoryBlock.getBaseOffset() + runningLength, data2.length);
        runningLength += data2.length;
        break;
      case STRUCT:
        CarbonRowSchema[] childSchemas =
            ((CarbonRowSchema.StructCarbonRowSchema) schema).getChildSchemas();
        DataMapRow struct = row.getRow(index);
        for (int i = 0; i < childSchemas.length; i++) {
          addToUnsafe(childSchemas[i], struct, i);
        }
        break;
      default:
        throw new UnsupportedOperationException(
            "unsupported data type for unsafe storage: " + schema.getDataType());
    }
  }

  public DataMapRow getDataMapRow(CarbonRowSchema[] schema, int index) {
    assert (index < rowCount);
    return new UnsafeDataMapRow(schema, memoryBlock, pointers[index]);
  }

  public void finishWriting() throws MemoryException {
    if (runningLength < allocatedSize) {
      MemoryBlock allocate =
          UnsafeMemoryManager.allocateMemoryWithRetry(taskId, runningLength, true);
      getUnsafe().copyMemory(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset(),
          allocate.getBaseObject(), allocate.getBaseOffset(), runningLength);
      UnsafeMemoryManager.INSTANCE.freeMemory(taskId, memoryBlock, true);
      memoryBlock = allocate;
    }
    // Compact pointers.
    if (rowCount < pointers.length) {
      int[] newPointer = new int[rowCount];
      System.arraycopy(pointers, 0, newPointer, 0, rowCount);
      this.pointers = newPointer;
    }
  }

  public void freeMemory() {
    if (!isMemoryFreed) {
      UnsafeMemoryManager.INSTANCE.freeMemory(taskId, memoryBlock, true);
      isMemoryFreed = true;
    }
  }

  public int getMemoryUsed() {
    return runningLength;
  }

  public int getRowCount() {
    return rowCount;
  }

}

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	package org.apache.carbondata.core.indexstore;
18
19	import org.apache.carbondata.core.indexstore.row.DataMapRow;
20	import org.apache.carbondata.core.indexstore.row.UnsafeDataMapRow;
21	import org.apache.carbondata.core.indexstore.schema.CarbonRowSchema;
22	import org.apache.carbondata.core.memory.MemoryBlock;
23	import org.apache.carbondata.core.memory.MemoryException;
24	import org.apache.carbondata.core.memory.UnsafeMemoryManager;
25	import org.apache.carbondata.core.metadata.datatype.DataType;
26	import org.apache.carbondata.core.metadata.datatype.DataTypes;
27
28	import static org.apache.carbondata.core.memory.CarbonUnsafe.BYTE_ARRAY_OFFSET;
29	import static org.apache.carbondata.core.memory.CarbonUnsafe.getUnsafe;
30
31	/**
32	* Store the data map row @{@link DataMapRow} data to unsafe.
33	*/
34	public class UnsafeMemoryDMStore extends AbstractMemoryDMStore {	1✔
35
36	private static final long serialVersionUID = -5344592407101055335L;
37
38	private transient MemoryBlock memoryBlock;
39
40	private static int capacity = 8 * 1024;	1✔
41
42	private int allocatedSize;
43
44	private int runningLength;
45
46	private int[] pointers;
47
48	private int rowCount;
49
50	public UnsafeMemoryDMStore() throws MemoryException {	1✔
51	this.allocatedSize = capacity;	1✔
52	this.memoryBlock = UnsafeMemoryManager.allocateMemoryWithRetry(taskId, allocatedSize, true);	1✔
53	this.pointers = new int[1000];	1✔
54	}	1✔
55
56	/**
57	* Check memory is sufficient or not, if not sufficient allocate more memory and copy old data to
58	* new one.
59	*
60	* @param rowSize
61	*/
62	private void ensureSize(int rowSize) throws MemoryException {
63	if (runningLength + rowSize >= allocatedSize) {	1✔
64	increaseMemory(runningLength + rowSize);	×
65	}
66	if (this.pointers.length <= rowCount + 1) {	1✔
67	int[] newPointer = new int[pointers.length + 1000];	×
68	System.arraycopy(pointers, 0, newPointer, 0, pointers.length);	×
69	this.pointers = newPointer;	×
70	}
71	}	1✔
72
73	private void increaseMemory(int requiredMemory) throws MemoryException {
74	MemoryBlock newMemoryBlock =	×
75	UnsafeMemoryManager.allocateMemoryWithRetry(taskId, allocatedSize + requiredMemory, true);	×
76	getUnsafe().copyMemory(this.memoryBlock.getBaseObject(), this.memoryBlock.getBaseOffset(),	×
77	newMemoryBlock.getBaseObject(), newMemoryBlock.getBaseOffset(), runningLength);	×
78	UnsafeMemoryManager.INSTANCE.freeMemory(taskId, this.memoryBlock, true);	×
79	allocatedSize = allocatedSize + requiredMemory;	×
80	this.memoryBlock = newMemoryBlock;	×
81	}	×
82
83	/**
84	* Add the index row to unsafe.
85	*
86	* @param indexRow
87	* @return
88	*/
89	public void addIndexRow(CarbonRowSchema[] schema, DataMapRow indexRow) throws MemoryException {
90	// First calculate the required memory to keep the row in unsafe
91	int rowSize = indexRow.getTotalSizeInBytes();	1✔
92	// Check whether allocated memory is sufficient or not.
93	ensureSize(rowSize);	1✔
94	int pointer = runningLength;	1✔
95
96	for (int i = 0; i < schema.length; i++) {	1✔
97	addToUnsafe(schema[i], indexRow, i);	1✔
98	}
99	pointers[rowCount++] = pointer;	1✔
100	}	1✔
101
102	private void addToUnsafe(CarbonRowSchema schema, DataMapRow row, int index) {
103	switch (schema.getSchemaType()) {	1✔
104	case FIXED:
105	DataType dataType = schema.getDataType();	1✔
106	if (dataType == DataTypes.BYTE) {	1✔
107	getUnsafe()	×
108	.putByte(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,	×
109	row.getByte(index));	×
110	runningLength += row.getSizeInBytes(index);	×
111	} else if (dataType == DataTypes.BOOLEAN) {	1✔
112	getUnsafe()	1✔
113	.putBoolean(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,	1✔
114	row.getBoolean(index));	1✔
115	runningLength += row.getSizeInBytes(index);	1✔
116	} else if (dataType == DataTypes.SHORT) {	1✔
117	getUnsafe()	1✔
118	.putShort(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,	1✔
119	row.getShort(index));	1✔
120	runningLength += row.getSizeInBytes(index);	1✔
121	} else if (dataType == DataTypes.INT) {	1✔
122	getUnsafe()	1✔
123	.putInt(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,	1✔
124	row.getInt(index));	1✔
125	runningLength += row.getSizeInBytes(index);	1✔
126	} else if (dataType == DataTypes.LONG) {	1✔
127	getUnsafe()	1✔
128	.putLong(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,	1✔
129	row.getLong(index));	1✔
130	runningLength += row.getSizeInBytes(index);	1✔
131	} else if (dataType == DataTypes.FLOAT) {	1✔
132	getUnsafe()	×
133	.putFloat(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,	×
134	row.getFloat(index));	×
135	runningLength += row.getSizeInBytes(index);	×
136	} else if (dataType == DataTypes.DOUBLE) {	1✔
137	getUnsafe()	×
138	.putDouble(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset() + runningLength,	×
139	row.getDouble(index));	×
140	runningLength += row.getSizeInBytes(index);	×
141	} else if (dataType == DataTypes.BYTE_ARRAY) {	1✔
142	byte[] data = row.getByteArray(index);	1✔
143	getUnsafe().copyMemory(data, BYTE_ARRAY_OFFSET, memoryBlock.getBaseObject(),	1✔
144	memoryBlock.getBaseOffset() + runningLength, data.length);	1✔
145	runningLength += row.getSizeInBytes(index);	1✔
146	} else {	1✔
147	throw new UnsupportedOperationException(	×
148	"unsupported data type for unsafe storage: " + schema.getDataType());	×
149	}
150	break;
151	case VARIABLE_SHORT:
152	byte[] data = row.getByteArray(index);	1✔
153	getUnsafe().putShort(memoryBlock.getBaseObject(),	1✔
154	memoryBlock.getBaseOffset() + runningLength, (short) data.length);	1✔
155	runningLength += 2;	1✔
156	getUnsafe().copyMemory(data, BYTE_ARRAY_OFFSET, memoryBlock.getBaseObject(),	1✔
157	memoryBlock.getBaseOffset() + runningLength, data.length);	1✔
158	runningLength += data.length;	1✔
159	break;	1✔
160	case VARIABLE_INT:
161	byte[] data2 = row.getByteArray(index);	1✔
162	getUnsafe().putInt(memoryBlock.getBaseObject(),	1✔
163	memoryBlock.getBaseOffset() + runningLength, data2.length);	1✔
164	runningLength += 4;	1✔
165	getUnsafe().copyMemory(data2, BYTE_ARRAY_OFFSET, memoryBlock.getBaseObject(),	1✔
166	memoryBlock.getBaseOffset() + runningLength, data2.length);	1✔
167	runningLength += data2.length;	1✔
168	break;	1✔
169	case STRUCT:
170	CarbonRowSchema[] childSchemas =	1✔
171	((CarbonRowSchema.StructCarbonRowSchema) schema).getChildSchemas();	1✔
172	DataMapRow struct = row.getRow(index);	1✔
173	for (int i = 0; i < childSchemas.length; i++) {	1✔
174	addToUnsafe(childSchemas[i], struct, i);	1✔
175	}
176	break;	1✔
177	default:
178	throw new UnsupportedOperationException(	×
179	"unsupported data type for unsafe storage: " + schema.getDataType());	×
180	}
181	}	1✔
182
183	public DataMapRow getDataMapRow(CarbonRowSchema[] schema, int index) {
184	assert (index < rowCount);	1✔
185	return new UnsafeDataMapRow(schema, memoryBlock, pointers[index]);	1✔
186	}
187
188	public void finishWriting() throws MemoryException {
189	if (runningLength < allocatedSize) {	1✔
190	MemoryBlock allocate =	1✔
191	UnsafeMemoryManager.allocateMemoryWithRetry(taskId, runningLength, true);	1✔
192	getUnsafe().copyMemory(memoryBlock.getBaseObject(), memoryBlock.getBaseOffset(),	1✔
193	allocate.getBaseObject(), allocate.getBaseOffset(), runningLength);	1✔
194	UnsafeMemoryManager.INSTANCE.freeMemory(taskId, memoryBlock, true);	1✔
195	memoryBlock = allocate;	1✔
196	}
197	// Compact pointers.
198	if (rowCount < pointers.length) {	1✔
199	int[] newPointer = new int[rowCount];	1✔
200	System.arraycopy(pointers, 0, newPointer, 0, rowCount);	1✔
201	this.pointers = newPointer;	1✔
202	}
203	}	1✔
204
205	public void freeMemory() {
206	if (!isMemoryFreed) {	1✔
207	UnsafeMemoryManager.INSTANCE.freeMemory(taskId, memoryBlock, true);	1✔
208	isMemoryFreed = true;	1✔
209	}
210	}	1✔
211
212	public int getMemoryUsed() {
213	return runningLength;	1✔
214	}
215
216	public int getRowCount() {
217	return rowCount;	1✔
218	}
219
220	}

apache / carbondata / 685

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