You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by hy...@apache.org on 2013/07/02 16:16:22 UTC
[28/51] [partial] TAJO-22: The package prefix should be
org.apache.tajo. (DaeMyung Kang via hyunsik)
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/BinaryPhysicalExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/BinaryPhysicalExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/BinaryPhysicalExec.java
new file mode 100644
index 0000000..a17ddf4
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/BinaryPhysicalExec.java
@@ -0,0 +1,63 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.catalog.Schema;
+
+import java.io.IOException;
+
+public abstract class BinaryPhysicalExec extends PhysicalExec {
+ protected final PhysicalExec outerChild;
+ protected final PhysicalExec innerChild;
+
+ public BinaryPhysicalExec(final TaskAttemptContext context,
+ final Schema inSchema, final Schema outSchema,
+ final PhysicalExec outer, final PhysicalExec inner) {
+ super(context, inSchema, outSchema);
+ this.outerChild = outer;
+ this.innerChild = inner;
+ }
+
+ public PhysicalExec getOuterChild() {
+ return outerChild;
+ }
+
+ public PhysicalExec getInnerChild() {
+ return innerChild;
+ }
+
+ @Override
+ public void init() throws IOException {
+ outerChild.init();
+ innerChild.init();
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ outerChild.rescan();
+ innerChild.rescan();
+ }
+
+ @Override
+ public void close() throws IOException {
+ outerChild.close();
+ innerChild.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/EvalExprExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/EvalExprExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/EvalExprExec.java
new file mode 100644
index 0000000..5af4f18
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/EvalExprExec.java
@@ -0,0 +1,69 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.engine.eval.EvalContext;
+import org.apache.tajo.engine.parser.QueryBlock.Target;
+import org.apache.tajo.engine.planner.logical.EvalExprNode;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+
+import java.io.IOException;
+
+public class EvalExprExec extends PhysicalExec {
+ private final EvalExprNode plan;
+ private final EvalContext[] evalContexts;
+
+ public EvalExprExec(final TaskAttemptContext context, final EvalExprNode plan) {
+ super(context, plan.getInSchema(), plan.getOutSchema());
+ this.plan = plan;
+
+ evalContexts = new EvalContext[plan.getExprs().length];
+ for (int i = 0; i < plan.getExprs().length; i++) {
+ evalContexts[i] = plan.getExprs()[i].getEvalTree().newContext();
+ }
+ }
+
+ @Override
+ public void init() throws IOException {
+ }
+
+ /* (non-Javadoc)
+ * @see PhysicalExec#next()
+ */
+ @Override
+ public Tuple next() throws IOException {
+ Target [] targets = plan.getExprs();
+ Tuple t = new VTuple(targets.length);
+ for (int i = 0; i < targets.length; i++) {
+ targets[i].getEvalTree().eval(evalContexts[i], inSchema, null);
+ t.put(i, targets[i].getEvalTree().terminate(evalContexts[i]));
+ }
+ return t;
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ }
+
+ @Override
+ public void close() throws IOException {
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
new file mode 100644
index 0000000..7ba0d8a
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
@@ -0,0 +1,221 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
+import org.apache.tajo.conf.TajoConf.ConfVars;
+import org.apache.tajo.engine.planner.logical.SortNode;
+import org.apache.tajo.storage.*;
+
+import java.io.IOException;
+import java.util.*;
+
+public class ExternalSortExec extends SortExec {
+ private SortNode plan;
+
+ private final List<Tuple> tupleSlots;
+ private boolean sorted = false;
+ private RawFile.RawFileScanner result;
+ private RawFile.RawFileAppender appender;
+ private FileSystem localFS;
+
+ private final TableMeta meta;
+ private final Path sortTmpDir;
+ private int SORT_BUFFER_SIZE;
+
+ public ExternalSortExec(final TaskAttemptContext context,
+ final StorageManager sm, final SortNode plan, final PhysicalExec child)
+ throws IOException {
+ super(context, plan.getInSchema(), plan.getOutSchema(), child,
+ plan.getSortKeys());
+ this.plan = plan;
+
+ this.SORT_BUFFER_SIZE = context.getConf().getIntVar(ConfVars.EXT_SORT_BUFFER);
+ this.tupleSlots = new ArrayList<Tuple>(SORT_BUFFER_SIZE);
+
+ this.sortTmpDir = new Path(context.getWorkDir(), UUID.randomUUID().toString());
+ this.localFS = FileSystem.getLocal(context.getConf());
+ meta = CatalogUtil.newTableMeta(inSchema, StoreType.ROWFILE);
+ }
+
+ public void init() throws IOException {
+ super.init();
+ localFS.mkdirs(sortTmpDir);
+ }
+
+ public SortNode getPlan() {
+ return this.plan;
+ }
+
+ private void sortAndStoreChunk(int chunkId, List<Tuple> tupleSlots)
+ throws IOException {
+ TableMeta meta = CatalogUtil.newTableMeta(inSchema, StoreType.RAW);
+ Collections.sort(tupleSlots, getComparator());
+ // TODO - RawFile requires the local file path.
+ // So, I add the scheme 'file:/' to path. But, it should be improved.
+ Path localPath = new Path(sortTmpDir + "/0_" + chunkId);
+
+ appender = new RawFile.RawFileAppender(context.getConf(), meta, localPath);
+ appender.init();
+
+ for (Tuple t : tupleSlots) {
+ appender.addTuple(t);
+ }
+ appender.close();
+ tupleSlots.clear();
+ }
+
+ /**
+ * It divides all tuples into a number of chunks, then sort for each chunk.
+ * @return the number of stored chunks
+ * @throws java.io.IOException
+ */
+ private int sortAndStoreAllChunks() throws IOException {
+ int chunkId = 0;
+
+ Tuple tuple;
+ while ((tuple = child.next()) != null) { // partition sort start
+ tupleSlots.add(new VTuple(tuple));
+ if (tupleSlots.size() == SORT_BUFFER_SIZE) {
+ sortAndStoreChunk(chunkId, tupleSlots);
+ chunkId++;
+ }
+ }
+
+ if (tupleSlots.size() > 0) {
+ sortAndStoreChunk(chunkId, tupleSlots);
+ chunkId++;
+ }
+
+ return chunkId;
+ }
+
+ private Path getChunkPath(int level, int chunkId) {
+ return StorageUtil.concatPath(sortTmpDir, "" + level + "_" + chunkId);
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ if (!sorted) {
+
+ // the total number of chunks for zero level
+ int totalChunkNumForLevel = sortAndStoreAllChunks();
+
+ // if there are no chunk
+ if (totalChunkNumForLevel == 0) {
+ return null;
+ }
+
+ int level = 0;
+ int chunkId = 0;
+
+ // continue until the chunk remains only one
+ while (totalChunkNumForLevel > 1) {
+
+ while (chunkId < totalChunkNumForLevel) {
+
+ Path nextChunk = getChunkPath(level + 1, chunkId / 2);
+
+ // if number of chunkId is odd just copy it.
+ if (chunkId + 1 >= totalChunkNumForLevel) {
+
+ Path chunk = getChunkPath(level, chunkId);
+ localFS.moveFromLocalFile(chunk, nextChunk);
+
+ } else {
+
+ Path leftChunk = getChunkPath(level, chunkId);
+ Path rightChunk = getChunkPath(level, chunkId + 1);
+
+ appender = new RawFile.RawFileAppender(context.getConf(), meta, nextChunk);
+ appender.init();
+ merge(appender, leftChunk, rightChunk);
+
+ appender.flush();
+ appender.close();
+ }
+
+ chunkId += 2;
+ }
+
+ level++;
+ // init chunkId for each level
+ chunkId = 0;
+ // calculate the total number of chunks for next level
+ totalChunkNumForLevel = totalChunkNumForLevel / 2
+ + totalChunkNumForLevel % 2;
+ }
+
+ Path result = getChunkPath(level, 0);
+ this.result = new RawFile.RawFileScanner(context.getConf(), meta, result);
+ sorted = true;
+ }
+
+ return result.next();
+ }
+
+ private void merge(RawFile.RawFileAppender appender, Path left, Path right)
+ throws IOException {
+ RawFile.RawFileScanner leftScan = new RawFile.RawFileScanner(context.getConf(), meta, left);
+
+ RawFile.RawFileScanner rightScan =
+ new RawFile.RawFileScanner(context.getConf(), meta, right);
+
+ Tuple leftTuple = leftScan.next();
+ Tuple rightTuple = rightScan.next();
+
+ Comparator<Tuple> comparator = getComparator();
+ while (leftTuple != null && rightTuple != null) {
+ if (comparator.compare(leftTuple, rightTuple) < 0) {
+ appender.addTuple(leftTuple);
+ leftTuple = leftScan.next();
+ } else {
+ appender.addTuple(rightTuple);
+ rightTuple = rightScan.next();
+ }
+ }
+
+ if (leftTuple == null) {
+ appender.addTuple(rightTuple);
+ while ((rightTuple = rightScan.next()) != null) {
+ appender.addTuple(rightTuple);
+ }
+ } else {
+ appender.addTuple(leftTuple);
+ while ((leftTuple = leftScan.next()) != null) {
+ appender.addTuple(leftTuple);
+ }
+ }
+
+ leftScan.close();
+ rightScan.close();
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ if (result != null) {
+ result.reset();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashAggregateExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashAggregateExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashAggregateExec.java
new file mode 100644
index 0000000..427e6f4
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashAggregateExec.java
@@ -0,0 +1,108 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.engine.eval.EvalContext;
+import org.apache.tajo.engine.planner.logical.GroupbyNode;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+
+/**
+ * This is the hash-based GroupBy Operator.
+ */
+public class HashAggregateExec extends AggregationExec {
+ private Tuple tuple = null;
+ private Map<Tuple, EvalContext[]> tupleSlots;
+ private boolean computed = false;
+ private Iterator<Entry<Tuple, EvalContext []>> iterator = null;
+
+ /**
+ * @throws java.io.IOException
+ *
+ */
+ public HashAggregateExec(TaskAttemptContext ctx, GroupbyNode annotation,
+ PhysicalExec subOp) throws IOException {
+ super(ctx, annotation, subOp);
+ tupleSlots = new HashMap<Tuple, EvalContext[]>(10000);
+ this.tuple = new VTuple(outSchema.getColumnNum());
+ }
+
+ private void compute() throws IOException {
+ Tuple tuple;
+ Tuple keyTuple;
+ while((tuple = child.next()) != null && !context.isStopped()) {
+ keyTuple = new VTuple(keylist.length);
+ // build one key tuple
+ for(int i = 0; i < keylist.length; i++) {
+ keyTuple.put(i, tuple.get(keylist[i]));
+ }
+
+ if(tupleSlots.containsKey(keyTuple)) {
+ EvalContext [] tmpTuple = tupleSlots.get(keyTuple);
+ for(int i = 0; i < measureList.length; i++) {
+ evals[measureList[i]].eval(tmpTuple[measureList[i]], inSchema, tuple);
+ }
+ } else { // if the key occurs firstly
+ EvalContext evalCtx [] = new EvalContext[outSchema.getColumnNum()];
+ for(int i = 0; i < outSchema.getColumnNum(); i++) {
+ evalCtx[i] = evals[i].newContext();
+ evals[i].eval(evalCtx[i], inSchema, tuple);
+ }
+ tupleSlots.put(keyTuple, evalCtx);
+ }
+ }
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ if(!computed) {
+ compute();
+ iterator = tupleSlots.entrySet().iterator();
+ computed = true;
+ }
+
+ if(iterator.hasNext()) {
+ EvalContext [] ctx = iterator.next().getValue();
+ for (int i = 0; i < ctx.length; i++) {
+ tuple.put(i, evals[i].terminate(ctx[i]));
+ }
+ return tuple;
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ iterator = tupleSlots.entrySet().iterator();
+ }
+
+ @Override
+ public void close() throws IOException {
+ super.close();
+ tupleSlots.clear();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java
new file mode 100644
index 0000000..cce5587
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java
@@ -0,0 +1,196 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.engine.eval.EvalContext;
+import org.apache.tajo.engine.eval.EvalNode;
+import org.apache.tajo.engine.planner.PlannerUtil;
+import org.apache.tajo.engine.planner.Projector;
+import org.apache.tajo.engine.planner.logical.JoinNode;
+import org.apache.tajo.engine.utils.SchemaUtil;
+import org.apache.tajo.storage.FrameTuple;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+
+import java.io.IOException;
+import java.util.*;
+
+public class HashJoinExec extends BinaryPhysicalExec {
+ // from logical plan
+ private JoinNode plan;
+ private EvalNode joinQual;
+
+ private List<Column[]> joinKeyPairs;
+
+ // temporal tuples and states for nested loop join
+ private boolean first = true;
+ private FrameTuple frameTuple;
+ private Tuple outTuple = null;
+ private Map<Tuple, List<Tuple>> tupleSlots;
+ private Iterator<Tuple> iterator = null;
+ private EvalContext qualCtx;
+ private Tuple outerTuple;
+ private Tuple outerKeyTuple;
+
+ private int [] outerKeyList;
+ private int [] innerKeyList;
+
+ private boolean finished = false;
+ boolean nextOuter = true;
+
+ // projection
+ private final Projector projector;
+ private final EvalContext [] evalContexts;
+
+ public HashJoinExec(TaskAttemptContext context, JoinNode plan, PhysicalExec outer,
+ PhysicalExec inner) {
+ super(context, SchemaUtil.merge(outer.getSchema(), inner.getSchema()),
+ plan.getOutSchema(), outer, inner);
+ this.plan = plan;
+ this.joinQual = plan.getJoinQual();
+ this.qualCtx = joinQual.newContext();
+ this.tupleSlots = new HashMap<Tuple, List<Tuple>>(10000);
+
+ this.joinKeyPairs = PlannerUtil.getJoinKeyPairs(joinQual,
+ outer.getSchema(), inner.getSchema());
+
+ outerKeyList = new int[joinKeyPairs.size()];
+ innerKeyList = new int[joinKeyPairs.size()];
+
+ for (int i = 0; i < joinKeyPairs.size(); i++) {
+ outerKeyList[i] = outer.getSchema().getColumnId(joinKeyPairs.get(i)[0].getQualifiedName());
+ }
+
+ for (int i = 0; i < joinKeyPairs.size(); i++) {
+ innerKeyList[i] = inner.getSchema().getColumnId(joinKeyPairs.get(i)[1].getQualifiedName());
+ }
+
+ // for projection
+ this.projector = new Projector(inSchema, outSchema, plan.getTargets());
+ this.evalContexts = projector.renew();
+
+ // for join
+ frameTuple = new FrameTuple();
+ outTuple = new VTuple(outSchema.getColumnNum());
+ outerKeyTuple = new VTuple(outerKeyList.length);
+ }
+
+ private void getKeyOuterTuple(final Tuple outerTuple, Tuple keyTuple) {
+ for (int i = 0; i < outerKeyList.length; i++) {
+ keyTuple.put(i, outerTuple.get(outerKeyList[i]));
+ }
+ }
+
+ public Tuple next() throws IOException {
+ if (first) {
+ loadInnerTable();
+ }
+
+ Tuple innerTuple;
+ boolean found = false;
+
+ while(!finished) {
+
+ if (nextOuter) {
+ // getting new outer
+ outerTuple = outerChild.next();
+ if (outerTuple == null) {
+ finished = true;
+ return null;
+ }
+
+ // getting corresponding inner
+ getKeyOuterTuple(outerTuple, outerKeyTuple);
+ if (tupleSlots.containsKey(outerKeyTuple)) {
+ iterator = tupleSlots.get(outerKeyTuple).iterator();
+ nextOuter = false;
+ } else {
+ nextOuter = true;
+ continue;
+ }
+ }
+
+ // getting next inner tuple
+ innerTuple = iterator.next();
+ frameTuple.set(outerTuple, innerTuple);
+ joinQual.eval(qualCtx, inSchema, frameTuple);
+ if (joinQual.terminate(qualCtx).asBool()) {
+ projector.eval(evalContexts, frameTuple);
+ projector.terminate(evalContexts, outTuple);
+ found = true;
+ }
+
+ if (!iterator.hasNext()) { // no more inner tuple
+ nextOuter = true;
+ }
+
+ if (found) {
+ break;
+ }
+ }
+
+ return outTuple;
+ }
+
+ private void loadInnerTable() throws IOException {
+ Tuple tuple;
+ Tuple keyTuple;
+
+ while ((tuple = innerChild.next()) != null) {
+ keyTuple = new VTuple(joinKeyPairs.size());
+ List<Tuple> newValue;
+ for (int i = 0; i < innerKeyList.length; i++) {
+ keyTuple.put(i, tuple.get(innerKeyList[i]));
+ }
+
+ if (tupleSlots.containsKey(keyTuple)) {
+ newValue = tupleSlots.get(keyTuple);
+ newValue.add(tuple);
+ tupleSlots.put(keyTuple, newValue);
+ } else {
+ newValue = new ArrayList<Tuple>();
+ newValue.add(tuple);
+ tupleSlots.put(keyTuple, newValue);
+ }
+ }
+ first = false;
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ super.rescan();
+
+ tupleSlots.clear();
+ first = true;
+
+ finished = false;
+ iterator = null;
+ nextOuter = true;
+ }
+
+ public void close() throws IOException {
+ tupleSlots.clear();
+ }
+
+ public JoinNode getPlan() {
+ return this.plan;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashPartitioner.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashPartitioner.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashPartitioner.java
new file mode 100644
index 0000000..485e0d1
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashPartitioner.java
@@ -0,0 +1,44 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ *
+ */
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+
+public class HashPartitioner extends Partitioner {
+ private final Tuple keyTuple;
+
+ public HashPartitioner(final int [] keys, final int numPartitions) {
+ super(keys, numPartitions);
+ this.keyTuple = new VTuple(partitionKeys.length);
+ }
+
+ @Override
+ public int getPartition(Tuple tuple) {
+ // build one key tuple
+ for (int i = 0; i < partitionKeys.length; i++) {
+ keyTuple.put(i, tuple.get(partitionKeys[i]));
+ }
+ return (keyTuple.hashCode() & Integer.MAX_VALUE) %
+ (numPartitions == 32 ? numPartitions-1 : numPartitions);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexWriteExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexWriteExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexWriteExec.java
new file mode 100644
index 0000000..210a90c
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexWriteExec.java
@@ -0,0 +1,88 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ *
+ */
+package org.apache.tajo.engine.planner.physical;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.engine.planner.logical.IndexWriteNode;
+import org.apache.tajo.storage.*;
+import org.apache.tajo.storage.index.bst.BSTIndex;
+import org.apache.tajo.storage.index.bst.BSTIndex.BSTIndexWriter;
+import org.apache.tajo.util.IndexUtil;
+
+import java.io.IOException;
+
+public class IndexWriteExec extends UnaryPhysicalExec {
+ private int [] indexKeys = null;
+ private final BSTIndexWriter indexWriter;
+ private final TupleComparator comp;
+
+ public IndexWriteExec(final TaskAttemptContext context, final StorageManager sm,
+ final IndexWriteNode plan, final Fragment fragment,
+ final PhysicalExec child) throws IOException {
+ super(context, plan.getInSchema(), plan.getOutSchema(), child);
+ Preconditions.checkArgument(inSchema.equals(child.getSchema()));
+
+ indexKeys = new int[plan.getSortSpecs().length];
+ Schema keySchema = new Schema();
+ Column col;
+ for (int i = 0 ; i < plan.getSortSpecs().length; i++) {
+ col = plan.getSortSpecs()[i].getSortKey();
+ indexKeys[i] = inSchema.getColumnId(col.getQualifiedName());
+ keySchema.addColumn(inSchema.getColumn(col.getQualifiedName()));
+ }
+ this.comp = new TupleComparator(keySchema, plan.getSortSpecs());
+
+ BSTIndex bst = new BSTIndex(new TajoConf());
+ Path dir = new Path(sm.getTablePath(plan.getTableName()) , "index");
+ // TODO - to be improved
+ this.indexWriter = bst.getIndexWriter(new Path(dir,
+ IndexUtil.getIndexNameOfFrag(fragment, plan.getSortSpecs())),
+ BSTIndex.TWO_LEVEL_INDEX, keySchema, comp);
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ indexWriter.open();
+ Tuple tuple;
+
+ while ((tuple = child.next()) != null) {
+ Tuple keys = new VTuple(indexKeys.length);
+ for (int idx = 0; idx < indexKeys.length; idx++) {
+ keys.put(idx, tuple.get(indexKeys[idx]));
+ }
+ indexWriter.write(keys, tuple.getOffset());
+ }
+
+ indexWriter.flush();
+ indexWriter.close();
+ return null;
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexedStoreExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexedStoreExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexedStoreExec.java
new file mode 100644
index 0000000..03eeb51
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexedStoreExec.java
@@ -0,0 +1,118 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.catalog.*;
+import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.engine.planner.PlannerUtil;
+import org.apache.tajo.storage.*;
+import org.apache.tajo.storage.index.bst.BSTIndex;
+
+import java.io.IOException;
+
+public class IndexedStoreExec extends UnaryPhysicalExec {
+ private static Log LOG = LogFactory.getLog(IndexedStoreExec.class);
+ private final SortSpec[] sortSpecs;
+ private int [] indexKeys = null;
+ private Schema keySchema;
+
+ private BSTIndex.BSTIndexWriter indexWriter;
+ private TupleComparator comp;
+ private FileAppender appender;
+ private TableMeta meta;
+
+ public IndexedStoreExec(final TaskAttemptContext context, final StorageManager sm,
+ final PhysicalExec child, final Schema inSchema, final Schema outSchema,
+ final SortSpec[] sortSpecs) throws IOException {
+ super(context, inSchema, outSchema, child);
+ this.sortSpecs = sortSpecs;
+ }
+
+ public void init() throws IOException {
+ super.init();
+
+ indexKeys = new int[sortSpecs.length];
+ keySchema = PlannerUtil.sortSpecsToSchema(sortSpecs);
+
+ Column col;
+ for (int i = 0 ; i < sortSpecs.length; i++) {
+ col = sortSpecs[i].getSortKey();
+ indexKeys[i] = inSchema.getColumnId(col.getQualifiedName());
+ }
+
+ BSTIndex bst = new BSTIndex(new TajoConf());
+ this.comp = new TupleComparator(keySchema, sortSpecs);
+ Path storeTablePath = new Path(context.getWorkDir(), "output");
+ LOG.info("Output data directory: " + storeTablePath);
+ this.meta = CatalogUtil
+ .newTableMeta(this.outSchema, CatalogProtos.StoreType.CSV);
+ FileSystem fs = new RawLocalFileSystem();
+ fs.mkdirs(storeTablePath);
+ this.appender = (FileAppender) StorageManager.getAppender(context.getConf(), meta,
+ new Path(storeTablePath, "output"));
+ this.appender.enableStats();
+ this.appender.init();
+ this.indexWriter = bst.getIndexWriter(new Path(storeTablePath, "index"),
+ BSTIndex.TWO_LEVEL_INDEX, keySchema, comp);
+ this.indexWriter.setLoadNum(100);
+ this.indexWriter.open();
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ Tuple tuple;
+ Tuple keyTuple;
+ long offset;
+
+
+ while((tuple = child.next()) != null) {
+ offset = appender.getOffset();
+ appender.addTuple(tuple);
+ keyTuple = new VTuple(keySchema.getColumnNum());
+ RowStoreUtil.project(tuple, keyTuple, indexKeys);
+ indexWriter.write(keyTuple, offset);
+ }
+
+ return null;
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ }
+
+ public void close() throws IOException {
+ super.close();
+
+ appender.flush();
+ appender.close();
+ indexWriter.flush();
+ indexWriter.close();
+
+ // Collect statistics data
+ context.setResultStats(appender.getStats());
+ context.addRepartition(0, context.getTaskId().toString());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/JoinTupleComparator.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/JoinTupleComparator.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/JoinTupleComparator.java
new file mode 100644
index 0000000..0d4c47b
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/JoinTupleComparator.java
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import com.google.common.base.Preconditions;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.SortSpec;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.storage.Tuple;
+
+import java.util.Comparator;
+
+/**
+ * The Comparator class for Outer and Inner Tuples
+ *
+ * @see org.apache.tajo.storage.Tuple
+ */
+public class JoinTupleComparator implements Comparator<Tuple> {
+ private int numSortKey;
+ private final int[] outerSortKeyIds;
+ private final int[] innerSortKeyIds;
+
+ private Datum outer;
+ private Datum inner;
+ private int compVal;
+
+ public JoinTupleComparator(Schema leftschema, Schema rightschema, SortSpec[][] sortKeys) {
+ Preconditions.checkArgument(sortKeys.length == 2,
+ "The two of the sortspecs must be given, but " + sortKeys.length + " sortkeys are given.");
+ Preconditions.checkArgument(sortKeys[0].length == sortKeys[1].length,
+ "The number of both side sortkeys must be equals, but they are different: "
+ + sortKeys[0].length + " and " + sortKeys[1].length);
+
+ this.numSortKey = sortKeys[0].length; // because it is guaranteed that the number of sortspecs are equals
+ this.outerSortKeyIds = new int[numSortKey];
+ this.innerSortKeyIds = new int[numSortKey];
+
+ for (int i = 0; i < numSortKey; i++) {
+ this.outerSortKeyIds[i] = leftschema.getColumnId(sortKeys[0][i].getSortKey().getQualifiedName());
+ this.innerSortKeyIds[i] = rightschema.getColumnId(sortKeys[1][i].getSortKey().getQualifiedName());
+ }
+ }
+
+ @Override
+ public int compare(Tuple outerTuple, Tuple innerTuple) {
+ for (int i = 0; i < numSortKey; i++) {
+ outer = outerTuple.get(outerSortKeyIds[i]);
+ inner = innerTuple.get(innerSortKeyIds[i]);
+
+ if (outer instanceof NullDatum || inner instanceof NullDatum) {
+ if (!outer.equals(inner)) {
+ if (outer instanceof NullDatum) {
+ compVal = 1;
+ } else if (inner instanceof NullDatum) {
+ compVal = -1;
+ }
+ } else {
+ compVal = 0;
+ }
+ } else {
+ compVal = outer.compareTo(inner);
+ }
+
+ if (compVal < 0 || compVal > 0) {
+ return compVal;
+ }
+ }
+ return 0;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/LimitExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/LimitExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/LimitExec.java
new file mode 100644
index 0000000..eb3f3c8
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/LimitExec.java
@@ -0,0 +1,55 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.engine.planner.logical.LimitNode;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.IOException;
+
+public class LimitExec extends UnaryPhysicalExec {
+ private final long fetchFirstNum;
+ private long fetchCount;
+
+ public LimitExec(TaskAttemptContext context, Schema inSchema,
+ Schema outSchema, PhysicalExec child, LimitNode limit) {
+ super(context, inSchema, outSchema, child);
+ this.fetchFirstNum = limit.getFetchFirstNum();
+ this.fetchCount = 0;
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ Tuple tuple = child.next();
+ fetchCount++;
+
+ if (fetchCount > fetchFirstNum || tuple == null) {
+ return null;
+ }
+
+ return tuple;
+ }
+
+ public void rescan() throws IOException {
+ super.init();
+ fetchCount = 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/MemSortExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/MemSortExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/MemSortExec.java
new file mode 100644
index 0000000..1532307
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/MemSortExec.java
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.engine.planner.logical.SortNode;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+public class MemSortExec extends SortExec {
+ private final SortNode plan;
+ private List<Tuple> tupleSlots;
+ private boolean sorted = false;
+ private Iterator<Tuple> iterator;
+
+ public MemSortExec(final TaskAttemptContext context,
+ SortNode plan, PhysicalExec child) {
+ super(context, plan.getInSchema(), plan.getOutSchema(), child,
+ plan.getSortKeys());
+ this.plan = plan;
+ }
+
+ public void init() throws IOException {
+ super.init();
+ this.tupleSlots = new ArrayList<Tuple>(1000);
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+
+ if (!sorted) {
+ Tuple tuple;
+ while ((tuple = child.next()) != null) {
+ tupleSlots.add(new VTuple(tuple));
+ }
+
+ Collections.sort(tupleSlots, getComparator());
+ this.iterator = tupleSlots.iterator();
+ sorted = true;
+ }
+
+ if (iterator.hasNext()) {
+ return this.iterator.next();
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ super.rescan();
+ this.iterator = tupleSlots.iterator();
+ sorted = true;
+ }
+
+ public SortNode getPlan() {
+ return this.plan;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/MergeJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/MergeJoinExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/MergeJoinExec.java
new file mode 100644
index 0000000..0179da5
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/MergeJoinExec.java
@@ -0,0 +1,181 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import com.google.common.base.Preconditions;
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.catalog.SortSpec;
+import org.apache.tajo.engine.eval.EvalContext;
+import org.apache.tajo.engine.eval.EvalNode;
+import org.apache.tajo.engine.planner.PlannerUtil;
+import org.apache.tajo.engine.planner.Projector;
+import org.apache.tajo.engine.planner.logical.JoinNode;
+import org.apache.tajo.storage.FrameTuple;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.TupleComparator;
+import org.apache.tajo.storage.VTuple;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+public class MergeJoinExec extends BinaryPhysicalExec {
+ // from logical plan
+ private JoinNode joinNode;
+ private EvalNode joinQual;
+ private EvalContext qualCtx;
+
+ // temporal tuples and states for nested loop join
+ private FrameTuple frameTuple;
+ private Tuple outerTuple = null;
+ private Tuple innerTuple = null;
+ private Tuple outTuple = null;
+ private Tuple outerNext = null;
+
+ private final List<Tuple> outerTupleSlots;
+ private final List<Tuple> innerTupleSlots;
+ private Iterator<Tuple> outerIterator;
+ private Iterator<Tuple> innerIterator;
+
+ private JoinTupleComparator joincomparator = null;
+ private TupleComparator[] tupleComparator = null;
+
+ private final static int INITIAL_TUPLE_SLOT = 10000;
+
+ private boolean end = false;
+
+ // projection
+ private final Projector projector;
+ private final EvalContext [] evalContexts;
+
+ public MergeJoinExec(TaskAttemptContext context, JoinNode plan, PhysicalExec outer,
+ PhysicalExec inner, SortSpec[] outerSortKey, SortSpec[] innerSortKey) {
+ super(context, plan.getInSchema(), plan.getOutSchema(), outer, inner);
+ Preconditions.checkArgument(plan.hasJoinQual(), "Sort-merge join is only used for the equi-join, " +
+ "but there is no join condition");
+ this.joinNode = plan;
+ this.joinQual = plan.getJoinQual();
+ this.qualCtx = this.joinQual.newContext();
+
+ this.outerTupleSlots = new ArrayList<Tuple>(INITIAL_TUPLE_SLOT);
+ this.innerTupleSlots = new ArrayList<Tuple>(INITIAL_TUPLE_SLOT);
+ SortSpec[][] sortSpecs = new SortSpec[2][];
+ sortSpecs[0] = outerSortKey;
+ sortSpecs[1] = innerSortKey;
+
+ this.joincomparator = new JoinTupleComparator(outer.getSchema(),
+ inner.getSchema(), sortSpecs);
+ this.tupleComparator = PlannerUtil.getComparatorsFromJoinQual(
+ plan.getJoinQual(), outer.getSchema(), inner.getSchema());
+ this.outerIterator = outerTupleSlots.iterator();
+ this.innerIterator = innerTupleSlots.iterator();
+
+ // for projection
+ this.projector = new Projector(inSchema, outSchema, plan.getTargets());
+ this.evalContexts = projector.renew();
+
+ // for join
+ frameTuple = new FrameTuple();
+ outTuple = new VTuple(outSchema.getColumnNum());
+ }
+
+ public JoinNode getJoinNode(){
+ return this.joinNode;
+ }
+
+ public Tuple next() throws IOException {
+ Tuple previous;
+
+ for (;;) {
+ if (!outerIterator.hasNext() && !innerIterator.hasNext()) {
+ if(end){
+ return null;
+ }
+
+ if(outerTuple == null){
+ outerTuple = outerChild.next();
+ }
+ if(innerTuple == null){
+ innerTuple = innerChild.next();
+ }
+
+ outerTupleSlots.clear();
+ innerTupleSlots.clear();
+
+ int cmp;
+ while ((cmp = joincomparator.compare(outerTuple, innerTuple)) != 0) {
+ if (cmp > 0) {
+ innerTuple = innerChild.next();
+ } else if (cmp < 0) {
+ outerTuple = outerChild.next();
+ }
+ if (innerTuple == null || outerTuple == null) {
+ return null;
+ }
+ }
+
+ previous = outerTuple;
+ do {
+ outerTupleSlots.add(outerTuple);
+ outerTuple = outerChild.next();
+ if (outerTuple == null) {
+ end = true;
+ break;
+ }
+ } while (tupleComparator[0].compare(previous, outerTuple) == 0);
+ outerIterator = outerTupleSlots.iterator();
+ outerNext = outerIterator.next();
+
+ previous = innerTuple;
+ do {
+ innerTupleSlots.add(innerTuple);
+ innerTuple = innerChild.next();
+ if (innerTuple == null) {
+ end = true;
+ break;
+ }
+ } while (tupleComparator[1].compare(previous, innerTuple) == 0);
+ innerIterator = innerTupleSlots.iterator();
+ }
+
+ if(!innerIterator.hasNext()){
+ outerNext = outerIterator.next();
+ innerIterator = innerTupleSlots.iterator();
+ }
+
+ frameTuple.set(outerNext, innerIterator.next());
+ joinQual.eval(qualCtx, inSchema, frameTuple);
+ if (joinQual.terminate(qualCtx).asBool()) {
+ projector.eval(evalContexts, frameTuple);
+ projector.terminate(evalContexts, outTuple);
+ return outTuple;
+ }
+ }
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ super.rescan();
+ outerTupleSlots.clear();
+ innerTupleSlots.clear();
+ outerIterator = outerTupleSlots.iterator();
+ innerIterator = innerTupleSlots.iterator();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/NLJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/NLJoinExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/NLJoinExec.java
new file mode 100644
index 0000000..82e78f3
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/NLJoinExec.java
@@ -0,0 +1,108 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.engine.eval.EvalContext;
+import org.apache.tajo.engine.eval.EvalNode;
+import org.apache.tajo.engine.planner.Projector;
+import org.apache.tajo.engine.planner.logical.JoinNode;
+import org.apache.tajo.storage.FrameTuple;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+
+import java.io.IOException;
+
+public class NLJoinExec extends BinaryPhysicalExec {
+ // from logical plan
+ private JoinNode plan;
+ private EvalNode joinQual;
+
+
+ // temporal tuples and states for nested loop join
+ private boolean needNewOuter;
+ private FrameTuple frameTuple;
+ private Tuple outerTuple = null;
+ private Tuple innerTuple = null;
+ private Tuple outTuple = null;
+ private EvalContext qualCtx;
+
+ // projection
+ private final EvalContext [] evalContexts;
+ private final Projector projector;
+
+ public NLJoinExec(TaskAttemptContext context, JoinNode plan, PhysicalExec outer,
+ PhysicalExec inner) {
+ super(context, plan.getInSchema(), plan.getOutSchema(), outer, inner);
+ this.plan = plan;
+
+ if (plan.hasJoinQual()) {
+ this.joinQual = plan.getJoinQual();
+ this.qualCtx = this.joinQual.newContext();
+ }
+
+ // for projection
+ projector = new Projector(inSchema, outSchema, plan.getTargets());
+ evalContexts = projector.renew();
+
+ // for join
+ needNewOuter = true;
+ frameTuple = new FrameTuple();
+ outTuple = new VTuple(outSchema.getColumnNum());
+ }
+
+ public Tuple next() throws IOException {
+ for (;;) {
+ if (needNewOuter) {
+ outerTuple = outerChild.next();
+ if (outerTuple == null) {
+ return null;
+ }
+ needNewOuter = false;
+ }
+
+ innerTuple = innerChild.next();
+ if (innerTuple == null) {
+ needNewOuter = true;
+ innerChild.rescan();
+ continue;
+ }
+
+ frameTuple.set(outerTuple, innerTuple);
+ if (joinQual != null) {
+ joinQual.eval(qualCtx, inSchema, frameTuple);
+ if (joinQual.terminate(qualCtx).asBool()) {
+ projector.eval(evalContexts, frameTuple);
+ projector.terminate(evalContexts, outTuple);
+ return outTuple;
+ }
+ } else {
+ projector.eval(evalContexts, frameTuple);
+ projector.terminate(evalContexts, outTuple);
+ return outTuple;
+ }
+ }
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ super.rescan();
+ needNewOuter = true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PartitionedStoreExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PartitionedStoreExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PartitionedStoreExec.java
new file mode 100644
index 0000000..f70bf18
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PartitionedStoreExec.java
@@ -0,0 +1,153 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
+import org.apache.tajo.catalog.statistics.StatisticsUtil;
+import org.apache.tajo.catalog.statistics.TableStat;
+import org.apache.tajo.engine.planner.logical.StoreTableNode;
+import org.apache.tajo.storage.Appender;
+import org.apache.tajo.storage.StorageManager;
+import org.apache.tajo.storage.StorageUtil;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.IOException;
+import java.text.NumberFormat;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public final class PartitionedStoreExec extends UnaryPhysicalExec {
+ private static Log LOG = LogFactory.getLog(PartitionedStoreExec.class);
+ private static final NumberFormat numFormat = NumberFormat.getInstance();
+
+ static {
+ numFormat.setGroupingUsed(false);
+ numFormat.setMinimumIntegerDigits(6);
+ }
+
+ private final StoreTableNode plan;
+
+ private final int numPartitions;
+ private final int [] partitionKeys;
+
+ private final TableMeta meta;
+ private final Partitioner partitioner;
+ private final Path storeTablePath;
+ private final Map<Integer, Appender> appenderMap = new HashMap<Integer, Appender>();
+
+ public PartitionedStoreExec(TaskAttemptContext context, final StorageManager sm,
+ final StoreTableNode plan, final PhysicalExec child) throws IOException {
+ super(context, plan.getInSchema(), plan.getOutSchema(), child);
+ Preconditions.checkArgument(plan.hasPartitionKey());
+ this.plan = plan;
+ this.meta = CatalogUtil.newTableMeta(this.outSchema, StoreType.CSV);
+
+ // about the partitions
+ this.numPartitions = this.plan.getNumPartitions();
+ int i = 0;
+ this.partitionKeys = new int [this.plan.getPartitionKeys().length];
+ for (Column key : this.plan.getPartitionKeys()) {
+ partitionKeys[i] = inSchema.getColumnId(key.getQualifiedName());
+ i++;
+ }
+ this.partitioner = new HashPartitioner(partitionKeys, numPartitions);
+ storeTablePath = new Path(context.getWorkDir(), "output");
+ }
+
+ @Override
+ public void init() throws IOException {
+ super.init();
+ FileSystem fs = new RawLocalFileSystem();
+ fs.mkdirs(storeTablePath);
+ }
+
+ private Appender getAppender(int partition) throws IOException {
+ Appender appender = appenderMap.get(partition);
+
+ if (appender == null) {
+ Path dataFile = getDataFile(partition);
+ FileSystem fs = dataFile.getFileSystem(context.getConf());
+ if (fs.exists(dataFile)) {
+ LOG.info("File " + dataFile + " already exists!");
+ FileStatus status = fs.getFileStatus(dataFile);
+ LOG.info("File size: " + status.getLen());
+ }
+ appender = StorageManager.getAppender(context.getConf(), meta, dataFile);
+ appender.enableStats();
+ appender.init();
+ appenderMap.put(partition, appender);
+ } else {
+ appender = appenderMap.get(partition);
+ }
+
+ return appender;
+ }
+
+ private Path getDataFile(int partition) {
+ return StorageUtil.concatPath(storeTablePath, ""+partition);
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ Tuple tuple;
+ Appender appender;
+ int partition;
+ while ((tuple = child.next()) != null) {
+ partition = partitioner.getPartition(tuple);
+ appender = getAppender(partition);
+ appender.addTuple(tuple);
+ }
+
+ List<TableStat> statSet = new ArrayList<TableStat>();
+ for (Map.Entry<Integer, Appender> entry : appenderMap.entrySet()) {
+ int partNum = entry.getKey();
+ Appender app = entry.getValue();
+ app.flush();
+ app.close();
+ statSet.add(app.getStats());
+ if (app.getStats().getNumRows() > 0) {
+ context.addRepartition(partNum, getDataFile(partNum).getName());
+ }
+ }
+
+ // Collect and aggregated statistics data
+ TableStat aggregated = StatisticsUtil.aggregateTableStat(statSet);
+ context.setResultStats(aggregated);
+
+ return null;
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ // nothing to do
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/Partitioner.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/Partitioner.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/Partitioner.java
new file mode 100644
index 0000000..866738d
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/Partitioner.java
@@ -0,0 +1,43 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ *
+ */
+package org.apache.tajo.engine.planner.physical;
+
+import com.google.common.base.Preconditions;
+import org.apache.tajo.storage.Tuple;
+
+public abstract class Partitioner {
+ protected final int [] partitionKeys;
+ protected final int numPartitions;
+
+ public Partitioner(final int [] keyList, final int numPartitions) {
+ Preconditions.checkArgument(keyList != null,
+ "Partition keys must be given");
+ Preconditions.checkArgument(keyList.length >= 0,
+ "At least one partition key must be specified.");
+ Preconditions.checkArgument(numPartitions > 0,
+ "The number of partitions must be positive: %s", numPartitions);
+ this.partitionKeys = keyList;
+ this.numPartitions = numPartitions;
+ }
+
+ public abstract int getPartition(Tuple tuple);
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExec.java
new file mode 100644
index 0000000..1781304
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExec.java
@@ -0,0 +1,51 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.SchemaObject;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.IOException;
+
+public abstract class PhysicalExec implements SchemaObject {
+ protected final TaskAttemptContext context;
+ protected final Schema inSchema;
+ protected final Schema outSchema;
+
+ public PhysicalExec(final TaskAttemptContext context, final Schema inSchema,
+ final Schema outSchema) {
+ this.context = context;
+ this.inSchema = inSchema;
+ this.outSchema = outSchema;
+ }
+
+ public final Schema getSchema() {
+ return outSchema;
+ }
+
+ public abstract void init() throws IOException;
+
+ public abstract Tuple next() throws IOException;
+
+ public abstract void rescan() throws IOException;
+
+ public abstract void close() throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ProjectionExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ProjectionExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ProjectionExec.java
new file mode 100644
index 0000000..4aa8a1c
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ProjectionExec.java
@@ -0,0 +1,66 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ *
+ */
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.engine.eval.EvalContext;
+import org.apache.tajo.engine.planner.Projector;
+import org.apache.tajo.engine.planner.logical.ProjectionNode;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+
+import java.io.IOException;
+
+public class ProjectionExec extends UnaryPhysicalExec {
+ private final ProjectionNode plan;
+
+ // for projection
+ private Tuple outTuple;
+ private EvalContext[] evalContexts;
+ private Projector projector;
+
+ public ProjectionExec(TaskAttemptContext context, ProjectionNode plan,
+ PhysicalExec child) {
+ super(context, plan.getInSchema(), plan.getOutSchema(), child);
+ this.plan = plan;
+ }
+
+ public void init() throws IOException {
+ super.init();
+
+ this.outTuple = new VTuple(outSchema.getColumnNum());
+ this.projector = new Projector(inSchema, outSchema, this.plan.getTargets());
+ this.evalContexts = projector.renew();
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ Tuple tuple = child.next();
+ if (tuple == null) {
+ return null;
+ }
+
+ projector.eval(evalContexts, tuple);
+ projector.terminate(evalContexts, outTuple);
+ return outTuple;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SelectionExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SelectionExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SelectionExec.java
new file mode 100644
index 0000000..b435747
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SelectionExec.java
@@ -0,0 +1,70 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.engine.eval.EvalContext;
+import org.apache.tajo.engine.eval.EvalNode;
+import org.apache.tajo.engine.planner.logical.SelectionNode;
+import org.apache.tajo.storage.RowStoreUtil;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+
+import java.io.IOException;
+
+public class SelectionExec extends UnaryPhysicalExec {
+ private final EvalNode qual;
+ private final EvalContext qualCtx;
+ private final Tuple outputTuple;
+ // projection
+ private int [] targetIds;
+
+ public SelectionExec(TaskAttemptContext context,
+ SelectionNode plan,
+ PhysicalExec child) {
+ super(context, plan.getInSchema(), plan.getOutSchema(), child);
+
+ this.qual = plan.getQual();
+ this.qualCtx = this.qual.newContext();
+ // for projection
+ if (!inSchema.equals(outSchema)) {
+ targetIds = RowStoreUtil.getTargetIds(inSchema, outSchema);
+ }
+
+ this.outputTuple = new VTuple(outSchema.getColumnNum());
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ Tuple tuple;
+ while ((tuple = child.next()) != null) {
+ qual.eval(qualCtx, inSchema, tuple);
+ if (qual.terminate(qualCtx).asBool()) {
+ if (targetIds != null) {
+ RowStoreUtil.project(tuple, outputTuple, targetIds);
+ return outputTuple;
+ } else {
+ return tuple;
+ }
+ }
+ }
+
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
new file mode 100644
index 0000000..9279c01
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
@@ -0,0 +1,113 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.engine.eval.EvalContext;
+import org.apache.tajo.engine.eval.EvalNode;
+import org.apache.tajo.engine.planner.Projector;
+import org.apache.tajo.engine.planner.logical.ScanNode;
+import org.apache.tajo.storage.*;
+import org.apache.tajo.util.TUtil;
+
+import java.io.IOException;
+
+public class SeqScanExec extends PhysicalExec {
+ private final ScanNode plan;
+ private Scanner scanner = null;
+
+ private EvalNode qual = null;
+ private EvalContext qualCtx;
+
+ private Fragment [] fragments;
+
+ private Projector projector;
+ private EvalContext [] evalContexts;
+
+ public SeqScanExec(TaskAttemptContext context, StorageManager sm,
+ ScanNode plan, Fragment[] fragments) throws IOException {
+ super(context, plan.getInSchema(), plan.getOutSchema());
+
+ this.plan = plan;
+ this.qual = plan.getQual();
+ this.fragments = fragments;
+
+ if (qual == null) {
+ qualCtx = null;
+ } else {
+ qualCtx = this.qual.newContext();
+ }
+ }
+
+ public void init() throws IOException {
+ this.projector = new Projector(inSchema, outSchema, plan.getTargets());
+ this.evalContexts = projector.renew();
+
+ if (fragments.length > 1) {
+ this.scanner = new MergeScanner(context.getConf(), fragments[0].getMeta(),
+ TUtil.newList(fragments));
+ } else {
+ this.scanner = StorageManager.getScanner(context.getConf(), fragments[0].getMeta(),
+ fragments[0], plan.getOutSchema());
+ }
+
+ scanner.init();
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ Tuple tuple;
+ Tuple outTuple = new VTuple(outSchema.getColumnNum());
+
+ if (!plan.hasQual()) {
+ if ((tuple = scanner.next()) != null) {
+ projector.eval(evalContexts, tuple);
+ projector.terminate(evalContexts, outTuple);
+ outTuple.setOffset(tuple.getOffset());
+ return outTuple;
+ } else {
+ return null;
+ }
+ } else {
+ while ((tuple = scanner.next()) != null) {
+ qual.eval(qualCtx, inSchema, tuple);
+ if (qual.terminate(qualCtx).asBool()) {
+ projector.eval(evalContexts, tuple);
+ projector.terminate(evalContexts, outTuple);
+ return outTuple;
+ }
+ }
+ return null;
+ }
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ scanner.reset();
+ }
+
+ @Override
+ public void close() throws IOException {
+ scanner.close();
+ }
+
+ public String getTableName() {
+ return plan.getTableId();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SortAggregateExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SortAggregateExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SortAggregateExec.java
new file mode 100644
index 0000000..34c1d06
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SortAggregateExec.java
@@ -0,0 +1,99 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.engine.planner.logical.GroupbyNode;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+
+import java.io.IOException;
+
+/**
+ * This is the sort-based Aggregation Operator.
+ */
+public class SortAggregateExec extends AggregationExec {
+ private Tuple prevKey = null;
+ private boolean finished = false;
+
+
+ public SortAggregateExec(TaskAttemptContext context, GroupbyNode plan,
+ PhysicalExec child) throws IOException {
+ super(context, plan, child);
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ Tuple curKey;
+ Tuple tuple;
+ Tuple finalTuple = null;
+ while(!context.isStopped() && (tuple = child.next()) != null) {
+ // build a key tuple
+ curKey = new VTuple(keylist.length);
+ for(int i = 0; i < keylist.length; i++) {
+ curKey.put(i, tuple.get(keylist[i]));
+ }
+
+ if (prevKey == null || prevKey.equals(curKey)) {
+ if (prevKey == null) {
+ for(int i = 0; i < outSchema.getColumnNum(); i++) {
+ evalContexts[i] = evals[i].newContext();
+ evals[i].eval(evalContexts[i], inSchema, tuple);
+ }
+ prevKey = curKey;
+ } else {
+ // aggregate
+ for (int idx : measureList) {
+ evals[idx].eval(evalContexts[idx], inSchema, tuple);
+ }
+ }
+ } else {
+ // finalize aggregate and return
+ finalTuple = new VTuple(outSchema.getColumnNum());
+ for(int i = 0; i < outSchema.getColumnNum(); i++) {
+ finalTuple.put(i, evals[i].terminate(evalContexts[i]));
+ }
+
+ for(int i = 0; i < outSchema.getColumnNum(); i++) {
+ evalContexts[i] = evals[i].newContext();
+ evals[i].eval(evalContexts[i], inSchema, tuple);
+ }
+ prevKey = curKey;
+ return finalTuple;
+ }
+ } // while loop
+
+ if (!finished) {
+ finalTuple = new VTuple(outSchema.getColumnNum());
+ for(int i = 0; i < outSchema.getColumnNum(); i++) {
+ finalTuple.put(i, evals[i].terminate(evalContexts[i]));
+ }
+ finished = true;
+ }
+ return finalTuple;
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ super.rescan();
+
+ prevKey = null;
+ finished = false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SortExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SortExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SortExec.java
new file mode 100644
index 0000000..27214cb
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SortExec.java
@@ -0,0 +1,51 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.SortSpec;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.TupleComparator;
+
+import java.io.IOException;
+import java.util.Comparator;
+
+public abstract class SortExec extends UnaryPhysicalExec {
+ private final Comparator<Tuple> comparator;
+ private final SortSpec [] sortSpecs;
+
+ public SortExec(TaskAttemptContext context, Schema inSchema,
+ Schema outSchema, PhysicalExec child, SortSpec [] sortSpecs) {
+ super(context, inSchema, outSchema, child);
+ this.sortSpecs = sortSpecs;
+ this.comparator = new TupleComparator(inSchema, sortSpecs);
+ }
+
+ public SortSpec[] getSortSpecs() {
+ return sortSpecs;
+ }
+
+ public Comparator<Tuple> getComparator() {
+ return comparator;
+ }
+
+ @Override
+ abstract public Tuple next() throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/StoreTableExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/StoreTableExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/StoreTableExec.java
new file mode 100644
index 0000000..7d6a525
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/StoreTableExec.java
@@ -0,0 +1,108 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ *
+ */
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.engine.planner.logical.StoreTableNode;
+import org.apache.tajo.storage.Appender;
+import org.apache.tajo.storage.StorageManager;
+import org.apache.tajo.storage.StorageUtil;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.IOException;
+
+/**
+ * This physical operator stores a relation into a table.
+ */
+public class StoreTableExec extends UnaryPhysicalExec {
+ private final StoreTableNode plan;
+ private Appender appender;
+ private Tuple tuple;
+
+ /**
+ * @throws java.io.IOException
+ *
+ */
+ public StoreTableExec(TaskAttemptContext context, StorageManager sm,
+ StoreTableNode plan, PhysicalExec child) throws IOException {
+ super(context, plan.getInSchema(), plan.getOutSchema(), child);
+
+ this.plan = plan;
+ }
+
+ public void init() throws IOException {
+ super.init();
+
+ TableMeta meta;
+ if (plan.hasOptions()) {
+ meta = CatalogUtil.newTableMeta(outSchema, plan.getStorageType(), plan.getOptions());
+ } else {
+ meta = CatalogUtil.newTableMeta(outSchema, plan.getStorageType());
+ }
+
+ if (context.isInterQuery()) {
+ Path storeTablePath = new Path(context.getWorkDir(), "out");
+ FileSystem fs = new RawLocalFileSystem();
+ fs.mkdirs(storeTablePath);
+ appender = StorageManager.getAppender(context.getConf(), meta,
+ StorageUtil.concatPath(storeTablePath, "0"));
+ } else {
+ appender = StorageManager.getAppender(context.getConf(), meta, context.getOutputPath());
+ }
+ appender.enableStats();
+ appender.init();
+ }
+
+ /* (non-Javadoc)
+ * @see PhysicalExec#next()
+ */
+ @Override
+ public Tuple next() throws IOException {
+ while((tuple = child.next()) != null) {
+ appender.addTuple(tuple);
+ }
+
+ return null;
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ // nothing to do
+ }
+
+ public void close() throws IOException {
+ super.close();
+
+ appender.flush();
+ appender.close();
+
+ // Collect statistics data
+// ctx.addStatSet(annotation.getType().toString(), appender.getStats());
+ context.setResultStats(appender.getStats());
+ context.addRepartition(0, context.getTaskId().toString());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/TunnelExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/TunnelExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/TunnelExec.java
new file mode 100644
index 0000000..b629a49
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/TunnelExec.java
@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.IOException;
+
+public class TunnelExec extends UnaryPhysicalExec {
+
+ public TunnelExec (final TaskAttemptContext context,
+ final Schema outputSchema,
+ final PhysicalExec child) {
+ super(context, outputSchema, outputSchema, child);
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ return child.next();
+ }
+ @Override
+ public void rescan() throws IOException {
+ }
+}