You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by ti...@apache.org on 2016/10/22 01:40:57 UTC
asterixdb git commit: Fix Upsert Pipeline
Repository: asterixdb
Updated Branches:
refs/heads/master 2f5c318d2 -> 452c5d09f
Fix Upsert Pipeline
Change-Id: I5c19d448f9664ecaeac600668a6dbdcf40673c56
Reviewed-on: https://asterix-gerrit.ics.uci.edu/1308
Reviewed-by: Ian Maxon <im...@apache.org>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Till Westmann <ti...@apache.org>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Project: http://git-wip-us.apache.org/repos/asf/asterixdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/asterixdb/commit/452c5d09
Tree: http://git-wip-us.apache.org/repos/asf/asterixdb/tree/452c5d09
Diff: http://git-wip-us.apache.org/repos/asf/asterixdb/diff/452c5d09
Branch: refs/heads/master
Commit: 452c5d09f3399e934f25f1ff164c81a971960fc3
Parents: 2f5c318
Author: Abdullah Alamoudi <ba...@gmail.com>
Authored: Fri Oct 21 16:51:34 2016 -0700
Committer: Till Westmann <ti...@apache.org>
Committed: Fri Oct 21 18:40:38 2016 -0700
----------------------------------------------------------------------
.../feed/dataflow/FeedExceptionHandler.java | 5 -
.../metadata/declared/AqlMetadataProvider.java | 48 +--
...rixLSMPrimaryUpsertOperatorNodePushable.java | 69 +++--
.../algebra/metadata/IMetadataProvider.java | 2 +-
.../logical/InsertDeleteUpsertOperator.java | 4 +-
.../logical/visitors/SchemaVariableVisitor.java | 9 +-
.../physical/InsertDeleteUpsertPOperator.java | 2 +-
.../common/comm/io/FrameTupleAccessor.java | 130 --------
.../comm/io/ResultFrameTupleAccessor.java | 14 -
.../dataflow/common/util/FrameDebugUtils.java | 298 +++++++++++++++++++
.../tests/unit/RunMergingFrameReaderTest.java | 11 +-
11 files changed, 370 insertions(+), 222 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/452c5d09/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedExceptionHandler.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedExceptionHandler.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedExceptionHandler.java
index 304eb0c..64cca8c 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedExceptionHandler.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedExceptionHandler.java
@@ -43,11 +43,6 @@ public class FeedExceptionHandler implements IExceptionHandler {
this.fta = fta;
}
- public void prettyPrint(ByteBuffer frame) {
- fta.reset(frame);
- fta.prettyPrint();
- }
-
@Override
public ByteBuffer handle(HyracksDataException th, ByteBuffer frame) {
try {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/452c5d09/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/AqlMetadataProvider.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/AqlMetadataProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/AqlMetadataProvider.java
index e05aa25..f6eeb66 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/AqlMetadataProvider.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/AqlMetadataProvider.java
@@ -26,10 +26,10 @@ import java.util.List;
import java.util.Map;
import org.apache.asterix.common.config.AsterixStorageProperties;
-import org.apache.asterix.common.config.GlobalConfig;
import org.apache.asterix.common.config.DatasetConfig.DatasetType;
import org.apache.asterix.common.config.DatasetConfig.ExternalFilePendingOp;
import org.apache.asterix.common.config.DatasetConfig.IndexType;
+import org.apache.asterix.common.config.GlobalConfig;
import org.apache.asterix.common.context.AsterixVirtualBufferCacheProvider;
import org.apache.asterix.common.context.ITransactionSubsystemProvider;
import org.apache.asterix.common.context.TransactionSubsystemProvider;
@@ -42,8 +42,8 @@ import org.apache.asterix.common.ioopcallbacks.LSMBTreeWithBuddyIOOperationCallb
import org.apache.asterix.common.ioopcallbacks.LSMInvertedIndexIOOperationCallbackFactory;
import org.apache.asterix.common.ioopcallbacks.LSMRTreeIOOperationCallbackFactory;
import org.apache.asterix.common.library.ILibraryManager;
-import org.apache.asterix.common.transactions.JobId;
import org.apache.asterix.common.transactions.IRecoveryManager.ResourceType;
+import org.apache.asterix.common.transactions.JobId;
import org.apache.asterix.common.utils.StoragePathUtil;
import org.apache.asterix.dataflow.data.nontagged.valueproviders.AqlPrimitiveValueProviderFactory;
import org.apache.asterix.external.adapter.factory.LookupAdapterFactory;
@@ -1097,7 +1097,7 @@ public class AqlMetadataProvider implements IMetadataProvider<AqlSourceId, Strin
@Override
public Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> getUpsertRuntime(
- IDataSource<AqlSourceId> dataSource, IOperatorSchema propagatedSchema, IVariableTypeEnvironment typeEnv,
+ IDataSource<AqlSourceId> dataSource, IOperatorSchema inputSchema, IVariableTypeEnvironment typeEnv,
List<LogicalVariable> primaryKeys, LogicalVariable payload, List<LogicalVariable> filterKeys,
List<LogicalVariable> additionalNonFilterFields, RecordDescriptor recordDesc, JobGenContext context,
JobSpecification spec) throws AlgebricksException {
@@ -1119,22 +1119,22 @@ public class AqlMetadataProvider implements IMetadataProvider<AqlSourceId, Strin
int i = 0;
// set the keys' permutations
for (LogicalVariable varKey : primaryKeys) {
- int idx = propagatedSchema.findVariable(varKey);
+ int idx = inputSchema.findVariable(varKey);
fieldPermutation[i] = idx;
bloomFilterKeyFields[i] = i;
i++;
}
// set the record permutation
- fieldPermutation[i++] = propagatedSchema.findVariable(payload);
+ fieldPermutation[i++] = inputSchema.findVariable(payload);
// set the filters' permutations.
if (numFilterFields > 0) {
- int idx = propagatedSchema.findVariable(filterKeys.get(0));
+ int idx = inputSchema.findVariable(filterKeys.get(0));
fieldPermutation[i++] = idx;
}
if (additionalNonFilterFields != null) {
for (LogicalVariable var : additionalNonFilterFields) {
- int idx = propagatedSchema.findVariable(var);
+ int idx = inputSchema.findVariable(var);
fieldPermutation[i++] = idx;
}
}
@@ -1195,22 +1195,19 @@ public class AqlMetadataProvider implements IMetadataProvider<AqlSourceId, Strin
+ (dataset.hasMetaPart() ? 2 : 1) + numFilterFields];
ISerializerDeserializer[] outputSerDes = new ISerializerDeserializer[recordDesc.getFieldCount()
+ (dataset.hasMetaPart() ? 2 : 1) + numFilterFields];
- for (int j = 0; j < recordDesc.getFieldCount(); j++) {
- outputTypeTraits[j] = recordDesc.getTypeTraits()[j];
- outputSerDes[j] = recordDesc.getFields()[j];
- }
- outputSerDes[outputSerDes.length - (dataset.hasMetaPart() ? 2 : 1) - numFilterFields] = FormatUtils
- .getDefaultFormat().getSerdeProvider().getSerializerDeserializer(itemType);
- outputTypeTraits[outputTypeTraits.length - (dataset.hasMetaPart() ? 2 : 1) - numFilterFields] = FormatUtils
- .getDefaultFormat().getTypeTraitProvider().getTypeTrait(itemType);
+ // add the previous record first
+ int f = 0;
+ outputSerDes[f] = FormatUtils.getDefaultFormat().getSerdeProvider().getSerializerDeserializer(itemType);
+ f++;
+ // add the previous meta second
if (dataset.hasMetaPart()) {
- outputSerDes[outputSerDes.length - 1 - numFilterFields] = FormatUtils.getDefaultFormat()
- .getSerdeProvider().getSerializerDeserializer(metaItemType);
- outputTypeTraits[outputTypeTraits.length - 1 - numFilterFields] = FormatUtils.getDefaultFormat()
- .getTypeTraitProvider().getTypeTrait(metaItemType);
+ outputSerDes[f] = FormatUtils.getDefaultFormat().getSerdeProvider().getSerializerDeserializer(
+ metaItemType);
+ outputTypeTraits[f] = FormatUtils.getDefaultFormat().getTypeTraitProvider().getTypeTrait(metaItemType);
+ f++;
}
-
+ // add the previous filter third
int fieldIdx = -1;
if (numFilterFields > 0) {
String filterField = DatasetUtils.getFilterField(dataset).get(0);
@@ -1220,10 +1217,15 @@ public class AqlMetadataProvider implements IMetadataProvider<AqlSourceId, Strin
}
}
fieldIdx = i;
- outputTypeTraits[outputTypeTraits.length - 1] = FormatUtils.getDefaultFormat().getTypeTraitProvider()
- .getTypeTrait(itemType.getFieldTypes()[fieldIdx]);
- outputSerDes[outputSerDes.length - 1] = FormatUtils.getDefaultFormat().getSerdeProvider()
+ outputTypeTraits[f] = FormatUtils.getDefaultFormat().getTypeTraitProvider().getTypeTrait(itemType
+ .getFieldTypes()[fieldIdx]);
+ outputSerDes[f] = FormatUtils.getDefaultFormat().getSerdeProvider()
.getSerializerDeserializer(itemType.getFieldTypes()[fieldIdx]);
+ f++;
+ }
+ for (int j = 0; j < recordDesc.getFieldCount(); j++) {
+ outputTypeTraits[j + f] = recordDesc.getTypeTraits()[j];
+ outputSerDes[j + f] = recordDesc.getFields()[j];
}
RecordDescriptor outputRecordDesc = new RecordDescriptor(outputSerDes, outputTypeTraits);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/452c5d09/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/AsterixLSMPrimaryUpsertOperatorNodePushable.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/AsterixLSMPrimaryUpsertOperatorNodePushable.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/AsterixLSMPrimaryUpsertOperatorNodePushable.java
index afd6019..96f9e76 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/AsterixLSMPrimaryUpsertOperatorNodePushable.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/AsterixLSMPrimaryUpsertOperatorNodePushable.java
@@ -78,6 +78,8 @@ public class AsterixLSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertU
private ARecordPointable recPointable;
private DataOutput prevDos;
private final boolean hasMeta;
+ private final int filterFieldIndex;
+ private final int metaFieldIndex;
public AsterixLSMPrimaryUpsertOperatorNodePushable(IIndexOperatorDescriptor opDesc, IHyracksTaskContext ctx,
int partition, int[] fieldPermutation, IRecordDescriptorProvider recordDescProvider, int numOfPrimaryKeys,
@@ -93,6 +95,8 @@ public class AsterixLSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertU
key.setFieldPermutation(searchKeyPermutations);
hasMeta = (fieldPermutation.length > numOfPrimaryKeys + 1) && (filterFieldIndex < 0
|| (filterFieldIndex >= 0 && (fieldPermutation.length > numOfPrimaryKeys + 2)));
+ this.metaFieldIndex = numOfPrimaryKeys + 1;
+ this.filterFieldIndex = numOfPrimaryKeys + (hasMeta ? 2 : 1);
if (filterFieldIndex >= 0) {
isFiltered = true;
this.recordType = recordType;
@@ -101,7 +105,6 @@ public class AsterixLSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertU
this.prevRecWithPKWithFilterValue = new ArrayTupleBuilder(fieldPermutation.length + (hasMeta ? 1 : 0));
this.prevDos = prevRecWithPKWithFilterValue.getDataOutput();
}
-
}
// we have the permutation which has [pk locations, record location, optional:filter-location]
@@ -141,8 +144,8 @@ public class AsterixLSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertU
.createSearchOperationCallback(indexHelper.getResourceID(), ctx, this));
cursor = indexAccessor.createSearchCursor(false);
frameTuple = new FrameTupleReference();
- IAsterixAppRuntimeContext runtimeCtx =
- (IAsterixAppRuntimeContext) ctx.getJobletContext().getApplicationContext().getApplicationObject();
+ IAsterixAppRuntimeContext runtimeCtx = (IAsterixAppRuntimeContext) ctx.getJobletContext()
+ .getApplicationContext().getApplicationObject();
AsterixLSMIndexUtil.checkAndSetFirstLSN((AbstractLSMIndex) index,
runtimeCtx.getTransactionSubsystem().getLogManager());
} catch (Exception e) {
@@ -156,41 +159,12 @@ public class AsterixLSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertU
searchPred.reset(key, key, true, true, keySearchCmp, keySearchCmp);
}
- private void writeOutput(int tupleIndex, boolean recordWasInserted) throws IOException {
- boolean recordWasDeleted = prevTuple != null;
- tb.reset();
+ private void writeOutput(int tupleIndex, boolean recordWasInserted, boolean recordWasDeleted) throws IOException {
frameTuple.reset(accessor, tupleIndex);
for (int i = 0; i < frameTuple.getFieldCount(); i++) {
dos.write(frameTuple.getFieldData(i), frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
tb.addFieldEndOffset();
}
- if (recordWasDeleted) {
- dos.write(prevTuple.getFieldData(numOfPrimaryKeys), prevTuple.getFieldStart(numOfPrimaryKeys),
- prevTuple.getFieldLength(numOfPrimaryKeys));
- tb.addFieldEndOffset();
- // if has meta, then append meta
- if (hasMeta) {
- dos.write(prevTuple.getFieldData(numOfPrimaryKeys + 1), prevTuple.getFieldStart(numOfPrimaryKeys + 1),
- prevTuple.getFieldLength(numOfPrimaryKeys + 1));
- tb.addFieldEndOffset();
- }
- // if with filters, append the filter
- if (isFiltered) {
- dos.write(prevTuple.getFieldData(numOfPrimaryKeys + (hasMeta ? 2 : 1)),
- prevTuple.getFieldStart(numOfPrimaryKeys + (hasMeta ? 2 : 1)),
- prevTuple.getFieldLength(numOfPrimaryKeys + (hasMeta ? 2 : 1)));
- tb.addFieldEndOffset();
- }
- } else {
- addNullField();
- if (hasMeta) {
- addNullField();
- }
- // if with filters, append null
- if (isFiltered) {
- addNullField();
- }
- }
if (recordWasInserted || recordWasDeleted) {
FrameUtils.appendToWriter(writer, appender, tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
}
@@ -214,6 +188,7 @@ public class AsterixLSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertU
int i = 0;
try {
while (i < tupleCount) {
+ tb.reset();
boolean recordWasInserted = false;
tuple.reset(accessor, i);
resetSearchPredicate(i);
@@ -222,10 +197,26 @@ public class AsterixLSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertU
cursor.next();
prevTuple = cursor.getTuple();
cursor.reset();
- modCallback.setOp(Operation.DELETE);
if (isFiltered) {
prevTuple = getPrevTupleWithFilter(prevTuple);
}
+ dos.write(prevTuple.getFieldData(numOfPrimaryKeys), prevTuple.getFieldStart(numOfPrimaryKeys),
+ prevTuple.getFieldLength(numOfPrimaryKeys));
+ tb.addFieldEndOffset();
+ // if has meta, then append meta
+ if (hasMeta) {
+ dos.write(prevTuple.getFieldData(metaFieldIndex), prevTuple.getFieldStart(metaFieldIndex),
+ prevTuple.getFieldLength(metaFieldIndex));
+ tb.addFieldEndOffset();
+ }
+ // if with filters, append the filter
+ if (isFiltered) {
+ dos.write(prevTuple.getFieldData(filterFieldIndex),
+ prevTuple.getFieldStart(filterFieldIndex),
+ prevTuple.getFieldLength(filterFieldIndex));
+ tb.addFieldEndOffset();
+ }
+ modCallback.setOp(Operation.DELETE);
if (i == 0) {
lsmAccessor.delete(prevTuple);
} else {
@@ -233,6 +224,14 @@ public class AsterixLSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertU
}
} else {
prevTuple = null;
+ addNullField();
+ if (hasMeta) {
+ addNullField();
+ }
+ // if with filters, append null
+ if (isFiltered) {
+ addNullField();
+ }
cursor.reset();
}
if (!isNull(tuple, numOfPrimaryKeys)) {
@@ -244,7 +243,7 @@ public class AsterixLSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertU
}
recordWasInserted = true;
}
- writeOutput(i, recordWasInserted);
+ writeOutput(i, recordWasInserted, prevTuple != null);
i++;
}
appender.write(writer, true);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/452c5d09/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/metadata/IMetadataProvider.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/metadata/IMetadataProvider.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/metadata/IMetadataProvider.java
index 68710a5..6947942 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/metadata/IMetadataProvider.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/metadata/IMetadataProvider.java
@@ -193,7 +193,7 @@ public interface IMetadataProvider<S, I> {
public IFunctionInfo lookupFunction(FunctionIdentifier fid);
public Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> getUpsertRuntime(IDataSource<S> dataSource,
- IOperatorSchema propagatedSchema, IVariableTypeEnvironment typeEnv, List<LogicalVariable> keys,
+ IOperatorSchema inputSchema, IVariableTypeEnvironment typeEnv, List<LogicalVariable> keys,
LogicalVariable payLoadVar, List<LogicalVariable> additionalFilterFields,
List<LogicalVariable> additionalNonFilteringFields, RecordDescriptor recordDesc, JobGenContext context,
JobSpecification jobSpec) throws AlgebricksException;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/452c5d09/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/InsertDeleteUpsertOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/InsertDeleteUpsertOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/InsertDeleteUpsertOperator.java
index 5dc327a..9838c12 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/InsertDeleteUpsertOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/InsertDeleteUpsertOperator.java
@@ -84,7 +84,6 @@ public class InsertDeleteUpsertOperator extends AbstractLogicalOperator {
@Override
public void recomputeSchema() throws AlgebricksException {
schema = new ArrayList<LogicalVariable>();
- schema.addAll(inputs.get(0).getValue().getSchema());
if (operation == Kind.UPSERT) {
// The upsert case also produces the previous record
schema.add(prevRecordVar);
@@ -95,6 +94,7 @@ public class InsertDeleteUpsertOperator extends AbstractLogicalOperator {
schema.add(prevFilterVar);
}
}
+ schema.addAll(inputs.get(0).getValue().getSchema());
}
public void getProducedVariables(Collection<LogicalVariable> producedVariables) {
@@ -146,7 +146,6 @@ public class InsertDeleteUpsertOperator extends AbstractLogicalOperator {
@Override
public void propagateVariables(IOperatorSchema target, IOperatorSchema... sources)
throws AlgebricksException {
- target.addAllVariables(sources[0]);
if (operation == Kind.UPSERT) {
target.addVariable(prevRecordVar);
if (prevAdditionalNonFilteringVars != null) {
@@ -158,6 +157,7 @@ public class InsertDeleteUpsertOperator extends AbstractLogicalOperator {
target.addVariable(prevFilterVar);
}
}
+ target.addAllVariables(sources[0]);
}
};
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/452c5d09/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/SchemaVariableVisitor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/SchemaVariableVisitor.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/SchemaVariableVisitor.java
index 28f4e5e..59ccd84 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/SchemaVariableVisitor.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/SchemaVariableVisitor.java
@@ -34,11 +34,11 @@ import org.apache.hyracks.algebricks.core.algebra.expressions.VariableReferenceE
import org.apache.hyracks.algebricks.core.algebra.operators.logical.AggregateOperator;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.AssignOperator;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.DataSourceScanOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.DelegateOperator;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.DistinctOperator;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.DistributeResultOperator;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.EmptyTupleSourceOperator;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.ExchangeOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.DelegateOperator;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.GroupByOperator;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.IndexInsertDeleteUpsertOperator;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.InnerJoinOperator;
@@ -295,7 +295,12 @@ public class SchemaVariableVisitor implements ILogicalOperatorVisitor<Void, Void
@Override
public Void visitInsertDeleteUpsertOperator(InsertDeleteUpsertOperator op, Void arg) throws AlgebricksException {
- standardLayout(op);
+ // produced first
+ VariableUtilities.getProducedVariables(op, schemaVariables);
+ // then propagated
+ for (Mutable<ILogicalOperator> c : op.getInputs()) {
+ VariableUtilities.getLiveVariables(c.getValue(), schemaVariables);
+ }
return null;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/452c5d09/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InsertDeleteUpsertPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InsertDeleteUpsertPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InsertDeleteUpsertPOperator.java
index 3c9cddf..6ded4a3 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InsertDeleteUpsertPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InsertDeleteUpsertPOperator.java
@@ -114,7 +114,7 @@ public class InsertDeleteUpsertPOperator extends AbstractPhysicalOperator {
runtimeAndConstraints = mp.getDeleteRuntime(dataSource, propagatedSchema, typeEnv, keys, payload,
additionalFilteringKeys, inputDesc, context, spec);
} else if (operation == Kind.UPSERT) {
- runtimeAndConstraints = mp.getUpsertRuntime(dataSource, propagatedSchema, typeEnv, keys, payload,
+ runtimeAndConstraints = mp.getUpsertRuntime(dataSource, inputSchemas[0], typeEnv, keys, payload,
additionalFilteringKeys, additionalNonFilteringFields, inputDesc, context, spec);
} else {
throw new AlgebricksException("Unsupported Operation " + operation);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/452c5d09/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAccessor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAccessor.java b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAccessor.java
index d99e2f2..cefada7 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAccessor.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAccessor.java
@@ -18,18 +18,12 @@
*/
package org.apache.hyracks.dataflow.common.comm.io;
-import java.io.DataInputStream;
-import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.Arrays;
import org.apache.hyracks.api.comm.FrameConstants;
import org.apache.hyracks.api.comm.FrameHelper;
import org.apache.hyracks.api.comm.IFrameTupleAccessor;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.dataflow.common.comm.util.ByteBufferInputStream;
-import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
import org.apache.hyracks.util.IntSerDeUtils;
/**
@@ -118,132 +112,8 @@ public class FrameTupleAccessor implements IFrameTupleAccessor {
return getFieldCount() * FrameConstants.SIZE_LEN;
}
- public void prettyPrint(String prefix) {
- ByteBufferInputStream bbis = new ByteBufferInputStream();
- DataInputStream dis = new DataInputStream(bbis);
- int tc = getTupleCount();
- StringBuilder sb = new StringBuilder();
- sb.append(prefix).append("TC: " + tc).append("\n");
- for (int i = 0; i < tc; ++i) {
- prettyPrint(i, bbis, dis, sb);
- }
- System.err.println(sb.toString());
- }
-
- public void prettyPrint() {
- prettyPrint("");
- }
-
- protected void prettyPrint(int tid, ByteBufferInputStream bbis, DataInputStream dis, StringBuilder sb) {
- sb.append(" tid" + tid + ":(" + getTupleStartOffset(tid) + ", " + getTupleEndOffset(tid) + ")[");
- for (int j = 0; j < getFieldCount(); ++j) {
- sb.append(" ");
- if (j > 0) {
- sb.append("|");
- }
- sb.append("f" + j + ":(" + getFieldStartOffset(tid, j) + ", " + getFieldEndOffset(tid, j) + ") ");
- sb.append("{");
- bbis.setByteBuffer(buffer, getTupleStartOffset(tid) + getFieldSlotsLength() + getFieldStartOffset(tid, j));
- try {
- sb.append(recordDescriptor.getFields()[j].deserialize(dis));
- } catch (Exception e) {
- e.printStackTrace();
- sb.append("Failed to deserialize field" + j);
- }
- sb.append("}");
- }
- sb.append("\n");
- }
-
- public void prettyPrint(int tid) {
- ByteBufferInputStream bbis = new ByteBufferInputStream();
- DataInputStream dis = new DataInputStream(bbis);
- StringBuilder sb = new StringBuilder();
- prettyPrint(tid, bbis, dis, sb);
- System.err.println(sb.toString());
- }
-
@Override
public int getFieldCount() {
return recordDescriptor.getFieldCount();
}
-
- /*
- * The two methods below can be used for debugging.
- * They are safe as they don't print records. Printing records
- * using IserializerDeserializer can print incorrect results or throw exceptions.
- * A better way yet would be to use record pointable.
- */
- public void prettyPrint(String prefix, int[] recordFields) throws IOException {
- ByteBufferInputStream bbis = new ByteBufferInputStream();
- DataInputStream dis = new DataInputStream(bbis);
- int tc = getTupleCount();
- StringBuilder sb = new StringBuilder();
- sb.append(prefix).append("TC: " + tc).append("\n");
- for (int i = 0; i < tc; ++i) {
- prettyPrint(i, bbis, dis, sb, recordFields);
- }
- System.err.println(sb.toString());
- }
-
- public void prettyPrint(int tIdx, int[] recordFields) throws IOException {
- ByteBufferInputStream bbis = new ByteBufferInputStream();
- DataInputStream dis = new DataInputStream(bbis);
- StringBuilder sb = new StringBuilder();
- prettyPrint(tIdx, bbis, dis, sb, recordFields);
- System.err.println(sb.toString());
- }
-
- public void prettyPrint(ITupleReference tuple, int fieldsIdx, int descIdx) throws HyracksDataException {
- ByteBufferInputStream bbis = new ByteBufferInputStream();
- DataInputStream dis = new DataInputStream(bbis);
- StringBuilder sb = new StringBuilder();
- sb.append("[");
- sb.append("f" + fieldsIdx + ":(" + tuple.getFieldStart(fieldsIdx) + ", "
- + (tuple.getFieldLength(fieldsIdx) + tuple.getFieldStart(fieldsIdx)) + ") ");
- sb.append("{");
- ByteBuffer bytebuff = ByteBuffer.wrap(tuple.getFieldData(fieldsIdx));
- bbis.setByteBuffer(bytebuff, tuple.getFieldStart(fieldsIdx));
- sb.append(recordDescriptor.getFields()[descIdx].deserialize(dis));
- sb.append("}");
- sb.append("\n");
- System.err.println(sb.toString());
- }
-
- public void prettyPrint(ITupleReference tuple, int[] descF) throws HyracksDataException {
- ByteBufferInputStream bbis = new ByteBufferInputStream();
- DataInputStream dis = new DataInputStream(bbis);
- StringBuilder sb = new StringBuilder();
- sb.append("[");
- for (int j = 0; j < descF.length; ++j) {
- sb.append("f" + j + ":(" + tuple.getFieldStart(j) + ", "
- + (tuple.getFieldLength(j) + tuple.getFieldStart(j)) + ") ");
- sb.append("{");
- ByteBuffer bytebuff = ByteBuffer.wrap(tuple.getFieldData(j));
- bbis.setByteBuffer(bytebuff, tuple.getFieldStart(j));
- sb.append(recordDescriptor.getFields()[descF[j]].deserialize(dis));
- sb.append("}");
- }
- sb.append("\n");
- System.err.println(sb.toString());
- }
-
- protected void prettyPrint(int tid, ByteBufferInputStream bbis, DataInputStream dis, StringBuilder sb,
- int[] recordFields) throws IOException {
- Arrays.sort(recordFields);
- sb.append(" tid" + tid + ":(" + getTupleStartOffset(tid) + ", " + getTupleEndOffset(tid) + ")[");
- for (int j = 0; j < getFieldCount(); ++j) {
- sb.append("f" + j + ":(" + getFieldStartOffset(tid, j) + ", " + getFieldEndOffset(tid, j) + ") ");
- sb.append("{");
- bbis.setByteBuffer(buffer, getTupleStartOffset(tid) + getFieldSlotsLength() + getFieldStartOffset(tid, j));
- if (Arrays.binarySearch(recordFields, j) >= 0) {
- sb.append("{a record field: only print using pointable:");
- sb.append("tag->" + dis.readByte() + "}");
- } else {
- sb.append(recordDescriptor.getFields()[j].deserialize(dis));
- }
- sb.append("}");
- }
- sb.append("\n");
- }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/452c5d09/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/ResultFrameTupleAccessor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/ResultFrameTupleAccessor.java b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/ResultFrameTupleAccessor.java
index d05b3ed..874ac46 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/ResultFrameTupleAccessor.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/ResultFrameTupleAccessor.java
@@ -18,10 +18,6 @@
*/
package org.apache.hyracks.dataflow.common.comm.io;
-import java.io.DataInputStream;
-
-import org.apache.hyracks.dataflow.common.comm.util.ByteBufferInputStream;
-
public class ResultFrameTupleAccessor extends FrameTupleAccessor {
public ResultFrameTupleAccessor() {
@@ -29,16 +25,6 @@ public class ResultFrameTupleAccessor extends FrameTupleAccessor {
}
@Override
- protected void prettyPrint(int tid, ByteBufferInputStream bbis, DataInputStream dis, StringBuilder sb) {
- sb.append(tid + ":(" + getTupleStartOffset(tid) + ", " + getTupleEndOffset(tid) + ")[");
-
- bbis.setByteBuffer(getBuffer(), getTupleStartOffset(tid));
- sb.append(dis);
-
- sb.append("]\n");
- }
-
- @Override
public int getFieldCount() {
return 1;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/452c5d09/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/util/FrameDebugUtils.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/util/FrameDebugUtils.java b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/util/FrameDebugUtils.java
new file mode 100644
index 0000000..aa27c42
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/util/FrameDebugUtils.java
@@ -0,0 +1,298 @@
+/*
+ * 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.hyracks.dataflow.common.util;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+import org.apache.hyracks.api.comm.IFrameTupleAccessor;
+import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.dataflow.common.comm.util.ByteBufferInputStream;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+
+/**
+ * A Util class used for inspecting frames
+ * for debugging purposes
+ */
+public class FrameDebugUtils {
+ private FrameDebugUtils() {
+ }
+
+ /**
+ * Debugging method
+ * @param fta
+ * @param recordDescriptor
+ * @param prefix
+ */
+ public void prettyPrint(IFrameTupleAccessor fta, RecordDescriptor recordDescriptor, String prefix) {
+ try (ByteBufferInputStream bbis = new ByteBufferInputStream();
+ DataInputStream dis = new DataInputStream(bbis)) {
+ int tc = fta.getTupleCount();
+ StringBuilder sb = new StringBuilder();
+ sb.append(prefix).append("TC: " + tc).append("\n");
+ for (int i = 0; i < tc; ++i) {
+ prettyPrint(fta, recordDescriptor, i, bbis, dis, sb);
+ }
+ System.err.println(sb.toString());
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * Debugging method
+ * @param fta
+ * @param recordDescriptor
+ */
+ public void prettyPrint(IFrameTupleAccessor fta, RecordDescriptor recordDescriptor) {
+ prettyPrint(fta, recordDescriptor, "");
+ }
+
+ /**
+ * Debugging method
+ * @param fta
+ * @param operator
+ */
+ public void prettyPrintTags(IFrameTupleAccessor fta, String operator) {
+ try (ByteBufferInputStream bbis = new ByteBufferInputStream();
+ DataInputStream dis = new DataInputStream(bbis)) {
+ int tc = fta.getTupleCount();
+ StringBuilder sb = new StringBuilder();
+ sb.append(operator + ":");
+ sb.append("TC: " + tc).append("\n");
+ for (int i = 0; i < tc; ++i) {
+ prettyPrintTag(fta, i, bbis, dis, sb);
+ }
+ System.err.println(sb.toString());
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * Debugging method
+ * @param fta
+ * @param tid
+ * @param bbis
+ * @param dis
+ * @param sb
+ */
+ protected void prettyPrintTag(IFrameTupleAccessor fta, int tid, ByteBufferInputStream bbis, DataInputStream dis,
+ StringBuilder sb) {
+ sb.append(" tid" + tid + ":(" + fta.getTupleStartOffset(tid) + ", " + fta.getTupleEndOffset(tid) + ")[");
+ for (int j = 0; j < fta.getFieldCount(); ++j) {
+ sb.append(" ");
+ if (j > 0) {
+ sb.append("|");
+ }
+ sb.append("f" + j + ":(" + fta.getFieldStartOffset(tid, j) + ", " + fta.getFieldEndOffset(tid, j) + ") ");
+ sb.append("{");
+ sb.append(Byte.toString(fta.getBuffer().array()[fta.getTupleStartOffset(tid) + fta.getFieldSlotsLength()
+ + fta.getFieldStartOffset(tid, j)]));
+ sb.append("}");
+ }
+ sb.append("\n");
+ }
+
+ /**
+ * Debugging method
+ * @param fta
+ * @param recordDescriptor
+ * @param tid
+ * @param bbis
+ * @param dis
+ * @param sb
+ */
+ protected void prettyPrint(IFrameTupleAccessor fta, RecordDescriptor recordDescriptor, int tid,
+ ByteBufferInputStream bbis, DataInputStream dis,
+ StringBuilder sb) {
+ sb.append(" tid" + tid + ":(" + fta.getTupleStartOffset(tid) + ", " + fta.getTupleEndOffset(tid) + ")[");
+ for (int j = 0; j < fta.getFieldCount(); ++j) {
+ sb.append(" ");
+ if (j > 0) {
+ sb.append("|");
+ }
+ sb.append("f" + j + ":(" + fta.getFieldStartOffset(tid, j) + ", " + fta.getFieldEndOffset(tid, j) + ") ");
+ sb.append("{");
+ bbis.setByteBuffer(fta.getBuffer(), fta.getTupleStartOffset(tid) + fta.getFieldSlotsLength() + fta
+ .getFieldStartOffset(tid, j));
+ try {
+ sb.append(recordDescriptor.getFields()[j].deserialize(dis));
+ } catch (Exception e) {
+ e.printStackTrace();
+ sb.append("Failed to deserialize field" + j);
+ }
+ sb.append("}");
+ }
+ sb.append("\n");
+ }
+
+
+ /**
+ * Debugging method
+ * @param fta
+ * @param recordDescriptor
+ * @param tid
+ */
+ public void prettyPrint(IFrameTupleAccessor fta, RecordDescriptor recordDescriptor, int tid) {
+ try (ByteBufferInputStream bbis = new ByteBufferInputStream();
+ DataInputStream dis = new DataInputStream(bbis)) {
+ StringBuilder sb = new StringBuilder();
+ prettyPrint(fta, recordDescriptor, tid, bbis, dis, sb);
+ System.err.println(sb.toString());
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * Debugging method
+ * They are safe as they don't print records. Printing records
+ * using IserializerDeserializer can print incorrect results or throw exceptions.
+ * A better way yet would be to use record pointable.
+ * @param fta
+ * @param recordDescriptor
+ * @param prefix
+ * @param recordFields
+ * @throws IOException
+ */
+ public void prettyPrint(IFrameTupleAccessor fta, RecordDescriptor recordDescriptor, String prefix,
+ int[] recordFields) throws IOException {
+ try (ByteBufferInputStream bbis = new ByteBufferInputStream();
+ DataInputStream dis = new DataInputStream(bbis)) {
+ int tc = fta.getTupleCount();
+ StringBuilder sb = new StringBuilder();
+ sb.append(prefix).append("TC: " + tc).append("\n");
+ for (int i = 0; i < tc; ++i) {
+ prettyPrint(fta, recordDescriptor, i, bbis, dis, sb, recordFields);
+ }
+ System.err.println(sb.toString());
+ }
+ }
+
+ /**
+ * Debugging method
+ * @param fta
+ * @param recordDescriptor
+ * @param tIdx
+ * @param recordFields
+ * @throws IOException
+ */
+ public void prettyPrint(IFrameTupleAccessor fta, RecordDescriptor recordDescriptor, int tIdx, int[] recordFields)
+ throws IOException {
+ try (ByteBufferInputStream bbis = new ByteBufferInputStream();
+ DataInputStream dis = new DataInputStream(bbis)) {
+ StringBuilder sb = new StringBuilder();
+ prettyPrint(fta, recordDescriptor, tIdx, bbis, dis, sb, recordFields);
+ System.err.println(sb.toString());
+ }
+ }
+
+ /**
+ * Debugging method
+ * @param tuple
+ * @param fieldsIdx
+ * @param descIdx
+ * @throws HyracksDataException
+ */
+ public void prettyPrint(IFrameTupleAccessor fta, RecordDescriptor recordDescriptor, ITupleReference tuple,
+ int fieldsIdx, int descIdx)
+ throws HyracksDataException {
+ try (ByteBufferInputStream bbis = new ByteBufferInputStream();
+ DataInputStream dis = new DataInputStream(bbis)) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("[");
+ sb.append("f" + fieldsIdx + ":(" + tuple.getFieldStart(fieldsIdx) + ", "
+ + (tuple.getFieldLength(fieldsIdx) + tuple.getFieldStart(fieldsIdx)) + ") ");
+ sb.append("{");
+ ByteBuffer bytebuff = ByteBuffer.wrap(tuple.getFieldData(fieldsIdx));
+ bbis.setByteBuffer(bytebuff, tuple.getFieldStart(fieldsIdx));
+ sb.append(recordDescriptor.getFields()[descIdx].deserialize(dis));
+ sb.append("}");
+ sb.append("\n");
+ System.err.println(sb.toString());
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * Debugging method
+ * @param tuple
+ * @param descF
+ * @throws HyracksDataException
+ */
+ public void prettyPrint(IFrameTupleAccessor fta, RecordDescriptor recordDescriptor, ITupleReference tuple,
+ int[] descF) throws HyracksDataException {
+ try (ByteBufferInputStream bbis = new ByteBufferInputStream();
+ DataInputStream dis = new DataInputStream(bbis)) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("[");
+ for (int j = 0; j < descF.length; ++j) {
+ sb.append("f" + j + ":(" + tuple.getFieldStart(j) + ", "
+ + (tuple.getFieldLength(j) + tuple.getFieldStart(j)) + ") ");
+ sb.append("{");
+ ByteBuffer bytebuff = ByteBuffer.wrap(tuple.getFieldData(j));
+ bbis.setByteBuffer(bytebuff, tuple.getFieldStart(j));
+ sb.append(recordDescriptor.getFields()[descF[j]].deserialize(dis));
+ sb.append("}");
+ }
+ sb.append("\n");
+ System.err.println(sb.toString());
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * Debugging method
+ * @param fta
+ * @param recordDescriptor
+ * @param tid
+ * @param bbis
+ * @param dis
+ * @param sb
+ * @param recordFields
+ * @throws IOException
+ */
+ protected void prettyPrint(IFrameTupleAccessor fta, RecordDescriptor recordDescriptor, int tid,
+ ByteBufferInputStream bbis, DataInputStream dis,
+ StringBuilder sb,
+ int[] recordFields) throws IOException {
+ Arrays.sort(recordFields);
+ sb.append(" tid" + tid + ":(" + fta.getTupleStartOffset(tid) + ", " + fta.getTupleEndOffset(tid) + ")[");
+ for (int j = 0; j < fta.getFieldCount(); ++j) {
+ sb.append("f" + j + ":(" + fta.getFieldStartOffset(tid, j) + ", " + fta.getFieldEndOffset(tid, j) + ") ");
+ sb.append("{");
+ bbis.setByteBuffer(fta.getBuffer(), fta.getTupleStartOffset(tid) + fta
+ .getFieldSlotsLength() + fta.getFieldStartOffset(tid, j));
+ if (Arrays.binarySearch(recordFields, j) >= 0) {
+ sb.append("{a record field: only print using pointable:");
+ sb.append("tag->" + dis.readByte() + "}");
+ } else {
+ sb.append(recordDescriptor.getFields()[j].deserialize(dis));
+ }
+ sb.append("}");
+ }
+ sb.append("\n");
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/452c5d09/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/unit/RunMergingFrameReaderTest.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/unit/RunMergingFrameReaderTest.java b/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/unit/RunMergingFrameReaderTest.java
index 567b7df..c68d59d 100644
--- a/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/unit/RunMergingFrameReaderTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/unit/RunMergingFrameReaderTest.java
@@ -31,7 +31,6 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.DataInputStream;
-import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
@@ -40,8 +39,6 @@ import java.util.List;
import java.util.Map;
import java.util.TreeMap;
-import junit.extensions.PA;
-
import org.apache.hyracks.api.comm.FrameHelper;
import org.apache.hyracks.api.comm.IFrame;
import org.apache.hyracks.api.comm.IFrameReader;
@@ -62,6 +59,8 @@ import org.apache.hyracks.dataflow.std.sort.RunMergingFrameReader;
import org.apache.hyracks.dataflow.std.sort.util.GroupVSizeFrame;
import org.junit.Test;
+import junit.extensions.PA;
+
public class RunMergingFrameReaderTest {
static IBinaryComparator[] Comparators = new IBinaryComparator[] { ComparatorFactories[0].createBinaryComparator(),
ComparatorFactories[1].createBinaryComparator(), };
@@ -125,12 +124,6 @@ public class RunMergingFrameReaderTest {
return true;
}
- private void printFrame(ByteBuffer buffer) {
- FrameTupleAccessor fta = new FrameTupleAccessor(RecordDesc);
- fta.reset(buffer);
- fta.prettyPrint();
- }
-
@Override
public void close() throws HyracksDataException {
}