flink-libraries/flink-table-common/src/main/java/org/apache/flink/table/dataformat/util/BinaryRowUtil.java - flink - Git at Google

 /*
  * 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.flink.table.dataformat.util;

 import org.apache.flink.api.common.typeutils.base.ComparatorUtil;
 import org.apache.flink.api.common.typeutils.base.DateComparator;
 import org.apache.flink.core.memory.MemorySegment;
 import org.apache.flink.core.memory.MemorySegmentFactory;
 import org.apache.flink.core.memory.MemoryUtils;
 import org.apache.flink.table.api.types.InternalType;
 import org.apache.flink.table.dataformat.BinaryRow;
 import org.apache.flink.table.dataformat.BinaryString;
 import org.apache.flink.table.dataformat.Decimal;
 import org.apache.flink.table.util.hash.Murmur32;

 import java.nio.ByteOrder;
 import java.sql.Date;
 import java.sql.Time;
 import java.sql.Timestamp;

 import static org.apache.flink.table.util.hash.Murmur32.DEFAULT_SEED;

 /**
  * Util for binary row. Many of the methods in this class are used in code generation.
  * So ignore IDE warnings.
  */
 public class BinaryRowUtil {

 	public static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;
 	public static final int BYTE_ARRAY_BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
 	public static final int BOOLEAN_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(boolean[].class);
 	public static final int SHORT_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(short[].class);
 	public static final int INT_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(int[].class);
 	public static final int LONG_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(long[].class);
 	public static final int FLOAT_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(float[].class);
 	public static final int DOUBLE_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(double[].class);
 	public static final boolean LITTLE_ENDIAN = ByteOrder.nativeOrder() == ByteOrder.LITTLE_ENDIAN;
 	private static final int LONG_BYTES = 8;

 	public static final BinaryRow EMPTY_ROW = new BinaryRow(0);

 	static {
 		int size = EMPTY_ROW.getFixedLengthPartSize();
 		byte[] bytes = new byte[size];
 		EMPTY_ROW.pointTo(MemorySegmentFactory.wrap(bytes), 0, size);
 	}

 	public static int compareBoolean(boolean a, boolean b) {
 		return Boolean.compare(a, b);
 	}

 	public static int compareByte(byte a, byte b) {
 		return Byte.compare(a, b);
 	}

 	public static int compareShort(short a, short b) {
 		return Short.compare(a, b);
 	}

 	public static int compareInt(int a, int b) {
 		return Integer.compare(a, b);
 	}

 	public static int compareLong(long a, long b) {
 		return Long.compare(a, b);
 	}

 	public static int compareFloat(float a, float b) {
 		return Float.compare(a, b);
 	}

 	public static int compareDouble(double a, double b) {
 		return Double.compare(a, b);
 	}

 	public static int compareChar(char a, char b) {
 		return Character.compare(a, b);
 	}

 	public static int compareBinaryString(BinaryString a, BinaryString b) {
 		return a.compareTo(b);
 	}

 	public static int compareDecimal(Decimal a, Decimal b) {
 		return a.compareTo(b);
 	}

 	public static int compareDate(Date a, Date b) {
 		return a.compareTo(b);
 	}

 	public static int compareTime(Time a, Time b) {
 		return a.compareTo(b);
 	}

 	public static int compareTimestamp(Timestamp a, Timestamp b) {
 		return a.compareTo(b);
 	}

 	public static int compareByteArray(byte[] a, byte[] b) {
 		return compareByteArray(a, 0, a.length, b, 0, b.length);
 	}

 	public static int compareByteArray(
 			byte[] buffer1, int offset1, int length1,
 			byte[] buffer2, int offset2, int length2) {
 		// Short circuit equal case
 		if (buffer1 == buffer2 &&
 				offset1 == offset2 &&
 				length1 == length2) {
 			return 0;
 		}
 		int minLength = Math.min(length1, length2);
 		int minWords = minLength / LONG_BYTES;
 		int offset1Adj = offset1 + BYTE_ARRAY_BASE_OFFSET;
 		int offset2Adj = offset2 + BYTE_ARRAY_BASE_OFFSET;

         /*
          * Compare 8 bytes at a time. Benchmarking shows comparing 8 bytes at a
          * time is no slower than comparing 4 bytes at a time even on 32-bit.
          * On the other hand, it is substantially faster on 64-bit.
          */
 		for (int i = 0; i < minWords * LONG_BYTES; i += LONG_BYTES) {
 			long lw = UNSAFE.getLong(buffer1, offset1Adj + (long) i);
 			long rw = UNSAFE.getLong(buffer2, offset2Adj + (long) i);
 			long diff = lw ^ rw;

 			if (diff != 0) {
 				if (!LITTLE_ENDIAN) {
 					return lessThanUnsigned(lw, rw) ? -1 : 1;
 				}

 				// Use binary search
 				int n = 0;
 				int y;
 				int x = (int) diff;
 				if (x == 0) {
 					x = (int) (diff >>> 32);
 					n = 32;
 				}

 				y = x << 16;
 				if (y == 0) {
 					n += 16;
 				} else {
 					x = y;
 				}

 				y = x << 8;
 				if (y == 0) {
 					n += 8;
 				}
 				return (int) (((lw >>> n) & 0xFFL) - ((rw >>> n) & 0xFFL));
 			}
 		}

 		// The epilogue to cover the last (minLength % 8) elements.
 		for (int i = minWords * LONG_BYTES; i < minLength; i++) {
 			int result = unsignedByteToInt(buffer1[offset1 + i]) -
 					unsignedByteToInt(buffer2[offset2 + i]);
 			if (result != 0) {
 				return result;
 			}
 		}
 		return length1 - length2;
 	}

 	private static int unsignedByteToInt(byte value) {
 		return value & 0xff;
 	}

 	private static boolean lessThanUnsigned(long x1, long x2) {
 		return (x1 + Long.MIN_VALUE) < (x2 + Long.MIN_VALUE);
 	}

 	public static int hashInt(int value) {
 		return Integer.hashCode(value);
 	}

 	public static int hashLong(long value) {
 		return Long.hashCode(value);
 	}

 	public static int hashShort(short value) {
 		return Short.hashCode(value);
 	}

 	public static int hashByte(byte value) {
 		return Byte.hashCode(value);
 	}

 	public static int hashFloat(float value) {
 		return Float.hashCode(value);
 	}

 	public static int hashDouble(double value) {
 		return Double.hashCode(value);
 	}

 	public static int hashBoolean(boolean value) {
 		return Boolean.hashCode(value);
 	}

 	public static int hashChar(char value) {
 		return Character.hashCode(value);
 	}

 	public static int hashObject(Object value) {
 		return value.hashCode();
 	}

 	public static int hashString(BinaryString value) {
 		return value.hashCode();
 	}

 	public static int hashDecimal(Decimal value) {
 		return value.hashCode();
 	}

 	public static int hashByteArray(byte[] value) {
 		return Murmur32.hashUnsafeBytes(value, BYTE_ARRAY_BASE_OFFSET, value.length, DEFAULT_SEED);
 	}

 	public static long hashByteArray64(byte[] value) {
 		return Murmur32.hashUnsafe64(value, BYTE_ARRAY_BASE_OFFSET, value.length, DEFAULT_SEED);
 	}

 	public static long hashDecimal64(Decimal value) {
 		// TODO real hash64.
 		return value.hashCode();
 	}

 	public static void minNormalizedKey(MemorySegment target, int offset, int numBytes) {
 		//write min value.
 		for (int i = 0; i < numBytes; i++) {
 			target.put(offset + i, (byte) 0);
 		}
 	}

 	public static void maxNormalizedKey(MemorySegment target, int offset, int numBytes) {
 		//write max value.
 		for (int i = 0; i < numBytes; i++) {
 			target.put(offset + i, (byte) -1);
 		}
 	}

 	public static void putShortNormalizedKey(short value, MemorySegment target, int offset,
 			int numBytes) {
 		ComparatorUtil.putShortNormalizedKey(value, target, offset, numBytes);
 	}

 	public static void putByteNormalizedKey(byte value, MemorySegment target, int offset,
 			int numBytes) {
 		ComparatorUtil.putByteNormalizedKey(value, target, offset, numBytes);
 	}

 	public static void putBooleanNormalizedKey(boolean value, MemorySegment target, int offset,
 			int numBytes) {
 		ComparatorUtil.putBooleanNormalizedKey(value, target, offset, numBytes);
 	}

 	public static void putBinaryStringNormalizedKey(
 			BinaryString value, MemorySegment target, int offset, int numBytes) {
 		final int limit = offset + numBytes;
 		final int end = value.numBytes();
 		for (int i = 0; i < end && offset < limit; i++) {
 			target.put(offset++, value.getByte(i));
 		}

 		for (int i = offset; i < limit; i++) {
 			target.put(i, (byte) 0);
 		}
 	}

 	public static void putDecimalNormalizedKey(
 			Decimal record, MemorySegment target, int offset, int len) {
 		assert record.getPrecision() <= Decimal.MAX_COMPACT_PRECISION;
 		putLongNormalizedKey(record.toUnscaledLong(), target, offset, len);
 	}

 	public static void putByteArrayNormalizedKey(
 			byte[] value, MemorySegment target, int offset, int numBytes) {
 		final int limit = offset + numBytes;
 		final int end = value.length;
 		for (int i = 0; i < end && offset < limit; i++) {
 			target.put(offset++, value[i]);
 		}

 		for (int i = offset; i < limit; i++) {
 			target.put(i, (byte) 0);
 		}
 	}

 	public static void putIntNormalizedKey(int value, MemorySegment target, int offset, int numBytes) {
 		ComparatorUtil.putIntNormalizedKey(value, target, offset, numBytes);
 	}

 	public static void putLongNormalizedKey(long value, MemorySegment target, int offset,
 			int numBytes) {
 		ComparatorUtil.putLongNormalizedKey(value, target, offset, numBytes);
 	}

 	/**
 	 * See http://stereopsis.com/radix.html/ for more details.
 	 */
 	public static void putFloatNormalizedKey(float value, MemorySegment target, int offset,
 			int numBytes) {
 		int iValue = Float.floatToIntBits(value);
 		iValue ^= ((iValue >> (Integer.SIZE - 1)) | Integer.MIN_VALUE);
 		ComparatorUtil.putUnsignedIntegerNormalizedKey(iValue, target, offset, numBytes);
 	}

 	public static void putDoubleNormalizedKey(double value, MemorySegment target, int offset,
 			int numBytes) {
 		long lValue = Double.doubleToLongBits(value);
 		lValue ^= ((lValue >> (Long.SIZE - 1)) | Long.MIN_VALUE);
 		ComparatorUtil.putUnsignedLongNormalizedKey(lValue, target, offset, numBytes);
 	}

 	public static void putCharNormalizedKey(char value, MemorySegment target, int offset,
 			int numBytes) {
 		ComparatorUtil.putCharNormalizedKey(value, target, offset, numBytes);
 	}

 	public static void putDateNormalizedKey(Date value, MemorySegment target, int offset,
 			int numBytes) {
 		DateComparator.putNormalizedKeyDate(value, target, offset, numBytes);
 	}

 	public static void putTimeNormalizedKey(Time value, MemorySegment target, int offset,
 			int numBytes) {
 		DateComparator.putNormalizedKeyDate(value, target, offset, numBytes);
 	}

 	public static void putTimestampNormalizedKey(Timestamp value, MemorySegment target, int offset,
 			int numBytes) {
 		// put Date key
 		DateComparator.putNormalizedKeyDate(value, target, offset, numBytes > 8 ? 8 : numBytes);
 		numBytes -= 8;
 		offset += 8;
 		if (numBytes <= 0) {
 			// nothing to do
 		}
 		// put nanos
 		else if (numBytes < 4) {
 			final int nanos = value.getNanos();
 			for (int i = 0; numBytes > 0; numBytes--, i++) {
 				target.put(offset + i, (byte) (nanos >>> ((3 - i) << 3)));
 			}
 		}
 		// put nanos with padding
 		else {
 			final int nanos = value.getNanos();
 			target.putIntBigEndian(offset, nanos);
 			for (int i = 4; i < numBytes; i++) {
 				target.put(offset + i, (byte) 0);
 			}
 		}
 	}

 	public static boolean byteArrayEquals(byte[] left, byte[] right, int length) {
 		return byteArrayEquals(
 				left, BYTE_ARRAY_BASE_OFFSET, right, BYTE_ARRAY_BASE_OFFSET, length);
 	}

 	public static boolean byteArrayEquals(
 			Object left, long leftOffset, Object right, long rightOffset, int length) {
 		int i = 0;

 		while (i <= length - 8) {
 			if (UNSAFE.getLong(left, leftOffset + i) !=
 					UNSAFE.getLong(right, rightOffset + i)) {
 				return false;
 			}
 			i += 8;
 		}

 		while (i < length) {
 			if (UNSAFE.getByte(left, leftOffset + i) !=
 					UNSAFE.getByte(right, rightOffset + i)) {
 				return false;
 			}
 			i += 1;
 		}
 		return true;
 	}

 	public static boolean equals(
 			MemorySegment[] segments1, int offset1,
 			MemorySegment[] segments2, int offset2, int len) {
 		if (allInFirstSeg(segments1, offset1, len) && allInFirstSeg(segments2, offset2, len)) {
 			return segments1[0].equalTo(segments2[0], offset1, offset2, len);
 		} else {
 			return equalsSlow(segments1, offset1, segments2, offset2, len);
 		}
 	}

 	public static boolean equalsSlow(
 			MemorySegment[] segments1, int offset1,
 			MemorySegment[] segments2, int offset2, int len) {
 		if (len == 0) {
 			// quick way and avoid segSize is zero.
 			return true;
 		}

 		int segSize1 = segments1[0].size();
 		int segSize2 = segments2[0].size();

 		// find first segIndex and segOffset of segments.
 		int segIndex1 = offset1 / segSize1;
 		int segIndex2 = offset2 / segSize2;
 		int segOffset1 = offset1 - segSize1 * segIndex1; // equal to %
 		int segOffset2 = offset2 - segSize2 * segIndex2; // equal to %

 		while (len > 0) {
 			int equalLen = Math.min(Math.min(len, segSize1 - segOffset1), segSize2 - segOffset2);
 			if (!segments1[segIndex1].equalTo(segments2[segIndex2], segOffset1, segOffset2, equalLen)) {
 				return false;
 			}
 			len -= equalLen;
 			segOffset1 += equalLen;
 			if (segOffset1 == segSize1) {
 				segOffset1 = 0;
 				segIndex1++;
 			}
 			segOffset2 += equalLen;
 			if (segOffset2 == segSize2) {
 				segOffset2 = 0;
 				segIndex2++;
 			}
 		}
 		return true;
 	}

 	public static byte[] copy(MemorySegment[] segments, int offset, int numBytes) {
 		return copy(segments, offset, new byte[numBytes], 0, numBytes);
 	}

 	public static byte[] copy(MemorySegment[] segments, int offset, byte[] bytes,
 			int bytesOffset, int numBytes) {
 		if (allInFirstSeg(segments, offset, numBytes)) {
 			segments[0].get(offset, bytes, bytesOffset, numBytes);
 		} else {
 			copySlow(segments, offset, bytes, bytesOffset, numBytes);
 		}
 		return bytes;
 	}

 	public static void copySlow(MemorySegment[] segments, int offset, byte[] bytes, int numBytes) {
 		copySlow(segments, offset, bytes, 0, numBytes);
 	}

 	public static void copySlow(MemorySegment[] segments, int offset, byte[] bytes,
 			int bytesOffset, int numBytes) {
 		int remainSize = numBytes;
 		for (MemorySegment segment : segments) {
 			int remain = segment.size() - offset;
 			if (remain > 0) {
 				int nCopy = Math.min(remain, remainSize);
 				segment.get(offset, bytes, numBytes - remainSize + bytesOffset, nCopy);
 				remainSize -= nCopy;
 				// next new segment.
 				offset = 0;
 				if (remainSize == 0) {
 					return;
 				}
 			} else {
 				// remain is negative, let's advance to next segment
 				// now the offset = offset - segmentSize (-remain)
 				offset = -remain;
 			}
 		}
 	}

 	public static void copyToUnsafe(
 			MemorySegment[] segments, int offset,
 			Object target, int pointer, int numBytes) {
 		if (segments.length == 1) {
 			segments[0].copyToUnsafe(offset, target, pointer, numBytes);
 		} else {
 			copyToUnsafeSlow(segments, offset, target, pointer, numBytes);
 		}
 	}

 	private static void copyToUnsafeSlow(
 			MemorySegment[] segments, int offset,
 			Object target, int pointer, int numBytes) {
 		int remainSize = numBytes;
 		for (MemorySegment segment : segments) {
 			int remain = segment.size() - offset;
 			if (remain > 0) {
 				int nCopy = Math.min(remain, remainSize);
 				segment.copyToUnsafe(offset, target, numBytes - remainSize + pointer, nCopy);
 				remainSize -= nCopy;
 				// next new segment.
 				offset = 0;
 				if (remainSize == 0) {
 					return;
 				}
 			} else {
 				// remain is negative, let's advance to next segment
 				// now the offset = offset - segmentSize (-remain)
 				offset = -remain;
 			}
 		}
 	}

 	public static void copyFromBytes(
 			MemorySegment[] segments, int offset,
 			byte[] target, int pointer, int numBytes) {
 		copyFromUnsafe(segments, offset, target, BYTE_ARRAY_BASE_OFFSET + pointer, numBytes);
 	}

 	public static void copyFromUnsafe(
 			MemorySegment[] segments, int offset,
 			Object target, int pointer, int numBytes) {
 		if (segments.length == 1) {
 			segments[0].copyFromUnsafe(offset, target, pointer, numBytes);
 		} else {
 			copyFromUnsafeSlow(segments, offset, target, pointer, numBytes);
 		}
 	}

 	private static void copyFromUnsafeSlow(
 			MemorySegment[] segments, int offset,
 			Object target, int pointer, int numBytes) {
 		int remainSize = numBytes;
 		for (MemorySegment segment : segments) {
 			int remain = segment.size() - offset;
 			if (remain > 0) {
 				int nCopy = Math.min(remain, remainSize);
 				segment.copyFromUnsafe(offset, target, numBytes - remainSize + pointer, nCopy);
 				remainSize -= nCopy;
 				// next new segment.
 				offset = 0;
 				if (remainSize == 0) {
 					return;
 				}
 			} else {
 				// remain is negative, let's advance to next segment
 				// now the offset = offset - segmentSize (-remain)
 				offset = -remain;
 			}
 		}
 	}

 	public static int getVariableLength(InternalType[] types) {
 		int length = 0;
 		for (InternalType type : types) {
 			if (!BinaryRow.isFixedLength(type)) {
 				// find a better way of computing generic type field variable-length
 				// right now we use a small value assumption
 				length += 16;
 			}
 		}
 		return length;
 	}

 	private static boolean allInFirstSeg(MemorySegment[] segments, int offset, int numBytes) {
 		return numBytes + offset <= segments[0].size();
 	}

 	/**
 	 * Find equal segments2 in segments1.
 	 * @param segments1 segs to find.
 	 * @param segments2 sub segs.
 	 * @return Return the found offset, return -1 if not find.
 	 */
 	public static int find(
 			MemorySegment[] segments1, int offset1, int numBytes1,
 			MemorySegment[] segments2, int offset2, int numBytes2) {
 		if (numBytes2 == 0) { // quick way 1.
 			return offset1;
 		}
 		if (allInFirstSeg(segments1, offset1, numBytes1) &&
 				allInFirstSeg(segments2, offset2, numBytes2)) {
 			byte first = segments2[0].get(offset2);
 			int end = numBytes1 - numBytes2 + offset1;
 			for (int i = offset1; i <= end; i++) {
 				// quick way 2: equal first byte.
 				if (segments1[0].get(i) == first &&
 						segments1[0].equalTo(segments2[0], i, offset2, numBytes2)) {
 					return i;
 				}
 			}
 			return -1;
 		} else {
 			return findSlow(segments1, offset1, numBytes1, segments2, offset2, numBytes2);
 		}
 	}

 	private static int findSlow(
 			MemorySegment[] segments1, int offset1, int numBytes1,
 			MemorySegment[] segments2, int offset2, int numBytes2) {
 		int end = numBytes1 - numBytes2 + offset1;
 		for (int i = offset1; i <= end; i++) {
 			if (equalsSlow(segments1, i, segments2, offset2, numBytes2)) {
 				return i;
 			}
 		}
 		return -1;
 	}
 }
	/*
	* 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.flink.table.dataformat.util;

	import org.apache.flink.api.common.typeutils.base.ComparatorUtil;
	import org.apache.flink.api.common.typeutils.base.DateComparator;
	import org.apache.flink.core.memory.MemorySegment;
	import org.apache.flink.core.memory.MemorySegmentFactory;
	import org.apache.flink.core.memory.MemoryUtils;
	import org.apache.flink.table.api.types.InternalType;
	import org.apache.flink.table.dataformat.BinaryRow;
	import org.apache.flink.table.dataformat.BinaryString;
	import org.apache.flink.table.dataformat.Decimal;
	import org.apache.flink.table.util.hash.Murmur32;

	import java.nio.ByteOrder;
	import java.sql.Date;
	import java.sql.Time;
	import java.sql.Timestamp;

	import static org.apache.flink.table.util.hash.Murmur32.DEFAULT_SEED;

	/**
	* Util for binary row. Many of the methods in this class are used in code generation.
	* So ignore IDE warnings.
	*/
	public class BinaryRowUtil {

	public static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;
	public static final int BYTE_ARRAY_BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
	public static final int BOOLEAN_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(boolean[].class);
	public static final int SHORT_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(short[].class);
	public static final int INT_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(int[].class);
	public static final int LONG_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(long[].class);
	public static final int FLOAT_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(float[].class);
	public static final int DOUBLE_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(double[].class);
	public static final boolean LITTLE_ENDIAN = ByteOrder.nativeOrder() == ByteOrder.LITTLE_ENDIAN;
	private static final int LONG_BYTES = 8;

	public static final BinaryRow EMPTY_ROW = new BinaryRow(0);

	static {
	int size = EMPTY_ROW.getFixedLengthPartSize();
	byte[] bytes = new byte[size];
	EMPTY_ROW.pointTo(MemorySegmentFactory.wrap(bytes), 0, size);
	}

	public static int compareBoolean(boolean a, boolean b) {
	return Boolean.compare(a, b);
	}

	public static int compareByte(byte a, byte b) {
	return Byte.compare(a, b);
	}

	public static int compareShort(short a, short b) {
	return Short.compare(a, b);
	}

	public static int compareInt(int a, int b) {
	return Integer.compare(a, b);
	}

	public static int compareLong(long a, long b) {
	return Long.compare(a, b);
	}

	public static int compareFloat(float a, float b) {
	return Float.compare(a, b);
	}

	public static int compareDouble(double a, double b) {
	return Double.compare(a, b);
	}

	public static int compareChar(char a, char b) {
	return Character.compare(a, b);
	}

	public static int compareBinaryString(BinaryString a, BinaryString b) {
	return a.compareTo(b);
	}

	public static int compareDecimal(Decimal a, Decimal b) {
	return a.compareTo(b);
	}

	public static int compareDate(Date a, Date b) {
	return a.compareTo(b);
	}

	public static int compareTime(Time a, Time b) {
	return a.compareTo(b);
	}

	public static int compareTimestamp(Timestamp a, Timestamp b) {
	return a.compareTo(b);
	}

	public static int compareByteArray(byte[] a, byte[] b) {
	return compareByteArray(a, 0, a.length, b, 0, b.length);
	}

	public static int compareByteArray(
	byte[] buffer1, int offset1, int length1,
	byte[] buffer2, int offset2, int length2) {
	// Short circuit equal case
	if (buffer1 == buffer2 &&
	offset1 == offset2 &&
	length1 == length2) {
	return 0;
	}
	int minLength = Math.min(length1, length2);
	int minWords = minLength / LONG_BYTES;
	int offset1Adj = offset1 + BYTE_ARRAY_BASE_OFFSET;
	int offset2Adj = offset2 + BYTE_ARRAY_BASE_OFFSET;

	/*
	* Compare 8 bytes at a time. Benchmarking shows comparing 8 bytes at a
	* time is no slower than comparing 4 bytes at a time even on 32-bit.
	* On the other hand, it is substantially faster on 64-bit.
	*/
	for (int i = 0; i < minWords * LONG_BYTES; i += LONG_BYTES) {
	long lw = UNSAFE.getLong(buffer1, offset1Adj + (long) i);
	long rw = UNSAFE.getLong(buffer2, offset2Adj + (long) i);
	long diff = lw ^ rw;

	if (diff != 0) {
	if (!LITTLE_ENDIAN) {
	return lessThanUnsigned(lw, rw) ? -1 : 1;
	}

	// Use binary search
	int n = 0;
	int y;
	int x = (int) diff;
	if (x == 0) {
	x = (int) (diff >>> 32);
	n = 32;
	}

	y = x << 16;
	if (y == 0) {
	n += 16;
	} else {
	x = y;
	}

	y = x << 8;
	if (y == 0) {
	n += 8;
	}
	return (int) (((lw >>> n) & 0xFFL) - ((rw >>> n) & 0xFFL));
	}
	}

	// The epilogue to cover the last (minLength % 8) elements.
	for (int i = minWords * LONG_BYTES; i < minLength; i++) {
	int result = unsignedByteToInt(buffer1[offset1 + i]) -
	unsignedByteToInt(buffer2[offset2 + i]);
	if (result != 0) {
	return result;
	}
	}
	return length1 - length2;
	}

	private static int unsignedByteToInt(byte value) {
	return value & 0xff;
	}

	private static boolean lessThanUnsigned(long x1, long x2) {
	return (x1 + Long.MIN_VALUE) < (x2 + Long.MIN_VALUE);
	}

	public static int hashInt(int value) {
	return Integer.hashCode(value);
	}

	public static int hashLong(long value) {
	return Long.hashCode(value);
	}

	public static int hashShort(short value) {
	return Short.hashCode(value);
	}

	public static int hashByte(byte value) {
	return Byte.hashCode(value);
	}

	public static int hashFloat(float value) {
	return Float.hashCode(value);
	}

	public static int hashDouble(double value) {
	return Double.hashCode(value);
	}

	public static int hashBoolean(boolean value) {
	return Boolean.hashCode(value);
	}

	public static int hashChar(char value) {
	return Character.hashCode(value);
	}

	public static int hashObject(Object value) {
	return value.hashCode();
	}

	public static int hashString(BinaryString value) {
	return value.hashCode();
	}

	public static int hashDecimal(Decimal value) {
	return value.hashCode();
	}

	public static int hashByteArray(byte[] value) {
	return Murmur32.hashUnsafeBytes(value, BYTE_ARRAY_BASE_OFFSET, value.length, DEFAULT_SEED);
	}

	public static long hashByteArray64(byte[] value) {
	return Murmur32.hashUnsafe64(value, BYTE_ARRAY_BASE_OFFSET, value.length, DEFAULT_SEED);
	}

	public static long hashDecimal64(Decimal value) {
	// TODO real hash64.
	return value.hashCode();
	}

	public static void minNormalizedKey(MemorySegment target, int offset, int numBytes) {
	//write min value.
	for (int i = 0; i < numBytes; i++) {
	target.put(offset + i, (byte) 0);
	}
	}

	public static void maxNormalizedKey(MemorySegment target, int offset, int numBytes) {
	//write max value.
	for (int i = 0; i < numBytes; i++) {
	target.put(offset + i, (byte) -1);
	}
	}

	public static void putShortNormalizedKey(short value, MemorySegment target, int offset,
	int numBytes) {
	ComparatorUtil.putShortNormalizedKey(value, target, offset, numBytes);
	}

	public static void putByteNormalizedKey(byte value, MemorySegment target, int offset,
	int numBytes) {
	ComparatorUtil.putByteNormalizedKey(value, target, offset, numBytes);
	}

	public static void putBooleanNormalizedKey(boolean value, MemorySegment target, int offset,
	int numBytes) {
	ComparatorUtil.putBooleanNormalizedKey(value, target, offset, numBytes);
	}

	public static void putBinaryStringNormalizedKey(
	BinaryString value, MemorySegment target, int offset, int numBytes) {
	final int limit = offset + numBytes;
	final int end = value.numBytes();
	for (int i = 0; i < end && offset < limit; i++) {
	target.put(offset++, value.getByte(i));
	}

	for (int i = offset; i < limit; i++) {
	target.put(i, (byte) 0);
	}
	}

	public static void putDecimalNormalizedKey(
	Decimal record, MemorySegment target, int offset, int len) {
	assert record.getPrecision() <= Decimal.MAX_COMPACT_PRECISION;
	putLongNormalizedKey(record.toUnscaledLong(), target, offset, len);
	}

	public static void putByteArrayNormalizedKey(
	byte[] value, MemorySegment target, int offset, int numBytes) {
	final int limit = offset + numBytes;
	final int end = value.length;
	for (int i = 0; i < end && offset < limit; i++) {
	target.put(offset++, value[i]);
	}

	for (int i = offset; i < limit; i++) {
	target.put(i, (byte) 0);
	}
	}

	public static void putIntNormalizedKey(int value, MemorySegment target, int offset, int numBytes) {
	ComparatorUtil.putIntNormalizedKey(value, target, offset, numBytes);
	}

	public static void putLongNormalizedKey(long value, MemorySegment target, int offset,
	int numBytes) {
	ComparatorUtil.putLongNormalizedKey(value, target, offset, numBytes);
	}

	/**
	* See http://stereopsis.com/radix.html/ for more details.
	*/
	public static void putFloatNormalizedKey(float value, MemorySegment target, int offset,
	int numBytes) {
	int iValue = Float.floatToIntBits(value);
	iValue ^= ((iValue >> (Integer.SIZE - 1)) \| Integer.MIN_VALUE);
	ComparatorUtil.putUnsignedIntegerNormalizedKey(iValue, target, offset, numBytes);
	}

	public static void putDoubleNormalizedKey(double value, MemorySegment target, int offset,
	int numBytes) {
	long lValue = Double.doubleToLongBits(value);
	lValue ^= ((lValue >> (Long.SIZE - 1)) \| Long.MIN_VALUE);
	ComparatorUtil.putUnsignedLongNormalizedKey(lValue, target, offset, numBytes);
	}

	public static void putCharNormalizedKey(char value, MemorySegment target, int offset,
	int numBytes) {
	ComparatorUtil.putCharNormalizedKey(value, target, offset, numBytes);
	}

	public static void putDateNormalizedKey(Date value, MemorySegment target, int offset,
	int numBytes) {
	DateComparator.putNormalizedKeyDate(value, target, offset, numBytes);
	}

	public static void putTimeNormalizedKey(Time value, MemorySegment target, int offset,
	int numBytes) {
	DateComparator.putNormalizedKeyDate(value, target, offset, numBytes);
	}

	public static void putTimestampNormalizedKey(Timestamp value, MemorySegment target, int offset,
	int numBytes) {
	// put Date key
	DateComparator.putNormalizedKeyDate(value, target, offset, numBytes > 8 ? 8 : numBytes);
	numBytes -= 8;
	offset += 8;
	if (numBytes <= 0) {
	// nothing to do
	}
	// put nanos
	else if (numBytes < 4) {
	final int nanos = value.getNanos();
	for (int i = 0; numBytes > 0; numBytes--, i++) {
	target.put(offset + i, (byte) (nanos >>> ((3 - i) << 3)));
	}
	}
	// put nanos with padding
	else {
	final int nanos = value.getNanos();
	target.putIntBigEndian(offset, nanos);
	for (int i = 4; i < numBytes; i++) {
	target.put(offset + i, (byte) 0);
	}
	}
	}

	public static boolean byteArrayEquals(byte[] left, byte[] right, int length) {
	return byteArrayEquals(
	left, BYTE_ARRAY_BASE_OFFSET, right, BYTE_ARRAY_BASE_OFFSET, length);
	}

	public static boolean byteArrayEquals(
	Object left, long leftOffset, Object right, long rightOffset, int length) {
	int i = 0;

	while (i <= length - 8) {
	if (UNSAFE.getLong(left, leftOffset + i) !=
	UNSAFE.getLong(right, rightOffset + i)) {
	return false;
	}
	i += 8;
	}

	while (i < length) {
	if (UNSAFE.getByte(left, leftOffset + i) !=
	UNSAFE.getByte(right, rightOffset + i)) {
	return false;
	}
	i += 1;
	}
	return true;
	}

	public static boolean equals(
	MemorySegment[] segments1, int offset1,
	MemorySegment[] segments2, int offset2, int len) {
	if (allInFirstSeg(segments1, offset1, len) && allInFirstSeg(segments2, offset2, len)) {
	return segments1[0].equalTo(segments2[0], offset1, offset2, len);
	} else {
	return equalsSlow(segments1, offset1, segments2, offset2, len);
	}
	}

	public static boolean equalsSlow(
	MemorySegment[] segments1, int offset1,
	MemorySegment[] segments2, int offset2, int len) {
	if (len == 0) {
	// quick way and avoid segSize is zero.
	return true;
	}

	int segSize1 = segments1[0].size();
	int segSize2 = segments2[0].size();

	// find first segIndex and segOffset of segments.
	int segIndex1 = offset1 / segSize1;
	int segIndex2 = offset2 / segSize2;
	int segOffset1 = offset1 - segSize1 * segIndex1; // equal to %
	int segOffset2 = offset2 - segSize2 * segIndex2; // equal to %

	while (len > 0) {
	int equalLen = Math.min(Math.min(len, segSize1 - segOffset1), segSize2 - segOffset2);
	if (!segments1[segIndex1].equalTo(segments2[segIndex2], segOffset1, segOffset2, equalLen)) {
	return false;
	}
	len -= equalLen;
	segOffset1 += equalLen;
	if (segOffset1 == segSize1) {
	segOffset1 = 0;
	segIndex1++;
	}
	segOffset2 += equalLen;
	if (segOffset2 == segSize2) {
	segOffset2 = 0;
	segIndex2++;
	}
	}
	return true;
	}

	public static byte[] copy(MemorySegment[] segments, int offset, int numBytes) {
	return copy(segments, offset, new byte[numBytes], 0, numBytes);
	}

	public static byte[] copy(MemorySegment[] segments, int offset, byte[] bytes,
	int bytesOffset, int numBytes) {
	if (allInFirstSeg(segments, offset, numBytes)) {
	segments[0].get(offset, bytes, bytesOffset, numBytes);
	} else {
	copySlow(segments, offset, bytes, bytesOffset, numBytes);
	}
	return bytes;
	}

	public static void copySlow(MemorySegment[] segments, int offset, byte[] bytes, int numBytes) {
	copySlow(segments, offset, bytes, 0, numBytes);
	}

	public static void copySlow(MemorySegment[] segments, int offset, byte[] bytes,
	int bytesOffset, int numBytes) {
	int remainSize = numBytes;
	for (MemorySegment segment : segments) {
	int remain = segment.size() - offset;
	if (remain > 0) {
	int nCopy = Math.min(remain, remainSize);
	segment.get(offset, bytes, numBytes - remainSize + bytesOffset, nCopy);
	remainSize -= nCopy;
	// next new segment.
	offset = 0;
	if (remainSize == 0) {
	return;
	}
	} else {
	// remain is negative, let's advance to next segment
	// now the offset = offset - segmentSize (-remain)
	offset = -remain;
	}
	}
	}

	public static void copyToUnsafe(
	MemorySegment[] segments, int offset,
	Object target, int pointer, int numBytes) {
	if (segments.length == 1) {
	segments[0].copyToUnsafe(offset, target, pointer, numBytes);
	} else {
	copyToUnsafeSlow(segments, offset, target, pointer, numBytes);
	}
	}

	private static void copyToUnsafeSlow(
	MemorySegment[] segments, int offset,
	Object target, int pointer, int numBytes) {
	int remainSize = numBytes;
	for (MemorySegment segment : segments) {
	int remain = segment.size() - offset;
	if (remain > 0) {
	int nCopy = Math.min(remain, remainSize);
	segment.copyToUnsafe(offset, target, numBytes - remainSize + pointer, nCopy);
	remainSize -= nCopy;
	// next new segment.
	offset = 0;
	if (remainSize == 0) {
	return;
	}
	} else {
	// remain is negative, let's advance to next segment
	// now the offset = offset - segmentSize (-remain)
	offset = -remain;
	}
	}
	}

	public static void copyFromBytes(
	MemorySegment[] segments, int offset,
	byte[] target, int pointer, int numBytes) {
	copyFromUnsafe(segments, offset, target, BYTE_ARRAY_BASE_OFFSET + pointer, numBytes);
	}

	public static void copyFromUnsafe(
	MemorySegment[] segments, int offset,
	Object target, int pointer, int numBytes) {
	if (segments.length == 1) {
	segments[0].copyFromUnsafe(offset, target, pointer, numBytes);
	} else {
	copyFromUnsafeSlow(segments, offset, target, pointer, numBytes);
	}
	}

	private static void copyFromUnsafeSlow(
	MemorySegment[] segments, int offset,
	Object target, int pointer, int numBytes) {
	int remainSize = numBytes;
	for (MemorySegment segment : segments) {
	int remain = segment.size() - offset;
	if (remain > 0) {
	int nCopy = Math.min(remain, remainSize);
	segment.copyFromUnsafe(offset, target, numBytes - remainSize + pointer, nCopy);
	remainSize -= nCopy;
	// next new segment.
	offset = 0;
	if (remainSize == 0) {
	return;
	}
	} else {
	// remain is negative, let's advance to next segment
	// now the offset = offset - segmentSize (-remain)
	offset = -remain;
	}
	}
	}

	public static int getVariableLength(InternalType[] types) {
	int length = 0;
	for (InternalType type : types) {
	if (!BinaryRow.isFixedLength(type)) {
	// find a better way of computing generic type field variable-length
	// right now we use a small value assumption
	length += 16;
	}
	}
	return length;
	}

	private static boolean allInFirstSeg(MemorySegment[] segments, int offset, int numBytes) {
	return numBytes + offset <= segments[0].size();
	}

	/**
	* Find equal segments2 in segments1.
	* @param segments1 segs to find.
	* @param segments2 sub segs.
	* @return Return the found offset, return -1 if not find.
	*/
	public static int find(
	MemorySegment[] segments1, int offset1, int numBytes1,
	MemorySegment[] segments2, int offset2, int numBytes2) {
	if (numBytes2 == 0) { // quick way 1.
	return offset1;
	}
	if (allInFirstSeg(segments1, offset1, numBytes1) &&
	allInFirstSeg(segments2, offset2, numBytes2)) {
	byte first = segments2[0].get(offset2);
	int end = numBytes1 - numBytes2 + offset1;
	for (int i = offset1; i <= end; i++) {
	// quick way 2: equal first byte.
	if (segments1[0].get(i) == first &&
	segments1[0].equalTo(segments2[0], i, offset2, numBytes2)) {
	return i;
	}
	}
	return -1;
	} else {
	return findSlow(segments1, offset1, numBytes1, segments2, offset2, numBytes2);
	}
	}

	private static int findSlow(
	MemorySegment[] segments1, int offset1, int numBytes1,
	MemorySegment[] segments2, int offset2, int numBytes2) {
	int end = numBytes1 - numBytes2 + offset1;
	for (int i = offset1; i <= end; i++) {
	if (equalsSlow(segments1, i, segments2, offset2, numBytes2)) {
	return i;
	}
	}
	return -1;
	}
	}