You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@drill.apache.org by amithadke <gi...@git.apache.org> on 2015/10/08 03:26:06 UTC
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
GitHub user amithadke opened a pull request:
https://github.com/apache/drill/pull/190
Initial version of RecordIterator and MergeJoin rewrite.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/amithadke/drill mergejoin-rewrite
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/drill/pull/190.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #190
----
commit b92c56c3739bd416aca3609edc696527c723e3db
Author: Amit Hadke <am...@gmail.com>
Date: 2015-10-08T01:23:21Z
Initial version of RecordIterator and MergeJoin rewrite.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41717983
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java ---
@@ -403,14 +325,14 @@ private JoinWorker generateNewWorker() throws ClassTransformationException, IOEx
outputType = inputType;
}
JVar vvIn = cg.declareVectorValueSetupAndMember("incomingLeft",
- new TypedFieldId(inputType, vectorId));
+ new TypedFieldId(inputType, vectorId));
--- End diff --
Can you look if we can factor this out? I believe we have a utility method that builds vector to vector copies rather than by doing it manually.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41717923
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinTemplate.java ---
@@ -87,125 +42,68 @@ public void setupJoin(FragmentContext context, JoinStatus status, VectorContaine
*/
public final boolean doJoin(final JoinStatus status) {
while(!status.isOutgoingBatchFull()) {
- // for each record
- // validate input iterators (advancing to the next record batch if necessary)
- if (!status.isRightPositionAllowed()) {
+ if (status.rightFinished()) {
if (((MergeJoinPOP)status.outputBatch.getPopConfig()).getJoinType() == JoinRelType.LEFT) {
--- End diff --
let's grab this property at the beginning of the join rather than each time we doJoin.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41717952
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java ---
@@ -216,34 +206,34 @@ public IterOutcome innerNext() {
// get the outcome of the join.
switch (status.getOutcome()) {
- case BATCH_RETURNED:
- // only return new schema if new worker has been setup.
- logger.debug("BATCH RETURNED; returning {}", (first ? "OK_NEW_SCHEMA" : "OK"));
- setRecordCountInContainer();
- return first ? IterOutcome.OK_NEW_SCHEMA : IterOutcome.OK;
- case FAILURE:
- kill(false);
- return IterOutcome.STOP;
- case NO_MORE_DATA:
- logger.debug("NO MORE DATA; returning {}", (status.getOutPosition() > 0 ? (first ? "OK_NEW_SCHEMA" : "OK") : (first ? "OK_NEW_SCHEMA" :"NONE")));
- setRecordCountInContainer();
- state = BatchState.DONE;
- return status.getOutPosition() > 0 ? (first ? IterOutcome.OK_NEW_SCHEMA : IterOutcome.OK): (first ? IterOutcome.OK_NEW_SCHEMA : IterOutcome.NONE);
- case SCHEMA_CHANGED:
- worker = null;
- if (status.getOutPosition() > 0) {
- // if we have current data, let's return that.
- logger.debug("SCHEMA CHANGED; returning {} ", (first ? "OK_NEW_SCHEMA" : "OK"));
+ case BATCH_RETURNED:
+ // only return new schema if new worker has been setup.
+ logger.debug("BATCH RETURNED; returning {}", (first ? "OK_NEW_SCHEMA" : "OK"));
setRecordCountInContainer();
return first ? IterOutcome.OK_NEW_SCHEMA : IterOutcome.OK;
- }else{
- // loop again to rebuild worker.
- continue;
- }
- case WAITING:
- return IterOutcome.NOT_YET;
- default:
- throw new IllegalStateException();
+ case FAILURE:
+ kill(false);
+ return IterOutcome.STOP;
+ case NO_MORE_DATA:
--- End diff --
This seems like a location where we need to make sure we deplete the incoming iterators.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by mehant <gi...@git.apache.org>.
Github user mehant commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41701799
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java ---
@@ -0,0 +1,252 @@
+/*
+ * 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.drill.exec.record;
+
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.record.RecordBatch.IterOutcome;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.Lists;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Range;
+import com.google.common.collect.Sets;
+import com.google.common.collect.TreeRangeMap;
+
+
+public class RecordIterator implements VectorAccessible {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RecordIterator.class);
+
+ private final RecordBatch incoming;
+ private final AbstractRecordBatch<?> outgoing;
+ private int outerPosition; // Tracks total records consumed so far works across batches.
--- End diff --
should this be a long instead of int?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by amithadke <gi...@git.apache.org>.
Github user amithadke commented on the pull request:
https://github.com/apache/drill/pull/190#issuecomment-146626167
@jacques-n @StevenMPhillips @jaltekruse
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41717876
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinStatus.java ---
@@ -81,114 +58,82 @@ public JoinStatus(RecordBatch left, RecordBatch right, MergeJoinBatch output) {
@Override
public String toString() {
return
- super.toString()
+ super.toString()
+ "["
- + "leftPosition = " + leftPosition
- + ", rightPosition = " + rightPosition
- + ", svRightPosition = " + svRightPosition
+ + "leftPosition = " + left.getCurrentPosition()
+ + ", rightPosition = " + right.getCurrentPosition()
+ ", outputPosition = " + outputPosition
- + ", lastLeft = " + lastLeft
- + ", lastRight = " + lastRight
- + ", rightSourceMode = " + rightSourceMode
- + ", sv4 = " + sv4
+ ", joinType = " + joinType
+ ", ok = " + ok
+ ", initialSet = " + initialSet
- + ", leftRepeating = " + leftRepeating
+ ", left = " + left
+ ", right = " + right
+ ", outputBatch = " + outputBatch
+ "]";
}
- public boolean hasIntermediateData() {
- return hasIntermediateData;
- }
-
- public void resetIntermediateData() {
- hasIntermediateData = false;
- }
-
- public void setIntermediateData(int initialRightPosition, boolean crossedBatchBoundaries) {
- this.initialRightPosition = initialRightPosition;
- this.crossedBatchBoundaries = crossedBatchBoundaries;
- this.hasIntermediateData = true;
- }
-
- public int getInitialRightPosition() {
- return initialRightPosition;
- }
-
- public boolean getCrossedBatchBoundaries() {
- return crossedBatchBoundaries;
- }
-
- private final IterOutcome nextLeft() {
- return outputBatch.next(LEFT_INPUT, left);
- }
-
- private final IterOutcome nextRight() {
- return outputBatch.next(RIGHT_INPUT, right);
- }
public final void ensureInitial() {
switch(initialSet) {
case INIT:
- this.lastLeft = nextLeft();
- this.lastRight = nextRight();
+ advanceLeft();
+ advanceRight();
initialSet = InitState.CHECK;
break;
case CHECK:
- if (lastLeft != IterOutcome.NONE && left.getRecordCount() == 0) {
- this.lastLeft = nextLeft();
+ if (getLeftStatus() != IterOutcome.NONE && left.getInnerRecordCount() == 0) {
--- End diff --
can you add comments on this? Both the method and these checks. What happens if the iteroutcome is something like STOP?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41717996
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/record/HyperVectorWrapper.java ---
@@ -159,4 +159,14 @@ public void addVectors(ValueVector[] vv) {
vectors = (T[]) ArrayUtils.add(vectors, vv);
}
+ public void transfer(VectorWrapper<?> destination) {
--- End diff --
Javadocs
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by mehant <gi...@git.apache.org>.
Github user mehant commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41701806
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java ---
@@ -0,0 +1,252 @@
+/*
+ * 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.drill.exec.record;
+
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.record.RecordBatch.IterOutcome;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.Lists;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Range;
+import com.google.common.collect.Sets;
+import com.google.common.collect.TreeRangeMap;
+
+
+public class RecordIterator implements VectorAccessible {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RecordIterator.class);
+
+ private final RecordBatch incoming;
+ private final AbstractRecordBatch<?> outgoing;
+ private int outerPosition; // Tracks total records consumed so far works across batches.
+ private int innerPosition; // Index within current vector container.
+ private int innerRecordCount; // Records in current container.
+ private int totalRecordCount; // Total records read so far.
+ private int startBatchPosition; // Start offset of current batch.
+ private int markedOuterPosition;
+ private int markedInnerPosition;
+ private IterOutcome lastOutcome;
+ private int inputIndex; // For two way merge join 0:left, 1:right
+ private boolean lastBatchRead; // True if all batches are consumed.
+ private boolean initialized;
+
+
+ private VectorContainer container; // Holds VectorContainer of current record batch
+ TreeRangeMap<Integer, RecordBatchData> batches = TreeRangeMap.create();
+
+ public RecordIterator(RecordBatch incoming,
+ AbstractRecordBatch<?> outgoing,
+ OperatorContext oContext,
+ int inputIndex) {
+ this.incoming = incoming;
+ this.innerPosition = this.outerPosition = this.startBatchPosition = -1;
+ this.totalRecordCount = this.innerRecordCount = 0;
+ this.markedInnerPosition = this.markedOuterPosition = -1;
+ this.outgoing = outgoing;
+ this.inputIndex = inputIndex;
+ this.lastBatchRead = false;
+ this.container = new VectorContainer(oContext);
+ this.initialized = false;
+ }
+
+ // Get next record batch.
+ private void nextBatch() {
+ // We have already seen last batch.
+ if (lastBatchRead) {
+ return;
+ }
+ lastOutcome = outgoing.next(inputIndex, incoming);
+ }
+
+ public void mark() {
+ // Release all batches before current batch. [0 to startBatchPosition).
+ final Map<Range<Integer>,RecordBatchData> oldBatches = batches.subRangeMap(
+ Range.closedOpen(0, startBatchPosition)).asMapOfRanges();
+ for (Range<Integer> range : oldBatches.keySet()) {
+ oldBatches.get(range.lowerEndpoint()).clear();
+ }
+ batches.remove(Range.closedOpen(0, startBatchPosition));
+ markedInnerPosition = innerPosition;
+ markedOuterPosition = outerPosition;
+ }
+
+ public void reset() {
+ if (markedOuterPosition >= 0) {
+ // Move to rbd for markedOuterPosition.
+ final RecordBatchData rbdNew = batches.get(markedOuterPosition);
+ final RecordBatchData rbdOld = batches.get(startBatchPosition);
+ Preconditions.checkArgument(rbdOld != null);
+ Preconditions.checkArgument(rbdNew != null);
+ if (rbdNew != rbdOld) {
+ container.transferOut(rbdOld.getContainer());
+ container.transferIn(rbdNew.getContainer());
+ }
+ innerPosition = markedInnerPosition;
+ outerPosition = markedOuterPosition;
+ startBatchPosition = batches.getEntry(outerPosition).getKey().lowerEndpoint();
+ innerRecordCount = batches.getEntry(outerPosition).getKey().upperEndpoint() - startBatchPosition;
+ markedInnerPosition = markedOuterPosition = -1;
+ }
+ }
+
+ // TODO: Experimental: Move forward by delta (may cross one or more record batches)
+ public void forward(int delta) {
+ Preconditions.checkArgument(delta >= 0);
+ Preconditions.checkArgument(delta + outerPosition < totalRecordCount);
+ final int nextOuterPosition = delta + outerPosition;
+ final RecordBatchData rbdNew = batches.get(nextOuterPosition);
+ final RecordBatchData rbdOld = batches.get(outerPosition);
+ Preconditions.checkArgument(rbdNew != null);
+ Preconditions.checkArgument(rbdOld != null);
+ container.transferOut(rbdOld.getContainer());
+ // Get vectors from new position.
+ container.transferIn(rbdNew.getContainer());
+ outerPosition = nextOuterPosition;
+ startBatchPosition = batches.getEntry(outerPosition).getKey().lowerEndpoint();
+ innerPosition = outerPosition - startBatchPosition;
+ innerRecordCount = batches.getEntry(outerPosition).getKey().upperEndpoint() - startBatchPosition;
+ }
+
+ public IterOutcome next() {
--- End diff --
javadocs.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r43559169
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java ---
@@ -403,14 +325,14 @@ private JoinWorker generateNewWorker() throws ClassTransformationException, IOEx
outputType = inputType;
}
JVar vvIn = cg.declareVectorValueSetupAndMember("incomingLeft",
- new TypedFieldId(inputType, vectorId));
+ new TypedFieldId(inputType, vectorId));
--- End diff --
Probably. I didn't check the code but that sounds like the right thing.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by mehant <gi...@git.apache.org>.
Github user mehant commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41701556
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java ---
@@ -0,0 +1,252 @@
+/*
+ * 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.drill.exec.record;
+
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.record.RecordBatch.IterOutcome;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.Lists;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Range;
+import com.google.common.collect.Sets;
+import com.google.common.collect.TreeRangeMap;
+
+
+public class RecordIterator implements VectorAccessible {
--- End diff --
RecordIterator seems to provide two useful abstractions
1. Provide a abstraction to iterate over records (hiding away details about batch boundaries etc)
2. Provide an abstraction to be able to mark a record and go back and reiterate from that record.
What do you think about building the second abstraction on top of the first (MarkAndResetIterator and RecordIterator)? I feel the first abstraction is very generic and can be used in most operators cleaning out some of the redundant code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/drill/pull/190
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on the pull request:
https://github.com/apache/drill/pull/190#issuecomment-147266034
Can you add some tests for the iterator? For example, I want to make sure we're not holding onto batches longer than is necessary, that having schema changes or weird iteroutcomes works well, etc.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by mehant <gi...@git.apache.org>.
Github user mehant commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41701797
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java ---
@@ -0,0 +1,252 @@
+/*
+ * 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.drill.exec.record;
+
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.record.RecordBatch.IterOutcome;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.Lists;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Range;
+import com.google.common.collect.Sets;
+import com.google.common.collect.TreeRangeMap;
+
+
+public class RecordIterator implements VectorAccessible {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RecordIterator.class);
+
+ private final RecordBatch incoming;
+ private final AbstractRecordBatch<?> outgoing;
+ private int outerPosition; // Tracks total records consumed so far works across batches.
+ private int innerPosition; // Index within current vector container.
+ private int innerRecordCount; // Records in current container.
+ private int totalRecordCount; // Total records read so far.
+ private int startBatchPosition; // Start offset of current batch.
+ private int markedOuterPosition;
+ private int markedInnerPosition;
+ private IterOutcome lastOutcome;
+ private int inputIndex; // For two way merge join 0:left, 1:right
+ private boolean lastBatchRead; // True if all batches are consumed.
+ private boolean initialized;
+
+
+ private VectorContainer container; // Holds VectorContainer of current record batch
+ TreeRangeMap<Integer, RecordBatchData> batches = TreeRangeMap.create();
--- End diff --
private final.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41717893
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinStatus.java ---
@@ -81,114 +58,82 @@ public JoinStatus(RecordBatch left, RecordBatch right, MergeJoinBatch output) {
@Override
public String toString() {
return
- super.toString()
+ super.toString()
+ "["
- + "leftPosition = " + leftPosition
- + ", rightPosition = " + rightPosition
- + ", svRightPosition = " + svRightPosition
+ + "leftPosition = " + left.getCurrentPosition()
+ + ", rightPosition = " + right.getCurrentPosition()
+ ", outputPosition = " + outputPosition
- + ", lastLeft = " + lastLeft
- + ", lastRight = " + lastRight
- + ", rightSourceMode = " + rightSourceMode
- + ", sv4 = " + sv4
+ ", joinType = " + joinType
+ ", ok = " + ok
+ ", initialSet = " + initialSet
- + ", leftRepeating = " + leftRepeating
+ ", left = " + left
+ ", right = " + right
+ ", outputBatch = " + outputBatch
+ "]";
}
- public boolean hasIntermediateData() {
- return hasIntermediateData;
- }
-
- public void resetIntermediateData() {
- hasIntermediateData = false;
- }
-
- public void setIntermediateData(int initialRightPosition, boolean crossedBatchBoundaries) {
- this.initialRightPosition = initialRightPosition;
- this.crossedBatchBoundaries = crossedBatchBoundaries;
- this.hasIntermediateData = true;
- }
-
- public int getInitialRightPosition() {
- return initialRightPosition;
- }
-
- public boolean getCrossedBatchBoundaries() {
- return crossedBatchBoundaries;
- }
-
- private final IterOutcome nextLeft() {
- return outputBatch.next(LEFT_INPUT, left);
- }
-
- private final IterOutcome nextRight() {
- return outputBatch.next(RIGHT_INPUT, right);
- }
public final void ensureInitial() {
switch(initialSet) {
case INIT:
- this.lastLeft = nextLeft();
- this.lastRight = nextRight();
+ advanceLeft();
+ advanceRight();
initialSet = InitState.CHECK;
break;
case CHECK:
- if (lastLeft != IterOutcome.NONE && left.getRecordCount() == 0) {
- this.lastLeft = nextLeft();
+ if (getLeftStatus() != IterOutcome.NONE && left.getInnerRecordCount() == 0) {
+ advanceLeft();
}
- if (lastRight != IterOutcome.NONE && right.getRecordCount() == 0) {
- this.lastRight = nextRight();
+ if (getRightStatus() != IterOutcome.NONE && right.getInnerRecordCount() == 0) {
+ advanceRight();
}
initialSet = InitState.READY;
- // fall through
+ break;
default:
break;
}
}
public final void advanceLeft() {
- leftPosition++;
+ left.next();
--- End diff --
My movement, I mean iterator movement.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by amithadke <gi...@git.apache.org>.
Github user amithadke commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r43549769
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java ---
@@ -403,14 +325,14 @@ private JoinWorker generateNewWorker() throws ClassTransformationException, IOEx
outputType = inputType;
}
JVar vvIn = cg.declareVectorValueSetupAndMember("incomingLeft",
- new TypedFieldId(inputType, vectorId));
+ new TypedFieldId(inputType, vectorId));
--- End diff --
@jacques-n did you mean generateCopies from CopyUtil?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41717849
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java ---
@@ -0,0 +1,252 @@
+/*
+ * 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.drill.exec.record;
+
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.record.RecordBatch.IterOutcome;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.Lists;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Range;
+import com.google.common.collect.Sets;
+import com.google.common.collect.TreeRangeMap;
+
+
+public class RecordIterator implements VectorAccessible {
--- End diff --
Wouldn't that be the same as the current RecordIterator except that mark and reset are unused? Not sure what the benefit of the simpler one is. I guess the code is little simpler. On the flipside, it seems like it would have to be a separate implementation since we don't need to transfer in the Simple one.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41718052
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java ---
@@ -0,0 +1,252 @@
+/*
+ * 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.drill.exec.record;
+
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.record.RecordBatch.IterOutcome;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.Lists;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Range;
+import com.google.common.collect.Sets;
+import com.google.common.collect.TreeRangeMap;
+
+
+public class RecordIterator implements VectorAccessible {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RecordIterator.class);
+
+ private final RecordBatch incoming;
+ private final AbstractRecordBatch<?> outgoing;
+ private int outerPosition; // Tracks total records consumed so far works across batches.
+ private int innerPosition; // Index within current vector container.
+ private int innerRecordCount; // Records in current container.
+ private int totalRecordCount; // Total records read so far.
+ private int startBatchPosition; // Start offset of current batch.
+ private int markedOuterPosition;
+ private int markedInnerPosition;
+ private IterOutcome lastOutcome;
+ private int inputIndex; // For two way merge join 0:left, 1:right
+ private boolean lastBatchRead; // True if all batches are consumed.
+ private boolean initialized;
+
+
+ private VectorContainer container; // Holds VectorContainer of current record batch
+ TreeRangeMap<Integer, RecordBatchData> batches = TreeRangeMap.create();
+
+ public RecordIterator(RecordBatch incoming,
+ AbstractRecordBatch<?> outgoing,
+ OperatorContext oContext,
+ int inputIndex) {
+ this.incoming = incoming;
+ this.innerPosition = this.outerPosition = this.startBatchPosition = -1;
+ this.totalRecordCount = this.innerRecordCount = 0;
+ this.markedInnerPosition = this.markedOuterPosition = -1;
+ this.outgoing = outgoing;
+ this.inputIndex = inputIndex;
+ this.lastBatchRead = false;
+ this.container = new VectorContainer(oContext);
+ this.initialized = false;
+ }
+
+ // Get next record batch.
+ private void nextBatch() {
+ // We have already seen last batch.
+ if (lastBatchRead) {
+ return;
+ }
+ lastOutcome = outgoing.next(inputIndex, incoming);
+ }
+
+ public void mark() {
+ // Release all batches before current batch. [0 to startBatchPosition).
+ final Map<Range<Integer>,RecordBatchData> oldBatches = batches.subRangeMap(
+ Range.closedOpen(0, startBatchPosition)).asMapOfRanges();
+ for (Range<Integer> range : oldBatches.keySet()) {
+ oldBatches.get(range.lowerEndpoint()).clear();
+ }
+ batches.remove(Range.closedOpen(0, startBatchPosition));
+ markedInnerPosition = innerPosition;
+ markedOuterPosition = outerPosition;
+ }
+
+ public void reset() {
+ if (markedOuterPosition >= 0) {
+ // Move to rbd for markedOuterPosition.
+ final RecordBatchData rbdNew = batches.get(markedOuterPosition);
+ final RecordBatchData rbdOld = batches.get(startBatchPosition);
+ Preconditions.checkArgument(rbdOld != null);
+ Preconditions.checkArgument(rbdNew != null);
+ if (rbdNew != rbdOld) {
+ container.transferOut(rbdOld.getContainer());
+ container.transferIn(rbdNew.getContainer());
+ }
+ innerPosition = markedInnerPosition;
+ outerPosition = markedOuterPosition;
+ startBatchPosition = batches.getEntry(outerPosition).getKey().lowerEndpoint();
+ innerRecordCount = batches.getEntry(outerPosition).getKey().upperEndpoint() - startBatchPosition;
+ markedInnerPosition = markedOuterPosition = -1;
+ }
+ }
+
+ // TODO: Experimental: Move forward by delta (may cross one or more record batches)
+ public void forward(int delta) {
+ Preconditions.checkArgument(delta >= 0);
+ Preconditions.checkArgument(delta + outerPosition < totalRecordCount);
+ final int nextOuterPosition = delta + outerPosition;
+ final RecordBatchData rbdNew = batches.get(nextOuterPosition);
+ final RecordBatchData rbdOld = batches.get(outerPosition);
+ Preconditions.checkArgument(rbdNew != null);
+ Preconditions.checkArgument(rbdOld != null);
+ container.transferOut(rbdOld.getContainer());
+ // Get vectors from new position.
+ container.transferIn(rbdNew.getContainer());
+ outerPosition = nextOuterPosition;
+ startBatchPosition = batches.getEntry(outerPosition).getKey().lowerEndpoint();
+ innerPosition = outerPosition - startBatchPosition;
+ innerRecordCount = batches.getEntry(outerPosition).getKey().upperEndpoint() - startBatchPosition;
+ }
+
+ public IterOutcome next() {
+ if (finished()) {
+ return lastOutcome;
+ }
+ final int nextOuterPosition = outerPosition + 1;
+ final int nextInnerPosition = innerPosition + 1;
+ if (!initialized || nextOuterPosition >= totalRecordCount) {
+ nextBatch();
+ switch (lastOutcome) {
+ case NONE:
+ case STOP:
+ // No more data, disallow reads unless reset is called.
+ outerPosition = nextOuterPosition;
+ lastBatchRead = true;
+ break;
+ case OK_NEW_SCHEMA:
+ case OK:
+ // Transfer vectors from incoming record batch.
+ final RecordBatchData rbd = new RecordBatchData(incoming);
+ innerRecordCount = incoming.getRecordCount();
+ if (!initialized) {
+ for (VectorWrapper<?> w : rbd.getContainer()) {
+ container.addOrGet(w.getField());
+ }
+ container.buildSchema(rbd.getContainer().getSchema().getSelectionVectorMode());
+ initialized = true;
+ }
+ if (innerRecordCount > 0) {
+ // Transfer vectors back to old batch.
+ if (startBatchPosition != -1 && batches.get(startBatchPosition) != null) {
+ container.transferOut(batches.get(outerPosition).getContainer());
+ }
+ container.transferIn(rbd.getContainer());
+ startBatchPosition = nextOuterPosition;
+ batches.put(Range.closedOpen(nextOuterPosition, nextOuterPosition + innerRecordCount), rbd);
+ innerPosition = 0;
+ outerPosition = nextOuterPosition;
+ totalRecordCount += innerRecordCount;
+ } else {
+ // Release schema/empty batches.
+ rbd.clear();
+ }
+ break;
+ default:
+ // On errors return lastOutcome.
--- End diff --
Shouldn't this through UnsupportedOperationException?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41717886
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinStatus.java ---
@@ -81,114 +58,82 @@ public JoinStatus(RecordBatch left, RecordBatch right, MergeJoinBatch output) {
@Override
public String toString() {
return
- super.toString()
+ super.toString()
+ "["
- + "leftPosition = " + leftPosition
- + ", rightPosition = " + rightPosition
- + ", svRightPosition = " + svRightPosition
+ + "leftPosition = " + left.getCurrentPosition()
+ + ", rightPosition = " + right.getCurrentPosition()
+ ", outputPosition = " + outputPosition
- + ", lastLeft = " + lastLeft
- + ", lastRight = " + lastRight
- + ", rightSourceMode = " + rightSourceMode
- + ", sv4 = " + sv4
+ ", joinType = " + joinType
+ ", ok = " + ok
+ ", initialSet = " + initialSet
- + ", leftRepeating = " + leftRepeating
+ ", left = " + left
+ ", right = " + right
+ ", outputBatch = " + outputBatch
+ "]";
}
- public boolean hasIntermediateData() {
- return hasIntermediateData;
- }
-
- public void resetIntermediateData() {
- hasIntermediateData = false;
- }
-
- public void setIntermediateData(int initialRightPosition, boolean crossedBatchBoundaries) {
- this.initialRightPosition = initialRightPosition;
- this.crossedBatchBoundaries = crossedBatchBoundaries;
- this.hasIntermediateData = true;
- }
-
- public int getInitialRightPosition() {
- return initialRightPosition;
- }
-
- public boolean getCrossedBatchBoundaries() {
- return crossedBatchBoundaries;
- }
-
- private final IterOutcome nextLeft() {
- return outputBatch.next(LEFT_INPUT, left);
- }
-
- private final IterOutcome nextRight() {
- return outputBatch.next(RIGHT_INPUT, right);
- }
public final void ensureInitial() {
switch(initialSet) {
case INIT:
- this.lastLeft = nextLeft();
- this.lastRight = nextRight();
+ advanceLeft();
+ advanceRight();
initialSet = InitState.CHECK;
break;
case CHECK:
- if (lastLeft != IterOutcome.NONE && left.getRecordCount() == 0) {
- this.lastLeft = nextLeft();
+ if (getLeftStatus() != IterOutcome.NONE && left.getInnerRecordCount() == 0) {
+ advanceLeft();
}
- if (lastRight != IterOutcome.NONE && right.getRecordCount() == 0) {
- this.lastRight = nextRight();
+ if (getRightStatus() != IterOutcome.NONE && right.getInnerRecordCount() == 0) {
+ advanceRight();
}
initialSet = InitState.READY;
- // fall through
+ break;
default:
break;
}
}
public final void advanceLeft() {
- leftPosition++;
+ left.next();
--- End diff --
Can you movement back into the template. I can't see what benefit it has in the JoinStatus.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41718038
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java ---
@@ -0,0 +1,252 @@
+/*
+ * 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.drill.exec.record;
+
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.record.RecordBatch.IterOutcome;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.Lists;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Range;
+import com.google.common.collect.Sets;
+import com.google.common.collect.TreeRangeMap;
+
+
+public class RecordIterator implements VectorAccessible {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RecordIterator.class);
+
+ private final RecordBatch incoming;
+ private final AbstractRecordBatch<?> outgoing;
+ private int outerPosition; // Tracks total records consumed so far works across batches.
+ private int innerPosition; // Index within current vector container.
+ private int innerRecordCount; // Records in current container.
+ private int totalRecordCount; // Total records read so far.
+ private int startBatchPosition; // Start offset of current batch.
+ private int markedOuterPosition;
+ private int markedInnerPosition;
+ private IterOutcome lastOutcome;
+ private int inputIndex; // For two way merge join 0:left, 1:right
+ private boolean lastBatchRead; // True if all batches are consumed.
+ private boolean initialized;
+
+
+ private VectorContainer container; // Holds VectorContainer of current record batch
+ TreeRangeMap<Integer, RecordBatchData> batches = TreeRangeMap.create();
+
+ public RecordIterator(RecordBatch incoming,
+ AbstractRecordBatch<?> outgoing,
+ OperatorContext oContext,
+ int inputIndex) {
+ this.incoming = incoming;
+ this.innerPosition = this.outerPosition = this.startBatchPosition = -1;
+ this.totalRecordCount = this.innerRecordCount = 0;
+ this.markedInnerPosition = this.markedOuterPosition = -1;
+ this.outgoing = outgoing;
+ this.inputIndex = inputIndex;
+ this.lastBatchRead = false;
+ this.container = new VectorContainer(oContext);
+ this.initialized = false;
+ }
+
+ // Get next record batch.
+ private void nextBatch() {
+ // We have already seen last batch.
+ if (lastBatchRead) {
+ return;
+ }
+ lastOutcome = outgoing.next(inputIndex, incoming);
+ }
+
+ public void mark() {
+ // Release all batches before current batch. [0 to startBatchPosition).
+ final Map<Range<Integer>,RecordBatchData> oldBatches = batches.subRangeMap(
+ Range.closedOpen(0, startBatchPosition)).asMapOfRanges();
+ for (Range<Integer> range : oldBatches.keySet()) {
+ oldBatches.get(range.lowerEndpoint()).clear();
+ }
+ batches.remove(Range.closedOpen(0, startBatchPosition));
+ markedInnerPosition = innerPosition;
+ markedOuterPosition = outerPosition;
+ }
+
+ public void reset() {
+ if (markedOuterPosition >= 0) {
+ // Move to rbd for markedOuterPosition.
+ final RecordBatchData rbdNew = batches.get(markedOuterPosition);
+ final RecordBatchData rbdOld = batches.get(startBatchPosition);
+ Preconditions.checkArgument(rbdOld != null);
+ Preconditions.checkArgument(rbdNew != null);
+ if (rbdNew != rbdOld) {
+ container.transferOut(rbdOld.getContainer());
+ container.transferIn(rbdNew.getContainer());
+ }
+ innerPosition = markedInnerPosition;
+ outerPosition = markedOuterPosition;
+ startBatchPosition = batches.getEntry(outerPosition).getKey().lowerEndpoint();
+ innerRecordCount = batches.getEntry(outerPosition).getKey().upperEndpoint() - startBatchPosition;
+ markedInnerPosition = markedOuterPosition = -1;
+ }
+ }
+
+ // TODO: Experimental: Move forward by delta (may cross one or more record batches)
+ public void forward(int delta) {
+ Preconditions.checkArgument(delta >= 0);
+ Preconditions.checkArgument(delta + outerPosition < totalRecordCount);
+ final int nextOuterPosition = delta + outerPosition;
+ final RecordBatchData rbdNew = batches.get(nextOuterPosition);
+ final RecordBatchData rbdOld = batches.get(outerPosition);
+ Preconditions.checkArgument(rbdNew != null);
+ Preconditions.checkArgument(rbdOld != null);
+ container.transferOut(rbdOld.getContainer());
+ // Get vectors from new position.
+ container.transferIn(rbdNew.getContainer());
+ outerPosition = nextOuterPosition;
+ startBatchPosition = batches.getEntry(outerPosition).getKey().lowerEndpoint();
+ innerPosition = outerPosition - startBatchPosition;
+ innerRecordCount = batches.getEntry(outerPosition).getKey().upperEndpoint() - startBatchPosition;
+ }
+
+ public IterOutcome next() {
+ if (finished()) {
+ return lastOutcome;
+ }
+ final int nextOuterPosition = outerPosition + 1;
+ final int nextInnerPosition = innerPosition + 1;
+ if (!initialized || nextOuterPosition >= totalRecordCount) {
+ nextBatch();
+ switch (lastOutcome) {
+ case NONE:
+ case STOP:
+ // No more data, disallow reads unless reset is called.
+ outerPosition = nextOuterPosition;
+ lastBatchRead = true;
+ break;
+ case OK_NEW_SCHEMA:
--- End diff --
In the case of ok new schema, you should remove vectors that have been removed from upstream.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: Initial version of RecordIterator and MergeJoi...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on a diff in the pull request:
https://github.com/apache/drill/pull/190#discussion_r41717967
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java ---
@@ -255,84 +245,20 @@ private void setRecordCountInContainer() {
}
}
- public void resetBatchBuilder() {
- batchBuilder.close();
- batchBuilder = new MergeJoinBatchBuilder(oContext.getAllocator(), status);
- }
-
- public void addRightToBatchBuilder() {
- batchBuilder.add(right);
+ @Override
+ public void close() {
+ super.close();
+ leftIterator.close();
+ rightIterator.close();
}
@Override
protected void killIncoming(boolean sendUpstream) {
+ close();
--- End diff --
You shouldn't call close here. I believe that should come from the FragmentExecutor.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---