address review comments

apache · Mar 2, 2018 · c9f401a · c9f401a
1 parent ee5a798
commit c9f401a
Show file tree

Hide file tree

Showing 22 changed files with 152 additions and 184 deletions.
diff --git a/common/unsafe/src/main/java/org/apache/spark/sql/catalyst/expressions/HiveHasher.java b/common/unsafe/src/main/java/org/apache/spark/sql/catalyst/expressions/HiveHasher.java
@@ -39,8 +39,8 @@ public static int hashLong(long input) {
     return (int) ((input >>> 32) ^ input);
   }
 
-  public static int hashUnsafeBytesBlock(MemoryBlock base, long offset, int lengthInBytes) {
-    return hashUnsafeBytes(base.getBaseObject(), offset, lengthInBytes);
+  public static int hashUnsafeBytesBlock(MemoryBlock base) {
+    return hashUnsafeBytes(base.getBaseObject(), base.getBaseOffset(), (int)base.size());
   }
 
   public static int hashUnsafeBytes(Object base, long offset, int lengthInBytes) {

diff --git a/common/unsafe/src/main/java/org/apache/spark/unsafe/hash/Murmur3_x86_32.java b/common/unsafe/src/main/java/org/apache/spark/unsafe/hash/Murmur3_x86_32.java
@@ -64,7 +64,9 @@ public static int hashUnsafeWords(Object base, long offset, int lengthInBytes, i
     return fmix(h1, lengthInBytes);
   }
 
-  public static int hashUnsafeBytesBlock(MemoryBlock base, long offset, int lengthInBytes, int seed) {
+  public static int hashUnsafeBytesBlock(MemoryBlock base, int seed) {
+    long offset = base.getBaseOffset();
+    int lengthInBytes = (int)base.size();
     assert (lengthInBytes >= 0): "lengthInBytes cannot be negative";
     int lengthAligned = lengthInBytes - lengthInBytes % 4;
     int h1 = hashBytesByIntBlock(base, offset, lengthAligned, seed);

diff --git a/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/ByteArrayMemoryBlock.java b/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/ByteArrayMemoryBlock.java
@@ -27,12 +27,16 @@ public final class ByteArrayMemoryBlock extends MemoryBlock {
   private final byte[] array;
 
   public ByteArrayMemoryBlock(byte[] obj, long offset, long length) {
-    super(obj, offset, length);
+    super(obj, offset, (long)length);
     this.array = obj;
   }
 
+  public ByteArrayMemoryBlock(long length) {
+    this(new byte[(int)length], Platform.BYTE_ARRAY_OFFSET, length);
+  }
+
   @Override
-  public MemoryBlock allocate(long offset, long size) {
+  public MemoryBlock subBlock(long offset, long size) {
     return new ByteArrayMemoryBlock(array, this.offset + offset, size);
   }
 
@@ -45,7 +49,6 @@ public static ByteArrayMemoryBlock fromArray(final byte[] array) {
     return new ByteArrayMemoryBlock(array, Platform.BYTE_ARRAY_OFFSET, array.length);
   }
 
-
   public final int getInt(long offset) {
     return Platform.getInt(array, offset);
   }

diff --git a/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/MemoryBlock.java b/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/MemoryBlock.java
@@ -50,7 +50,7 @@ public abstract class MemoryBlock {
 
   protected long offset;
 
-  protected long length;
+  protected final long length;
 
   /**
    * Optional page number; used when this MemoryBlock represents a page allocated by a
@@ -125,9 +125,10 @@ public final static MemoryBlock allocateFromObject(Object obj, long offset, long
   }
 
   /**
-   * Instantiate the same type of MemoryBlock with new offset and size
+   * Just instantiate the same type of MemoryBlock with new offset and size. The data is not
+   * copied.
    */
-  public abstract MemoryBlock allocate(long offset, long size);
+  public abstract MemoryBlock subBlock(long offset, long size);
 
 
   public abstract int getInt(long offset);

diff --git a/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/OffHeapMemoryBlock.java b/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/OffHeapMemoryBlock.java
@@ -26,13 +26,8 @@ public OffHeapMemoryBlock(long address, long size) {
     super(null, address, size);
   }
 
-  public void setAddressAndSize(long address, long size) {
-    this.offset = address;
-    this.length = size;
-  }
-
   @Override
-  public MemoryBlock allocate(long offset, long size) {
+  public MemoryBlock subBlock(long offset, long size) {
     return new OffHeapMemoryBlock(this.offset + offset, size);
   }
 

diff --git a/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/OnHeapMemoryBlock.java b/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/OnHeapMemoryBlock.java
@@ -32,8 +32,13 @@ public OnHeapMemoryBlock(long[] obj, long offset, long size) {
     assert(offset - Platform.LONG_ARRAY_OFFSET + size <= obj.length * 8L);
   }
 
+  public OnHeapMemoryBlock(long size) {
+    this(new long[(int)((size + 7) / 8)], Platform.LONG_ARRAY_OFFSET,
+      ((size + 7) / 8) * 8L);
+  }
+
   @Override
-  public MemoryBlock allocate(long offset, long size) {
+  public MemoryBlock subBlock(long offset, long size) {
     return new OnHeapMemoryBlock(array, this.offset + offset, size);
   }
 
@@ -43,80 +48,80 @@ public MemoryBlock allocate(long offset, long size) {
    * Creates a memory block pointing to the memory used by the long array.
    */
   public static OnHeapMemoryBlock fromArray(final long[] array) {
-    return new OnHeapMemoryBlock(array, Platform.LONG_ARRAY_OFFSET, array.length * 8);
+    return new OnHeapMemoryBlock(array, Platform.LONG_ARRAY_OFFSET, array.length * 8L);
   }
 
   public static OnHeapMemoryBlock fromArray(final long[] array, long size) {
     return new OnHeapMemoryBlock(array, Platform.LONG_ARRAY_OFFSET, size);
   }
 
   public final int getInt(long offset) {
-    assert(offset + 4 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 4 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     return Platform.getInt(array, offset);
   }
 
   public final void putInt(long offset, int value) {
-    assert(offset + 4 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 4 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     Platform.putInt(array, offset, value);
   }
 
   public final boolean getBoolean(long offset) {
-    assert(offset + 1 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 1 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     return Platform.getBoolean(array, offset);
   }
 
   public final void putBoolean(long offset, boolean value) {
-    assert(offset + 1 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 1 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     Platform.putBoolean(array, offset, value);
   }
 
   public final byte getByte(long offset) {
-    assert(offset + 1 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 1 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     return Platform.getByte(array, offset);
   }
 
   public final void putByte(long offset, byte value) {
-    assert(offset + 1 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 1 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     Platform.putByte(array, offset, value);
   }
 
   public final short getShort(long offset) {
-    assert(offset + 2 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 2 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     return Platform.getShort(array, offset);
   }
 
   public final void putShort(long offset, short value) {
-    assert(offset + 2 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 2 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     Platform.putShort(array, offset, value);
   }
 
   public final long getLong(long offset) {
-    assert(offset + 8 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 8 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     return Platform.getLong(array, offset);
   }
 
   public final void putLong(long offset, long value) {
-    assert(offset + 8 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 8 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     Platform.putLong(array, offset, value);
   }
 
   public final float getFloat(long offset) {
-    assert(offset + 4 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 4 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     return Platform.getFloat(array, offset);
   }
 
   public final void putFloat(long offset, float value) {
-    assert(offset + 4 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 4 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     Platform.putFloat(array, offset, value);
   }
 
   public final double getDouble(long offset) {
-    assert(offset + 8 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 8 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     return Platform.getDouble(array, offset);
   }
 
   public final void putDouble(long offset, double value) {
-    assert(offset + 8 - Platform.LONG_ARRAY_OFFSET <= array.length * 8);
+    assert(offset + 8 - Platform.LONG_ARRAY_OFFSET <= array.length * 8L);
     Platform.putDouble(array, offset, value);
   }
 }
diff --git a/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/UnsafeMemoryAllocator.java b/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/UnsafeMemoryAllocator.java
@@ -69,7 +69,8 @@ public OffHeapMemoryBlock reallocate(OffHeapMemoryBlock block, long oldSize, lon
     OffHeapMemoryBlock mb = this.allocate(newSize);
     if (block.getBaseOffset() != 0)
       MemoryBlock.copyMemory(block, mb, oldSize);
-    free(block);
+    if (block != mb)
+      free(block);
     return mb;
   }
 }