[DO-NOT-MERGE][SPARK-27127][SQL] Deduplicate codes reading from/writing to unsafe object

common/unsafe/src/main/java/org/apache/spark/unsafe/UnsafeHelper.java

@@ -0,0 +1,54 @@
+/*
+ * 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.spark.unsafe;
+
+import java.nio.ByteBuffer;
+
+public final class UnsafeHelper {
+  private UnsafeHelper() {}
+
+  public static void writeTo(
+      ByteBuffer buffer,
+      Object baseObject,
+      long baseOffset,
+      int sizeInBytes) {
+    assert(buffer.hasArray());
+    byte[] target = buffer.array();
+    int offset = buffer.arrayOffset();
+    int pos = buffer.position();
+    writeToMemory(baseObject, baseOffset, target,
+        Platform.BYTE_ARRAY_OFFSET + offset + pos, sizeInBytes);
+    buffer.position(pos + sizeInBytes);
+  }
+
+  public static void writeToMemory(
+      Object baseObject,
+      long baseOffset,
+      Object target,
+      long targetOffset,
+      int sizeInBytes) {
+    Platform.copyMemory(baseObject, baseOffset, target, targetOffset, sizeInBytes);
+  }
+
+  public static byte[] copyToMemory(Object baseObject, long baseOffset, int sizeInBytes) {
+    final byte[] arrayDataCopy = new byte[sizeInBytes];
+    Platform.copyMemory(baseObject, baseOffset, arrayDataCopy, Platform.BYTE_ARRAY_OFFSET,
+        sizeInBytes);
+    return arrayDataCopy;
+  }
+}

common/unsafe/src/main/java/org/apache/spark/unsafe/types/UTF8String.java

@@ -32,6 +32,7 @@
 import com.google.common.primitives.Ints;
 
 import org.apache.spark.unsafe.Platform;
+import org.apache.spark.unsafe.UnsafeHelper;
 import org.apache.spark.unsafe.array.ByteArrayMethods;
 import org.apache.spark.unsafe.hash.Murmur3_x86_32;
 
@@ -168,16 +169,11 @@ public UTF8String() {
    * bytes in this string.
    */
   public void writeToMemory(Object target, long targetOffset) {
-    Platform.copyMemory(base, offset, target, targetOffset, numBytes);
+    UnsafeHelper.writeToMemory(base, offset, target, targetOffset, numBytes);
   }
 
   public void writeTo(ByteBuffer buffer) {
-    assert(buffer.hasArray());
-    byte[] target = buffer.array();
-    int offset = buffer.arrayOffset();
-    int pos = buffer.position();
-    writeToMemory(target, Platform.BYTE_ARRAY_OFFSET + offset + pos);
-    buffer.position(pos + numBytes);
+    UnsafeHelper.writeTo(buffer, base, offset, numBytes);
   }
 
   /**

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/SqlTypesUnsafeHelper.java

@@ -0,0 +1,104 @@
+/*
+ * 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.spark.sql.catalyst.expressions;
+
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.Platform;
+import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+
+public final class SqlTypesUnsafeHelper {
+  private SqlTypesUnsafeHelper() {}
+
+  public static byte[] getBinary(long offsetAndSize, Object baseObject, long baseOffset) {
+    final int offset = getOffsetFromOffsetAndSize(offsetAndSize);
+    final int size = getSizeFromOffsetAndSize(offsetAndSize);
+    final byte[] bytes = new byte[size];
+    Platform.copyMemory(baseObject, baseOffset + offset, bytes, Platform.BYTE_ARRAY_OFFSET, size);
+    return bytes;
+  }
+
+  public static UTF8String getUTF8String(long offsetAndSize, Object baseObject, long baseOffset) {
+    final int offset = getOffsetFromOffsetAndSize(offsetAndSize);
+    final int size = getSizeFromOffsetAndSize(offsetAndSize);
+    return UTF8String.fromAddress(baseObject, baseOffset + offset, size);
+  }
+
+  public static CalendarInterval getInterval(
+      long offsetAndSize,
+      Object baseObject,
+      long baseOffset) {
+    final int offset = getOffsetFromOffsetAndSize(offsetAndSize);
+    final int months = (int) Platform.getLong(baseObject, baseOffset + offset);
+    final long microseconds = Platform.getLong(baseObject, baseOffset + offset + 8);
+    return new CalendarInterval(months, microseconds);
+  }
+
+  public static Decimal getDecimalExceedingLong(
+      byte[] bytes,
+      int precision,
+      int scale,
+      boolean wrapWithScalaBigDecimal) {
+    final BigInteger bigInteger = new BigInteger(bytes);
+    BigDecimal decimal = new BigDecimal(bigInteger, scale);
+    if (wrapWithScalaBigDecimal) {
+      return Decimal.apply(new scala.math.BigDecimal(decimal), precision, scale);
+    } else {
+      return Decimal.apply(decimal, precision, scale);
+    }
+  }
+
+  public static UnsafeRow getStruct(
+      long offsetAndSize,
+      Object baseObject,
+      long baseOffset,
+      int numFields) {
+    final int offset = getOffsetFromOffsetAndSize(offsetAndSize);
+    final int size = getSizeFromOffsetAndSize(offsetAndSize);
+    final UnsafeRow row = new UnsafeRow(numFields);
+    row.pointTo(baseObject, baseOffset + offset, size);
+    return row;
+  }
+
+  public static UnsafeArrayData getArray(long offsetAndSize, Object baseObject, long baseOffset) {
+    final int offset = getOffsetFromOffsetAndSize(offsetAndSize);
+    final int size = getSizeFromOffsetAndSize(offsetAndSize);
+    final UnsafeArrayData array = new UnsafeArrayData();
+    array.pointTo(baseObject, baseOffset + offset, size);
+    return array;
+  }
+
+  public static UnsafeMapData getMap(long offsetAndSize, Object baseObject, long baseOffset) {
+    final int offset = getOffsetFromOffsetAndSize(offsetAndSize);
+    final int size = getSizeFromOffsetAndSize(offsetAndSize);
+    final UnsafeMapData map = new UnsafeMapData();
+    map.pointTo(baseObject, baseOffset + offset, size);
+    return map;
+  }
+
+  public static int getOffsetFromOffsetAndSize(long offsetAndSize) {
+    return (int) (offsetAndSize >> 32);
+  }
+
+  public static int getSizeFromOffsetAndSize(long offsetAndSize) {
+    return (int) offsetAndSize;
+  }
+}

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeArrayData.java

@@ -17,13 +17,12 @@
 
 package org.apache.spark.sql.catalyst.expressions;
 
-import java.math.BigDecimal;
-import java.math.BigInteger;
 import java.nio.ByteBuffer;
 
 import org.apache.spark.sql.catalyst.util.ArrayData;
 import org.apache.spark.sql.types.*;
 import org.apache.spark.unsafe.Platform;
+import org.apache.spark.unsafe.UnsafeHelper;
 import org.apache.spark.unsafe.array.ByteArrayMethods;
 import org.apache.spark.unsafe.bitset.BitSetMethods;
 import org.apache.spark.unsafe.hash.Murmur3_x86_32;
@@ -227,74 +226,45 @@ public Decimal getDecimal(int ordinal, int precision, int scale) {
     if (precision <= Decimal.MAX_LONG_DIGITS()) {
       return Decimal.apply(getLong(ordinal), precision, scale);
     } else {
-      final byte[] bytes = getBinary(ordinal);
-      final BigInteger bigInteger = new BigInteger(bytes);
-      final BigDecimal javaDecimal = new BigDecimal(bigInteger, scale);
-      return Decimal.apply(new scala.math.BigDecimal(javaDecimal), precision, scale);
+      return SqlTypesUnsafeHelper.getDecimalExceedingLong(getBinary(ordinal), precision, scale,
+          true);
     }
   }
 
   @Override
   public UTF8String getUTF8String(int ordinal) {
     if (isNullAt(ordinal)) return null;
-    final long offsetAndSize = getLong(ordinal);
-    final int offset = (int) (offsetAndSize >> 32);
-    final int size = (int) offsetAndSize;
-    return UTF8String.fromAddress(baseObject, baseOffset + offset, size);
+    return SqlTypesUnsafeHelper.getUTF8String(getLong(ordinal), baseObject, baseOffset);
   }
 
   @Override
   public byte[] getBinary(int ordinal) {
     if (isNullAt(ordinal)) return null;
-    final long offsetAndSize = getLong(ordinal);
-    final int offset = (int) (offsetAndSize >> 32);
-    final int size = (int) offsetAndSize;
-    final byte[] bytes = new byte[size];
-    Platform.copyMemory(baseObject, baseOffset + offset, bytes, Platform.BYTE_ARRAY_OFFSET, size);
-    return bytes;
+    return SqlTypesUnsafeHelper.getBinary(getLong(ordinal), baseObject, baseOffset);
   }
 
   @Override
   public CalendarInterval getInterval(int ordinal) {
     if (isNullAt(ordinal)) return null;
-    final long offsetAndSize = getLong(ordinal);
-    final int offset = (int) (offsetAndSize >> 32);
-    final int months = (int) Platform.getLong(baseObject, baseOffset + offset);
-    final long microseconds = Platform.getLong(baseObject, baseOffset + offset + 8);
-    return new CalendarInterval(months, microseconds);
+    return SqlTypesUnsafeHelper.getInterval(getLong(ordinal), baseObject, baseOffset);
   }
 
   @Override
   public UnsafeRow getStruct(int ordinal, int numFields) {
     if (isNullAt(ordinal)) return null;
-    final long offsetAndSize = getLong(ordinal);
-    final int offset = (int) (offsetAndSize >> 32);
-    final int size = (int) offsetAndSize;
-    final UnsafeRow row = new UnsafeRow(numFields);
-    row.pointTo(baseObject, baseOffset + offset, size);
-    return row;
+    return SqlTypesUnsafeHelper.getStruct(getLong(ordinal), baseObject, baseOffset, numFields);
   }
 
   @Override
   public UnsafeArrayData getArray(int ordinal) {
     if (isNullAt(ordinal)) return null;
-    final long offsetAndSize = getLong(ordinal);
-    final int offset = (int) (offsetAndSize >> 32);
-    final int size = (int) offsetAndSize;
-    final UnsafeArrayData array = new UnsafeArrayData();
-    array.pointTo(baseObject, baseOffset + offset, size);
-    return array;
+    return SqlTypesUnsafeHelper.getArray(getLong(ordinal), baseObject, baseOffset);
   }
 
   @Override
   public UnsafeMapData getMap(int ordinal) {
     if (isNullAt(ordinal)) return null;
-    final long offsetAndSize = getLong(ordinal);
-    final int offset = (int) (offsetAndSize >> 32);
-    final int size = (int) offsetAndSize;
-    final UnsafeMapData map = new UnsafeMapData();
-    map.pointTo(baseObject, baseOffset + offset, size);
-    return map;
+    return SqlTypesUnsafeHelper.getMap(getLong(ordinal), baseObject, baseOffset);
   }
 
   @Override
@@ -363,25 +333,14 @@ public boolean equals(Object other) {
     return false;
   }
 
-  public void writeToMemory(Object target, long targetOffset) {
-    Platform.copyMemory(baseObject, baseOffset, target, targetOffset, sizeInBytes);
-  }
-
   public void writeTo(ByteBuffer buffer) {
-    assert(buffer.hasArray());
-    byte[] target = buffer.array();
-    int offset = buffer.arrayOffset();
-    int pos = buffer.position();
-    writeToMemory(target, Platform.BYTE_ARRAY_OFFSET + offset + pos);
-    buffer.position(pos + sizeInBytes);
+    UnsafeHelper.writeTo(buffer, baseObject, baseOffset, sizeInBytes);
   }
 
   @Override
   public UnsafeArrayData copy() {
     UnsafeArrayData arrayCopy = new UnsafeArrayData();
-    final byte[] arrayDataCopy = new byte[sizeInBytes];
-    Platform.copyMemory(
-      baseObject, baseOffset, arrayDataCopy, Platform.BYTE_ARRAY_OFFSET, sizeInBytes);
+    final byte[] arrayDataCopy = UnsafeHelper.copyToMemory(baseObject, baseOffset, sizeInBytes);
     arrayCopy.pointTo(arrayDataCopy, Platform.BYTE_ARRAY_OFFSET, sizeInBytes);
     return arrayCopy;
   }
@@ -447,14 +406,7 @@ public static UnsafeArrayData fromPrimitiveArray(
     final long headerInBytes = calculateHeaderPortionInBytes(length);
     final long valueRegionInBytes = (long)elementSize * length;
     final long totalSizeInLongs = (headerInBytes + valueRegionInBytes + 7) / 8;
-    if (totalSizeInLongs > Integer.MAX_VALUE / 8) {
-      throw new UnsupportedOperationException("Cannot convert this array to unsafe format as " +
-        "it's too big.");
-    }
-
-    final long[] data = new long[(int)totalSizeInLongs];
-
-    Platform.putLong(data, Platform.LONG_ARRAY_OFFSET, length);
+    final long[] data = initializeUnderlyingArray(length, totalSizeInLongs);
     if (arr != null) {
       Platform.copyMemory(arr, offset, data,
         Platform.LONG_ARRAY_OFFSET + headerInBytes, valueRegionInBytes);
@@ -466,28 +418,32 @@ public static UnsafeArrayData fromPrimitiveArray(
   }
 
   public static UnsafeArrayData createFreshArray(int length, int elementSize) {
-    final long headerInBytes = calculateHeaderPortionInBytes(length);
-    final long valueRegionInBytes = (long)elementSize * length;
-    final long totalSizeInLongs = (headerInBytes + valueRegionInBytes + 7) / 8;
+    final long totalSizeInLongs = getTotalSize(elementSize, length);
+    final long[] data = initializeUnderlyingArray(length, totalSizeInLongs);
+    UnsafeArrayData result = new UnsafeArrayData();
+    result.pointTo(data, Platform.LONG_ARRAY_OFFSET, (int)totalSizeInLongs * 8);
+    return result;
+  }
+
+  private static long[] initializeUnderlyingArray(int length, long totalSizeInLongs) {
     if (totalSizeInLongs > Integer.MAX_VALUE / 8) {
       throw new UnsupportedOperationException("Cannot convert this array to unsafe format as " +
-        "it's too big.");
+          "it's too big.");
     }
 
     final long[] data = new long[(int)totalSizeInLongs];
-
     Platform.putLong(data, Platform.LONG_ARRAY_OFFSET, length);
-
-    UnsafeArrayData result = new UnsafeArrayData();
-    result.pointTo(data, Platform.LONG_ARRAY_OFFSET, (int)totalSizeInLongs * 8);
-    return result;
+    return data;
   }
 
-  public static boolean shouldUseGenericArrayData(int elementSize, long length) {
+  private static long getTotalSize(int elementSize, long length) {
     final long headerInBytes = calculateHeaderPortionInBytes(length);
     final long valueRegionInBytes = elementSize * length;
-    final long totalSizeInLongs = (headerInBytes + valueRegionInBytes + 7) / 8;
-    return totalSizeInLongs > Integer.MAX_VALUE / 8;
+    return (headerInBytes + valueRegionInBytes + 7) / 8;
+  }
+
+  public static boolean shouldUseGenericArrayData(int elementSize, long length) {
+    return getTotalSize(elementSize, length) > Integer.MAX_VALUE / 8;
   }
 
   public static UnsafeArrayData fromPrimitiveArray(boolean[] arr) {