You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by im...@apache.org on 2015/04/24 20:45:32 UTC
[05/85] [abbrv] [partial] incubator-asterixdb-hyracks git commit:
Move Pregelix and Hivesterix codebase to new repositories: 1. Move Pregelix
codebase to https://github.com/pregelix/pregelix;
2. Move Hivesterix codebase to https://code.google.com/p/hives
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/ffc967fd/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/ISerializableAggregateFunctionFactory.java
----------------------------------------------------------------------
diff --git a/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/ISerializableAggregateFunctionFactory.java b/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/ISerializableAggregateFunctionFactory.java
deleted file mode 100644
index 3abbb3b..0000000
--- a/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/ISerializableAggregateFunctionFactory.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * Copyright 2009-2013 by The Regents of the University of California
- * Licensed 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 from
- *
- * 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 edu.uci.ics.pregelix.dataflow.std.base;
-
-import java.io.Serializable;
-
-import edu.uci.ics.hyracks.api.comm.IFrameWriter;
-import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-import edu.uci.ics.hyracks.api.exceptions.HyracksException;
-
-public interface ISerializableAggregateFunctionFactory extends Serializable {
- public ISerializableAggregateFunction createAggregateFunction(IHyracksTaskContext ctx, IFrameWriter writer)
- throws HyracksException;
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/ffc967fd/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/ISerializerDeserializerFactory.java
----------------------------------------------------------------------
diff --git a/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/ISerializerDeserializerFactory.java b/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/ISerializerDeserializerFactory.java
deleted file mode 100644
index 894cba5..0000000
--- a/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/ISerializerDeserializerFactory.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Copyright 2009-2013 by The Regents of the University of California
- * Licensed 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 from
- *
- * 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 edu.uci.ics.pregelix.dataflow.std.base;
-
-import java.io.Serializable;
-
-import org.apache.hadoop.conf.Configuration;
-
-import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
-
-public interface ISerializerDeserializerFactory<T> extends Serializable {
-
- public ISerializerDeserializer<T> getSerializerDeserializer(Configuration conf);
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/ffc967fd/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/IUpdateFunction.java
----------------------------------------------------------------------
diff --git a/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/IUpdateFunction.java b/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/IUpdateFunction.java
deleted file mode 100644
index 081b3bc..0000000
--- a/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/IUpdateFunction.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * Copyright 2009-2013 by The Regents of the University of California
- * Licensed 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 from
- *
- * 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 edu.uci.ics.pregelix.dataflow.std.base;
-
-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.data.accessors.ITupleReference;
-import edu.uci.ics.hyracks.storage.am.common.api.IIndexCursor;
-
-public interface IUpdateFunction extends IFunction {
-
- /**
- * update the tuple pointed by tupleRef called after process,
- * one-input-tuple-at-a-time
- *
- * @param tupleRef
- * @throws HyracksDataException
- */
- public void update(ITupleReference tupleRef, ArrayTupleBuilder cloneUpdateTb, IIndexCursor cursor)
- throws HyracksDataException;
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/ffc967fd/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/IUpdateFunctionFactory.java
----------------------------------------------------------------------
diff --git a/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/IUpdateFunctionFactory.java b/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/IUpdateFunctionFactory.java
deleted file mode 100644
index 504bd4c..0000000
--- a/pregelix/pregelix-dataflow-std-base/src/main/java/edu/uci/ics/pregelix/dataflow/std/base/IUpdateFunctionFactory.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * Copyright 2009-2013 by The Regents of the University of California
- * Licensed 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 from
- *
- * 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 edu.uci.ics.pregelix.dataflow.std.base;
-
-import java.io.Serializable;
-
-public interface IUpdateFunctionFactory extends Serializable {
-
- public IUpdateFunction createFunction();
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/ffc967fd/pregelix/pregelix-dataflow-std/pom.xml
----------------------------------------------------------------------
diff --git a/pregelix/pregelix-dataflow-std/pom.xml b/pregelix/pregelix-dataflow-std/pom.xml
deleted file mode 100644
index 725a282..0000000
--- a/pregelix/pregelix-dataflow-std/pom.xml
+++ /dev/null
@@ -1,153 +0,0 @@
-<!--
- ! Copyright 2009-2013 by The Regents of the University of California
- ! Licensed 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 from
- !
- ! 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.
- !-->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
- <artifactId>pregelix-dataflow-std</artifactId>
- <packaging>jar</packaging>
- <name>pregelix-dataflow-std</name>
-
- <parent>
- <groupId>edu.uci.ics.hyracks</groupId>
- <artifactId>pregelix</artifactId>
- <version>0.2.14-SNAPSHOT</version>
- </parent>
-
-
- <properties>
- <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
- </properties>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-compiler-plugin</artifactId>
- <version>2.0.2</version>
- <configuration>
- <source>1.7</source>
- <target>1.7</target>
- <fork>true</fork>
- </configuration>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-clean-plugin</artifactId>
- <version>2.4.1</version>
- <configuration>
- <filesets>
- <fileset>
- <directory>.</directory>
- <includes>
- <include>teststore*</include>
- <include>edu*</include>
- <include>actual*</include>
- <include>build*</include>
- <include>expect*</include>
- <include>ClusterController*</include>
- </includes>
- </fileset>
- </filesets>
- </configuration>
- </plugin>
- </plugins>
- </build>
-
- <dependencies>
- <dependency>
- <groupId>edu.uci.ics.hyracks</groupId>
- <artifactId>pregelix-dataflow-std-base</artifactId>
- <version>0.2.14-SNAPSHOT</version>
- <type>jar</type>
- <scope>compile</scope>
- </dependency>
- <dependency>
- <groupId>edu.uci.ics.hyracks</groupId>
- <artifactId>hyracks-dataflow-std</artifactId>
- <version>0.2.14-SNAPSHOT</version>
- <type>jar</type>
- <scope>compile</scope>
- </dependency>
- <dependency>
- <groupId>edu.uci.ics.hyracks</groupId>
- <artifactId>hyracks-api</artifactId>
- <version>0.2.14-SNAPSHOT</version>
- <type>jar</type>
- <scope>compile</scope>
- </dependency>
- <dependency>
- <groupId>edu.uci.ics.hyracks</groupId>
- <artifactId>hyracks-dataflow-common</artifactId>
- <version>0.2.14-SNAPSHOT</version>
- <type>jar</type>
- <scope>compile</scope>
- </dependency>
- <dependency>
- <groupId>edu.uci.ics.hyracks</groupId>
- <artifactId>hyracks-data-std</artifactId>
- <version>0.2.14-SNAPSHOT</version>
- <type>jar</type>
- <scope>compile</scope>
- </dependency>
- <dependency>
- <groupId>edu.uci.ics.hyracks</groupId>
- <artifactId>hyracks-hdfs-core</artifactId>
- <version>0.2.14-SNAPSHOT</version>
- <type>jar</type>
- <scope>compile</scope>
- </dependency>
- <dependency>
- <groupId>edu.uci.ics.hyracks</groupId>
- <artifactId>hyracks-storage-am-common</artifactId>
- <version>0.2.14-SNAPSHOT</version>
- <type>jar</type>
- <scope>compile</scope>
- </dependency>
- <dependency>
- <groupId>edu.uci.ics.hyracks</groupId>
- <artifactId>hyracks-storage-am-btree</artifactId>
- <version>0.2.14-SNAPSHOT</version>
- <type>jar</type>
- <scope>compile</scope>
- </dependency>
- <dependency>
- <groupId>edu.uci.ics.hyracks</groupId>
- <artifactId>hyracks-storage-am-lsm-btree</artifactId>
- <version>0.2.14-SNAPSHOT</version>
- <type>jar</type>
- <scope>compile</scope>
- </dependency>
- <dependency>
- <groupId>edu.uci.ics.hyracks</groupId>
- <artifactId>hyracks-control-cc</artifactId>
- <version>0.2.14-SNAPSHOT</version>
- <type>jar</type>
- <scope>compile</scope>
- </dependency>
- <dependency>
- <groupId>edu.uci.ics.hyracks</groupId>
- <artifactId>hyracks-control-nc</artifactId>
- <version>0.2.14-SNAPSHOT</version>
- <type>jar</type>
- <scope>compile</scope>
- </dependency>
- <dependency>
- <groupId>edu.uci.ics.hyracks</groupId>
- <artifactId>hyracks-ipc</artifactId>
- <version>0.2.14-SNAPSHOT</version>
- <type>jar</type>
- <scope>compile</scope>
- </dependency>
- </dependencies>
-</project>
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/ffc967fd/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/FunctionCallOperatorDescriptor.java
----------------------------------------------------------------------
diff --git a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/FunctionCallOperatorDescriptor.java b/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/FunctionCallOperatorDescriptor.java
deleted file mode 100644
index 5ba2c91..0000000
--- a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/FunctionCallOperatorDescriptor.java
+++ /dev/null
@@ -1,114 +0,0 @@
-/*
- * Copyright 2009-2013 by The Regents of the University of California
- * Licensed 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 from
- *
- * 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 edu.uci.ics.pregelix.dataflow.std;
-
-import java.nio.ByteBuffer;
-
-import edu.uci.ics.hyracks.api.comm.IFrameWriter;
-import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-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.JobSpecification;
-import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameDeserializer;
-import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
-import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputOperatorNodePushable;
-import edu.uci.ics.pregelix.dataflow.std.base.IFunction;
-import edu.uci.ics.pregelix.dataflow.std.base.IFunctionFactory;
-import edu.uci.ics.pregelix.dataflow.std.base.IRecordDescriptorFactory;
-import edu.uci.ics.pregelix.dataflow.std.base.IRuntimeHookFactory;
-
-public class FunctionCallOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
- private static final long serialVersionUID = 1L;
- private final IFunctionFactory functionFactory;
- private final IRuntimeHookFactory preHookFactory;
- private final IRuntimeHookFactory postHookFactory;
- private final IRecordDescriptorFactory inputRdFactory;
-
- public FunctionCallOperatorDescriptor(JobSpecification spec, IRecordDescriptorFactory inputRdFactory,
- int outputArity, IFunctionFactory functionFactory, IRuntimeHookFactory preHookFactory,
- IRuntimeHookFactory postHookFactory, RecordDescriptor... rDescs) {
- super(spec, 1, outputArity);
- this.functionFactory = functionFactory;
- this.preHookFactory = preHookFactory;
- this.postHookFactory = postHookFactory;
- this.inputRdFactory = inputRdFactory;
-
- for (int i = 0; i < rDescs.length; i++) {
- this.recordDescriptors[i] = rDescs[i];
- }
- }
-
- @Override
- public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
- final IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions)
- throws HyracksDataException {
- return new AbstractUnaryInputOperatorNodePushable() {
- private RecordDescriptor rd0;
- private FrameDeserializer frameDeserializer;
- private final IFrameWriter[] writers = new IFrameWriter[outputArity];
- private final IFunction function = functionFactory.createFunction();
- private ClassLoader ctxCL = Thread.currentThread().getContextClassLoader();
-
- @Override
- public void open() throws HyracksDataException {
- rd0 = inputRdFactory == null ? recordDescProvider.getInputRecordDescriptor(getActivityId(), 0)
- : inputRdFactory.createRecordDescriptor(ctx);
- frameDeserializer = new FrameDeserializer(ctx.getFrameSize(), rd0);
- ctxCL = Thread.currentThread().getContextClassLoader();
- Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
- for (IFrameWriter writer : writers) {
- writer.open();
- }
- if (preHookFactory != null)
- preHookFactory.createRuntimeHook().configure(ctx);
- function.open(ctx, rd0, writers);
- }
-
- @Override
- public void nextFrame(ByteBuffer frame) throws HyracksDataException {
- frameDeserializer.reset(frame);
- while (!frameDeserializer.done()) {
- Object[] tuple = frameDeserializer.deserializeRecord();
- function.process(tuple);
- }
- }
-
- @Override
- public void close() throws HyracksDataException {
- if (postHookFactory != null)
- postHookFactory.createRuntimeHook().configure(ctx);
- function.close();
- for (IFrameWriter writer : writers) {
- writer.close();
- }
- Thread.currentThread().setContextClassLoader(ctxCL);
- }
-
- @Override
- public void fail() throws HyracksDataException {
- for (IFrameWriter writer : writers) {
- writer.fail();
- }
- }
-
- @Override
- public void setOutputFrameWriter(int index, IFrameWriter writer, RecordDescriptor recordDesc) {
- writers[index] = writer;
- }
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/ffc967fd/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinFunctionUpdateOperatorDescriptor.java
----------------------------------------------------------------------
diff --git a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinFunctionUpdateOperatorDescriptor.java b/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinFunctionUpdateOperatorDescriptor.java
deleted file mode 100644
index 5890514..0000000
--- a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinFunctionUpdateOperatorDescriptor.java
+++ /dev/null
@@ -1,177 +0,0 @@
-/*
- * Copyright 2009-2013 by The Regents of the University of California
- * Licensed 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 from
- *
- * 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 edu.uci.ics.pregelix.dataflow.std;
-
-import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
-import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
-import edu.uci.ics.hyracks.api.dataflow.value.INullWriter;
-import edu.uci.ics.hyracks.api.dataflow.value.INullWriterFactory;
-import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
-import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
-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.std.file.IFileSplitProvider;
-import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.AbstractTreeIndexOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexDataflowHelperFactory;
-import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallbackFactory;
-import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
-import edu.uci.ics.hyracks.storage.common.file.TransientLocalResourceFactoryProvider;
-import edu.uci.ics.pregelix.dataflow.std.base.IRecordDescriptorFactory;
-import edu.uci.ics.pregelix.dataflow.std.base.IRuntimeHookFactory;
-import edu.uci.ics.pregelix.dataflow.std.base.IUpdateFunctionFactory;
-
-public class IndexNestedLoopJoinFunctionUpdateOperatorDescriptor extends AbstractTreeIndexOperatorDescriptor {
- private static final long serialVersionUID = 1L;
-
- private boolean isForward;
- 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
- private boolean lowKeyInclusive;
- private boolean highKeyInclusive;
-
- // right outer join
- private boolean isRightOuter = false;
- private INullWriterFactory[] nullWriterFactories;
-
- // set union
- private boolean isSetUnion = false;
-
- private final IUpdateFunctionFactory functionFactory;
- private final IRuntimeHookFactory preHookFactory;
- private final IRuntimeHookFactory postHookFactory;
- private final IRecordDescriptorFactory inputRdFactory;
-
- private final int outputArity;
-
- public IndexNestedLoopJoinFunctionUpdateOperatorDescriptor(JobSpecification spec,
- IStorageManagerInterface storageManager, IIndexLifecycleManagerProvider lcManagerProvider,
- IFileSplitProvider fileSplitProvider, ITypeTraits[] typeTraits,
- IBinaryComparatorFactory[] comparatorFactories, boolean isForward, int[] lowKeyFields, int[] highKeyFields,
- boolean lowKeyInclusive, boolean highKeyInclusive, IIndexDataflowHelperFactory opHelperFactory,
- IRecordDescriptorFactory inputRdFactory, int outputArity, IUpdateFunctionFactory functionFactory,
- IRuntimeHookFactory preHookFactory, IRuntimeHookFactory postHookFactory, RecordDescriptor... rDescs) {
- super(spec, 1, outputArity, rDescs[0], storageManager, lcManagerProvider, fileSplitProvider, typeTraits,
- comparatorFactories, highKeyFields, opHelperFactory, null, false,
- false, null,
- new TransientLocalResourceFactoryProvider(), NoOpOperationCallbackFactory.INSTANCE, NoOpOperationCallbackFactory.INSTANCE);
- this.isForward = isForward;
- this.lowKeyFields = lowKeyFields;
- this.highKeyFields = highKeyFields;
- this.lowKeyInclusive = lowKeyInclusive;
- this.highKeyInclusive = highKeyInclusive;
-
- this.functionFactory = functionFactory;
- this.preHookFactory = preHookFactory;
- this.postHookFactory = postHookFactory;
- this.inputRdFactory = inputRdFactory;
-
- for (int i = 0; i < rDescs.length; i++) {
- this.recordDescriptors[i] = rDescs[i];
- }
-
- this.outputArity = outputArity;
- }
-
- public IndexNestedLoopJoinFunctionUpdateOperatorDescriptor(JobSpecification spec,
- IStorageManagerInterface storageManager, IIndexLifecycleManagerProvider lcManagerProvider,
- IFileSplitProvider fileSplitProvider, ITypeTraits[] typeTraits,
- IBinaryComparatorFactory[] comparatorFactories, boolean isForward, int[] lowKeyFields, int[] highKeyFields,
- boolean lowKeyInclusive, boolean highKeyInclusive, IIndexDataflowHelperFactory opHelperFactory,
- boolean isRightOuter, INullWriterFactory[] nullWriterFactories, IRecordDescriptorFactory inputRdFactory,
- int outputArity, IUpdateFunctionFactory functionFactory, IRuntimeHookFactory preHookFactory,
- IRuntimeHookFactory postHookFactory, RecordDescriptor... rDescs) {
- super(spec, 1, outputArity, rDescs[0], storageManager, lcManagerProvider, fileSplitProvider, typeTraits,
- comparatorFactories, highKeyFields, opHelperFactory, null, false,
- false, null,
- new TransientLocalResourceFactoryProvider(), NoOpOperationCallbackFactory.INSTANCE, NoOpOperationCallbackFactory.INSTANCE);
- this.isForward = isForward;
- this.lowKeyFields = lowKeyFields;
- this.highKeyFields = highKeyFields;
- this.lowKeyInclusive = lowKeyInclusive;
- this.highKeyInclusive = highKeyInclusive;
-
- this.isRightOuter = isRightOuter;
- this.nullWriterFactories = nullWriterFactories;
-
- this.functionFactory = functionFactory;
- this.preHookFactory = preHookFactory;
- this.postHookFactory = postHookFactory;
- this.inputRdFactory = inputRdFactory;
-
- for (int i = 0; i < rDescs.length; i++) {
- this.recordDescriptors[i] = rDescs[i];
- }
-
- this.outputArity = outputArity;
- }
-
- public IndexNestedLoopJoinFunctionUpdateOperatorDescriptor(JobSpecification spec,
- IStorageManagerInterface storageManager, IIndexLifecycleManagerProvider lcManagerProvider,
- IFileSplitProvider fileSplitProvider, ITypeTraits[] typeTraits,
- IBinaryComparatorFactory[] comparatorFactories, boolean isForward, int[] lowKeyFields, int[] highKeyFields,
- boolean lowKeyInclusive, boolean highKeyInclusive, IIndexDataflowHelperFactory opHelperFactory,
- boolean isSetUnion, IRecordDescriptorFactory inputRdFactory, int outputArity,
- IUpdateFunctionFactory functionFactory, IRuntimeHookFactory preHookFactory,
- IRuntimeHookFactory postHookFactory, RecordDescriptor... rDescs) {
- super(spec, 1, outputArity, rDescs[0], storageManager, lcManagerProvider, fileSplitProvider, typeTraits,
- comparatorFactories, highKeyFields, opHelperFactory, null, false,
- false, null,
- new TransientLocalResourceFactoryProvider(), NoOpOperationCallbackFactory.INSTANCE, NoOpOperationCallbackFactory.INSTANCE);
- this.isForward = isForward;
- this.lowKeyFields = lowKeyFields;
- this.highKeyFields = highKeyFields;
- this.lowKeyInclusive = lowKeyInclusive;
- this.highKeyInclusive = highKeyInclusive;
-
- this.isSetUnion = isSetUnion;
-
- this.functionFactory = functionFactory;
- this.preHookFactory = preHookFactory;
- this.postHookFactory = postHookFactory;
- this.inputRdFactory = inputRdFactory;
-
- for (int i = 0; i < rDescs.length; i++) {
- this.recordDescriptors[i] = rDescs[i];
- }
-
- this.outputArity = outputArity;
- }
-
- @Override
- public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
- IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) throws HyracksDataException {
- if (isRightOuter) {
- INullWriter[] nullWriters = new INullWriter[nullWriterFactories.length];
- for (int i = 0; i < nullWriters.length; i++)
- nullWriters[i] = nullWriterFactories[i].createNullWriter();
- return new IndexNestedLoopRightOuterJoinFunctionUpdateOperatorNodePushable(this, ctx, partition,
- recordDescProvider, isForward, lowKeyFields, highKeyFields, nullWriters, functionFactory,
- preHookFactory, postHookFactory, inputRdFactory, outputArity);
- } else if (isSetUnion) {
- return new IndexNestedLoopSetUnionFunctionUpdateOperatorNodePushable(this, ctx, partition,
- recordDescProvider, isForward, lowKeyFields, highKeyFields, functionFactory, preHookFactory,
- postHookFactory, inputRdFactory, outputArity);
- } else {
- return new IndexNestedLoopJoinFunctionUpdateOperatorNodePushable(this, ctx, partition, recordDescProvider,
- isForward, lowKeyFields, highKeyFields, lowKeyInclusive, highKeyInclusive, functionFactory,
- preHookFactory, postHookFactory, inputRdFactory, outputArity);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/ffc967fd/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinFunctionUpdateOperatorNodePushable.java
----------------------------------------------------------------------
diff --git a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinFunctionUpdateOperatorNodePushable.java b/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinFunctionUpdateOperatorNodePushable.java
deleted file mode 100644
index 3512a23..0000000
--- a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinFunctionUpdateOperatorNodePushable.java
+++ /dev/null
@@ -1,265 +0,0 @@
-/*
- * Copyright 2009-2013 by The Regents of the University of California
- * Licensed 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 from
- *
- * 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 edu.uci.ics.pregelix.dataflow.std;
-
-import java.nio.ByteBuffer;
-
-import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
-import edu.uci.ics.hyracks.api.comm.IFrameWriter;
-import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-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.data.accessors.ITupleReference;
-import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputOperatorNodePushable;
-import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
-import edu.uci.ics.hyracks.storage.am.common.api.IIndexAccessor;
-import edu.uci.ics.hyracks.storage.am.common.api.IIndexCursor;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.AbstractTreeIndexOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IndexDataflowHelper;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.TreeIndexDataflowHelper;
-import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
-import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
-import edu.uci.ics.hyracks.storage.am.common.tuples.PermutingFrameTupleReference;
-import edu.uci.ics.pregelix.dataflow.std.base.IRecordDescriptorFactory;
-import edu.uci.ics.pregelix.dataflow.std.base.IRuntimeHookFactory;
-import edu.uci.ics.pregelix.dataflow.std.base.IUpdateFunctionFactory;
-import edu.uci.ics.pregelix.dataflow.std.util.CopyUpdateUtil;
-import edu.uci.ics.pregelix.dataflow.std.util.FunctionProxy;
-import edu.uci.ics.pregelix.dataflow.std.util.SearchKeyTupleReference;
-import edu.uci.ics.pregelix.dataflow.std.util.StorageType;
-import edu.uci.ics.pregelix.dataflow.std.util.UpdateBuffer;
-
-public class IndexNestedLoopJoinFunctionUpdateOperatorNodePushable extends AbstractUnaryInputOperatorNodePushable {
- private IndexDataflowHelper treeIndexOpHelper;
- private FrameTupleAccessor accessor;
-
- private ByteBuffer writeBuffer;
- private FrameTupleAppender appender;
- private ITreeIndex index;
- private PermutingFrameTupleReference lowKey;
- private PermutingFrameTupleReference highKey;
- private boolean lowKeyInclusive;
- private boolean highKeyInclusive;
- private RangePredicate rangePred;
- private MultiComparator lowKeySearchCmp;
- private MultiComparator highKeySearchCmp;
- private IIndexCursor cursor;
- protected IIndexAccessor indexAccessor;
-
- private RecordDescriptor recDesc;
- private final IFrameWriter[] writers;
- private final FunctionProxy functionProxy;
- private ArrayTupleBuilder cloneUpdateTb;
- private final UpdateBuffer updateBuffer;
- private final SearchKeyTupleReference tempTupleReference = new SearchKeyTupleReference();
- private final StorageType storageType;
-
- public IndexNestedLoopJoinFunctionUpdateOperatorNodePushable(AbstractTreeIndexOperatorDescriptor opDesc,
- IHyracksTaskContext ctx, int partition, IRecordDescriptorProvider recordDescProvider, boolean isForward,
- int[] lowKeyFields, int[] highKeyFields, boolean lowKeyInclusive, boolean highKeyInclusive,
- IUpdateFunctionFactory functionFactory, IRuntimeHookFactory preHookFactory,
- IRuntimeHookFactory postHookFactory, IRecordDescriptorFactory inputRdFactory, int outputArity)
- throws HyracksDataException {
- treeIndexOpHelper = (IndexDataflowHelper) opDesc.getIndexDataflowHelperFactory().createIndexDataflowHelper(
- opDesc, ctx, partition);
- if (treeIndexOpHelper instanceof TreeIndexDataflowHelper) {
- storageType = StorageType.TreeIndex;
- } else {
- storageType = StorageType.LSMIndex;
- }
- this.lowKeyInclusive = lowKeyInclusive;
- this.highKeyInclusive = highKeyInclusive;
- this.recDesc = recordDescProvider.getInputRecordDescriptor(opDesc.getActivityId(), 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);
- }
-
- this.writers = new IFrameWriter[outputArity];
- this.functionProxy = new FunctionProxy(ctx, functionFactory, preHookFactory, postHookFactory, inputRdFactory,
- writers);
- this.updateBuffer = new UpdateBuffer(ctx, 2);
- }
-
- protected void setCursor() {
- cursor = indexAccessor.createSearchCursor(true);
- }
-
- @Override
- public void open() throws HyracksDataException {
- /**
- * open the function
- */
- functionProxy.functionOpen();
- accessor = new FrameTupleAccessor(treeIndexOpHelper.getTaskContext().getFrameSize(), recDesc);
-
- try {
- treeIndexOpHelper.open();
- index = (ITreeIndex) treeIndexOpHelper.getIndexInstance();
-
- // TODO: Can we construct the multicmps using helper methods?
- int lowKeySearchFields = index.getComparatorFactories().length;
- int highKeySearchFields = index.getComparatorFactories().length;
- if (lowKey != null)
- lowKeySearchFields = lowKey.getFieldCount();
- if (highKey != null)
- highKeySearchFields = highKey.getFieldCount();
-
- IBinaryComparator[] lowKeySearchComparators = new IBinaryComparator[lowKeySearchFields];
- for (int i = 0; i < lowKeySearchFields; i++) {
- lowKeySearchComparators[i] = index.getComparatorFactories()[i].createBinaryComparator();
- }
- lowKeySearchCmp = new MultiComparator(lowKeySearchComparators);
-
- if (lowKeySearchFields == highKeySearchFields) {
- highKeySearchCmp = lowKeySearchCmp;
- } else {
- IBinaryComparator[] highKeySearchComparators = new IBinaryComparator[highKeySearchFields];
- for (int i = 0; i < highKeySearchFields; i++) {
- highKeySearchComparators[i] = index.getComparatorFactories()[i].createBinaryComparator();
- }
- highKeySearchCmp = new MultiComparator(highKeySearchComparators);
- }
-
- rangePred = new RangePredicate(null, null, lowKeyInclusive, highKeyInclusive, lowKeySearchCmp,
- highKeySearchCmp);
- writeBuffer = treeIndexOpHelper.getTaskContext().allocateFrame();
- appender = new FrameTupleAppender(treeIndexOpHelper.getTaskContext().getFrameSize());
- appender.reset(writeBuffer, true);
-
- indexAccessor = index.createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
- setCursor();
- cloneUpdateTb = new ArrayTupleBuilder(index.getFieldCount());
- updateBuffer.setFieldCount(index.getFieldCount());
- } catch (Exception e) {
- closeResource();
- throw new HyracksDataException(e);
- }
- }
-
- private void writeSearchResults(IFrameTupleAccessor leftAccessor, int tIndex) throws Exception {
- while (cursor.hasNext()) {
- cursor.next();
- ITupleReference tupleRef = cursor.getTuple();
-
- /**
- * merge with updated tuple
- */
- ITupleReference indexEntryTuple = tupleRef;
- ITupleReference cachedUpdatedLastTuple = updateBuffer.getLastTuple();
- if (cachedUpdatedLastTuple != null) {
- if (compare(cachedUpdatedLastTuple, tupleRef) == 0) {
- indexEntryTuple = cachedUpdatedLastTuple;
- }
- }
-
- /**
- * call the update function
- */
- functionProxy.functionCall(leftAccessor, tIndex, indexEntryTuple, cloneUpdateTb, cursor);
-
- /**
- * doing copy update
- */
- CopyUpdateUtil.copyUpdate(tempTupleReference, indexEntryTuple, updateBuffer, cloneUpdateTb, indexAccessor,
- cursor, rangePred, false, storageType);
- }
- }
-
- /** compare tuples */
- private int compare(ITupleReference left, ITupleReference right) throws Exception {
- return lowKeySearchCmp.compare(left, right);
- }
-
- @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, lowKeyInclusive);
- rangePred.setHighKey(highKey, highKeyInclusive);
-
- cursor.reset();
- indexAccessor.search(cursor, rangePred);
- writeSearchResults(accessor, i);
- }
- } catch (Exception e) {
- closeResource();
- throw new HyracksDataException(e);
- }
- }
-
- @Override
- public void close() throws HyracksDataException {
- try {
- cursor.close();
- //batch update
- updateBuffer.updateIndex(indexAccessor);
- } catch (Exception e) {
- closeResource();
- throw new HyracksDataException(e);
- } finally {
- treeIndexOpHelper.close();
- /**
- * close the update function
- */
- functionProxy.functionClose();
- }
- }
-
- @Override
- public void fail() throws HyracksDataException {
- closeResource();
- populateFailure();
- }
-
- private void closeResource() throws HyracksDataException {
- try {
- cursor.close();
- } catch (Exception e) {
- throw new HyracksDataException(e);
- } finally {
- treeIndexOpHelper.close();
- }
- }
-
- private void populateFailure() throws HyracksDataException {
- for (IFrameWriter writer : writers) {
- writer.fail();
- }
- }
-
- @Override
- public void setOutputFrameWriter(int index, IFrameWriter writer, RecordDescriptor recordDesc) {
- writers[index] = writer;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/ffc967fd/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinOperatorDescriptor.java
----------------------------------------------------------------------
diff --git a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinOperatorDescriptor.java b/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinOperatorDescriptor.java
deleted file mode 100644
index 0f003ba..0000000
--- a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinOperatorDescriptor.java
+++ /dev/null
@@ -1,123 +0,0 @@
-/*
- * Copyright 2009-2013 by The Regents of the University of California
- * Licensed 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 from
- *
- * 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 edu.uci.ics.pregelix.dataflow.std;
-
-import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
-import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
-import edu.uci.ics.hyracks.api.dataflow.value.INullWriter;
-import edu.uci.ics.hyracks.api.dataflow.value.INullWriterFactory;
-import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
-import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
-import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
-import edu.uci.ics.hyracks.api.job.JobSpecification;
-import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
-import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.AbstractTreeIndexOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexDataflowHelperFactory;
-import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallbackFactory;
-import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
-import edu.uci.ics.hyracks.storage.common.file.TransientLocalResourceFactoryProvider;
-
-public class IndexNestedLoopJoinOperatorDescriptor extends AbstractTreeIndexOperatorDescriptor {
-
- private static final long serialVersionUID = 1L;
-
- private boolean isForward;
- 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
- private boolean lowKeyInclusive;
- private boolean highKeyInclusive;
-
- // right outer join
- private boolean isRightOuter = false;
- private INullWriterFactory[] nullWriterFactories;
-
- // set union
- private boolean isSetUnion = false;
-
- public IndexNestedLoopJoinOperatorDescriptor(JobSpecification spec, RecordDescriptor recDesc,
- IStorageManagerInterface storageManager, IIndexLifecycleManagerProvider lcManagerProvider,
- IFileSplitProvider fileSplitProvider, ITypeTraits[] typeTraits,
- IBinaryComparatorFactory[] comparatorFactories, boolean isForward, int[] lowKeyFields, int[] highKeyFields,
- boolean lowKeyInclusive, boolean highKeyInclusive, IIndexDataflowHelperFactory opHelperFactory) {
- super(spec, 1, 1, recDesc, storageManager, lcManagerProvider, fileSplitProvider, typeTraits,
- comparatorFactories, null, opHelperFactory, null, false, false,
- null, new TransientLocalResourceFactoryProvider(), NoOpOperationCallbackFactory.INSTANCE, NoOpOperationCallbackFactory.INSTANCE);
- this.isForward = isForward;
- this.lowKeyFields = lowKeyFields;
- this.highKeyFields = highKeyFields;
- this.lowKeyInclusive = lowKeyInclusive;
- this.highKeyInclusive = highKeyInclusive;
- }
-
- public IndexNestedLoopJoinOperatorDescriptor(JobSpecification spec, RecordDescriptor recDesc,
- IStorageManagerInterface storageManager, IIndexLifecycleManagerProvider lcManagerProvider,
- IFileSplitProvider fileSplitProvider, ITreeIndexFrameFactory interiorFrameFactory,
- ITreeIndexFrameFactory leafFrameFactory, ITypeTraits[] typeTraits,
- IBinaryComparatorFactory[] comparatorFactories, boolean isForward, int[] lowKeyFields, int[] highKeyFields,
- boolean lowKeyInclusive, boolean highKeyInclusive, IIndexDataflowHelperFactory opHelperFactory,
- boolean isRightOuter, INullWriterFactory[] nullWriterFactories) {
- super(spec, 1, 1, recDesc, storageManager, lcManagerProvider, fileSplitProvider, typeTraits,
- comparatorFactories, null, opHelperFactory, null, false, false,
- null, new TransientLocalResourceFactoryProvider(), NoOpOperationCallbackFactory.INSTANCE, NoOpOperationCallbackFactory.INSTANCE);
- this.isForward = isForward;
- this.lowKeyFields = lowKeyFields;
- this.highKeyFields = highKeyFields;
- this.lowKeyInclusive = lowKeyInclusive;
- this.highKeyInclusive = highKeyInclusive;
-
- this.isRightOuter = isRightOuter;
- this.nullWriterFactories = nullWriterFactories;
- }
-
- public IndexNestedLoopJoinOperatorDescriptor(JobSpecification spec, RecordDescriptor recDesc,
- IStorageManagerInterface storageManager, IIndexLifecycleManagerProvider lcManagerProvider,
- IFileSplitProvider fileSplitProvider, ITypeTraits[] typeTraits,
- IBinaryComparatorFactory[] comparatorFactories, boolean isForward, int[] lowKeyFields, int[] highKeyFields,
- boolean lowKeyInclusive, boolean highKeyInclusive, IIndexDataflowHelperFactory opHelperFactory,
- boolean isSetUnion) {
- super(spec, 1, 1, recDesc, storageManager, lcManagerProvider, fileSplitProvider, typeTraits,
- comparatorFactories, null, opHelperFactory, null, false, false,
- null, new TransientLocalResourceFactoryProvider(), NoOpOperationCallbackFactory.INSTANCE, NoOpOperationCallbackFactory.INSTANCE);
- this.isForward = isForward;
- this.lowKeyFields = lowKeyFields;
- this.highKeyFields = highKeyFields;
- this.lowKeyInclusive = lowKeyInclusive;
- this.highKeyInclusive = highKeyInclusive;
-
- this.isSetUnion = isSetUnion;
- }
-
- @Override
- public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
- IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
- if (isRightOuter) {
- INullWriter[] nullWriters = new INullWriter[nullWriterFactories.length];
- for (int i = 0; i < nullWriters.length; i++)
- nullWriters[i] = nullWriterFactories[i].createNullWriter();
- return new IndexNestedLoopRightOuterJoinOperatorNodePushable(this, ctx, partition, recordDescProvider,
- isForward, lowKeyFields, highKeyFields, nullWriters);
- } else if (isSetUnion) {
- return new IndexNestedLoopSetUnionOperatorNodePushable(this, ctx, partition, recordDescProvider, isForward,
- lowKeyFields, highKeyFields);
- } else {
- return new IndexNestedLoopJoinOperatorNodePushable(this, ctx, partition, recordDescProvider, isForward,
- lowKeyFields, highKeyFields, lowKeyInclusive, highKeyInclusive);
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/ffc967fd/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinOperatorNodePushable.java
----------------------------------------------------------------------
diff --git a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinOperatorNodePushable.java b/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinOperatorNodePushable.java
deleted file mode 100644
index b5b21b6..0000000
--- a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopJoinOperatorNodePushable.java
+++ /dev/null
@@ -1,208 +0,0 @@
-/*
- * Copyright 2009-2013 by The Regents of the University of California
- * Licensed 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 from
- *
- * 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 edu.uci.ics.pregelix.dataflow.std;
-
-import java.io.DataOutput;
-import java.nio.ByteBuffer;
-
-import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
-import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-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.AbstractUnaryInputUnaryOutputOperatorNodePushable;
-import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
-import edu.uci.ics.hyracks.storage.am.common.api.IIndexAccessor;
-import edu.uci.ics.hyracks.storage.am.common.api.IIndexCursor;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.AbstractTreeIndexOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IndexDataflowHelper;
-import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
-import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
-import edu.uci.ics.hyracks.storage.am.common.tuples.PermutingFrameTupleReference;
-
-public class IndexNestedLoopJoinOperatorNodePushable extends AbstractUnaryInputUnaryOutputOperatorNodePushable {
- private IndexDataflowHelper treeIndexOpHelper;
- private FrameTupleAccessor accessor;
-
- private ByteBuffer writeBuffer;
- private FrameTupleAppender appender;
- private ArrayTupleBuilder tb;
- private DataOutput dos;
-
- private ITreeIndex index;
- private PermutingFrameTupleReference lowKey;
- private PermutingFrameTupleReference highKey;
- private boolean lowKeyInclusive;
- private boolean highKeyInclusive;
- private RangePredicate rangePred;
- private MultiComparator lowKeySearchCmp;
- private MultiComparator highKeySearchCmp;
- private IIndexCursor cursor;
- protected IIndexAccessor indexAccessor;
-
- private RecordDescriptor recDesc;
- private final RecordDescriptor inputRecDesc;
-
- public IndexNestedLoopJoinOperatorNodePushable(AbstractTreeIndexOperatorDescriptor opDesc, IHyracksTaskContext ctx,
- int partition, IRecordDescriptorProvider recordDescProvider, boolean isForward, int[] lowKeyFields,
- int[] highKeyFields, boolean lowKeyInclusive, boolean highKeyInclusive) {
- inputRecDesc = recordDescProvider.getInputRecordDescriptor(opDesc.getActivityId(), 0);
- treeIndexOpHelper = (IndexDataflowHelper) opDesc.getIndexDataflowHelperFactory().createIndexDataflowHelper(
- opDesc, ctx, partition);
- this.lowKeyInclusive = lowKeyInclusive;
- this.highKeyInclusive = highKeyInclusive;
- this.recDesc = recordDescProvider.getInputRecordDescriptor(opDesc.getActivityId(), 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);
- }
- }
-
- protected void setCursor() {
- cursor = indexAccessor.createSearchCursor(false);
- }
-
- @Override
- public void open() throws HyracksDataException {
- accessor = new FrameTupleAccessor(treeIndexOpHelper.getTaskContext().getFrameSize(), recDesc);
-
- try {
- treeIndexOpHelper.open();
- index = (ITreeIndex) treeIndexOpHelper.getIndexInstance();
- writer.open();
-
- int lowKeySearchFields = index.getComparatorFactories().length;
- int highKeySearchFields = index.getComparatorFactories().length;
- if (lowKey != null)
- lowKeySearchFields = lowKey.getFieldCount();
- if (highKey != null)
- highKeySearchFields = highKey.getFieldCount();
-
- IBinaryComparator[] lowKeySearchComparators = new IBinaryComparator[lowKeySearchFields];
- for (int i = 0; i < lowKeySearchFields; i++) {
- lowKeySearchComparators[i] = index.getComparatorFactories()[i].createBinaryComparator();
- }
- lowKeySearchCmp = new MultiComparator(lowKeySearchComparators);
-
- if (lowKeySearchFields == highKeySearchFields) {
- highKeySearchCmp = lowKeySearchCmp;
- } else {
- IBinaryComparator[] highKeySearchComparators = new IBinaryComparator[highKeySearchFields];
- for (int i = 0; i < highKeySearchFields; i++) {
- highKeySearchComparators[i] = index.getComparatorFactories()[i].createBinaryComparator();
- }
- highKeySearchCmp = new MultiComparator(highKeySearchComparators);
- }
-
- rangePred = new RangePredicate(null, null, lowKeyInclusive, highKeyInclusive, lowKeySearchCmp,
- highKeySearchCmp);
- writeBuffer = treeIndexOpHelper.getTaskContext().allocateFrame();
- tb = new ArrayTupleBuilder(inputRecDesc.getFields().length + index.getFieldCount());
- dos = tb.getDataOutput();
- appender = new FrameTupleAppender(treeIndexOpHelper.getTaskContext().getFrameSize());
- appender.reset(writeBuffer, true);
- indexAccessor = index.createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
- setCursor();
- } catch (Exception e) {
- treeIndexOpHelper.close();
- throw new HyracksDataException(e);
- }
- }
-
- private void writeSearchResults(IFrameTupleAccessor leftAccessor, int tIndex) throws Exception {
- while (cursor.hasNext()) {
- tb.reset();
- cursor.next();
-
- ITupleReference frameTuple = cursor.getTuple();
- for (int i = 0; i < inputRecDesc.getFields().length; i++) {
- int tupleStart = leftAccessor.getTupleStartOffset(tIndex);
- int fieldStart = leftAccessor.getFieldStartOffset(tIndex, i);
- int offset = leftAccessor.getFieldSlotsLength() + tupleStart + fieldStart;
- int len = leftAccessor.getFieldEndOffset(tIndex, i) - fieldStart;
- dos.write(leftAccessor.getBuffer().array(), offset, len);
- tb.addFieldEndOffset();
- }
- 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, lowKeyInclusive);
- rangePred.setHighKey(highKey, highKeyInclusive);
-
- cursor.reset();
- indexAccessor.search(cursor, rangePred);
- writeSearchResults(accessor, i);
- }
- } catch (Exception e) {
- throw new HyracksDataException(e);
- }
- }
-
- @Override
- public void close() throws HyracksDataException {
- try {
- if (appender.getTupleCount() > 0) {
- FrameUtils.flushFrame(writeBuffer, writer);
- }
- writer.close();
- try {
- cursor.close();
- } catch (Exception e) {
- throw new HyracksDataException(e);
- }
- } finally {
- treeIndexOpHelper.close();
- }
- }
-
- @Override
- public void fail() throws HyracksDataException {
- writer.fail();
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/ffc967fd/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopRightOuterJoinFunctionUpdateOperatorNodePushable.java
----------------------------------------------------------------------
diff --git a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopRightOuterJoinFunctionUpdateOperatorNodePushable.java b/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopRightOuterJoinFunctionUpdateOperatorNodePushable.java
deleted file mode 100644
index 18675a1..0000000
--- a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopRightOuterJoinFunctionUpdateOperatorNodePushable.java
+++ /dev/null
@@ -1,332 +0,0 @@
-/*
- * Copyright 2009-2013 by The Regents of the University of California
- * Licensed 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 from
- *
- * 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 edu.uci.ics.pregelix.dataflow.std;
-
-import java.io.DataOutput;
-import java.nio.ByteBuffer;
-
-import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
-import edu.uci.ics.hyracks.api.comm.IFrameWriter;
-import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
-import edu.uci.ics.hyracks.api.dataflow.value.INullWriter;
-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.data.accessors.ITupleReference;
-import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputOperatorNodePushable;
-import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
-import edu.uci.ics.hyracks.storage.am.common.api.IIndexAccessor;
-import edu.uci.ics.hyracks.storage.am.common.api.IIndexCursor;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.AbstractTreeIndexOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IndexDataflowHelper;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.TreeIndexDataflowHelper;
-import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
-import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
-import edu.uci.ics.hyracks.storage.am.common.tuples.PermutingFrameTupleReference;
-import edu.uci.ics.pregelix.dataflow.std.base.IRecordDescriptorFactory;
-import edu.uci.ics.pregelix.dataflow.std.base.IRuntimeHookFactory;
-import edu.uci.ics.pregelix.dataflow.std.base.IUpdateFunctionFactory;
-import edu.uci.ics.pregelix.dataflow.std.util.CopyUpdateUtil;
-import edu.uci.ics.pregelix.dataflow.std.util.FunctionProxy;
-import edu.uci.ics.pregelix.dataflow.std.util.SearchKeyTupleReference;
-import edu.uci.ics.pregelix.dataflow.std.util.StorageType;
-import edu.uci.ics.pregelix.dataflow.std.util.UpdateBuffer;
-
-public class IndexNestedLoopRightOuterJoinFunctionUpdateOperatorNodePushable extends
- AbstractUnaryInputOperatorNodePushable {
- private IndexDataflowHelper treeIndexOpHelper;
- private FrameTupleAccessor accessor;
-
- private ByteBuffer writeBuffer;
- private FrameTupleAppender appender;
- private ArrayTupleBuilder nullTupleBuilder;
- private DataOutput dos;
-
- private ITreeIndex index;
- private RangePredicate rangePred;
- private MultiComparator lowKeySearchCmp;
- private MultiComparator highKeySearchCmp;
- private IIndexCursor cursor;
- protected IIndexAccessor indexAccessor;
-
- private RecordDescriptor recDesc;
- private final RecordDescriptor inputRecDesc;
-
- private PermutingFrameTupleReference lowKey;
- private PermutingFrameTupleReference highKey;
-
- private INullWriter[] nullWriter;
- private ITupleReference currentTopTuple;
- private boolean match;
-
- private final IFrameWriter[] writers;
- private final FunctionProxy functionProxy;
- private ArrayTupleBuilder cloneUpdateTb;
- private final UpdateBuffer updateBuffer;
- private final SearchKeyTupleReference tempTupleReference = new SearchKeyTupleReference();
- private final StorageType storageType;
-
- public IndexNestedLoopRightOuterJoinFunctionUpdateOperatorNodePushable(AbstractTreeIndexOperatorDescriptor opDesc,
- IHyracksTaskContext ctx, int partition, IRecordDescriptorProvider recordDescProvider, boolean isForward,
- int[] lowKeyFields, int[] highKeyFields, INullWriter[] nullWriter, IUpdateFunctionFactory functionFactory,
- IRuntimeHookFactory preHookFactory, IRuntimeHookFactory postHookFactory,
- IRecordDescriptorFactory inputRdFactory, int outputArity) throws HyracksDataException {
- inputRecDesc = recordDescProvider.getInputRecordDescriptor(opDesc.getActivityId(), 0);
- treeIndexOpHelper = (IndexDataflowHelper) opDesc.getIndexDataflowHelperFactory().createIndexDataflowHelper(
- opDesc, ctx, partition);
- if (treeIndexOpHelper instanceof TreeIndexDataflowHelper) {
- storageType = StorageType.TreeIndex;
- } else {
- storageType = StorageType.LSMIndex;
- }
- this.recDesc = recordDescProvider.getInputRecordDescriptor(opDesc.getActivityId(), 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);
- }
- this.nullWriter = nullWriter;
-
- this.writers = new IFrameWriter[outputArity];
- this.functionProxy = new FunctionProxy(ctx, functionFactory, preHookFactory, postHookFactory, inputRdFactory,
- writers);
- this.updateBuffer = new UpdateBuffer(ctx, 2);
- }
-
- protected void setCursor() {
- cursor = indexAccessor.createSearchCursor(true);
- }
-
- @Override
- public void open() throws HyracksDataException {
- /**
- * function open
- */
- functionProxy.functionOpen();
- accessor = new FrameTupleAccessor(treeIndexOpHelper.getTaskContext().getFrameSize(), recDesc);
-
- try {
- treeIndexOpHelper.open();
- index = (ITreeIndex) treeIndexOpHelper.getIndexInstance();
-
- // construct range predicate
- // TODO: Can we construct the multicmps using helper methods?
- int lowKeySearchFields = index.getComparatorFactories().length;
- int highKeySearchFields = index.getComparatorFactories().length;
-
- IBinaryComparator[] lowKeySearchComparators = new IBinaryComparator[lowKeySearchFields];
- for (int i = 0; i < lowKeySearchFields; i++) {
- lowKeySearchComparators[i] = index.getComparatorFactories()[i].createBinaryComparator();
- }
- lowKeySearchCmp = new MultiComparator(lowKeySearchComparators);
-
- if (lowKeySearchFields == highKeySearchFields) {
- highKeySearchCmp = lowKeySearchCmp;
- } else {
- IBinaryComparator[] highKeySearchComparators = new IBinaryComparator[highKeySearchFields];
- for (int i = 0; i < highKeySearchFields; i++) {
- highKeySearchComparators[i] = index.getComparatorFactories()[i].createBinaryComparator();
- }
- highKeySearchCmp = new MultiComparator(highKeySearchComparators);
- }
-
- rangePred = new RangePredicate(null, null, true, true, lowKeySearchCmp, highKeySearchCmp);
-
- writeBuffer = treeIndexOpHelper.getTaskContext().allocateFrame();
-
- nullTupleBuilder = new ArrayTupleBuilder(inputRecDesc.getFields().length);
- dos = nullTupleBuilder.getDataOutput();
- nullTupleBuilder.reset();
- for (int i = 0; i < inputRecDesc.getFields().length; i++) {
- nullWriter[i].writeNull(dos);
- nullTupleBuilder.addFieldEndOffset();
- }
-
- appender = new FrameTupleAppender(treeIndexOpHelper.getTaskContext().getFrameSize());
- appender.reset(writeBuffer, true);
-
- indexAccessor = index.createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
- setCursor();
-
- /** set the search cursor */
- rangePred.setLowKey(null, true);
- rangePred.setHighKey(null, true);
- cursor.reset();
- indexAccessor.search(cursor, rangePred);
-
- /** set up current top tuple */
- if (cursor.hasNext()) {
- cursor.next();
- currentTopTuple = cursor.getTuple();
- match = false;
- }
-
- cloneUpdateTb = new ArrayTupleBuilder(index.getFieldCount());
- updateBuffer.setFieldCount(index.getFieldCount());
- } catch (Exception e) {
- closeResource();
- throw new HyracksDataException(e);
- }
- }
-
- @Override
- public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
- accessor.reset(buffer);
- int tupleCount = accessor.getTupleCount();
- try {
- for (int i = 0; i < tupleCount && currentTopTuple != null;) {
- if (lowKey != null)
- lowKey.reset(accessor, i);
- if (highKey != null)
- highKey.reset(accessor, i);
- // TODO: currently use low key only, check what they mean
- int cmp = compare(lowKey, currentTopTuple);
- if (cmp <= 0) {
- if (cmp == 0) {
- outputMatch(i);
- currentTopTuple = cursor.getTuple();
- }
- i++;
- } else {
- moveTreeCursor();
- }
- }
- } catch (Exception e) {
- closeResource();
- throw new HyracksDataException(e);
- }
- }
-
- private void outputMatch(int i) throws Exception {
- writeResults(accessor, i, currentTopTuple);
- match = true;
- }
-
- private void moveTreeCursor() throws Exception {
- if (!match) {
- writeResults(currentTopTuple);
- }
- if (cursor.hasNext()) {
- cursor.next();
- currentTopTuple = cursor.getTuple();
- match = false;
- } else {
- currentTopTuple = null;
- }
- }
-
- @Override
- public void close() throws HyracksDataException {
- try {
- while (currentTopTuple != null) {
- moveTreeCursor();
- }
- try {
- cursor.close();
- //batch update
- updateBuffer.updateIndex(indexAccessor);
- } catch (Exception e) {
- throw new HyracksDataException(e);
- }
-
- /**
- * function close
- */
- functionProxy.functionClose();
- } catch (Exception e) {
- throw new HyracksDataException(e);
- } finally {
- treeIndexOpHelper.close();
- }
- }
-
- @Override
- public void fail() throws HyracksDataException {
- closeResource();
- populateFailure();
- }
-
- private void closeResource() throws HyracksDataException {
- try {
- cursor.close();
- } catch (Exception e) {
- throw new HyracksDataException(e);
- } finally {
- treeIndexOpHelper.close();
- }
- }
-
- private void populateFailure() throws HyracksDataException {
- for (IFrameWriter writer : writers) {
- writer.fail();
- }
- }
-
- /** compare tuples */
- private int compare(ITupleReference left, ITupleReference right) throws Exception {
- return lowKeySearchCmp.compare(left, right);
- }
-
- //for the join match casesos
- private void writeResults(IFrameTupleAccessor leftAccessor, int tIndex, ITupleReference indexTuple)
- throws Exception {
- /**
- * merge with the cached tuple, if any
- */
- ITupleReference indexEntryTuple = indexTuple;
- ITupleReference cachedUpdatedLastTuple = updateBuffer.getLastTuple();
- if (cachedUpdatedLastTuple != null) {
- if (compare(cachedUpdatedLastTuple, indexTuple) == 0) {
- indexEntryTuple = cachedUpdatedLastTuple;
- }
- }
- /**
- * function call
- */
- functionProxy.functionCall(leftAccessor, tIndex, indexEntryTuple, cloneUpdateTb, cursor);
-
- /**
- * doing clone update
- */
- CopyUpdateUtil.copyUpdate(tempTupleReference, indexEntryTuple, updateBuffer, cloneUpdateTb, indexAccessor,
- cursor, rangePred, true, storageType);
- }
-
- /** write result for outer case */
- private void writeResults(ITupleReference frameTuple) throws Exception {
- /**
- * function call
- */
- functionProxy.functionCall(nullTupleBuilder, frameTuple, cloneUpdateTb, cursor, true);
-
- //doing clone update
- CopyUpdateUtil.copyUpdate(tempTupleReference, frameTuple, updateBuffer, cloneUpdateTb, indexAccessor, cursor,
- rangePred, true, storageType);
- }
-
- @Override
- public void setOutputFrameWriter(int index, IFrameWriter writer, RecordDescriptor recordDesc) {
- writers[index] = writer;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/ffc967fd/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopRightOuterJoinOperatorNodePushable.java
----------------------------------------------------------------------
diff --git a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopRightOuterJoinOperatorNodePushable.java b/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopRightOuterJoinOperatorNodePushable.java
deleted file mode 100644
index 6539828..0000000
--- a/pregelix/pregelix-dataflow-std/src/main/java/edu/uci/ics/pregelix/dataflow/std/IndexNestedLoopRightOuterJoinOperatorNodePushable.java
+++ /dev/null
@@ -1,283 +0,0 @@
-/*
- * Copyright 2009-2013 by The Regents of the University of California
- * Licensed 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 from
- *
- * 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 edu.uci.ics.pregelix.dataflow.std;
-
-import java.io.DataOutput;
-import java.nio.ByteBuffer;
-
-import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
-import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
-import edu.uci.ics.hyracks.api.dataflow.value.INullWriter;
-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.AbstractUnaryInputUnaryOutputOperatorNodePushable;
-import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
-import edu.uci.ics.hyracks.storage.am.common.api.IIndexAccessor;
-import edu.uci.ics.hyracks.storage.am.common.api.IIndexCursor;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.AbstractTreeIndexOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IndexDataflowHelper;
-import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
-import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
-import edu.uci.ics.hyracks.storage.am.common.tuples.PermutingFrameTupleReference;
-
-public class IndexNestedLoopRightOuterJoinOperatorNodePushable extends
- AbstractUnaryInputUnaryOutputOperatorNodePushable {
- private IndexDataflowHelper treeIndexOpHelper;
- private FrameTupleAccessor accessor;
-
- private ByteBuffer writeBuffer;
- private FrameTupleAppender appender;
- private ArrayTupleBuilder tb;
- private DataOutput dos;
-
- private ITreeIndex index;
- private boolean isForward;
- private RangePredicate rangePred;
- private MultiComparator lowKeySearchCmp;
- private MultiComparator highKeySearchCmp;
- private IIndexCursor cursor;
- protected IIndexAccessor indexAccessor;
-
- private RecordDescriptor recDesc;
- private final RecordDescriptor inputRecDesc;
-
- private PermutingFrameTupleReference lowKey;
- private PermutingFrameTupleReference highKey;
-
- private INullWriter[] nullWriter;
- private ITupleReference currentTopTuple;
- private boolean match;
-
- public IndexNestedLoopRightOuterJoinOperatorNodePushable(AbstractTreeIndexOperatorDescriptor opDesc,
- IHyracksTaskContext ctx, int partition, IRecordDescriptorProvider recordDescProvider, boolean isForward,
- int[] lowKeyFields, int[] highKeyFields, INullWriter[] nullWriter) {
- inputRecDesc = recordDescProvider.getInputRecordDescriptor(opDesc.getActivityId(), 0);
- treeIndexOpHelper = (IndexDataflowHelper) opDesc.getIndexDataflowHelperFactory().createIndexDataflowHelper(
- opDesc, ctx, partition);
- this.isForward = isForward;
- this.recDesc = recordDescProvider.getInputRecordDescriptor(opDesc.getActivityId(), 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);
- }
- this.nullWriter = nullWriter;
- }
-
- protected void setCursor() {
- cursor = indexAccessor.createSearchCursor(false);
- }
-
- @Override
- public void open() throws HyracksDataException {
- accessor = new FrameTupleAccessor(treeIndexOpHelper.getTaskContext().getFrameSize(), recDesc);
- try {
- treeIndexOpHelper.open();
- index = (ITreeIndex) treeIndexOpHelper.getIndexInstance();
- writer.open();
-
- // construct range predicate
- // TODO: Can we construct the multicmps using helper methods?
- int lowKeySearchFields = index.getComparatorFactories().length;
- int highKeySearchFields = index.getComparatorFactories().length;
-
- IBinaryComparator[] lowKeySearchComparators = new IBinaryComparator[lowKeySearchFields];
- for (int i = 0; i < lowKeySearchFields; i++) {
- lowKeySearchComparators[i] = index.getComparatorFactories()[i].createBinaryComparator();
- }
- lowKeySearchCmp = new MultiComparator(lowKeySearchComparators);
-
- if (lowKeySearchFields == highKeySearchFields) {
- highKeySearchCmp = lowKeySearchCmp;
- } else {
- IBinaryComparator[] highKeySearchComparators = new IBinaryComparator[highKeySearchFields];
- for (int i = 0; i < highKeySearchFields; i++) {
- highKeySearchComparators[i] = index.getComparatorFactories()[i].createBinaryComparator();
- }
- highKeySearchCmp = new MultiComparator(highKeySearchComparators);
-
- }
-
- rangePred = new RangePredicate(null, null, true, true, lowKeySearchCmp, highKeySearchCmp);
-
- writeBuffer = treeIndexOpHelper.getTaskContext().allocateFrame();
- tb = new ArrayTupleBuilder(inputRecDesc.getFields().length + index.getFieldCount());
- dos = tb.getDataOutput();
- appender = new FrameTupleAppender(treeIndexOpHelper.getTaskContext().getFrameSize());
- appender.reset(writeBuffer, true);
-
- indexAccessor = index.createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
- setCursor();
-
- /** set the search cursor */
- rangePred.setLowKey(null, true);
- rangePred.setHighKey(null, true);
- cursor.reset();
- indexAccessor.search(cursor, rangePred);
-
- /** set up current top tuple */
- if (cursor.hasNext()) {
- cursor.next();
- currentTopTuple = cursor.getTuple();
- match = false;
- }
-
- } catch (Exception e) {
- treeIndexOpHelper.close();
- throw new HyracksDataException(e);
- }
- }
-
- private void writeResults(IFrameTupleAccessor leftAccessor, int tIndex, ITupleReference frameTuple)
- throws Exception {
- tb.reset();
- for (int i = 0; i < inputRecDesc.getFields().length; i++) {
- int tupleStart = leftAccessor.getTupleStartOffset(tIndex);
- int fieldStart = leftAccessor.getFieldStartOffset(tIndex, i);
- int offset = leftAccessor.getFieldSlotsLength() + tupleStart + fieldStart;
- int len = leftAccessor.getFieldEndOffset(tIndex, i) - fieldStart;
- dos.write(leftAccessor.getBuffer().array(), offset, len);
- tb.addFieldEndOffset();
- }
- 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 && currentTopTuple != null;) {
- if (lowKey != null)
- lowKey.reset(accessor, i);
- if (highKey != null)
- highKey.reset(accessor, i);
- // TODO: currently use low key only, check what they mean
- int cmp = compare(lowKey, currentTopTuple);
- if ((cmp <= 0 && isForward) || (cmp >= 0 && !isForward)) {
- if (cmp == 0)
- outputMatch(i);
- i++;
- } else {
- moveTreeCursor();
- }
- }
- } catch (Exception e) {
- throw new HyracksDataException(e);
- }
- }
-
- private void outputMatch(int i) throws Exception {
- writeResults(accessor, i, currentTopTuple);
- match = true;
- }
-
- private void moveTreeCursor() throws Exception {
- if (!match) {
- writeResults(currentTopTuple);
- }
- if (cursor.hasNext()) {
- cursor.next();
- currentTopTuple = cursor.getTuple();
- match = false;
- } else {
- currentTopTuple = null;
- }
- }
-
- @Override
- public void close() throws HyracksDataException {
- try {
- while (currentTopTuple != null) {
- moveTreeCursor();
- }
-
- if (appender.getTupleCount() > 0) {
- FrameUtils.flushFrame(writeBuffer, writer);
- }
- writer.close();
- try {
- cursor.close();
- } catch (Exception e) {
- throw new HyracksDataException(e);
- }
- } catch (Exception e) {
- throw new HyracksDataException(e);
- } finally {
- treeIndexOpHelper.close();
- }
- }
-
- @Override
- public void fail() throws HyracksDataException {
- try {
- cursor.close();
- } catch (Exception e) {
- throw new HyracksDataException(e);
- } finally {
- treeIndexOpHelper.close();
- }
- writer.fail();
- }
-
- /** compare tuples */
- private int compare(ITupleReference left, ITupleReference right) throws Exception {
- return lowKeySearchCmp.compare(left, right);
- }
-
- /** write result for outer case */
- private void writeResults(ITupleReference frameTuple) throws Exception {
- tb.reset();
- for (int i = 0; i < inputRecDesc.getFields().length; i++) {
- nullWriter[i].writeNull(dos);
- tb.addFieldEndOffset();
- }
- 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();
- }
- }
- }
-}
\ No newline at end of file