asterixdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From prest...@apache.org
Subject [36/50] [abbrv] asterixdb git commit: Deletable frame tuple appender with reuse of tuple index slots
Date Mon, 17 Oct 2016 19:55:26 GMT
Deletable frame tuple appender with reuse of tuple index slots


Project: http://git-wip-us.apache.org/repos/asf/asterixdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/asterixdb/commit/fd514a0e
Tree: http://git-wip-us.apache.org/repos/asf/asterixdb/tree/fd514a0e
Diff: http://git-wip-us.apache.org/repos/asf/asterixdb/diff/fd514a0e

Branch: refs/heads/ecarm002/interval_join_merge
Commit: fd514a0ed598219faf0ecdfa19e3c894f3b706f1
Parents: 6c31214
Author: Preston Carman <prestonc@apache.org>
Authored: Mon Sep 19 17:59:17 2016 -0700
Committer: Preston Carman <prestonc@apache.org>
Committed: Mon Sep 19 17:59:17 2016 -0700

----------------------------------------------------------------------
 .../common/comm/io/FrameTupleAccessor.java      |   5 +-
 .../sort/util/DeletableFrameTupleAppender.java  | 259 ++++++++++++++-----
 2 files changed, 204 insertions(+), 60 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fd514a0e/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAccessor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAccessor.java
