Github user JamesRTaylor commented on a diff in the pull request:
https://github.com/apache/incubator-phoenix/pull/8#discussion_r10156795
--- Diff:
phoenix-core/src/main/java/org/apache/phoenix/schema/PArrayDataType.java ---
@@ -214,150 +281,205 @@ public Object toObject(Object object, PDataType
actualType, SortOrder sortOrder)
return toObject(object, actualType);
}
- // Making this private
/**
- * The format of the byte buffer looks like this for variable length
array elements
- * <noofelements><Offset of the index array><elements><offset array>
- * where <noOfelements> - vint
- * <offset of the index array> - int
- * <elements> - these are the values
- * <offset array> - offset of every element written as INT/SHORT
- *
+ * ser
+ * @param byteStream
+ * @param oStream
* @param array
- * @param buffer
* @param noOfElements
- * @param byteSize
- * @param capacity
+ * @param baseType
+ * @param capacity
* @return
*/
- private byte[] bytesFromByteBuffer(PhoenixArray array, ByteBuffer
buffer,
- int noOfElements, PDataType baseType, int capacity) {
- int temp = noOfElements;
- if (buffer == null) return null;
- buffer.put(ARRAY_SERIALIZATION_VERSION);
- buffer.putInt(noOfElements);
- if (!baseType.isFixedWidth() ||
baseType.isCoercibleTo(PDataType.VARCHAR)) {
- int fillerForOffsetByteArray = buffer.position();
- buffer.position(fillerForOffsetByteArray + Bytes.SIZEOF_INT);
- ByteBuffer offsetArray = ByteBuffer.allocate(capacity);
- if(noOfElements < 0){
- noOfElements = -noOfElements;
+ private byte[] createArrayBytes(TrustedByteArrayOutputStream
byteStream, DataOutputStream oStream,
+ PhoenixArray array, int noOfElements, PDataType baseType) {
+ try {
+ if (!baseType.isFixedWidth()) {
+ int[] offsetPos = new int[noOfElements];
+ int nulls = 0;
+ for (int i = 0; i < noOfElements; i++) {
+ byte[] bytes = array.toBytes(i);
+ if (bytes.length == 0) {
+ offsetPos[i] = byteStream.size();
+ nulls++;
+ } else {
+ nulls = serializeNulls(oStream, nulls);
+ offsetPos[i] = byteStream.size();
+ oStream.write(bytes, 0, bytes.length);
+ oStream.write(QueryConstants.SEPARATOR_BYTE);
+ }
+ }
+ // Double seperator byte to show end of the non null array
+ PArrayDataType.writeEndSeperatorForVarLengthArray(oStream);
+ noOfElements =
PArrayDataType.serailizeOffsetArrayIntoStream(oStream, byteStream, noOfElements,
+ offsetPos[offsetPos.length - 1], offsetPos);
+ } else {
+ for (int i = 0; i < noOfElements; i++) {
+ byte[] bytes = array.toBytes(i);
+ int length = bytes.length;
+ if(baseType == PDataType.CHAR) {
+ byte[] bytesWithPad = new
byte[array.getMaxLength()];
+ Arrays.fill(bytesWithPad, StringUtil.SPACE_UTF8);
+ System.arraycopy(bytes, 0, bytesWithPad, 0,
length);
+ oStream.write(bytesWithPad, 0,
bytesWithPad.length);
+ } else {
+ oStream.write(bytes, 0, length);
+ }
+ }
}
- for (int i = 0; i < noOfElements; i++) {
- // Not fixed width
- if (temp < 0) {
- offsetArray.putInt(buffer.position());
- } else {
-
offsetArray.putShort((short)(buffer.position() - Short.MAX_VALUE));
- }
- byte[] bytes = array.toBytes(i);
- buffer.put(bytes);
+ serializeHeaderInfoIntoStream(oStream, noOfElements);
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ ptr.set(byteStream.getBuffer(), 0, byteStream.size());
+ return ByteUtil.copyKeyBytesIfNecessary(ptr);
+ } catch (IOException e) {
+ try {
+ byteStream.close();
+ oStream.close();
+ } catch (IOException ioe) {
+
}
- int offsetArrayPosition = buffer.position();
- buffer.put(offsetArray.array());
- buffer.position(fillerForOffsetByteArray);
- buffer.putInt(offsetArrayPosition);
+ }
+ // This should not happen
+ return null;
+ }
+
+ public static int serailizeOffsetArrayIntoStream(DataOutputStream
oStream, TrustedByteArrayOutputStream byteStream,
+ int noOfElements, int maxOffset, int[] offsetPos) throws
IOException {
+ int offsetPosition = (byteStream.size());
+ byte[] offsetArr = null;
+ boolean useInt = true;
+ if (PArrayDataType.useShortForOffsetArray(maxOffset)) {
+ offsetArr = new
byte[PArrayDataType.initOffsetArray(noOfElements, Bytes.SIZEOF_SHORT)];
+ useInt = false;
} else {
- for (int i = 0; i < noOfElements; i++) {
- byte[] bytes = array.toBytes(i);
- buffer.put(bytes);
+ offsetArr = new
byte[PArrayDataType.initOffsetArray(noOfElements, Bytes.SIZEOF_INT)];
+ noOfElements = -noOfElements;
+ }
+ int off = 0;
+ if(useInt) {
+ for (int pos : offsetPos) {
+ Bytes.putInt(offsetArr, off, pos);
+ off += Bytes.SIZEOF_INT;
+ }
+ } else {
+ for (int pos : offsetPos) {
+ Bytes.putShort(offsetArr, off, (short)(pos -
Short.MAX_VALUE));
+ off += Bytes.SIZEOF_SHORT;
}
}
- return buffer.array();
- }
+ oStream.write(offsetArr);
+ oStream.writeInt(offsetPosition);
+ return noOfElements;
+ }
- private static int initOffsetArray(int noOfElements, int baseSize) {
+ public static void serializeHeaderInfoIntoBuffer(ByteBuffer buffer,
int noOfElements) {
+ // No of elements
+ buffer.putInt(noOfElements);
+ // Version of the array
+ buffer.put(ARRAY_SERIALIZATION_VERSION);
+ }
+
+ public static void serializeHeaderInfoIntoStream(DataOutputStream
oStream, int noOfElements) throws IOException {
+ // No of elements
+ oStream.writeInt(noOfElements);
+ // Version of the array
+ oStream.write(ARRAY_SERIALIZATION_VERSION);
+ }
+
+ public static int initOffsetArray(int noOfElements, int baseSize) {
// for now create an offset array equal to the noofelements
return noOfElements * baseSize;
}
- private Object createPhoenixArray(byte[] bytes, int offset, int length,
- SortOrder sortOrder, PDataType baseDataType) {
- if(bytes == null || bytes.length == 0) {
- return null;
- }
- ByteBuffer buffer = ByteBuffer.wrap(bytes, offset, length);
- int initPos = buffer.position();
- buffer.get();
- int noOfElements = buffer.getInt();
- boolean useShort = true;
- int baseSize = Bytes.SIZEOF_SHORT;
- if(noOfElements < 0) {
- noOfElements = -noOfElements;
- baseSize = Bytes.SIZEOF_INT;
- useShort = false;
- }
- Object[] elements = (Object[])
java.lang.reflect.Array.newInstance(
- baseDataType.getJavaClass(), noOfElements);
- if (!baseDataType.isFixedWidth() ||
baseDataType.isCoercibleTo(PDataType.VARCHAR)) {
- int indexOffset = buffer.getInt();
- int valArrayPostion = buffer.position();
- buffer.position(indexOffset + initPos);
- ByteBuffer indexArr = ByteBuffer
- .allocate(initOffsetArray(noOfElements,
baseSize));
- byte[] array = indexArr.array();
- buffer.get(array);
- int countOfElementsRead = 0;
- int i = 0;
- int currOff = -1;
- int nextOff = -1;
- if (noOfElements > 1) {
- while (indexArr.hasRemaining()) {
- if (countOfElementsRead < noOfElements)
{
- if (currOff == -1) {
- if
((indexArr.position() + 2 * baseSize) <= indexArr
-
.capacity()) {
- if (useShort) {
- currOff
= indexArr.getShort() + Short.MAX_VALUE;
- nextOff
= indexArr.getShort() + Short.MAX_VALUE;
- } else {
- currOff
= indexArr.getInt();
- nextOff
= indexArr.getInt();
- }
-
countOfElementsRead += 2;
- }
- } else {
- currOff = nextOff;
- if(useShort) {
- nextOff =
indexArr.getShort() + Short.MAX_VALUE;
- } else {
- nextOff =
indexArr.getInt();
- }
- countOfElementsRead +=
1;
- }
- int elementLength = nextOff -
currOff;
- buffer.position(currOff +
initPos);
- byte[] val = new
byte[elementLength];
- buffer.get(val);
- elements[i++] =
baseDataType.toObject(val,
- sortOrder);
- }
- }
- buffer.position(nextOff + initPos);
- byte[] val = new byte[indexOffset - nextOff];
- buffer.get(val);
- elements[i++] = baseDataType.toObject(val,
sortOrder);
- } else {
- byte[] val = new byte[indexOffset -
valArrayPostion];
- buffer.position(valArrayPostion + initPos);
- buffer.get(val);
- elements[i++] = baseDataType.toObject(val,
sortOrder);
- }
- } else {
- for (int i = 0; i < noOfElements; i++) {
- byte[] val;
- if (baseDataType.getByteSize() == null) {
- val = new byte[length];
- } else {
- val = new
byte[baseDataType.getByteSize()];
- }
- buffer.get(val);
- elements[i] = baseDataType.toObject(val,
sortOrder);
- }
- }
- return PArrayDataType
- .instantiatePhoenixArray(baseDataType,
elements);
- }
+ // Any variable length array would follow the below order
+ // Every element would be seperated by a seperator byte '0'
+ // Null elements are counted and once a first non null element appears
we
+ // write the count of the nulls prefixed with a seperator byte
+ // Trailing nulls are not taken into account
+ // The last non null element is followed by two seperator bytes
+ // For eg
+ // a, b, null, null, c, null would be
+ // 65 0 66 0 0 2 67 0 0 0
+ // a null null null b c null d would be
+ // 65 0 0 3 66 0 67 0 0 1 68 0 0 0
+ // Follow the above example to understand how this works
+ private Object createPhoenixArray(byte[] bytes, int offset, int
length, SortOrder sortOrder, PDataType baseDataType) {
--- End diff --
Pass maxLength in here and pass through to
PArrayDataType.instantiatePhoenixArray() instead of having a setMaxLength() on
our array impl.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
