asterixdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From prest...@apache.org
Subject [01/50] [abbrv] asterixdb git commit: fix for indexing memory error.
Date Mon, 17 Oct 2016 19:54:51 GMT
Repository: asterixdb
Updated Branches:
  refs/heads/ecarm002/interval_join_merge [created] faa3bb457


fix for indexing memory error.


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

Branch: refs/heads/ecarm002/interval_join_merge
Commit: 199bddd3c68ef8e8c3ee0cccf6d8cdcc38fd8189
Parents: f9443fa
Author: Preston Carman <prestonc@apache.org>
Authored: Wed Jun 15 19:06:51 2016 -0700
Committer: Preston Carman <prestonc@apache.org>
Committed: Wed Jun 15 19:06:51 2016 -0700

----------------------------------------------------------------------
 .../joins/intervalindex/ActiveSweepManager.java | 10 +++----
 .../IntervalIndexJoinOperatorDescriptor.java    |  1 +
 .../intervalindex/IntervalIndexJoiner.java      | 29 ++++++--------------
 .../dataflow/std/join/AbstractMergeJoiner.java  | 12 ++++----
 .../dataflow/std/join/MergeBranchStatus.java    |  6 ++++
 .../std/join/MergeJoinOperatorDescriptor.java   |  1 +
 6 files changed, 29 insertions(+), 30 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/asterixdb/blob/199bddd3/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/ActiveSweepManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/ActiveSweepManager.java