b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAccessor.java
index d99e2f2..1c702be 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAccessor.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAccessor.java
@@ -35,7 +35,8 @@ import org.apache.hyracks.util.IntSerDeUtils;
 /**
  * FrameTupleCursor is used to navigate over tuples in a Frame. A frame is
  * formatted with tuple data concatenated starting at offset 1, one tuple after
- * another. The first byte is used to notify how big the frame is, so the maximum frame size
is 255 * initialFrameSetting.
+ * another. The first byte is used to notify how big the frame is,
+ * so the maximum frame size is 255 * initialFrameSetting.
  * Offset FS - 4 holds an int indicating the number of tuples (N) in
  * the frame. FS - ((i + 1) * 4) for i from 0 to N - 1 holds an int indicating
  * the offset of the (i + 1)^th tuple. Every tuple is organized as a sequence of
@@ -184,6 +185,7 @@ public class FrameTupleAccessor implements IFrameTupleAccessor {
             prettyPrint(i, bbis, dis, sb, recordFields);
         }
         System.err.println(sb.toString());
+        bbis.close();
     }
 
     public void prettyPrint(int tIdx, int[] recordFields) throws IOException {
@@ -192,6 +194,7 @@ public class FrameTupleAccessor implements IFrameTupleAccessor {
         StringBuilder sb = new StringBuilder();
         prettyPrint(tIdx, bbis, dis, sb, recordFields);
         System.err.println(sb.toString());
+        bbis.close();
     }
 
     public void prettyPrint(ITupleReference tuple, int fieldsIdx, int descIdx) throws HyracksDataException
{

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fd514a0e/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/util/DeletableFrameTupleAppender.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/util/DeletableFrameTupleAppender.java
b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/util/DeletableFrameTupleAppender.java
index 79aba3e..8cae721 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/util/DeletableFrameTupleAppender.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/util/DeletableFrameTupleAppender.java
@@ -21,8 +21,11 @@ package org.apache.hyracks.dataflow.std.sort.util;
 
 import java.io.PrintStream;
 import java.nio.ByteBuffer;
+import java.util.Comparator;
+import java.util.PriorityQueue;
 
-import org.apache.hyracks.api.comm.FrameHelper;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hyracks.api.comm.IFrameTupleAccessor;
 import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -32,59 +35,153 @@ import org.apache.hyracks.util.IntSerDeUtils;
  * This is a special frame which is used in TupleMemoryBuffer.
  * This frame has a special structure to organize the deleted spaces.
  * Specifically, the endOffset of the deleted tuple will be set as negative number.
- * And we add a special <code>deleted_space</code> field at the last 4 bytes
to remember how many bytes has been deleted.
+ * And we add a special <code>deleted_space</code> field at the last 4 bytes
to
+ * remember how many bytes has been deleted.
+ * The offsets also store both the start and end values because tuples may be out of
+ * after several add, remove and reorganize operations.
+ * A frame is formatted with tuple data concatenated starting at offset 0,
+ * one tuple after another.
+ * Offset FS - 4 holds an int indicating the amount of <code>deleted_space</code>
in the frame.
+ * Offset FS - 4 holds an int indicating the number of tuples (N) in the frame.
+ * FS - ((i + 1) * (4 + 4)) for i from 0 to N - 1 holds an two ints indicating
+ * the offset and length of the (i + 1)^th tuple.
+ * The tuple references are organized as a sequence of ints indicating the start of the field
+ * followed by the length of each tuple.
+ * The offset if the start of the frame.
+ * The tuple has been deleted if the length is 0.
+ *
+ * <pre>
+ * [ *tuple_1_bytes*,
+ *   *tuple_2_bytes*,
+ *   ...
+ *   int length, int offset, # tuple 2
+ *   int length, int offset, # tuple 1
+ *   int tuple_append,
+ *   int next_index,
+ *   int deleted_space,
+ *   int index_count,
+ * ]
+ * </pre>
+ *
+ * <pre>
+ * [ *tuple_1_bytes*,
+ *   *tuple_2_bytes*,
+ *   ...
+ *   int end_offset,
+ *   int offset, # tuple 2
+ *   int offset, # tuple 1
+ *   int deleted_space,
+ *   int tuple_count,
+ * ]
+ * </pre>
  */
 public class DeletableFrameTupleAppender implements IAppendDeletableFrameTupleAccessor {
 
+    public static final Comparator<Pair<Integer, Integer>> INDEX_OFFSET_ASC_COMPARATOR
= new Comparator<Pair<Integer, Integer>>() {
+        @Override
+        public int compare(Pair<Integer, Integer> p1, Pair<Integer, Integer>
p2) {
+            return p1.getValue() - p2.getValue();
+        }
+
+    };
+
+    private static final int SIZE_INDEX_COUNT = 4;
     private static final int SIZE_DELETED_SPACE = 4;
+    private static final int SIZE_NEXT_INDEX = 4;
+    private static final int SIZE_TUPLE_APPEND = 4;
+
+    private static final int SIZE_START_OFFSET = 4;
+    private static final int SIZE_END_OFFSET = 4;
+    private static final int SIZE_OFFSET_GROUP = SIZE_END_OFFSET + SIZE_START_OFFSET;
+
     private final RecordDescriptor recordDescriptor;
     private ByteBuffer buffer;
-    private int tupleCountOffset;
-    private int tupleCount;
-    private int freeDataEndOffset;
+    private int indexSlotsOffset;
+    private int indexCount;
+    private int tupleAppend;
     private int deletedSpace;
+    private int nextIndex;
     private byte[] array; // to speed up the array visit a little
 
+    private final PriorityQueue<Pair<Integer, Integer>> reorganizeQueue;
+
     public DeletableFrameTupleAppender(RecordDescriptor recordDescriptor) {
         this.recordDescriptor = recordDescriptor;
+        reorganizeQueue = new PriorityQueue<>(16, INDEX_OFFSET_ASC_COMPARATOR);
     }
 
-    private int getTupleCountOffset() {
-        return FrameHelper.getTupleCountOffset(buffer.capacity()) - SIZE_DELETED_SPACE;
+    private int getIndexCount() {
+        return IntSerDeUtils.getInt(array, getIndexCountOffset());
     }
 
-    private int getFreeDataEndOffset() {
-        return tupleCount == 0 ? 0 : Math.abs(IntSerDeUtils.getInt(array, tupleCountOffset
- tupleCount * 4));
+    private void setIndexCount(int count) {
+        IntSerDeUtils.putInt(array, getIndexCountOffset(), count);
     }
 
-    private void setFreeDataEndOffset(int offset) {
-        assert (offset >= 0);
-        IntSerDeUtils.putInt(array, tupleCountOffset - tupleCount * 4, offset);
+    private int getIndexCountOffset() {
+        return buffer.capacity() - SIZE_INDEX_COUNT;
     }
 
-    private void setTupleCount(int count) {
-        IntSerDeUtils.putInt(array, tupleCountOffset, count);
+    private int getDeletedSpace() {
+        return IntSerDeUtils.getInt(array, getDeletedSpaceOffset());
     }
 
-    private void setDeleteSpace(int count) {
-        IntSerDeUtils.putInt(array, buffer.capacity() - SIZE_DELETED_SPACE, count);
+    private void setDeletedSpace(int space) {
+        IntSerDeUtils.putInt(array, getDeletedSpaceOffset(), space);
     }
 
-    private int getPhysicalTupleCount() {
-        return IntSerDeUtils.getInt(array, tupleCountOffset);
+    private int getDeletedSpaceOffset() {
+        return getIndexCountOffset() - SIZE_DELETED_SPACE;
     }
 
-    private int getDeletedSpace() {
-        return IntSerDeUtils.getInt(array, buffer.capacity() - SIZE_DELETED_SPACE);
+    private int getNextIndex() {
+        return IntSerDeUtils.getInt(array, getNextIndexOffset());
+    }
+
+    private void setNextIndex(int index) {
+        IntSerDeUtils.putInt(array, getNextIndexOffset(), index);
+    }
+
+    private int getNextIndexOffset() {
+        return getDeletedSpaceOffset() - SIZE_NEXT_INDEX;
+    }
+
+    private int getAndUpdateNextIndex() {
+        int index = nextIndex;
+        nextIndex = index + 1;
+        while (nextIndex < indexCount) {
+            if (getTupleEndOffset(nextIndex) <= 0) {
+                break;
+            }
+            nextIndex++;
+        }
+        setNextIndex(nextIndex);
+        return index;
+    }
+
+    private int getTupleAppend() {
+        return IntSerDeUtils.getInt(array, getTupleAppendOffset());
+    }
+
+    private void setTupleAppend(int offset) {
+        IntSerDeUtils.putInt(array, getTupleAppendOffset(), offset);
+    }
+
+    private int getTupleAppendOffset() {
+        return getNextIndexOffset() - SIZE_TUPLE_APPEND;
+    }
+
+    private int getIndexSlotOffset() {
+        return getTupleAppendOffset();
     }
 
     @Override
     public void clear(ByteBuffer buffer) throws HyracksDataException {
         this.buffer = buffer;
         this.array = buffer.array();
-        tupleCountOffset = getTupleCountOffset();
-        setTupleCount(0);
-        setDeleteSpace(0);
+        setIndexCount(0);
+        setDeletedSpace(0);
+        setTupleAppend(0);
         resetCounts();
     }
 
@@ -92,14 +189,15 @@ public class DeletableFrameTupleAppender implements IAppendDeletableFrameTupleAc
     public void reset(ByteBuffer buffer) {
         this.buffer = buffer;
         this.array = buffer.array();
-        tupleCountOffset = getTupleCountOffset();
         resetCounts();
     }
 
     private void resetCounts() {
+        indexSlotsOffset = getIndexSlotOffset();
         deletedSpace = getDeletedSpace();
-        tupleCount = getPhysicalTupleCount();
-        freeDataEndOffset = getFreeDataEndOffset();
+        indexCount = getIndexCount();
+        tupleAppend = getTupleAppend();
+        nextIndex = getNextIndex();
     }
 
     /**
@@ -116,11 +214,18 @@ public class DeletableFrameTupleAppender implements IAppendDeletableFrameTupleAc
         byte[] src = tupleAccessor.getBuffer().array();
         int tStartOffset = tupleAccessor.getTupleStartOffset(tIndex);
         int length = tupleAccessor.getTupleLength(tIndex);
-        System.arraycopy(src, tStartOffset, array, freeDataEndOffset, length);
-        setTupleCount(++tupleCount);
-        freeDataEndOffset += length;
-        setFreeDataEndOffset(freeDataEndOffset);
-        return tupleCount - 1;
+        System.arraycopy(src, tStartOffset, array, tupleAppend, length);
+        int index = getAndUpdateNextIndex();
+        if (index < indexCount) {
+            // Don't change index count
+        } else {
+            // Increment count
+            setIndexCount(++indexCount);
+        }
+        setTupleOffsets(index, tupleAppend, length);
+        tupleAppend += length;
+        setTupleAppend(tupleAppend);
+        return index;
     }
 
     @Override
@@ -129,7 +234,11 @@ public class DeletableFrameTupleAppender implements IAppendDeletableFrameTupleAc
         if (endOffset > 0) {
             setTupleEndOffset(tupleIndex, -endOffset);
             deletedSpace += endOffset - getTupleStartOffset(tupleIndex);
-            setDeleteSpace(deletedSpace);
+            setDeletedSpace(deletedSpace);
+            if (nextIndex > tupleIndex) {
+                nextIndex = tupleIndex;
+                setNextIndex(nextIndex);
+            }
         }
     }
 
@@ -140,36 +249,53 @@ public class DeletableFrameTupleAppender implements IAppendDeletableFrameTupleAc
         }
         reclaimDeletedEnding();
 
-        freeDataEndOffset = 0;
-        int endOffset = 0;
-        for (int i = 0; i < tupleCount; i++) {
-            int startOffset = Math.abs(endOffset);
+        // Build reorganize queue
+        int endOffset;
+        int startOffset;
+        for (int i = 0; i < indexCount; i++) {
             endOffset = getTupleEndOffset(i);
+            if (endOffset > 0) {
+                reorganizeQueue.add(new ImmutablePair<Integer, Integer>(i, getTupleStartOffset(i)));
+            }
+        }
+
+        int index;
+        tupleAppend = 0;
+        while (!reorganizeQueue.isEmpty()) {
+            index = reorganizeQueue.remove().getKey();
+            startOffset = getTupleStartOffset(index);
+            endOffset = getTupleEndOffset(index);
             if (endOffset >= 0) {
                 int length = endOffset - startOffset;
                 assert length >= 0;
-                if (freeDataEndOffset != startOffset) {
-                    System.arraycopy(array, startOffset, array, freeDataEndOffset, length);
+                if (tupleAppend != startOffset) {
+                    System.arraycopy(array, startOffset, array, tupleAppend, length);
                 }
-                freeDataEndOffset += length;
+                setTupleOffsets(index, tupleAppend, length);
+                tupleAppend += length;
             }
-            setTupleEndOffset(i, freeDataEndOffset);
         }
-        setFreeDataEndOffset(freeDataEndOffset);
+        setTupleAppend(tupleAppend);
         deletedSpace = 0;
-        setDeleteSpace(0);
+        setDeletedSpace(0);
+
+        // Clean up
+        reorganizeQueue.clear();
     }
 
     private void reclaimDeletedEnding() {
-        for (int i = tupleCount - 1; i >= 0; i--) {
+        for (int i = indexCount - 1; i >= 0; i--) {
             int endOffset = getTupleEndOffset(i);
             if (endOffset <= 0) {
-                tupleCount--;
+                indexCount--;
             } else {
                 break;
             }
         }
-        setTupleCount(tupleCount);
+        setIndexCount(indexCount);
+        if (nextIndex > indexCount) {
+            setNextIndex(indexCount);
+        }
     }
 
     @Override
@@ -179,7 +305,8 @@ public class DeletableFrameTupleAppender implements IAppendDeletableFrameTupleAc
 
     @Override
     public int getContiguousFreeSpace() {
-        return getTupleCountOffset() - tupleCount * 4 - freeDataEndOffset;
+        int slotSpace = indexCount * SIZE_OFFSET_GROUP;
+        return indexSlotsOffset - tupleAppend - slotSpace;
     }
 
     @Override
@@ -203,6 +330,11 @@ public class DeletableFrameTupleAppender implements IAppendDeletableFrameTupleAc
     }
 
     @Override
+    public int getAbsoluteFieldStartOffset(int tupleIndex, int fIdx) {
+        return getTupleStartOffset(tupleIndex) + getFieldSlotsLength() + getFieldStartOffset(tupleIndex,
fIdx);
+    }
+
+    @Override
     public int getFieldLength(int tupleIndex, int fIdx) {
         return getFieldEndOffset(tupleIndex, fIdx) - getFieldStartOffset(tupleIndex, fIdx);
     }
@@ -210,40 +342,49 @@ public class DeletableFrameTupleAppender implements IAppendDeletableFrameTupleAc
     @Override
     public int getTupleLength(int tupleIndex) {
         int endOffset = getTupleEndOffset(tupleIndex);
-        if (endOffset < 0) {
-            return endOffset + getTupleStartOffset(tupleIndex);
-        }
+        assert endOffset > 0;
         return endOffset - getTupleStartOffset(tupleIndex);
     }
 
+    private void setTupleOffsets(int tupleIndex, int start, int length) {
+        setTupleStartOffset(tupleIndex, start);
+        setTupleEndOffset(tupleIndex, start + length);
+    }
+
     @Override
     public int getTupleEndOffset(int tupleIndex) {
-        return IntSerDeUtils.getInt(array, tupleCountOffset - 4 * (tupleIndex + 1));
+        return IntSerDeUtils.getInt(array, getTupleEndSlotOffset(tupleIndex));
     }
 
     private void setTupleEndOffset(int tupleIndex, int offset) {
-        IntSerDeUtils.putInt(array, tupleCountOffset - 4 * (tupleIndex + 1), offset);
+        IntSerDeUtils.putInt(array, getTupleEndSlotOffset(tupleIndex), offset);
     }
 
     @Override
     public int getTupleStartOffset(int tupleIndex) {
-        int offset = tupleIndex == 0 ? 0 : IntSerDeUtils.getInt(array, tupleCountOffset -
4 * tupleIndex);
-        return Math.abs(offset);
+        return IntSerDeUtils.getInt(array, getTupleStartSlotOffset(tupleIndex));
     }
 
-    @Override
-    public int getAbsoluteFieldStartOffset(int tupleIndex, int fIdx) {
-        return getTupleStartOffset(tupleIndex) + getFieldSlotsLength() + getFieldStartOffset(tupleIndex,
fIdx);
+    public void setTupleStartOffset(int tupleIndex, int offset) {
+        IntSerDeUtils.putInt(array, getTupleStartSlotOffset(tupleIndex), offset);
+    }
+
+    public int getTupleStartSlotOffset(int tupleIndex) {
+        return indexSlotsOffset - SIZE_OFFSET_GROUP * tupleIndex - SIZE_START_OFFSET;
+    }
+
+    public int getTupleEndSlotOffset(int tupleIndex) {
+        return getTupleStartSlotOffset(tupleIndex) - SIZE_END_OFFSET;
     }
 
     @Override
     public int getTupleCount() {
-        return tupleCount;
+        return indexCount;
     }
 
     private int getLiveTupleCount() {
         int live = 0;
-        for (int i = tupleCount - 1; i >= 0; i--) {
+        for (int i = 0; i < indexCount; ++i) {
             int endOffset = getTupleEndOffset(i);
             if (endOffset > 0) {
                 live++;
@@ -262,7 +403,7 @@ public class DeletableFrameTupleAppender implements IAppendDeletableFrameTupleAc
         if (getLiveTupleCount() == 0) {
             ps.print("");
         }
-        ps.printf("(%d, %d)", getLiveTupleCount(), getPhysicalTupleCount());
+        ps.printf("(%d, %d)", getLiveTupleCount(), getIndexCount());
     }
 
 }


Mime
View raw message