You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by pa...@apache.org on 2018/04/23 14:15:52 UTC
[6/8] drill git commit: DRILL-6324: Unnest - Initial Implementation
DRILL-6324: Unnest - Initial Implementation
- Based on Flatten
- Implement unnestRecords in UnnestTemplate
- Remove unnecessary code inherited from Flatten/Project. Add schema change handling.
- Fix build failure after rebase since RecordBatchSizer used by UNNEST was relocated to a different package
- Add unit tests
- Handling of input row splitting across multiple batches. Also do not kill incoming in killIncoming.
- Schema change generated by Unnest
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/51bf1dd3
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/51bf1dd3
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/51bf1dd3
Branch: refs/heads/master
Commit: 51bf1dd3f6f5054e1922f0b22b0364ef1c0c6e1f
Parents: 69ec019
Author: Parth Chandra <pa...@apache.org>
Authored: Thu Feb 8 10:43:13 2018 +0530
Committer: Parth Chandra <pa...@apache.org>
Committed: Sun Apr 22 22:20:48 2018 -0700
----------------------------------------------------------------------
.../physical/base/AbstractPhysicalVisitor.java | 6 +
.../exec/physical/base/PhysicalVisitor.java | 2 +
.../drill/exec/physical/config/UnnestPOP.java | 90 +++
.../drill/exec/physical/impl/unnest/Unnest.java | 44 ++
.../impl/unnest/UnnestBatchCreator.java | 38 +
.../exec/physical/impl/unnest/UnnestImpl.java | 134 ++++
.../physical/impl/unnest/UnnestRecordBatch.java | 376 ++++++++++
.../exec/store/mock/MockLateralJoinPOP.java | 72 ++
.../exec/physical/impl/MockRecordBatch.java | 6 +-
.../impl/unnest/MockLateralJoinBatch.java | 237 +++++++
.../impl/unnest/TestUnnestCorrectness.java | 691 +++++++++++++++++++
11 files changed, 1693 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java
index 66affeb..86a31c1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java
@@ -34,6 +34,7 @@ import org.apache.drill.exec.physical.config.Sort;
import org.apache.drill.exec.physical.config.StreamingAggregate;
import org.apache.drill.exec.physical.config.Trace;
import org.apache.drill.exec.physical.config.UnionAll;
+import org.apache.drill.exec.physical.config.UnnestPOP;
import org.apache.drill.exec.physical.config.UnorderedReceiver;
import org.apache.drill.exec.physical.config.Values;
import org.apache.drill.exec.physical.config.WindowPOP;
@@ -175,6 +176,11 @@ public abstract class AbstractPhysicalVisitor<T, X, E extends Throwable> impleme
}
@Override
+ public T visitUnnest(UnnestPOP unnest, X value) throws E {
+ return visitOp(unnest, value);
+ }
+
+ @Override
public T visitIteratorValidator(IteratorValidator op, X value) throws E {
return visitOp(op, value);
}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java
index 8480a07..5c926ad 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java
@@ -34,6 +34,7 @@ import org.apache.drill.exec.physical.config.Sort;
import org.apache.drill.exec.physical.config.StreamingAggregate;
import org.apache.drill.exec.physical.config.Trace;
import org.apache.drill.exec.physical.config.UnionAll;
+import org.apache.drill.exec.physical.config.UnnestPOP;
import org.apache.drill.exec.physical.config.UnorderedReceiver;
import org.apache.drill.exec.physical.config.Values;
import org.apache.drill.exec.physical.config.WindowPOP;
@@ -76,6 +77,7 @@ public interface PhysicalVisitor<RETURN, EXTRA, EXCEP extends Throwable> {
public RETURN visitSingleSender(SingleSender op, EXTRA value) throws EXCEP;
public RETURN visitWindowFrame(WindowPOP op, EXTRA value) throws EXCEP;
public RETURN visitProducerConsumer(ProducerConsumer op, EXTRA value) throws EXCEP;
+ public RETURN visitUnnest(UnnestPOP unnest, EXTRA value) throws EXCEP;
public RETURN visitIteratorValidator(IteratorValidator op, EXTRA value) throws EXCEP;
}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnnestPOP.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnnestPOP.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnnestPOP.java
new file mode 100644
index 0000000..2fb2b51
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnnestPOP.java
@@ -0,0 +1,90 @@
+/*******************************************************************************
+ * 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.physical.config;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.google.common.collect.Iterators;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.physical.base.AbstractSingle;
+import org.apache.drill.exec.physical.base.LateralContract;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.PhysicalVisitor;
+import org.apache.drill.exec.proto.beans.CoreOperatorType;
+
+import java.util.Iterator;
+import java.util.List;
+
+import static org.apache.drill.exec.proto.UserBitShared.CoreOperatorType.UNNEST_VALUE;
+import static org.apache.drill.exec.proto.beans.CoreOperatorType.UNNEST;
+
+@JsonTypeName("unnest")
+public class UnnestPOP extends AbstractSingle {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UnnestPOP.class);
+
+ private SchemaPath column;
+
+ @JsonProperty("lateral")
+ private LateralContract lateral ;
+
+ @JsonCreator
+ public UnnestPOP(
+ @JsonProperty("child") PhysicalOperator child, // Operator with incoming record batch
+ @JsonProperty("column") SchemaPath column) {
+ super(child);
+ this.column = column;
+ this.lateral = null;
+ }
+
+
+ @Override
+ public Iterator<PhysicalOperator> iterator() {
+ return Iterators.singletonIterator(child);
+ }
+
+ public SchemaPath getColumn() {
+ return column;
+ }
+
+ @Override
+ public <T, X, E extends Throwable> T accept(PhysicalVisitor<T, X, E> physicalVisitor, X value) throws E {
+ return physicalVisitor.visitUnnest(this, value);
+ }
+
+ @Override
+ public PhysicalOperator getNewWithChild(PhysicalOperator child) {
+ UnnestPOP unnest = new UnnestPOP(child, column);
+ unnest.setLateral(this.getLateral());
+ return unnest;
+ }
+
+ @JsonProperty("lateral")
+ public LateralContract getLateral() {
+ return lateral;
+ }
+
+ public void setLateral(LateralContract l) {
+ this.lateral = l;
+ }
+
+ @Override
+ public int getOperatorType() {
+ return UNNEST_VALUE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/Unnest.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/Unnest.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/Unnest.java
new file mode 100644
index 0000000..6e2e4c4
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/Unnest.java
@@ -0,0 +1,44 @@
+/**
+ * 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.physical.impl.unnest;
+
+import org.apache.drill.exec.compile.TemplateClassDefinition;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.base.LateralContract;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.TransferPair;
+import org.apache.drill.exec.vector.complex.RepeatedValueVector;
+
+import java.util.List;
+
+public interface Unnest {
+ TemplateClassDefinition<Unnest> TEMPLATE_DEFINITION = new TemplateClassDefinition<Unnest>(Unnest.class, UnnestImpl.class);
+
+ void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing, List<TransferPair> transfers,
+ LateralContract lateral) throws SchemaChangeException;
+
+ int unnestRecords(int recordCount, int firstOutputIndex);
+
+ void setUnnestField(RepeatedValueVector repeatedColumn);
+ void setOutputCount(int outputCount);
+
+ RepeatedValueVector getUnnestField();
+
+ void resetGroupIndex();
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestBatchCreator.java
new file mode 100644
index 0000000..5048d29
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestBatchCreator.java
@@ -0,0 +1,38 @@
+/**
+ * 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.physical.impl.unnest;
+
+import com.google.common.base.Preconditions;
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
+import org.apache.drill.exec.physical.config.FlattenPOP;
+import org.apache.drill.exec.physical.config.UnnestPOP;
+import org.apache.drill.exec.physical.impl.BatchCreator;
+import org.apache.drill.exec.physical.impl.flatten.FlattenRecordBatch;
+import org.apache.drill.exec.record.RecordBatch;
+
+import java.util.List;
+
+public class UnnestBatchCreator implements BatchCreator<UnnestPOP> {
+ @Override
+ public UnnestRecordBatch getBatch(ExecutorFragmentContext context, UnnestPOP config, List<RecordBatch> children)
+ throws ExecutionSetupException {
+ Preconditions.checkArgument(children.size() == 1);
+ return new UnnestRecordBatch(config, children.get(0), context);
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestImpl.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestImpl.java
new file mode 100644
index 0000000..d6f7f81
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestImpl.java
@@ -0,0 +1,134 @@
+/*
+ * 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.physical.impl.unnest;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.base.LateralContract;
+import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.TransferPair;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.exec.vector.complex.RepeatedValueVector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ * Contains the actual unnest operation. Unnest is a simple transfer operation in this impelementation.
+ * For use as a table function, we will need to change the logic of the unnest method to operate on
+ * more than one row at a time and remove any dependence on Lateral
+ * {@link org.apache.drill.exec.physical.impl.flatten.FlattenTemplate}.
+ * This class follows the pattern of other operators that generate code at runtime. Normally this class
+ * would be abstract and have placeholders for doSetup and doEval. Unnest however, doesn't require code
+ * generation so we can simply implement the code in a simple class that looks similar to the code gen
+ * templates used by other operators but does not implement the doSetup and doEval methods.
+ */
+public class UnnestImpl implements Unnest {
+ private static final Logger logger = LoggerFactory.getLogger(UnnestImpl.class);
+
+ private static final int OUTPUT_ROW_COUNT = ValueVector.MAX_ROW_COUNT;
+
+ private ImmutableList<TransferPair> transfers;
+ private LateralContract lateral; // corresponding lateral Join (or other operator implementing the Lateral Contract)
+ private SelectionVectorMode svMode;
+ private RepeatedValueVector fieldToUnnest;
+ private RepeatedValueVector.RepeatedAccessor accessor;
+
+ /**
+ * The output batch limit starts at OUTPUT_ROW_COUNT, but may be decreased
+ * if records are found to be large.
+ */
+ private int outputLimit = OUTPUT_ROW_COUNT;
+
+
+ // The index in the unnest column that is being processed.We start at zero and continue until
+ // InnerValueCount is reached or if the batch limit is reached
+ // this allows for groups to be written between batches if we run out of space, for cases where we have finished
+ // a batch on the boundary it will be set to 0
+ private int innerValueIndex = 0;
+
+ @Override
+ public void setUnnestField(RepeatedValueVector unnestField) {
+ this.fieldToUnnest = unnestField;
+ this.accessor = RepeatedValueVector.RepeatedAccessor.class.cast(unnestField.getAccessor());
+ }
+
+ @Override
+ public RepeatedValueVector getUnnestField() {
+ return fieldToUnnest;
+ }
+
+ @Override
+ public void setOutputCount(int outputCount) {
+ outputLimit = outputCount;
+ }
+
+ @Override
+ public final int unnestRecords(final int recordCount, final int firstOutputIndex) {
+ switch (svMode) {
+ case FOUR_BYTE:
+ throw new UnsupportedOperationException("Unnest does not support selection vector inputs.");
+
+ case TWO_BYTE:
+ throw new UnsupportedOperationException("Unnest does not support selection vector inputs.");
+
+ case NONE:
+ if (innerValueIndex == -1) {
+ innerValueIndex = 0;
+ }
+
+ // Current record being processed in the incoming record batch. We could keep
+ // track of it ourselves, but it is better to check with the Lateral Join and get the
+ // current record being processed thru the Lateral Join Contract.
+ final int currentRecord = lateral.getRecordIndex();
+ final int innerValueCount = accessor.getInnerValueCountAt(currentRecord);
+ final int count = Math.min(Math.min(innerValueCount, outputLimit), recordCount);
+
+ for (TransferPair t : transfers) {
+ t.splitAndTransfer(innerValueIndex, count);
+ }
+ innerValueIndex += count;
+ return count;
+
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ @Override public final void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing,
+ List<TransferPair> transfers, LateralContract lateral) throws SchemaChangeException {
+
+ this.svMode = incoming.getSchema().getSelectionVectorMode();
+ switch (svMode) {
+ case FOUR_BYTE:
+ throw new UnsupportedOperationException("Unnest does not support selection vector inputs.");
+ case TWO_BYTE:
+ throw new UnsupportedOperationException("Unnest does not support selection vector inputs.");
+ }
+ this.transfers = ImmutableList.copyOf(transfers);
+ this.lateral = lateral;
+ }
+
+ @Override public void resetGroupIndex() {
+ this.innerValueIndex = 0;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java
new file mode 100644
index 0000000..50dc99c
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java
@@ -0,0 +1,376 @@
+/*
+ * 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.physical.impl.unnest;
+
+import com.google.common.collect.Lists;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.expression.FieldReference;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.logical.data.NamedExpression;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.exception.ClassTransformationException;
+import org.apache.drill.exec.exception.OutOfMemoryException;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.expr.ClassGenerator;
+import org.apache.drill.exec.expr.CodeGenerator;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.base.LateralContract;
+import org.apache.drill.exec.physical.config.UnnestPOP;
+import org.apache.drill.exec.record.AbstractSingleRecordBatch;
+import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
+import org.apache.drill.exec.record.MaterializedField;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.RecordBatchSizer;
+import org.apache.drill.exec.record.TransferPair;
+import org.apache.drill.exec.record.TypedFieldId;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.exec.vector.complex.RepeatedMapVector;
+import org.apache.drill.exec.vector.complex.RepeatedValueVector;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.apache.drill.exec.record.RecordBatch.IterOutcome.OK;
+import static org.apache.drill.exec.record.RecordBatch.IterOutcome.OK_NEW_SCHEMA;
+
+// TODO - handle the case where a user tries to unnest a scalar, should just return the column as is
+public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UnnestRecordBatch.class);
+
+ private Unnest unnest;
+ private boolean hasRemainder = false; // set to true if there is data left over for the current row AND if we want
+ // to keep processing it. Kill may be called by a limit in a subquery that
+ // requires us to stop processing thecurrent row, but not stop processing
+ // the data.
+ // In some cases we need to return a predetermined state from a call to next. These are:
+ // 1) Kill is called due to an error occurring in the processing of the query. IterOutcome should be NONE
+ // 2) Kill is called by LIMIT to stop processing of the current row (This occurs when the LIMIT is part of a subquery
+ // between UNNEST and LATERAL. Iteroutcome should be EMIT
+ // 3) Kill is called by LIMIT downstream from LATERAL. IterOutcome should be NONE
+ private IterOutcome nextState = OK;
+ private int remainderIndex = 0;
+ private int recordCount;
+ private long outputBatchSize;
+ private LateralContract lateral;
+ private MaterializedField unnestFieldMetadata;
+
+
+
+ /**
+ * Memory manager for Unnest. Estimates the batch size exactly like we do for Flatten.
+ */
+ private class UnnestMemoryManager {
+ private final int outputRowCount;
+ private static final int OFFSET_VECTOR_WIDTH = 4;
+ private static final int WORST_CASE_FRAGMENTATION_FACTOR = 1;
+ private static final int MAX_NUM_ROWS = ValueVector.MAX_ROW_COUNT;
+ private static final int MIN_NUM_ROWS = 1;
+
+ private UnnestMemoryManager(RecordBatch incoming, long outputBatchSize, SchemaPath unnestColumn) {
+ // Get sizing information for the batch.
+ RecordBatchSizer sizer = new RecordBatchSizer(incoming);
+
+ final TypedFieldId typedFieldId = incoming.getValueVectorId(unnestColumn);
+ final MaterializedField field = incoming.getSchema().getColumn(typedFieldId.getFieldIds()[0]);
+
+ // Get column size of unnest column.
+ RecordBatchSizer.ColumnSize columnSize = RecordBatchSizer
+ .getColumn(incoming.getValueAccessorById(field.getValueClass(), typedFieldId.getFieldIds()).getValueVector(),
+ field.getName());
+
+ // Average rowWidth of single element in the unnest list.
+ // subtract the offset vector size from column data size.
+ final int avgRowWidthSingleUnnestEntry = RecordBatchSizer
+ .safeDivide(columnSize.netSize - (OFFSET_VECTOR_WIDTH * columnSize.valueCount), columnSize.elementCount);
+
+ // Average rowWidth of outgoing batch.
+ final int avgOutgoingRowWidth = avgRowWidthSingleUnnestEntry;
+
+ // Number of rows in outgoing batch
+ outputRowCount = Math.max(MIN_NUM_ROWS, Math.min(MAX_NUM_ROWS,
+ RecordBatchSizer.safeDivide((outputBatchSize / WORST_CASE_FRAGMENTATION_FACTOR), avgOutgoingRowWidth)));
+
+ logger.debug(
+ "unnest incoming batch sizer : {}, outputBatchSize : {}," + "avgOutgoingRowWidth : {}, outputRowCount : {}",
+ sizer, outputBatchSize, avgOutgoingRowWidth, outputRowCount);
+ }
+
+ public int getOutputRowCount() {
+ return outputRowCount;
+ }
+ }
+
+
+ public UnnestRecordBatch(UnnestPOP pop, RecordBatch incoming, FragmentContext context) throws OutOfMemoryException {
+ super(pop, context, incoming);
+ this.lateral = pop.getLateral();
+ // get the output batch size from config.
+ outputBatchSize = context.getOptions().getOption(ExecConstants.OUTPUT_BATCH_SIZE_VALIDATOR);
+ }
+
+ @Override
+ public int getRecordCount() {
+ return recordCount;
+ }
+
+
+ @Override
+ protected void killIncoming(boolean sendUpstream) {
+ // Do not call kill on incoming. Lateral Join has the responsibility for killing incoming
+ if (context.getExecutorState().isFailed() || lateral.getLeftOutcome() == IterOutcome.STOP) {
+ nextState = IterOutcome.NONE ;
+ } else {
+ // if we have already processed the record, then kill from a limit has no meaning.
+ // if, however, we have values remaining to be emitted, and limit has been reached,
+ // we abandon the remainder and send an empty batch with EMIT.
+ if(hasRemainder) {
+ nextState = IterOutcome.EMIT;
+ }
+ }
+ hasRemainder = false; // whatever the case, we need to stop processing the current row.
+ }
+
+
+ @Override
+ public IterOutcome innerNext() {
+
+ // Short circuit if record batch has already sent all data and is done
+ if (state == BatchState.DONE) {
+ return IterOutcome.NONE;
+ }
+
+ if (nextState == IterOutcome.NONE || nextState == IterOutcome.EMIT) {
+ return nextState;
+ }
+
+ if (hasRemainder) {
+ return handleRemainder();
+ }
+
+ // We do not need to call next() unlike the other operators.
+ // When unnest's innerNext is called, the LateralJoin would have already
+ // updated the incoming vector.
+ // We do, however, need to call doWork() to do the actual work.
+ // We also need to handle schema build if it is the first batch
+
+ if ((state == BatchState.FIRST)) {
+ state = BatchState.NOT_FIRST;
+ try {
+ stats.startSetup();
+ hasRemainder = true; // next call to next will handle the actual data.
+ schemaChanged(); // checks if schema has changed (redundant in this case becaause it has) AND saves the
+ // current field metadata for check in subsequent iterations
+ setupNewSchema();
+ } catch (SchemaChangeException ex) {
+ kill(false);
+ logger.error("Failure during query", ex);
+ context.getExecutorState().fail(ex);
+ return IterOutcome.STOP;
+ } finally {
+ stats.stopSetup();
+ }
+ return IterOutcome.OK_NEW_SCHEMA;
+ } else {
+ assert state != BatchState.FIRST : "First batch should be OK_NEW_SCHEMA";
+ container.zeroVectors();
+
+ // Check if schema has changed
+ if (lateral.getRecordIndex() == 0 && schemaChanged()) {
+ hasRemainder = true; // next call to next will handle the actual data.
+ try {
+ setupNewSchema();
+ } catch (SchemaChangeException ex) {
+ kill(false);
+ logger.error("Failure during query", ex);
+ context.getExecutorState().fail(ex);
+ return IterOutcome.STOP;
+ }
+ return OK_NEW_SCHEMA;
+ }
+ if (lateral.getRecordIndex() == 0) {
+ unnest.resetGroupIndex();
+ }
+ return doWork();
+ }
+
+ }
+
+ @Override
+ public VectorContainer getOutgoingContainer() {
+ return this.container;
+ }
+
+ @SuppressWarnings("resource") private void setUnnestVector() {
+ final TypedFieldId typedFieldId = incoming.getValueVectorId(popConfig.getColumn());
+ final MaterializedField field = incoming.getSchema().getColumn(typedFieldId.getFieldIds()[0]);
+ final RepeatedValueVector vector;
+ final ValueVector inVV =
+ incoming.getValueAccessorById(field.getValueClass(), typedFieldId.getFieldIds()).getValueVector();
+
+ if (!(inVV instanceof RepeatedValueVector)) {
+ if (incoming.getRecordCount() != 0) {
+ throw UserException.unsupportedError().message("Unnest does not support inputs of non-list values.")
+ .build(logger);
+ }
+ // Inherited from FLATTEN. When does this happen???
+ //when incoming recordCount is 0, don't throw exception since the type being seen here is not solid
+ logger.error("setUnnestVector cast failed and recordcount is 0, create empty vector anyway.");
+ vector = new RepeatedMapVector(field, oContext.getAllocator(), null);
+ } else {
+ vector = RepeatedValueVector.class.cast(inVV);
+ }
+ unnest.setUnnestField(vector);
+ }
+
+ @Override
+ protected IterOutcome doWork() {
+ final UnnestMemoryManager unnestMemoryManager = new UnnestMemoryManager(incoming, outputBatchSize,
+ popConfig.getColumn());
+ unnest.setOutputCount(unnestMemoryManager.getOutputRowCount());
+ final int incomingRecordCount = incoming.getRecordCount();
+ final int currentRecord = lateral.getRecordIndex();
+ // we call this in setupSchema, but we also need to call it here so we have a reference to the appropriate vector
+ // inside of the the unnest for the current batch
+ setUnnestVector();
+
+ //expected output count is the num of values in the unnest colum array for the current record
+ final int childCount =
+ incomingRecordCount == 0 ? 0 : unnest.getUnnestField().getAccessor().getInnerValueCountAt(currentRecord);
+
+ // unnest the data
+ final int outputRecords = childCount == 0 ? 0 : unnest.unnestRecords(childCount, 0);
+
+ // Keep track of any spill over into another batch. HAppens only if you artificially set the output batch
+ // size for unnest to a low number
+ if (outputRecords < childCount) {
+ hasRemainder = true;
+ remainderIndex = outputRecords;
+ this.recordCount = remainderIndex;
+ } else {
+ this.recordCount = outputRecords;
+ }
+
+ // If the current incoming record has spilled into two batches, we return
+ // IterOutcome.OK so that the Lateral Join can keep calling next() until the
+ // entire incoming recods has been unnested. If the entire records has been
+ // unnested, we return EMIT and any blocking operators in the pipeline will
+ // unblock.
+ return hasRemainder ? IterOutcome.OK : IterOutcome.EMIT;
+ }
+
+ private IterOutcome handleRemainder() {
+ final UnnestMemoryManager unnestMemoryManager = new UnnestMemoryManager(incoming, outputBatchSize,
+ popConfig.getColumn());
+ unnest.setOutputCount(unnestMemoryManager.getOutputRowCount());
+ final int currentRecord = lateral.getRecordIndex();
+ final int remainingRecordCount =
+ unnest.getUnnestField().getAccessor().getInnerValueCountAt(currentRecord) - remainderIndex;
+ final int projRecords = unnest.unnestRecords(remainingRecordCount, 0);
+ if (projRecords < remainingRecordCount) {
+ this.recordCount = projRecords;
+ this.remainderIndex += projRecords;
+ } else {
+ this.hasRemainder = false;
+ this.remainderIndex = 0;
+ this.recordCount = remainingRecordCount;
+ }
+ return hasRemainder ? IterOutcome.OK : IterOutcome.EMIT;
+ }
+
+ /**
+ * The data layout is the same for the actual data within a repeated field, as it is in a scalar vector for
+ * the same sql type. For example, a repeated int vector has a vector of offsets into a regular int vector to
+ * represent the lists. As the data layout for the actual values in the same in the repeated vector as in the
+ * scalar vector of the same type, we can avoid making individual copies for the column being unnested, and just
+ * use vector copies between the inner vector of the repeated field to the resulting scalar vector from the unnest
+ * operation. This is completed after we determine how many records will fit (as we will hit either a batch end, or
+ * the end of one of the other vectors while we are copying the data of the other vectors alongside each new unnested
+ * value coming out of the repeated field.)
+ */
+ @SuppressWarnings("resource") private TransferPair getUnnestFieldTransferPair(FieldReference reference) {
+ final TypedFieldId fieldId = incoming.getValueVectorId(popConfig.getColumn());
+ final Class<?> vectorClass = incoming.getSchema().getColumn(fieldId.getFieldIds()[0]).getValueClass();
+ final ValueVector unnestField = incoming.getValueAccessorById(vectorClass, fieldId.getFieldIds()).getValueVector();
+
+ TransferPair tp = null;
+ if (unnestField instanceof RepeatedMapVector) {
+ tp = ((RepeatedMapVector) unnestField)
+ .getTransferPairToSingleMap(reference.getAsNamePart().getName(), oContext.getAllocator());
+ } else if (!(unnestField instanceof RepeatedValueVector)) {
+ if (incoming.getRecordCount() != 0) {
+ throw UserException.unsupportedError().message("Unnest does not support inputs of non-list values.")
+ .build(logger);
+ }
+ logger.error("Cannot cast {} to RepeatedValueVector", unnestField);
+ //when incoming recordCount is 0, don't throw exception since the type being seen here is not solid
+ final ValueVector vv = new RepeatedMapVector(unnestField.getField(), oContext.getAllocator(), null);
+ tp = RepeatedValueVector.class.cast(vv)
+ .getTransferPair(reference.getAsNamePart().getName(), oContext.getAllocator());
+ } else {
+ final ValueVector vvIn = RepeatedValueVector.class.cast(unnestField).getDataVector();
+ // vvIn may be null because of fast schema return for repeated list vectors
+ if (vvIn != null) {
+ tp = vvIn.getTransferPair(reference.getAsNamePart().getName(), oContext.getAllocator());
+ }
+ }
+ return tp;
+ }
+
+ @Override protected boolean setupNewSchema() throws SchemaChangeException {
+ container.clear();
+ final List<TransferPair> transfers = Lists.newArrayList();
+
+ final NamedExpression unnestExpr =
+ new NamedExpression(popConfig.getColumn(), new FieldReference(popConfig.getColumn()));
+ final FieldReference fieldReference = unnestExpr.getRef();
+ final TransferPair transferPair = getUnnestFieldTransferPair(fieldReference);
+
+ final ValueVector unnestVector = transferPair.getTo();
+ transfers.add(transferPair);
+ container.add(unnestVector);
+ logger.debug("Added transfer for unnest expression.");
+ container.buildSchema(SelectionVectorMode.NONE);
+
+ this.unnest = new UnnestImpl();
+ unnest.setup(context, incoming, this, transfers, lateral);
+ setUnnestVector();
+ return true;
+ }
+
+ /**
+ * Compares the schema of the unnest column in the current incoming with the schema of
+ * the unnest column in the previous incoming.
+ * Also saves the schema for comparison in future iterations
+ *
+ * @return true if the schema has changed, false otherwise
+ */
+ private boolean schemaChanged() {
+ final TypedFieldId fieldId = incoming.getValueVectorId(popConfig.getColumn());
+ final MaterializedField thisField = incoming.getSchema().getColumn(fieldId.getFieldIds()[0]);
+ final MaterializedField prevField = unnestFieldMetadata;
+ unnestFieldMetadata = thisField;
+ // isEquivalent may return false if the order of the fields has changed. This usually does not
+ // happen but if it does we end up throwing a spurious schema change exeption
+ if (prevField == null || !prevField.isEquivalent(thisField)) {
+ return true;
+ }
+ return false;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/store/mock/MockLateralJoinPOP.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/mock/MockLateralJoinPOP.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/mock/MockLateralJoinPOP.java
new file mode 100644
index 0000000..6cc1b60
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/mock/MockLateralJoinPOP.java
@@ -0,0 +1,72 @@
+/*
+ * 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.store.mock;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.drill.exec.physical.EndpointAffinity;
+import org.apache.drill.exec.physical.base.AbstractStore;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.Store;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+
+import java.util.Collections;
+import java.util.List;
+
+@JsonTypeName("mock-lj")
+public class MockLateralJoinPOP extends AbstractStore {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MockLateralJoinPOP.class);
+
+ @JsonCreator
+ public MockLateralJoinPOP(@JsonProperty("child") PhysicalOperator child) {
+ super(child);
+ }
+
+ @Override
+ public int getMaxWidth() {
+ return 1;
+ }
+
+ @Override
+ public List<EndpointAffinity> getOperatorAffinity() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void applyAssignments(List<DrillbitEndpoint> endpoints) {
+
+ }
+
+ @Override
+ public Store getSpecificStore(PhysicalOperator child, int minorFragmentId) {
+ return new MockLateralJoinPOP(child);
+ }
+
+ @Override
+ protected PhysicalOperator getNewWithChild(PhysicalOperator child) {
+ return new MockLateralJoinPOP(child);
+ }
+
+ @Override
+ public int getOperatorType() {
+ throw new UnsupportedOperationException();
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java
index f9af887..9ed9848 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java
@@ -37,7 +37,7 @@ public class MockRecordBatch implements CloseableRecordBatch {
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MockRecordBatch.class);
// These resources are owned by this RecordBatch
- private VectorContainer container;
+ protected VectorContainer container;
private int currentContainerIndex;
private int currentOutcomeIndex;
private boolean isDone;
@@ -46,8 +46,8 @@ public class MockRecordBatch implements CloseableRecordBatch {
private final List<VectorContainer> allTestContainers;
private final List<IterOutcome> allOutcomes;
private final FragmentContext context;
- private final OperatorContext oContext;
- private final BufferAllocator allocator;
+ protected final OperatorContext oContext;
+ protected final BufferAllocator allocator;
public MockRecordBatch(FragmentContext context, OperatorContext oContext,
List<VectorContainer> testContainers, List<IterOutcome> iterOutcomes,
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/MockLateralJoinBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/MockLateralJoinBatch.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/MockLateralJoinBatch.java
new file mode 100644
index 0000000..98c9a9e
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/MockLateralJoinBatch.java
@@ -0,0 +1,237 @@
+/*
+* 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.physical.impl.unnest;
+
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.base.LateralContract;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.record.BatchSchema;
+import org.apache.drill.exec.record.CloseableRecordBatch;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.TypedFieldId;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.record.WritableBatch;
+import org.apache.drill.exec.record.selection.SelectionVector2;
+import org.apache.drill.exec.record.selection.SelectionVector4;
+import org.apache.drill.exec.vector.ValueVector;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * A mock lateral join implementation for testing unnest. This ignores all the other input and
+ * simply puts the unnest output into a results hypervector.
+ * Since Unnest returns an empty batch when it encounters a schema change, this implementation
+ * will also output an empty batch when it sees a schema change from unnest
+ */
+public class MockLateralJoinBatch implements LateralContract, CloseableRecordBatch {
+
+ private RecordBatch incoming;
+
+ private int recordIndex = 0;
+ private RecordBatch unnest;
+ private int unnestLimit = -1; // Unnest will EMIT if the number of records cross this limit
+
+ private boolean isDone;
+ private IterOutcome currentLeftOutcome = IterOutcome.NOT_YET;
+
+ private final FragmentContext context;
+ private final OperatorContext oContext;
+
+ private List<ValueVector> resultList = new ArrayList<>();
+
+ public MockLateralJoinBatch(FragmentContext context, OperatorContext oContext, RecordBatch incoming) {
+ this.context = context;
+ this.oContext = oContext;
+ this.incoming = incoming;
+ this.isDone = false;
+ }
+
+ @Override public RecordBatch getIncoming() {
+ return incoming; // don't need this
+ }
+
+ @Override public int getRecordIndex() {
+ return recordIndex;
+ }
+
+ @Override public IterOutcome getLeftOutcome() {
+ return currentLeftOutcome;
+ }
+
+ public void moveToNextRecord() {
+ recordIndex++;
+ }
+
+ public void reset() {
+ recordIndex = 0;
+ }
+
+ public void setUnnest(RecordBatch unnest){
+ this.unnest = unnest;
+ }
+
+ public void setUnnestLimit(int limit){
+ this.unnestLimit = limit;
+ }
+
+ public RecordBatch getUnnest() {
+ return unnest;
+ }
+
+ public IterOutcome next() {
+
+ IterOutcome currentOutcome = incoming.next();
+ currentLeftOutcome = currentOutcome;
+ recordIndex = 0;
+
+ switch (currentOutcome) {
+ case OK_NEW_SCHEMA:
+ // Nothing to do for this.
+ case OK:
+ IterOutcome outcome;
+ // consume all the outout from unnest until EMIT or end of
+ // incoming data
+ int unnestCount = 0; // number of values unnested for current record
+ while (recordIndex < incoming.getRecordCount()) {
+ outcome = unnest.next();
+ if (outcome == IterOutcome.OK_NEW_SCHEMA) {
+ // setup schema does nothing (this is just a place holder)
+ setupSchema();
+ // however unnest is also expected to return an empty batch
+ // which we will add to our output
+ }
+ // We put each batch output from unnest into a hypervector
+ // the calling test can match this against the baseline
+ //unnestCount +=
+ // unnest.getOutgoingContainer().hasRecordCount() ? unnest.getOutgoingContainer().getRecordCount() : 0;
+ unnestCount += addBatchToHyperContainer(unnest);
+ if (outcome == IterOutcome.EMIT) {
+ // reset unnest count
+ unnestCount = 0;
+ moveToNextRecord();
+ }
+ // Pretend that an operator somewhere between lateral and unnest
+ // wants to terminate processing of the record.
+ if(unnestLimit > 0 && unnestCount >= unnestLimit) {
+ unnest.kill(true);
+ }
+ }
+ return currentOutcome;
+ case NONE:
+ case STOP:
+ case OUT_OF_MEMORY:
+ isDone = true;
+ return currentOutcome;
+ case NOT_YET:
+ return currentOutcome;
+ default:
+ throw new UnsupportedOperationException("This state is not supported");
+ }
+ }
+
+ @Override public WritableBatch getWritableBatch() {
+ return null;
+ }
+
+ public List<ValueVector> getResultList() {
+ return resultList;
+ }
+
+ @Override
+ public void close() throws Exception {
+
+ }
+
+ @Override public int getRecordCount() {
+ return 0;
+ }
+
+ @Override
+ public SelectionVector2 getSelectionVector2() {
+ return null;
+ }
+
+ @Override
+ public SelectionVector4 getSelectionVector4() {
+ return null;
+ }
+
+ @Override
+ public FragmentContext getContext() {
+ return context;
+ }
+
+ @Override public BatchSchema getSchema() {
+ return null;
+ }
+
+ @Override public void kill(boolean sendUpstream) {
+ unnest.kill(sendUpstream);
+ }
+
+ @Override public VectorContainer getOutgoingContainer() {
+ return null;
+ }
+
+ @Override public TypedFieldId getValueVectorId(SchemaPath path) {
+ return null;
+ }
+
+ @Override public VectorWrapper<?> getValueAccessorById(Class<?> clazz, int... ids) {
+ return null;
+ }
+
+ private void setupSchema(){
+ // Nothing to do in this test
+ return;
+ }
+
+ public boolean isCompleted() {
+ return isDone;
+ }
+
+ // returns number of records added to output hyper container
+ private int addBatchToHyperContainer(RecordBatch inputBatch) {
+ int count = 0;
+ final RecordBatchData batchCopy = new RecordBatchData(inputBatch, oContext.getAllocator());
+ boolean success = false;
+ try {
+ for (VectorWrapper<?> w : batchCopy.getContainer()) {
+ ValueVector vv = w.getValueVector();
+ count += vv.getAccessor().getValueCount();
+ resultList.add(vv);
+ }
+ success = true;
+ } finally {
+ if (!success) {
+ batchCopy.clear();
+ }
+ }
+ return count;
+ }
+
+
+ @Override public Iterator<VectorWrapper<?>> iterator() {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestCorrectness.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestCorrectness.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestCorrectness.java
new file mode 100644
index 0000000..1a92ba0
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestCorrectness.java
@@ -0,0 +1,691 @@
+/*
+* 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.physical.impl.unnest;
+
+import org.apache.drill.categories.OperatorTest;
+import org.apache.drill.common.exceptions.DrillException;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.expression.PathSegment;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.types.TypeProtos;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.config.UnnestPOP;
+import org.apache.drill.exec.physical.impl.MockRecordBatch;
+import org.apache.drill.exec.physical.rowSet.impl.TestResultSetLoaderMapArray;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.TupleMetadata;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.store.mock.MockStorePOP;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.exec.vector.VarCharVector;
+import org.apache.drill.exec.vector.complex.MapVector;
+import org.apache.drill.test.SubOperatorTest;
+import org.apache.drill.test.rowSet.RowSet;
+import org.apache.drill.test.rowSet.RowSetBuilder;
+import org.apache.drill.test.rowSet.SchemaBuilder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static junit.framework.TestCase.fail;
+import static org.junit.Assert.assertTrue;
+
+@Category(OperatorTest.class) public class TestUnnestCorrectness extends SubOperatorTest {
+
+
+ // Operator Context for mock batch
+ public static OperatorContext operatorContext;
+
+ // use MockLateralJoinPop for MockRecordBatch ??
+ public static PhysicalOperator mockPopConfig;
+
+
+ @BeforeClass public static void setUpBeforeClass() throws Exception {
+ mockPopConfig = new MockStorePOP(null);
+ operatorContext = fixture.newOperatorContext(mockPopConfig);
+ }
+
+ @AfterClass public static void tearDownAfterClass() throws Exception {
+ operatorContext.close();
+ }
+
+ @Test
+ public void testUnnestFixedWidthColumn() {
+
+ Object[][] data = {
+ { (Object) new int[] {1, 2},
+ (Object) new int[] {3, 4, 5}},
+ { (Object) new int[] {6, 7, 8, 9},
+ (Object) new int[] {10, 11, 12, 13, 14}}
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema =
+ new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT)
+ .buildSchema();
+ TupleMetadata[] incomingSchemas = { incomingSchema, incomingSchema };
+
+ // First batch in baseline is an empty batch corresponding to OK_NEW_SCHEMA
+ Integer[][] baseline = {{}, {1, 2}, {3, 4, 5}, {6, 7, 8, 9}, {10, 11, 12, 13, 14}};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestVarWidthColumn() {
+
+ Object[][] data = {
+ { (Object) new String[] {"", "zero"},
+ (Object) new String[] {"one", "two", "three"}},
+ { (Object) new String[] {"four", "five", "six", "seven"},
+ (Object) new String[] {"eight", "nine", "ten", "eleven", "twelve"}}
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.VARCHAR).buildSchema();
+ TupleMetadata[] incomingSchemas = {incomingSchema, incomingSchema};
+
+ // First batch in baseline is an empty batch corresponding to OK_NEW_SCHEMA
+ String[][] baseline = {{}, {"", "zero"}, {"one", "two", "three"}, {"four", "five", "six", "seven"},
+ {"eight", "nine", "ten", "eleven", "twelve"}};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestMapColumn() {
+
+ Object[][] data = getMapData();
+
+ // Create input schema
+ TupleMetadata incomingSchema = getRepeatedMapSchema();
+ TupleMetadata[] incomingSchemas = {incomingSchema, incomingSchema};
+
+ // First batch in baseline is an empty batch corresponding to OK_NEW_SCHEMA
+ Object[][] baseline = getMapBaseline();
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestEmptyList() {
+
+ Object[][] data = {
+ { (Object) new String[] {},
+ (Object) new String[] {}
+ },
+ { (Object) new String[] {},
+ (Object) new String[] {}
+ }
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.VARCHAR).buildSchema();
+ TupleMetadata[] incomingSchemas = {incomingSchema, incomingSchema};
+
+ // First batch in baseline is an empty batch corresponding to OK_NEW_SCHEMA
+ // All subsequent batches are also empty
+ String[][] baseline = {{}, {}, {}, {}, {}};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestMultipleNewSchemaIncoming() {
+
+ // Schema changes in incoming have no effect on unnest unless the type of the
+ // unnest column itself has changed
+ Object[][] data = {
+ {
+ (Object) new String[] {"0", "1"},
+ (Object) new String[] {"2", "3", "4"}
+ },
+ {
+ (Object) new String[] {"5", "6" },
+ },
+ {
+ (Object) new String[] {"9"}
+ }
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.VARCHAR).buildSchema();
+ TupleMetadata[] incomingSchemas = {incomingSchema, incomingSchema, incomingSchema};
+
+ // First batch in baseline is an empty batch corresponding to OK_NEW_SCHEMA
+ String[][] baseline = {{}, {"0", "1"}, {"2", "3", "4"}, {"5", "6" }, {"9"} };
+
+ RecordBatch.IterOutcome[] iterOutcomes = {
+ RecordBatch.IterOutcome.OK_NEW_SCHEMA,
+ RecordBatch.IterOutcome.OK,
+ RecordBatch.IterOutcome.OK_NEW_SCHEMA};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestSchemaChange() {
+ Object[][] data = {
+ {
+ (Object) new String[] {"0", "1"},
+ (Object) new String[] {"2", "3", "4"}
+ },
+ {
+ (Object) new String[] {"5", "6" },
+ },
+ {
+ (Object) new int[] {9}
+ }
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema1 = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.VARCHAR).buildSchema();
+ TupleMetadata incomingSchema2 = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT).buildSchema();
+
+ TupleMetadata[] incomingSchemas = {incomingSchema1, incomingSchema1, incomingSchema2};
+
+ // First batch in baseline is an empty batch corresponding to OK_NEW_SCHEMA
+ // Another empty batch introduced by the schema change in the last batch
+ Object[][] baseline = {{}, {"0", "1"}, {"2", "3", "4"}, {"5", "6" }, {}, {9} };
+
+ RecordBatch.IterOutcome[] iterOutcomes = {
+ RecordBatch.IterOutcome.OK_NEW_SCHEMA,
+ RecordBatch.IterOutcome.OK,
+ RecordBatch.IterOutcome.OK_NEW_SCHEMA};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestLimitBatchSize() {
+
+ final int limitedOutputBatchSize = 1024;
+ final int limitedOutputBatchSizeBytes = 1024*4*1; // num rows * size of int
+ final int inputBatchSize = 1024+1;
+ // single record batch with single row. The unnest column has one
+ // more record than the batch size we want in the output
+ Object[][] data = new Object[1][1];
+
+ for (int i = 0; i < data.length; i++) {
+ for (int j = 0; j < data[i].length; j++) {
+ data[i][j] = new int[inputBatchSize];
+ for (int k =0; k < inputBatchSize; k++) {
+ ((int[])data[i][j])[k] = k;
+ }
+ }
+ }
+ Integer[][] baseline = new Integer[3][];
+ baseline[0] = new Integer[] {};
+ baseline[1] = new Integer[limitedOutputBatchSize];
+ baseline[2] = new Integer[1];
+ for (int i = 0; i < limitedOutputBatchSize; i++) {
+ baseline[1][i] = i;
+ }
+ baseline[2][0] = limitedOutputBatchSize;
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT).buildSchema();
+
+ TupleMetadata[] incomingSchemas = {incomingSchema};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK};
+
+ final long outputBatchSize = fixture.getFragmentContext().getOptions().getOption(ExecConstants
+ .OUTPUT_BATCH_SIZE_VALIDATOR);
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, limitedOutputBatchSizeBytes);
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ } finally {
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, outputBatchSize);
+ }
+
+ }
+
+ @Test
+ // Limit sends a kill. Unnest has more than one record batch for a record when
+ // the kill is sent.
+ public void testUnnestKillFromLimitSubquery1() {
+
+ // similar to previous test; we split a record across more than one batch.
+ // but we also set a limit less than the size of the batch so only one batch gets output.
+
+ final int limitedOutputBatchSize = 1024;
+ final int limitedOutputBatchSizeBytes = 1024*4*1; // num rows * size of int
+ final int inputBatchSize = 1024+1;
+ // single record batch with single row. The unnest column has one
+ // more record than the batch size we want in the output
+ Object[][] data = new Object[1][1];
+
+ for (int i = 0; i < data.length; i++) {
+ for (int j = 0; j < data[i].length; j++) {
+ data[i][j] = new int[inputBatchSize];
+ for (int k =0; k < inputBatchSize; k++) {
+ ((int[])data[i][j])[k] = k;
+ }
+ }
+ }
+ Integer[][] baseline = new Integer[3][];
+ baseline[0] = new Integer[] {};
+ baseline[1] = new Integer[limitedOutputBatchSize];
+ baseline[2] = new Integer[1];
+ for (int i = 0; i < limitedOutputBatchSize; i++) {
+ baseline[1][i] = i;
+ }
+ baseline[2] = new Integer[] {}; // because of kill the next batch is an empty batch
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT).buildSchema();
+
+ TupleMetadata[] incomingSchemas = {incomingSchema};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK};
+
+ final long outputBatchSize = fixture.getFragmentContext().getOptions().getOption(ExecConstants
+ .OUTPUT_BATCH_SIZE_VALIDATOR);
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, limitedOutputBatchSizeBytes);
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, 100, -1, data, baseline); // Limit of 100 values for unnest.
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ } finally {
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, outputBatchSize);
+ }
+
+ }
+
+ @Test
+ // Limit sends a kill. Unnest has exactly one record batch for a record when
+ // the kill is sent. This test is actually useless since it tests the behaviour of
+ // the mock lateral which doesn't send kill at all if it gets an EMIT. We expect limit
+ // to do the same so let's keep the test to demonstrate the expected behaviour.
+ public void testUnnestKillFromLimitSubquery2() {
+
+ // similar to previous test but the size of the array fits exactly into the record batch;
+
+
+ final int limitedOutputBatchSize = 1024;
+ final int limitedOutputBatchSizeBytes = 1024*4; // num rows * size of int
+ final int inputBatchSize = 1024;
+ // single record batch with single row. The unnest column has one
+ // more record than the batch size we want in the output
+ Object[][] data = new Object[1][1];
+
+ for (int i = 0; i < data.length; i++) {
+ for (int j = 0; j < data[i].length; j++) {
+ data[i][j] = new int[inputBatchSize];
+ for (int k =0; k < inputBatchSize; k++) {
+ ((int[])data[i][j])[k] = k;
+ }
+ }
+ }
+ Integer[][] baseline = new Integer[2][];
+ baseline[0] = new Integer[] {};
+ baseline[1] = new Integer[limitedOutputBatchSize];
+ for (int i = 0; i < limitedOutputBatchSize; i++) {
+ baseline[1][i] = i;
+ }
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT).buildSchema();
+
+ TupleMetadata[] incomingSchemas = {incomingSchema};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK};
+
+ final long outputBatchSize = fixture.getFragmentContext().getOptions().getOption(ExecConstants
+ .OUTPUT_BATCH_SIZE_VALIDATOR);
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, limitedOutputBatchSizeBytes);
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, 100, -1, data, baseline); // Limit of 100 values for unnest.
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ } finally {
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, outputBatchSize);
+ }
+
+ }
+
+
+ @Test
+ public void testUnnestNonArrayColumn() {
+
+ Object[][] data = {
+ { (Object) new Integer (1),
+ (Object) new Integer (3)},
+ { (Object) new Integer (6),
+ (Object) new Integer (10)}
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema =
+ new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .add("unnestColumn", TypeProtos.MinorType.INT)
+ .buildSchema();
+ TupleMetadata[] incomingSchemas = { incomingSchema, incomingSchema };
+
+ // We expect an Exception
+ Integer[][] baseline = {};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (UserException e) {
+ return; // succeeded
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+
+ // test unnest for various input conditions without invoking kill
+ private <T> void testUnnest(
+ TupleMetadata[] incomingSchemas,
+ RecordBatch.IterOutcome[] iterOutcomes,
+ T[][] data,
+ T[][] baseline ) throws Exception{
+ testUnnest(incomingSchemas, iterOutcomes, -1, -1, data, baseline);
+ }
+
+ // test unnest for various input conditions optionally invoking kill. if the kill or killBatch
+ // parameter is greater than 0 then the record batch is sent a kill after that many batches have been processed
+ private <T> void testUnnest( TupleMetadata[] incomingSchemas,
+ RecordBatch.IterOutcome[] iterOutcomes,
+ int unnestLimit, // kill unnest after every 'unnestLimit' number of values in every record
+ int execKill, // number of batches after which to kill the execution (!)
+ T[][] data,
+ T[][] baseline) throws Exception {
+
+ // Get the incoming container with dummy data for LJ
+ final List<VectorContainer> incomingContainer = new ArrayList<>(data.length);
+
+ // Create data
+ ArrayList<RowSet.SingleRowSet> rowSets = new ArrayList<>();
+
+ int batchNum = 0;
+ for ( Object[] recordBatch : data) {
+ RowSetBuilder rowSetBuilder = fixture.rowSetBuilder(incomingSchemas[batchNum]);
+ int rowNumber = 0;
+ for ( Object rowData : recordBatch) {
+ rowSetBuilder.addRow(++rowNumber, rowData);
+ }
+ RowSet.SingleRowSet rowSet = rowSetBuilder.build();
+ rowSets.add(rowSet);
+ incomingContainer.add(rowSet.container());
+ batchNum++;
+ }
+
+ // Get the unnest POPConfig
+ final UnnestPOP unnestPopConfig = new UnnestPOP(null, new SchemaPath(new PathSegment.NameSegment("unnestColumn")));
+
+ // Get the IterOutcomes for LJ
+ final List<RecordBatch.IterOutcome> outcomes = new ArrayList<>(iterOutcomes.length);
+ for(RecordBatch.IterOutcome o : iterOutcomes) {
+ outcomes.add(o);
+ }
+
+ // Create incoming MockRecordBatch
+ final MockRecordBatch incomingMockBatch =
+ new MockRecordBatch(fixture.getFragmentContext(), operatorContext, incomingContainer, outcomes,
+ incomingContainer.get(0).getSchema());
+
+ final MockLateralJoinBatch lateralJoinBatch =
+ new MockLateralJoinBatch(fixture.getFragmentContext(), operatorContext, incomingMockBatch);
+
+ // set pointer to Lateral in unnest pop config
+ unnestPopConfig.setLateral(lateralJoinBatch);
+
+ // setup Unnest record batch
+ final UnnestRecordBatch unnestBatch =
+ new UnnestRecordBatch(unnestPopConfig, incomingMockBatch, fixture.getFragmentContext());
+
+ // set backpointer to lateral join in unnest
+ lateralJoinBatch.setUnnest(unnestBatch);
+ lateralJoinBatch.setUnnestLimit(unnestLimit);
+
+ // Simulate the pipeline by calling next on the incoming
+ List<ValueVector> results = null;
+ int batchesProcessed = 0;
+ try {
+ while (!isTerminal(lateralJoinBatch.next())) {
+ batchesProcessed++;
+ if (batchesProcessed == execKill) {
+ lateralJoinBatch.getContext().getExecutorState().fail(new DrillException("Testing failure of execution."));
+ lateralJoinBatch.kill(true);
+ }
+ // else nothing to do
+ }
+
+ // Check results against baseline
+ results = lateralJoinBatch.getResultList();
+
+ int i = 0;
+ for (ValueVector vv : results) {
+ int valueCount = vv.getAccessor().getValueCount();
+ if (valueCount != baseline[i].length) {
+ fail("Test failed in validating unnest output. Value count mismatch.");
+ }
+ for (int j = 0; j < valueCount; j++) {
+
+ if (vv instanceof MapVector) {
+ if (!compareMapBaseline((Object) baseline[i][j], vv.getAccessor().getObject(j))) {
+ fail("Test failed in validating unnest(Map) output. Value mismatch");
+ }
+ } else if (vv instanceof VarCharVector) {
+ Object val = vv.getAccessor().getObject(j);
+ if (((String) baseline[i][j]).compareTo(val.toString()) != 0) {
+ fail("Test failed in validating unnest output. Value mismatch. Baseline value[]" + i + "][" + j + "]"
+ + ": " + baseline[i][j] + " VV.getObject(j): " + val);
+ }
+ } else {
+ Object val = vv.getAccessor().getObject(j);
+ if (!baseline[i][j].equals(val)) {
+ fail(
+ "Test failed in validating unnest output. Value mismatch. Baseline value[" + i + "][" + j + "]" + ": "
+ + baseline[i][j] + " VV.getObject(j): " + val);
+ }
+ }
+ }
+ i++;
+
+ }
+
+ assertTrue(((MockLateralJoinBatch) lateralJoinBatch).isCompleted());
+
+ } catch (UserException e) {
+ throw e; // Valid exception
+ } catch (Exception e) {
+ fail("Test failed in validating unnest output. Exception : " + e.getMessage());
+ } finally {
+ // Close all the resources for this test case
+ unnestBatch.close();
+ lateralJoinBatch.close();
+ incomingMockBatch.close();
+
+ if (results != null) {
+ for (ValueVector vv : results) {
+ vv.clear();
+ }
+ }
+ for(RowSet.SingleRowSet rowSet: rowSets) {
+ rowSet.clear();
+ }
+ }
+
+ }
+
+ /**
+ * Build a schema with a repeated map -
+ *
+ * {
+ * rowNum,
+ * mapColumn : [
+ * {
+ * colA,
+ * colB : [
+ * varcharCol
+ * ]
+ * }
+ * ]
+ * }
+ *
+ * @see TestResultSetLoaderMapArray TestResultSetLoaderMapArray for similar schema and data
+ * @return TupleMetadata corresponding to the schema
+ */
+ private TupleMetadata getRepeatedMapSchema() {
+ TupleMetadata schema = new SchemaBuilder()
+ .add("rowNum", TypeProtos.MinorType.INT)
+ .addMapArray("unnestColumn")
+ .add("colA", TypeProtos.MinorType.INT)
+ .addArray("colB", TypeProtos.MinorType.VARCHAR)
+ .buildMap()
+ .buildSchema();
+ return schema;
+ }
+
+ private Object[][] getMapData( ) {
+
+ Object[][] d = {
+ {
+ new Object[] {},
+ new Object[] {
+ new Object[] {11, new String[] {"1.1.1", "1.1.2" }},
+ new Object[] {12, new String[] {"1.2.1", "1.2.2" }}
+ },
+
+ new Object[] {
+ new Object[] {21, new String[] {"2.1.1", "2.1.2" }},
+ new Object[] {22, new String[] {}},
+ new Object[] {23, new String[] {"2.3.1", "2.3.2" }}
+ }
+ },
+ {
+ new Object[] {
+ new Object[] {31, new String[] {"3.1.1", "3.1.2" }},
+ new Object[] {32, new String[] {"3.2.1", "3.2.2" }}
+ }
+ }
+ };
+
+ return d;
+ }
+
+ private Object[][] getMapBaseline() {
+
+ Object[][] d = {
+ new Object[] {}, // Empty record batch returned by OK_NEW_SCHEMA
+ new Object[] {}, // First incoming batch is empty
+ new Object[] {
+ "{\"colA\":11,\"colB\":[\"1.1.1\",\"1.1.2\"]}",
+ "{\"colA\":12,\"colB\":[\"1.2.1\",\"1.2.2\"]}"
+ },
+ new Object[] {
+ "{\"colA\":21,\"colB\":[\"2.1.1\",\"2.1.2\"]}",
+ "{\"colA\":22,\"colB\":[]}",
+ "{\"colA\":23,\"colB\":[\"2.3.1\",\"2.3.2\"]}"
+ },
+ new Object[] {
+ "{\"colA\":31,\"colB\":[\"3.1.1\",\"3.1.2\"]}",
+ "{\"colA\":32,\"colB\":[\"3.2.1\",\"3.2.2\"]}"
+ }
+ };
+ return d;
+ }
+
+ private boolean compareMapBaseline(Object baselineValue, Object vector) {
+ String vv = vector.toString();
+ String b = (String)baselineValue;
+ return vv.equalsIgnoreCase(b);
+ }
+
+ private boolean isTerminal(RecordBatch.IterOutcome outcome) {
+ return (outcome == RecordBatch.IterOutcome.NONE || outcome == RecordBatch.IterOutcome.STOP) || (outcome
+ == RecordBatch.IterOutcome.OUT_OF_MEMORY);
+ }
+
+}
+