BTree search op now takes input tuples as search keys. Added ConstantTupleSourceProviderOp to feed constant search keys. Modified BTree app example to search secondary index and then retrieve corresponding records from primary index.
git-svn-id: https://hyracks.googlecode.com/svn/trunk/hyracks@171 123451ca-8445-de46-9d55-352943316053
diff --git a/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexBulkLoadExample.java b/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexBulkLoadExample.java
index 9f34737..0725929 100644
--- a/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexBulkLoadExample.java
+++ b/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexBulkLoadExample.java
@@ -61,7 +61,7 @@
@Option(name = "-port", usage = "Hyracks Cluster Controller Port (default: 1099)")
public int port = 1099;
-
+
@Option(name = "-app", usage = "Hyracks Application name", required = true)
public String app;
diff --git a/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexSearchExample.java b/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexSearchExample.java
index c84ed4e..0d419e6 100644
--- a/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexSearchExample.java
+++ b/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexSearchExample.java
@@ -14,6 +14,7 @@
*/
package edu.uci.ics.hyracks.examples.btree.client;
+import java.io.DataOutput;
import java.util.UUID;
import org.kohsuke.args4j.CmdLineParser;
@@ -28,7 +29,9 @@
import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.JobSpecification;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import edu.uci.ics.hyracks.dataflow.common.data.comparators.IntegerBinaryComparatorFactory;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
@@ -40,11 +43,10 @@
import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrameFactory;
import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.ConstantTupleSourceOperatorDescriptor;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBTreeRegistryProvider;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBufferCacheProvider;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.IFileMappingProviderProvider;
-import edu.uci.ics.hyracks.storage.am.btree.dataflow.ITupleReferenceFactory;
-import edu.uci.ics.hyracks.storage.am.btree.dataflow.NullTupleReferenceFactory;
import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
@@ -73,7 +75,7 @@
Options options = new Options();
CmdLineParser parser = new CmdLineParser(options);
parser.parseArgument(args);
-
+
IHyracksClientConnection hcc = new HyracksRMIConnection(options.host, options.port);
JobSpecification job = createJob(options);
@@ -86,7 +88,7 @@
System.err.println(start + " " + end + " " + (end - start));
}
- private static JobSpecification createJob(Options options) {
+ private static JobSpecification createJob(Options options) throws HyracksDataException {
JobSpecification spec = new JobSpecification();
@@ -111,12 +113,28 @@
IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[1];
comparatorFactories[0] = IntegerBinaryComparatorFactory.INSTANCE;
- // build search key factories
- ITupleReferenceFactory[] searchKeys = new ITupleReferenceFactory[2];
- searchKeys[0] = new NullTupleReferenceFactory(); // equivalent to -infinity
- searchKeys[1] = new NullTupleReferenceFactory(); // equivalent to +infinity
+ // build tuple containing low and high search keys
+ ArrayTupleBuilder tb = new ArrayTupleBuilder(comparatorFactories.length*2); // high key and low key
+ DataOutput dos = tb.getDataOutput();
+
+ tb.reset();
+ IntegerSerializerDeserializer.INSTANCE.serialize(100, dos); // low key
+ tb.addFieldEndOffset();
+ IntegerSerializerDeserializer.INSTANCE.serialize(200, dos); // build high key
+ tb.addFieldEndOffset();
+
+ ISerializerDeserializer[] keyRecDescSers = { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE };
+ RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
+
+ ConstantTupleSourceOperatorDescriptor keyProviderOp = new ConstantTupleSourceOperatorDescriptor(spec, keyRecDesc, tb.getFieldEndOffsets(), tb.getByteArray(), tb.getSize());
+ PartitionConstraint keyProviderPartitionConstraint = createPartitionConstraint(splitNCs);
+ keyProviderOp.setPartitionConstraint(keyProviderPartitionConstraint);
+
+
+ int[] lowKeyFields = { 0 }; // low key is in field 0 of tuples going into search op
+ int[] highKeyFields = { 1 }; // low key is in field 1 of tuples going into search op
- BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, options.btreeName, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, recDesc.getFields().length, comparatorFactories, true, searchKeys, comparatorFactories.length);
+ BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, options.btreeName, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, recDesc.getFields().length, comparatorFactories, true, lowKeyFields, highKeyFields);
PartitionConstraint btreeSearchConstraint = createPartitionConstraint(splitNCs);
btreeSearchOp.setPartitionConstraint(btreeSearchConstraint);
@@ -125,6 +143,8 @@
PartitionConstraint printerConstraint = createPartitionConstraint(splitNCs);
printer.setPartitionConstraint(printerConstraint);
+ spec.connect(new OneToOneConnectorDescriptor(spec), keyProviderOp, 0, btreeSearchOp, 0);
+
spec.connect(new OneToOneConnectorDescriptor(spec), btreeSearchOp, 0, printer, 0);
spec.addRoot(printer);
diff --git a/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexBulkLoadExample.java b/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexBulkLoadExample.java
index a39126d..a2efc18 100644
--- a/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexBulkLoadExample.java
+++ b/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexBulkLoadExample.java
@@ -68,7 +68,7 @@
@Option(name = "-primary-btreename", usage = "Name of primary-index B-Tree to load from", required = true)
public String primaryBtreeName;
- @Option(name = "-btreename", usage = "B-Tree file name for secondary index to be built", required = true)
+ @Option(name = "-secondary-btreename", usage = "B-Tree file name for secondary index to be built", required = true)
public String btreeName;
@Option(name = "-sortbuffer-size", usage = "Sort buffer size in frames (default: 32768)", required = false)
diff --git a/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexSearchExample.java b/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexSearchExample.java
index a74d7a0..3302674 100644
--- a/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexSearchExample.java
+++ b/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexSearchExample.java
@@ -15,7 +15,6 @@
package edu.uci.ics.hyracks.examples.btree.client;
import java.io.DataOutput;
-import java.nio.ByteBuffer;
import java.util.UUID;
import org.kohsuke.args4j.CmdLineParser;
@@ -27,15 +26,12 @@
import edu.uci.ics.hyracks.api.constraints.ExplicitPartitionConstraint;
import edu.uci.ics.hyracks.api.constraints.LocationConstraint;
import edu.uci.ics.hyracks.api.constraints.PartitionConstraint;
-import edu.uci.ics.hyracks.api.context.IHyracksContext;
import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.JobSpecification;
-import edu.uci.ics.hyracks.control.nc.runtime.RootHyracksContext;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
-import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.data.comparators.UTF8StringBinaryComparatorFactory;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
@@ -47,15 +43,15 @@
import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrameFactory;
import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.btree.dataflow.FrameTupleReferenceFactory;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.ConstantTupleSourceOperatorDescriptor;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBTreeRegistryProvider;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBufferCacheProvider;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.IFileMappingProviderProvider;
-import edu.uci.ics.hyracks.storage.am.btree.dataflow.ITupleReferenceFactory;
import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
// This example will perform range search on the secondary index
+// and then retrieve the corresponding source records from the primary index
public class SecondaryIndexSearchExample {
private static class Options {
@@ -71,8 +67,11 @@
@Option(name = "-target-ncs", usage = "Comma separated list of node-controller names to use", required = true)
public String ncs;
- @Option(name = "-btreename", usage = "B-Tree file name to search", required = true)
- public String btreeName;
+ @Option(name = "-primary-btreename", usage = "Primary B-Tree file name", required = true)
+ public String primaryBTreeName;
+
+ @Option(name = "-secondary-btreename", usage = "Secondary B-Tree file name to search", required = true)
+ public String secondaryBTreeName;
}
public static void main(String[] args) throws Exception {
@@ -106,60 +105,67 @@
IFileMappingProviderProvider fileMappingProviderProvider = FileMappingProviderProvider.INSTANCE;
// schema of tuples coming out of secondary index
- RecordDescriptor recDesc = new RecordDescriptor(new ISerializerDeserializer[] {
+ RecordDescriptor secondaryRecDesc = new RecordDescriptor(new ISerializerDeserializer[] {
UTF8StringSerializerDeserializer.INSTANCE,
IntegerSerializerDeserializer.INSTANCE
});
+ // schema of tuples coming out of primary index
+ RecordDescriptor primaryRecDesc = new RecordDescriptor(new ISerializerDeserializer[] {
+ IntegerSerializerDeserializer.INSTANCE,
+ UTF8StringSerializerDeserializer.INSTANCE,
+ IntegerSerializerDeserializer.INSTANCE,
+ UTF8StringSerializerDeserializer.INSTANCE,
+ });
+
// comparators for btree, note that we only need a comparator for the non-unique key
// i.e. we will have a range condition on the first field only (implying [-infinity, +infinity] for the second field)
IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[1];
comparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;
-
-
- // build search keys (which must be of type ITupleReference)
- // put search keys into frame and create tuplereference factories
- IHyracksContext ctx = new RootHyracksContext(32768); // WARNING: make sure frame size is same as on NCs
- ByteBuffer keyFrame = ctx.getResourceManager().allocateFrame();
- FrameTupleAppender appender = new FrameTupleAppender(ctx);
- appender.reset(keyFrame, true);
- ArrayTupleBuilder tb = new ArrayTupleBuilder(comparatorFactories.length);
+
+ // build tuple containing low and high search keys
+ ArrayTupleBuilder tb = new ArrayTupleBuilder(comparatorFactories.length*2); // low and high key
DataOutput dos = tb.getDataOutput();
-
- ISerializerDeserializer[] keyRecDescSers = { UTF8StringSerializerDeserializer.INSTANCE };
- RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
-
- // build low key
- tb.reset();
- UTF8StringSerializerDeserializer.INSTANCE.serialize("a", dos);
+
+ tb.reset();
+ UTF8StringSerializerDeserializer.INSTANCE.serialize("0", dos); // low key
tb.addFieldEndOffset();
-
- appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
-
- // build high key
- tb.reset();
- UTF8StringSerializerDeserializer.INSTANCE.serialize("f", dos);
+ UTF8StringSerializerDeserializer.INSTANCE.serialize("f", dos); // high key
tb.addFieldEndOffset();
+
+ ISerializerDeserializer[] keyRecDescSers = { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE };
+ RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
- appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
-
- // build search key factories
- ITupleReferenceFactory[] searchKeys = new ITupleReferenceFactory[2];
- // the low key is tuple 0 in the keyFrame
- searchKeys[0] = new FrameTupleReferenceFactory(keyFrame.array(), 0, keyRecDesc);
- // the high key is tuple 1 in the keyFrame
- searchKeys[1] = new FrameTupleReferenceFactory(keyFrame.array(), 1, keyRecDesc);
-
- BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, options.btreeName, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, recDesc.getFields().length, comparatorFactories, true, searchKeys, comparatorFactories.length);
- PartitionConstraint btreeSearchConstraint = createPartitionConstraint(splitNCs);
- btreeSearchOp.setPartitionConstraint(btreeSearchConstraint);
+ ConstantTupleSourceOperatorDescriptor keyProviderOp = new ConstantTupleSourceOperatorDescriptor(spec, keyRecDesc, tb.getFieldEndOffsets(), tb.getByteArray(), tb.getSize());
+ PartitionConstraint keyProviderPartitionConstraint = createPartitionConstraint(splitNCs);
+ keyProviderOp.setPartitionConstraint(keyProviderPartitionConstraint);
+
+ int[] secondaryLowKeyFields = { 0 }; // low key is in field 0 of tuples going into secondary index search op
+ int[] secondaryHighKeyFields = { 1 }; // high key is in field 1 of tuples going into secondary index search op
+
+ BTreeSearchOperatorDescriptor secondarySearchOp = new BTreeSearchOperatorDescriptor(spec, secondaryRecDesc, bufferCacheProvider, btreeRegistryProvider, options.secondaryBTreeName, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, secondaryRecDesc.getFields().length, comparatorFactories, true, secondaryLowKeyFields, secondaryHighKeyFields);
+ PartitionConstraint secondarySearchConstraint = createPartitionConstraint(splitNCs);
+ secondarySearchOp.setPartitionConstraint(secondarySearchConstraint);
+
+ // secondary index will output tuples with [UTF8String, Integer]
+ // the Integer field refers to the key in the primary index of the source data records
+ int[] primaryLowKeyFields = { 1 }; // low key is in field 0 of tuples going into primary index search op
+ int[] primaryHighKeyFields = { 1 }; // high key is in field 1 of tuples going into primary index search op
+
+ BTreeSearchOperatorDescriptor primarySearchOp = new BTreeSearchOperatorDescriptor(spec, primaryRecDesc, bufferCacheProvider, btreeRegistryProvider, options.primaryBTreeName, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, primaryRecDesc.getFields().length, comparatorFactories, true, primaryLowKeyFields, primaryHighKeyFields);
+ PartitionConstraint primarySearchConstraint = createPartitionConstraint(splitNCs);
+ primarySearchOp.setPartitionConstraint(primarySearchConstraint);
// have each node print the results of its respective B-Tree
PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
PartitionConstraint printerConstraint = createPartitionConstraint(splitNCs);
printer.setPartitionConstraint(printerConstraint);
- spec.connect(new OneToOneConnectorDescriptor(spec), btreeSearchOp, 0, printer, 0);
+ spec.connect(new OneToOneConnectorDescriptor(spec), keyProviderOp, 0, secondarySearchOp, 0);
+
+ spec.connect(new OneToOneConnectorDescriptor(spec), secondarySearchOp, 0, primarySearchOp, 0);
+
+ spec.connect(new OneToOneConnectorDescriptor(spec), primarySearchOp, 0, printer, 0);
spec.addRoot(printer);
diff --git a/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java b/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
index 7fca1d3..cf7c847 100644
--- a/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
+++ b/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
@@ -18,7 +18,6 @@
import java.io.DataOutput;
import java.io.File;
import java.io.RandomAccessFile;
-import java.nio.ByteBuffer;
import org.junit.Test;
@@ -26,16 +25,12 @@
import edu.uci.ics.hyracks.api.constraints.ExplicitPartitionConstraint;
import edu.uci.ics.hyracks.api.constraints.LocationConstraint;
import edu.uci.ics.hyracks.api.constraints.PartitionConstraint;
-import edu.uci.ics.hyracks.api.context.IHyracksContext;
-import edu.uci.ics.hyracks.api.dataflow.IConnectorDescriptor;
import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.job.JobSpecification;
-import edu.uci.ics.hyracks.control.nc.runtime.RootHyracksContext;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
-import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
import edu.uci.ics.hyracks.dataflow.common.data.comparators.UTF8StringBinaryComparatorFactory;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
@@ -59,12 +54,11 @@
import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeRegistryProvider;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.BufferCacheProvider;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.ConstantTupleSourceOperatorDescriptor;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.FileMappingProviderProvider;
-import edu.uci.ics.hyracks.storage.am.btree.dataflow.FrameTupleReferenceFactory;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBTreeRegistryProvider;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBufferCacheProvider;
import edu.uci.ics.hyracks.storage.am.btree.dataflow.IFileMappingProviderProvider;
-import edu.uci.ics.hyracks.storage.am.btree.dataflow.ITupleReferenceFactory;
import edu.uci.ics.hyracks.storage.am.btree.frames.MetaDataFrame;
import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
@@ -83,7 +77,7 @@
@Test
public void bulkLoadTest() throws Exception {
// relies on the fact that NCs are run from same process
- System.setProperty("NodeControllerDataPath", "/tmp/");
+ System.setProperty("NodeControllerDataPath", System.getProperty("java.io.tmpdir") + "/");
JobSpecification spec = new JobSpecification();
@@ -170,7 +164,7 @@
@Test
public void btreeSearchTest() throws Exception {
// relies on the fact that NCs are run from same process
- System.setProperty("NodeControllerDataPath", "/tmp/");
+ System.setProperty("NodeControllerDataPath", System.getProperty("java.io.tmpdir") + "/");
JobSpecification spec = new JobSpecification();
@@ -191,51 +185,32 @@
comparators[i] = comparatorFactories[i].createBinaryComparator();
}
MultiComparator cmp = new MultiComparator(fieldCount, comparators);
-
-
- // put search keys into frame and create tuplereference factories
- IHyracksContext ctx = new RootHyracksContext(32768); // WARNING: make sure frame size is same as on NCs
- ByteBuffer keyFrame = ctx.getResourceManager().allocateFrame();
- FrameTupleAppender appender = new FrameTupleAppender(ctx);
- appender.reset(keyFrame, true);
- ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+
+ // build tuple containing low and high search key
+ ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getKeyFieldCount()*2); // high key and low key
DataOutput dos = tb.getDataOutput();
- ISerializerDeserializer[] keyRecDescSers = { UTF8StringSerializerDeserializer.INSTANCE};
+ tb.reset();
+ UTF8StringSerializerDeserializer.INSTANCE.serialize("100", dos); // low key
+ tb.addFieldEndOffset();
+ UTF8StringSerializerDeserializer.INSTANCE.serialize("200", dos); // high key
+ tb.addFieldEndOffset();
+
+ ISerializerDeserializer[] keyRecDescSers = { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE };
RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
- // build low key
- tb.reset();
- UTF8StringSerializerDeserializer.INSTANCE.serialize("100", dos);
- tb.addFieldEndOffset();
-
- appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
-
- // build high key
- tb.reset();
- UTF8StringSerializerDeserializer.INSTANCE.serialize("200", dos);
- tb.addFieldEndOffset();
-
- appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
-
- // build search key factories
- ITupleReferenceFactory[] searchKeys = new ITupleReferenceFactory[2];
- searchKeys[0] = new FrameTupleReferenceFactory(keyFrame.array(), 0, keyRecDesc);
- searchKeys[1] = new FrameTupleReferenceFactory(keyFrame.array(), 1, keyRecDesc);
+ ConstantTupleSourceOperatorDescriptor keyProviderOp = new ConstantTupleSourceOperatorDescriptor(spec, keyRecDesc, tb.getFieldEndOffsets(), tb.getByteArray(), tb.getSize());
+ PartitionConstraint keyProviderPartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+ keyProviderOp.setPartitionConstraint(keyProviderPartitionConstraint);
IBufferCacheProvider bufferCacheProvider = new BufferCacheProvider();
IBTreeRegistryProvider btreeRegistryProvider = new BTreeRegistryProvider();
IFileMappingProviderProvider fileMappingProviderProvider = new FileMappingProviderProvider();
RecordDescriptor recDesc = new RecordDescriptor(
- new ISerializerDeserializer[] { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE });
-
- // TODO: hacky, manually adding file
- //String ncDataPath = System.getProperty("NodeControllerDataPath");
- //String fileName = ncDataPath + "btreetest.bin";
- //int blubb = fileMappingProviderProvider.getFileMappingProvider().mapNameToFileId(fileName, true);
-
- BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, "btreetest.bin", fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, fieldCount, comparatorFactories, true, searchKeys, comparatorFactories.length);
+ new ISerializerDeserializer[] { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE });
+
+ BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, "btreetest.bin", fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, fieldCount, comparatorFactories, true, new int[]{0}, new int[]{1});
//BTreeDiskOrderScanOperatorDescriptor btreeSearchOp = new BTreeDiskOrderScanOperatorDescriptor(spec, splitProvider, recDesc, bufferCacheProvider, btreeRegistryProvider, 0, "btreetest.bin", interiorFrameFactory, leafFrameFactory, cmp);
PartitionConstraint btreePartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
@@ -244,9 +219,9 @@
PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
PartitionConstraint printerPartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
printer.setPartitionConstraint(printerPartitionConstraint);
-
- IConnectorDescriptor conn = new OneToOneConnectorDescriptor(spec);
- spec.connect(conn, btreeSearchOp, 0, printer, 0);
+
+ spec.connect(new OneToOneConnectorDescriptor(spec), keyProviderOp, 0, btreeSearchOp, 0);
+ spec.connect(new OneToOneConnectorDescriptor(spec), btreeSearchOp, 0, printer, 0);
spec.addRoot(printer);
runTest(spec);
@@ -255,7 +230,7 @@
@Test
public void insertTest() throws Exception {
// relies on the fact that NCs are run from same process
- System.setProperty("NodeControllerDataPath", "/tmp/");
+ System.setProperty("NodeControllerDataPath", System.getProperty("java.io.tmpdir") + "/");
JobSpecification spec = new JobSpecification();
@@ -450,5 +425,5 @@
scanCursorC.close();
}
System.out.println();
- }
+ }
}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/AbstractBTreeOperatorDescriptor.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/AbstractBTreeOperatorDescriptor.java
index fa3a748..03f60a0 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/AbstractBTreeOperatorDescriptor.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/AbstractBTreeOperatorDescriptor.java
@@ -19,10 +19,8 @@
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.job.JobSpecification;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
-import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrameFactory;
import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
-import edu.uci.ics.hyracks.storage.common.file.IFileMappingProvider;
public abstract class AbstractBTreeOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorNodePushable.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorNodePushable.java
index 5aa239b..e1fedda 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorNodePushable.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorNodePushable.java
@@ -43,6 +43,8 @@
private PermutingFrameTupleReference tuple = new PermutingFrameTupleReference();
+ private ByteBuffer writeBuffer;
+
public BTreeInsertUpdateDeleteOperatorNodePushable(AbstractBTreeOperatorDescriptor opDesc, IHyracksContext ctx,
int[] fieldPermutation, IRecordDescriptorProvider recordDescProvider, BTreeOp op) {
btreeOpHelper = new BTreeOpHelper(opDesc, ctx, false);
@@ -93,7 +95,8 @@
}
// pass a copy of the frame to next op
- FrameUtils.flushFrame(buffer.duplicate(), writer);
+ System.arraycopy(buffer.array(), 0, writeBuffer.array(), 0, buffer.capacity());
+ FrameUtils.flushFrame(writeBuffer, writer);
}
@Override
@@ -101,6 +104,7 @@
AbstractBTreeOperatorDescriptor opDesc = btreeOpHelper.getOperatorDescriptor();
RecordDescriptor recDesc = recordDescProvider.getInputRecordDescriptor(opDesc.getOperatorId(), 0);
accessor = new FrameTupleAccessor(btreeOpHelper.getHyracksContext(), recDesc);
+ writeBuffer = btreeOpHelper.getHyracksContext().getResourceManager().allocateFrame();
try {
btreeOpHelper.init();
btreeOpHelper.getBTree().open(btreeOpHelper.getBTreeFileId());
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorDescriptor.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorDescriptor.java
index 1e4ffee..b75b3a9 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorDescriptor.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorDescriptor.java
@@ -30,19 +30,19 @@
private static final long serialVersionUID = 1L;
private boolean isForward;
- private ITupleReferenceFactory[] searchKeys; // create tuples for low and high keys
- private int searchKeyFieldCount;
+ private int[] lowKeyFields; // fields in input tuple to be used as low keys
+ private int[] highKeyFields; // fields in input tuple to be used as high keys
- public BTreeSearchOperatorDescriptor(JobSpecification spec, RecordDescriptor recDesc, IBufferCacheProvider bufferCacheProvider, IBTreeRegistryProvider btreeRegistryProvider, String btreeFileName, IFileMappingProviderProvider fileMappingProviderProvider, IBTreeInteriorFrameFactory interiorFactory, IBTreeLeafFrameFactory leafFactory, int fieldCount, IBinaryComparatorFactory[] comparatorFactories, boolean isForward, ITupleReferenceFactory[] searchKeys, int searchKeyFields) {
- super(spec, 0, 1, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeFileName, fileMappingProviderProvider, interiorFactory, leafFactory, fieldCount, comparatorFactories);
+ public BTreeSearchOperatorDescriptor(JobSpecification spec, RecordDescriptor recDesc, IBufferCacheProvider bufferCacheProvider, IBTreeRegistryProvider btreeRegistryProvider, String btreeFileName, IFileMappingProviderProvider fileMappingProviderProvider, IBTreeInteriorFrameFactory interiorFactory, IBTreeLeafFrameFactory leafFactory, int fieldCount, IBinaryComparatorFactory[] comparatorFactories, boolean isForward, int[] lowKeyFields, int[] highKeyFields) {
+ super(spec, 1, 1, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeFileName, fileMappingProviderProvider, interiorFactory, leafFactory, fieldCount, comparatorFactories);
this.isForward = isForward;
- this.searchKeys = searchKeys;
- this.searchKeyFieldCount = searchKeyFields;
+ this.lowKeyFields = lowKeyFields;
+ this.highKeyFields = highKeyFields;
}
@Override
public IOperatorNodePushable createPushRuntime(final IHyracksContext ctx, final IOperatorEnvironment env,
IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
- return new BTreeSearchOperatorNodePushable(this, ctx, isForward, searchKeys, searchKeyFieldCount);
+ return new BTreeSearchOperatorNodePushable(this, ctx, recordDescProvider, isForward, lowKeyFields, highKeyFields);
}
}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
index 5f5d916..80b04ac 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
@@ -19,12 +19,15 @@
import edu.uci.ics.hyracks.api.context.IHyracksContext;
import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
+import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
+import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperatorNodePushable;
import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeCursor;
import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
@@ -33,96 +36,139 @@
import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
import edu.uci.ics.hyracks.storage.am.btree.impls.RangeSearchCursor;
-public class BTreeSearchOperatorNodePushable extends AbstractUnaryOutputSourceOperatorNodePushable {
- private BTreeOpHelper btreeOpHelper;
- private boolean isForward;
- private ITupleReferenceFactory[] searchKeys;
- private int searchKeyFieldCount;
-
- public BTreeSearchOperatorNodePushable(AbstractBTreeOperatorDescriptor opDesc, IHyracksContext ctx,
- boolean isForward, ITupleReferenceFactory[] searchKeys, int searchKeyFields) {
+public class BTreeSearchOperatorNodePushable extends AbstractUnaryInputUnaryOutputOperatorNodePushable {
+ private BTreeOpHelper btreeOpHelper;
+ private boolean isForward;
+ private FrameTupleAccessor accessor;
+
+ private ByteBuffer writeBuffer;
+ private FrameTupleAppender appender;
+ private ArrayTupleBuilder tb;
+ private DataOutput dos;
+
+ private BTree btree;
+ private PermutingFrameTupleReference lowKey;
+ private PermutingFrameTupleReference highKey;
+ private RangePredicate rangePred;
+ private MultiComparator searchCmp;
+ private IBTreeCursor cursor;
+ private IBTreeLeafFrame leafFrame;
+ private IBTreeLeafFrame cursorFrame;
+ private IBTreeInteriorFrame interiorFrame;
+
+ private RecordDescriptor recDesc;
+
+ public BTreeSearchOperatorNodePushable(AbstractBTreeOperatorDescriptor opDesc, IHyracksContext ctx, IRecordDescriptorProvider recordDescProvider, boolean isForward, int[] lowKeyFields, int[] highKeyFields) {
btreeOpHelper = new BTreeOpHelper(opDesc, ctx, false);
- this.isForward = isForward;
- this.searchKeys = searchKeys;
- this.searchKeyFieldCount = searchKeyFields;
+ this.isForward = isForward;
+ this.recDesc = recordDescProvider.getInputRecordDescriptor(opDesc.getOperatorId(), 0);
+ if(lowKeyFields != null && lowKeyFields.length > 0) {
+ lowKey = new PermutingFrameTupleReference();
+ lowKey.setFieldPermutation(lowKeyFields);
+ }
+ if(highKeyFields != null && highKeyFields.length > 0) {
+ highKey = new PermutingFrameTupleReference();
+ highKey.setFieldPermutation(highKeyFields);
+ }
}
-
+
@Override
- public void initialize() throws HyracksDataException {
- AbstractBTreeOperatorDescriptor opDesc = btreeOpHelper.getOperatorDescriptor();
- IHyracksContext ctx = btreeOpHelper.getHyracksContext();
-
- IBTreeLeafFrame cursorFrame = opDesc.getLeafFactory().getFrame();
- IBTreeCursor cursor = new RangeSearchCursor(cursorFrame);
-
- BTree btree = null;
+ public void open() throws HyracksDataException {
+ AbstractBTreeOperatorDescriptor opDesc = btreeOpHelper.getOperatorDescriptor();
+ accessor = new FrameTupleAccessor(btreeOpHelper.getHyracksContext(), recDesc);
+
+ cursorFrame = opDesc.getLeafFactory().getFrame();
+ cursor = new RangeSearchCursor(cursorFrame);
+
try {
- btreeOpHelper.init();
- //btreeOpHelper.fill();
- btree = btreeOpHelper.getBTree();
+ btreeOpHelper.init();
+ } catch (Exception e) {
+ throw new HyracksDataException(e.getMessage());
+ }
+ btree = btreeOpHelper.getBTree();
- IBTreeLeafFrame leafFrame = btreeOpHelper.getLeafFrame();
- IBTreeInteriorFrame interiorFrame = btreeOpHelper.getInteriorFrame();
-
- // construct range predicate
- assert (searchKeyFieldCount <= btree.getMultiComparator().getKeyFieldCount());
- IBinaryComparator[] searchComparators = new IBinaryComparator[searchKeyFieldCount];
- for (int i = 0; i < searchKeyFieldCount; i++) {
- searchComparators[i] = btree.getMultiComparator().getComparators()[i];
- }
- MultiComparator searchCmp = new MultiComparator(btree.getMultiComparator().getFieldCount(),
- searchComparators);
-
- ITupleReference lowKey = searchKeys[0].createTuple(ctx);
- ITupleReference highKey = searchKeys[1].createTuple(ctx);
-
- RangePredicate rangePred = new RangePredicate(isForward, lowKey, highKey, searchCmp);
-
- btree.search(cursor, rangePred, leafFrame, interiorFrame);
- } catch (Exception e) {
- throw new HyracksDataException(e);
+ leafFrame = btreeOpHelper.getLeafFrame();
+ interiorFrame = btreeOpHelper.getInteriorFrame();
+
+ // construct range predicate
+
+ int numSearchFields = btree.getMultiComparator().getComparators().length;
+ if(lowKey != null) numSearchFields = lowKey.getFieldCount();
+ IBinaryComparator[] searchComparators = new IBinaryComparator[numSearchFields];
+ for (int i = 0; i < numSearchFields; i++) {
+ searchComparators[i] = btree.getMultiComparator().getComparators()[i];
}
-
- MultiComparator cmp = btree.getMultiComparator();
- ByteBuffer frame = ctx.getResourceManager().allocateFrame();
- FrameTupleAppender appender = new FrameTupleAppender(ctx);
- appender.reset(frame, true);
- ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFieldCount());
- DataOutput dos = tb.getDataOutput();
-
- try {
- while (cursor.hasNext()) {
- tb.reset();
- cursor.next();
-
- ITupleReference frameTuple = cursor.getTuple();
- for (int i = 0; i < frameTuple.getFieldCount(); i++) {
- dos.write(frameTuple.getFieldData(i), frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
- tb.addFieldEndOffset();
- }
-
- if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
- FrameUtils.flushFrame(frame, writer);
- appender.reset(frame, true);
- if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
- throw new IllegalStateException();
- }
- }
-
- //int recOffset = cursor.getOffset();
- //String rec = cmp.printRecord(array, recOffset);
- //System.out.println(rec);
- }
-
- if (appender.getTupleCount() > 0) {
- FrameUtils.flushFrame(frame, writer);
- }
-
- cursor.close();
- writer.close();
-
- } catch (Exception e) {
- throw new HyracksDataException(e);
- }
+ searchCmp = new MultiComparator(btree.getMultiComparator().getFieldCount(), searchComparators);
+
+ rangePred = new RangePredicate(isForward, null, null, searchCmp);
+
+ accessor = new FrameTupleAccessor(btreeOpHelper.getHyracksContext(), recDesc);
+
+ writeBuffer = btreeOpHelper.getHyracksContext().getResourceManager().allocateFrame();
+ tb = new ArrayTupleBuilder(btree.getMultiComparator().getFieldCount());
+ dos = tb.getDataOutput();
+ appender = new FrameTupleAppender(btreeOpHelper.getHyracksContext());
}
+
+ private void writeSearchResults() throws Exception {
+ while (cursor.hasNext()) {
+ tb.reset();
+ cursor.next();
+
+ ITupleReference frameTuple = cursor.getTuple();
+ for (int i = 0; i < frameTuple.getFieldCount(); i++) {
+ dos.write(frameTuple.getFieldData(i), frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
+ tb.addFieldEndOffset();
+ }
+
+ if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
+ FrameUtils.flushFrame(writeBuffer, writer);
+ appender.reset(writeBuffer, true);
+ if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
+ throw new IllegalStateException();
+ }
+ }
+ }
+ }
+
+ @Override
+ public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
+ accessor.reset(buffer);
+
+ int tupleCount = accessor.getTupleCount();
+ try {
+ for (int i = 0; i < tupleCount; i++) {
+ if(lowKey != null) lowKey.reset(accessor, i);
+ if(highKey != null) highKey.reset(accessor, i);
+ rangePred.setLowKey(lowKey);
+ rangePred.setHighKey(highKey);
+
+ cursor.reset();
+ btree.search(cursor, rangePred, leafFrame, interiorFrame);
+ appender.reset(writeBuffer, true);
+ writeSearchResults();
+ }
+
+ if (appender.getTupleCount() > 0) {
+ FrameUtils.flushFrame(writeBuffer, writer);
+ }
+
+ } catch (Exception e) {
+ throw new HyracksDataException(e.getMessage());
+ }
+ }
+
+ @Override
+ public void close() throws HyracksDataException {
+ writer.close();
+ try {
+ cursor.close();
+ } catch (Exception e) {
+ throw new HyracksDataException(e.getMessage());
+ }
+ }
+
+ @Override
+ public void flush() throws HyracksDataException {
+ }
}
\ No newline at end of file
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ConstantTupleSourceOperatorDescriptor.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ConstantTupleSourceOperatorDescriptor.java
new file mode 100644
index 0000000..9ef860b
--- /dev/null
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ConstantTupleSourceOperatorDescriptor.java
@@ -0,0 +1,36 @@
+package edu.uci.ics.hyracks.storage.am.btree.dataflow;
+
+import edu.uci.ics.hyracks.api.context.IHyracksContext;
+import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
+import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.api.job.IOperatorEnvironment;
+import edu.uci.ics.hyracks.api.job.JobSpecification;
+import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
+
+public class ConstantTupleSourceOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
+
+ private static final long serialVersionUID = 1L;
+
+ private int[] fieldSlots;
+ private byte[] tupleData;
+ private int tupleSize;
+
+ public ConstantTupleSourceOperatorDescriptor(JobSpecification spec, RecordDescriptor recDesc, int[] fieldSlots, byte[] tupleData, int tupleSize) {
+ super(spec, 0, 1);
+ this.tupleData = tupleData;
+ this.fieldSlots = fieldSlots;
+ this.tupleSize = tupleSize;
+ recordDescriptors[0] = recDesc;
+ }
+
+ @Override
+ public IOperatorNodePushable createPushRuntime(IHyracksContext ctx,
+ IOperatorEnvironment env,
+ IRecordDescriptorProvider recordDescProvider, int partition,
+ int nPartitions) throws HyracksDataException {
+ return new ConstantTupleSourceOperatorNodePushable(ctx, fieldSlots, tupleData, tupleSize);
+ }
+
+}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ConstantTupleSourceOperatorNodePushable.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ConstantTupleSourceOperatorNodePushable.java
new file mode 100644
index 0000000..cbdb713
--- /dev/null
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ConstantTupleSourceOperatorNodePushable.java
@@ -0,0 +1,37 @@
+package edu.uci.ics.hyracks.storage.am.btree.dataflow;
+
+import java.nio.ByteBuffer;
+
+import edu.uci.ics.hyracks.api.context.IHyracksContext;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
+import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
+
+public class ConstantTupleSourceOperatorNodePushable extends AbstractUnaryOutputSourceOperatorNodePushable {
+
+ private IHyracksContext ctx;
+
+ private int[] fieldSlots;
+ private byte[] tupleData;
+ private int tupleSize;
+
+
+ public ConstantTupleSourceOperatorNodePushable(IHyracksContext ctx, int[] fieldSlots, byte[] tupleData, int tupleSize) {
+ super();
+ this.fieldSlots = fieldSlots;
+ this.tupleData = tupleData;
+ this.tupleSize = tupleSize;
+ this.ctx = ctx;
+ }
+
+ @Override
+ public void initialize() throws HyracksDataException {
+ ByteBuffer writeBuffer = ctx.getResourceManager().allocateFrame();
+ FrameTupleAppender appender = new FrameTupleAppender(ctx);
+ appender.reset(writeBuffer, true);
+ appender.append(fieldSlots, tupleData, 0, tupleSize);
+ FrameUtils.flushFrame(writeBuffer, writer);
+ writer.close();
+ }
+}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/FrameTupleReferenceFactory.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/FrameTupleReferenceFactory.java
deleted file mode 100644
index 0702c29..0000000
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/FrameTupleReferenceFactory.java
+++ /dev/null
@@ -1,35 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.btree.dataflow;
-
-import java.nio.ByteBuffer;
-
-import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
-import edu.uci.ics.hyracks.api.context.IHyracksContext;
-import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
-import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.FrameTupleReference;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-
-public class FrameTupleReferenceFactory implements ITupleReferenceFactory {
-
- private static final long serialVersionUID = 1L;
-
- private byte[] frame;
- private int tupleIndex;
- private RecordDescriptor recDesc;
-
- public FrameTupleReferenceFactory(byte[] frame, int tupleIndex, RecordDescriptor recDesc) {
- this.frame = frame;
- this.tupleIndex = tupleIndex;
- this.recDesc = recDesc;
- }
-
- // TODO: lots of object creation, fix later
- @Override
- public ITupleReference createTuple(IHyracksContext ctx) {
- IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx, recDesc);
- accessor.reset(ByteBuffer.wrap(frame));
- FrameTupleReference tuple = new FrameTupleReference();
- tuple.reset(accessor, tupleIndex);
- return tuple;
- }
-}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ITupleReferenceFactory.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ITupleReferenceFactory.java
deleted file mode 100644
index a370f84..0000000
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ITupleReferenceFactory.java
+++ /dev/null
@@ -1,10 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.btree.dataflow;
-
-import java.io.Serializable;
-
-import edu.uci.ics.hyracks.api.context.IHyracksContext;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-
-public interface ITupleReferenceFactory extends Serializable {
- ITupleReference createTuple(IHyracksContext ctx);
-}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/NullTupleReferenceFactory.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/NullTupleReferenceFactory.java
deleted file mode 100644
index f2163a1..0000000
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/NullTupleReferenceFactory.java
+++ /dev/null
@@ -1,15 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.btree.dataflow;
-
-import edu.uci.ics.hyracks.api.context.IHyracksContext;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-
-public class NullTupleReferenceFactory implements ITupleReferenceFactory {
-
- private static final long serialVersionUID = 1L;
-
- @Override
- public ITupleReference createTuple(IHyracksContext ctx) {
- return null;
- }
-
-}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java
index 74f385c..02928be 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java
@@ -59,4 +59,12 @@
public ITupleReference getHighKey() {
return highKey;
}
+
+ public void setLowKey(ITupleReference lowKey) {
+ this.lowKey = lowKey;
+ }
+
+ public void setHighKey(ITupleReference highKey) {
+ this.highKey = highKey;
+ }
}
diff --git a/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java b/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java
index 77ef504..2746392 100644
--- a/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java
+++ b/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java
@@ -62,6 +62,8 @@
private static final int NUM_PAGES = 40;
private static final int HYRACKS_FRAME_SIZE = 128;
+ private String tmpDir = System.getProperty("java.io.tmpdir");
+
// to help with the logger madness
private void print(String str) {
System.out.print(str);
@@ -121,7 +123,7 @@
IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
- File f = new File("/tmp/btreetest.bin");
+ File f = new File(tmpDir + "/" + "btreetest.bin");
RandomAccessFile raf = new RandomAccessFile(f, "rw");
int fileId = 0;
FileInfo fi = new FileInfo(fileId, raf);
diff --git a/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java b/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
index 9d4f390..b4ed2d4 100644
--- a/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
+++ b/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
@@ -72,6 +72,8 @@
private static final int NUM_PAGES = 10;
private static final int HYRACKS_FRAME_SIZE = 128;
+ private String tmpDir = System.getProperty("java.io.tmpdir");
+
// to help with the logger madness
private void print(String str) {
System.out.print(str);
@@ -106,7 +108,7 @@
IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
- File f = new File("/tmp/btreetest.bin");
+ File f = new File(tmpDir + "/" + "btreetest.bin");
RandomAccessFile raf = new RandomAccessFile(f, "rw");
int fileId = 0;
FileInfo fi = new FileInfo(fileId, raf);
@@ -311,7 +313,7 @@
IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
- File f = new File("/tmp/btreetest.bin");
+ File f = new File(tmpDir + "/" + "btreetest.bin");
RandomAccessFile raf = new RandomAccessFile(f, "rw");
int fileId = 0;
FileInfo fi = new FileInfo(fileId, raf);
@@ -487,7 +489,7 @@
IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
- File f = new File("/tmp/btreetest.bin");
+ File f = new File(tmpDir + "/" + "btreetest.bin");
RandomAccessFile raf = new RandomAccessFile(f, "rw");
int fileId = 0;
FileInfo fi = new FileInfo(fileId, raf);
@@ -654,7 +656,7 @@
IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
- File f = new File("/tmp/btreetest.bin");
+ File f = new File(tmpDir + "/" + "btreetest.bin");
RandomAccessFile raf = new RandomAccessFile(f, "rw");
int fileId = 0;
FileInfo fi = new FileInfo(fileId, raf);
@@ -808,7 +810,7 @@
IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
- File f = new File("/tmp/btreetest.bin");
+ File f = new File(tmpDir + "/" + "btreetest.bin");
RandomAccessFile raf = new RandomAccessFile(f, "rw");
int fileId = 0;
FileInfo fi = new FileInfo(fileId, raf);
@@ -957,7 +959,7 @@
IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
- File f = new File("/tmp/btreetest.bin");
+ File f = new File(tmpDir + "/" + "btreetest.bin");
RandomAccessFile raf = new RandomAccessFile(f, "rw");
int fileId = 0;
FileInfo fi = new FileInfo(fileId, raf);