b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/ActiveSweepManager.java
index 70785ab..84bd262 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/ActiveSweepManager.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/ActiveSweepManager.java
@@ -44,18 +44,18 @@ public class ActiveSweepManager {
     private EndPointIndexItem item = null;
     private final LinkedList<TuplePointer> active = new LinkedList<>();
 
-    public ActiveSweepManager(IPartitionedDeletableTupleBufferManager bufferManager, int
key, int partition,
+    public ActiveSweepManager(IPartitionedDeletableTupleBufferManager bufferManager, int
key, int joinBranch,
             Comparator<EndPointIndexItem> endPointComparator) {
         this.bufferManager = bufferManager;
         this.key = key;
-        this.partition = partition;
+        this.partition = joinBranch;
         indexQueue = new PriorityQueue<>(16, endPointComparator);
     }
 
-    public boolean addTuple(ITupleAccessor leftInputAccessor, TuplePointer tp) throws HyracksDataException
{
-        if (bufferManager.insertTuple(partition, leftInputAccessor, leftInputAccessor.getTupleId(),
tp)) {
+    public boolean addTuple(ITupleAccessor accessor, TuplePointer tp) throws HyracksDataException
{
+        if (bufferManager.insertTuple(partition, accessor, accessor.getTupleId(), tp)) {
             EndPointIndexItem e = new EndPointIndexItem(tp, EndPointIndexItem.END_POINT,
-                    IntervalJoinUtil.getIntervalEnd(leftInputAccessor, leftInputAccessor.getTupleId(),
key));
+                    IntervalJoinUtil.getIntervalEnd(accessor, accessor.getTupleId(), key));
             indexQueue.add(e);
             active.add(tp);
             item = indexQueue.peek();

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/199bddd3/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/IntervalIndexJoinOperatorDescriptor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/IntervalIndexJoinOperatorDescriptor.java
b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/IntervalIndexJoinOperatorDescriptor.java
index c248f72..be44df3 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/IntervalIndexJoinOperatorDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/IntervalIndexJoinOperatorDescriptor.java
@@ -258,6 +258,7 @@ public class IntervalIndexJoinOperatorDescriptor extends AbstractOperatorDescrip
                         locks.getRight(partition).await();
                     }
                     state.indexJoiner.setFrame(RIGHT_ACTIVITY_ID, buffer);
+                    state.status.continueRightLoad = false;
                     locks.getLeft(partition).signal();
                 } catch (InterruptedException e) {
                     Thread.currentThread().interrupt();

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/199bddd3/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/IntervalIndexJoiner.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/IntervalIndexJoiner.java
b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/IntervalIndexJoiner.java
index cc164b7..e53f9ae 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/IntervalIndexJoiner.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/joins/intervalindex/IntervalIndexJoiner.java
@@ -125,7 +125,6 @@ public class IntervalIndexJoiner extends AbstractMergeJoiner {
             LOGGER.fine("IntervalIndexJoiner has started partition " + partition + " with
" + memorySize
                     + " frames of memory.");
         }
-
     }
 
     private void addToResult(IFrameTupleAccessor accessor1, int index1, IFrameTupleAccessor
accessor2, int index2,
@@ -134,7 +133,6 @@ public class IntervalIndexJoiner extends AbstractMergeJoiner {
             FrameUtils.appendConcatToWriter(writer, resultAppender, accessor2, index2, accessor1,
index1);
         } else {
             FrameUtils.appendConcatToWriter(writer, resultAppender, accessor1, index1, accessor2,
index2);
-
         }
     }
 
@@ -200,7 +198,8 @@ public class IntervalIndexJoiner extends AbstractMergeJoiner {
     public void processMergeUsingLeftTuple(IFrameWriter writer) throws HyracksDataException
{
         TupleStatus leftTs = loadLeftTuple();
         TupleStatus rightTs = loadRightTuple();
-        while (checkHasMoreProcessing(leftTs, LEFT_PARTITION) && checkHasMoreProcessing(rightTs,
RIGHT_PARTITION)) {
+        while (leftTs.isKnown() && checkHasMoreProcessing(leftTs, LEFT_PARTITION,
RIGHT_PARTITION)
+                && checkHasMoreProcessing(rightTs, RIGHT_PARTITION, LEFT_PARTITION))
{
             if (status.branch[RIGHT_PARTITION].isRunFileWriting()) {
                 // Right side from disk
                 rightTs = processRightTupleSpill(writer);
@@ -221,9 +220,13 @@ public class IntervalIndexJoiner extends AbstractMergeJoiner {
         }
     }
 
-    private boolean checkHasMoreProcessing(TupleStatus ts, int partition) {
-        return ts.isLoaded() || (ts.isEmpty()
-                && (activeManager[partition].hasRecords() || status.branch[partition].isRunFileWriting()));
+    private boolean checkHasMoreProcessing(TupleStatus ts, int partition, int joinPartition)
{
+        return ts.isLoaded() || status.branch[partition].isRunFileWriting()
+                || (checkHasMoreTuples(joinPartition) && activeManager[partition].hasRecords());
+    }
+
+    private boolean checkHasMoreTuples(int partition) {
+        return status.branch[partition].hasMore() || status.branch[partition].isRunFileReading();
     }
 
     private boolean checkToProcessRightTuple() {
@@ -244,7 +247,6 @@ public class IntervalIndexJoiner extends AbstractMergeJoiner {
 
     private TupleStatus processLeftTupleSpill(IFrameWriter writer) throws HyracksDataException
{
         // Process left tuples one by one, check them with active memory from the right branch.
-        //        System.err.println("  -  Start processLeftTupleSpill");
         int count = 0;
         TupleStatus ts = loadLeftTuple();
         while (ts.isLoaded() && activeManager[RIGHT_PARTITION].hasRecords()) {
@@ -273,12 +275,10 @@ public class IntervalIndexJoiner extends AbstractMergeJoiner {
             unfreezeAndContinue(LEFT_PARTITION, inputAccessor[LEFT_PARTITION], RIGHT_PARTITION);
             ts = loadLeftTuple();
         }
-        //        System.err.println("  -  End processLeftTupleSpill");
         return ts;
     }
 
     private TupleStatus processRightTupleSpill(IFrameWriter writer) throws HyracksDataException
{
-        //        System.err.println("  -  Start processRightTupleSpill");
         // Process left tuples one by one, check them with active memory from the right branch.
         int count = 0;
         TupleStatus ts = loadRightTuple();
@@ -308,12 +308,10 @@ public class IntervalIndexJoiner extends AbstractMergeJoiner {
             unfreezeAndContinue(RIGHT_PARTITION, inputAccessor[RIGHT_PARTITION], LEFT_PARTITION);
             ts = loadRightTuple();
         }
-        //        System.err.println("  -  End processRightTupleSpill");
         return ts;
     }
 
     private void processLeftTuple(IFrameWriter writer) throws HyracksDataException {
-        //        System.err.println(" +++ Start processLeftTuple");
         // Process endpoints
         do {
             if ((!activeManager[LEFT_PARTITION].hasRecords()
@@ -341,11 +339,9 @@ public class IntervalIndexJoiner extends AbstractMergeJoiner {
             processActiveJoin(activeManager[RIGHT_PARTITION].getActiveList(), memoryAccessor[RIGHT_PARTITION],
buffer,
                     memoryAccessor[LEFT_PARTITION], true, writer);
         }
-        //        System.err.println(" +++ End processLeftTuple");
     }
 
     private void processRightTuple(IFrameWriter writer) throws HyracksDataException {
-        //        System.err.println(" +++ Start processRightTuple");
         // Process endpoints
         do {
             if ((!activeManager[RIGHT_PARTITION].hasRecords()
@@ -373,7 +369,6 @@ public class IntervalIndexJoiner extends AbstractMergeJoiner {
             processActiveJoin(activeManager[LEFT_PARTITION].getActiveList(), memoryAccessor[LEFT_PARTITION],
buffer,
                     memoryAccessor[RIGHT_PARTITION], false, writer);
         }
-        //        System.err.println(" +++ End processRightTuple");
     }
 
     private void processActiveJoin(List<TuplePointer> outer, ITuplePointerAccessor
outerAccessor,
@@ -383,11 +378,8 @@ public class IntervalIndexJoiner extends AbstractMergeJoiner {
             outerAccessor.reset(outerTp);
             for (TuplePointer innerTp : inner) {
                 innerAccessor.reset(innerTp);
-                //                TuplePrinterUtil.printTuple("     --- A outer", outerAccessor,
outerTp.tupleIndex);
-                //                TuplePrinterUtil.printTuple("     --- A inner", innerAccessor,
innerTp.tupleIndex);
                 if (imjc.checkToSaveInResult(outerAccessor, outerTp.tupleIndex, innerAccessor,
innerTp.tupleIndex,
                         reversed)) {
-                    //                    System.err.println("  -- Matched --");
                     addToResult(outerAccessor, outerTp.tupleIndex, innerAccessor, innerTp.tupleIndex,
reversed, writer);
                 }
             }
@@ -402,11 +394,8 @@ public class IntervalIndexJoiner extends AbstractMergeJoiner {
             ITupleAccessor tupleAccessor, boolean reversed, IFrameWriter writer) throws HyracksDataException
{
         for (TuplePointer outerTp : outer) {
             outerAccessor.reset(outerTp);
-            //            TuplePrinterUtil.printTuple("     --- outer", outerAccessor, outerTp.tupleIndex);
-            //            TuplePrinterUtil.printTuple("     --- inner", tupleAccessor);
             if (imjc.checkToSaveInResult(outerAccessor, outerTp.tupleIndex, tupleAccessor,
tupleAccessor.getTupleId(),
                     reversed)) {
-                //                System.err.println("  -- Matched --");
                 addToResult(outerAccessor, outerTp.tupleIndex, tupleAccessor, tupleAccessor.getTupleId(),
reversed,
                         writer);
             }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/199bddd3/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/AbstractMergeJoiner.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/AbstractMergeJoiner.java
b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/AbstractMergeJoiner.java
index e9ed084..f8d328b 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/AbstractMergeJoiner.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/AbstractMergeJoiner.java
@@ -27,7 +27,6 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
 import org.apache.hyracks.dataflow.std.buffermanager.ITupleAccessor;
 import org.apache.hyracks.dataflow.std.buffermanager.TupleAccessor;
-import org.apache.hyracks.dataflow.std.join.AbstractMergeJoiner.TupleStatus;
 import org.apache.hyracks.dataflow.std.join.MergeBranchStatus.Stage;
 
 public abstract class AbstractMergeJoiner implements IMergeJoiner {
@@ -44,6 +43,10 @@ public abstract class AbstractMergeJoiner implements IMergeJoiner {
         public boolean isEmpty() {
             return this.equals(EMPTY);
         }
+
+        public boolean isKnown() {
+            return !this.equals(UNKNOWN);
+        }
     }
 
     protected static final int JOIN_PARTITIONS = 2;
@@ -97,12 +100,12 @@ public abstract class AbstractMergeJoiner implements IMergeJoiner {
         return TupleStatus.LOADED;
     }
 
-    protected TupleStatus loadMemoryTuple(int partition) {
+    protected TupleStatus loadMemoryTuple(int joinId) {
         TupleStatus loaded;
-        if (inputAccessor[partition] != null && inputAccessor[partition].exists())
{
+        if (inputAccessor[joinId] != null && inputAccessor[joinId].exists()) {
             // Still processing frame.
             loaded = TupleStatus.LOADED;
-        } else if (status.branch[partition].hasMore()) {
+        } else if (status.branch[joinId].hasMore()) {
             loaded = TupleStatus.UNKNOWN;
         } else {
             // No more frames or tuples to process.
@@ -120,7 +123,6 @@ public abstract class AbstractMergeJoiner implements IMergeJoiner {
         inputBuffer[partition].put(buffer.array(), 0, buffer.capacity());
         inputAccessor[partition].reset(inputBuffer[partition]);
         inputAccessor[partition].next();
-        status.continueRightLoad = false;
     }
 
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/199bddd3/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/MergeBranchStatus.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/MergeBranchStatus.java
b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/MergeBranchStatus.java
index f7f9737..4f69189 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/MergeBranchStatus.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/MergeBranchStatus.java
@@ -92,4 +92,10 @@ public class MergeBranchStatus implements IRunFileStreamStatus, Serializable
{
         this.runFileReading = runFileReading;
     }
 
+    @Override
+    public String toString() {
+        return "Branch status is " + stage + ": the stream " + (hasMore ? "has more" : "is
empty")
+                + " and the run file is " + (runFileWriting ? "WRITING " : "") + (runFileReading
? "READING " : "");
+
+    }
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/199bddd3/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/MergeJoinOperatorDescriptor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/MergeJoinOperatorDescriptor.java
b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/MergeJoinOperatorDescriptor.java
index 24abddf..649247e 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/MergeJoinOperatorDescriptor.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/MergeJoinOperatorDescriptor.java
@@ -260,6 +260,7 @@ public class MergeJoinOperatorDescriptor extends AbstractOperatorDescriptor
{
                         locks.getRight(partition).await();
                     }
                     state.joiner.setFrame(RIGHT_ACTIVITY_ID, buffer);
+                    state.status.continueRightLoad = false;
                     locks.getLeft(partition).signal();
                 } catch (InterruptedException e) {
                     Thread.currentThread().interrupt();


Mime
View raw message