You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by hj...@apache.org on 2014/10/27 06:54:15 UTC
[5/5] git commit: TAJO-1123: Use Fragment instead of FileFragment.
TAJO-1123: Use Fragment instead of FileFragment.
Closes #207
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/cc46aeac
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/cc46aeac
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/cc46aeac
Branch: refs/heads/hbase_storage
Commit: cc46aeacd6f43d9872b6d676dfb21567e127890a
Parents: 78c4e31
Author: HyoungJun Kim <ba...@babokim-mbp.server.gruter.com>
Authored: Mon Oct 27 14:28:46 2014 +0900
Committer: HyoungJun Kim <ba...@babokim-mbp.server.gruter.com>
Committed: Mon Oct 27 14:28:46 2014 +0900
----------------------------------------------------------------------
CHANGES | 2 +
.../org/apache/tajo/jdbc/TajoResultSet.java | 7 +-
.../java/org/apache/tajo/TajoConstants.java | 1 +
.../engine/planner/PhysicalPlannerImpl.java | 58 +-
.../planner/physical/BSTIndexScanExec.java | 7 +-
.../planner/physical/ColPartitionStoreExec.java | 3 +-
.../planner/physical/ExternalSortExec.java | 14 +-
.../physical/HashShuffleFileWriteExec.java | 2 +-
.../physical/PartitionMergeScanExec.java | 7 +-
.../planner/physical/PhysicalPlanUtil.java | 3 +-
.../physical/RangeShuffleFileWriteExec.java | 6 +-
.../engine/planner/physical/SeqScanExec.java | 16 +-
.../engine/planner/physical/StoreTableExec.java | 5 +-
.../tajo/engine/query/QueryUnitRequest.java | 3 +-
.../tajo/engine/query/QueryUnitRequestImpl.java | 4 +-
.../DefaultFragmentScheduleAlgorithm.java | 6 +-
.../tajo/master/DefaultTaskScheduler.java | 5 +-
.../org/apache/tajo/master/FragmentPair.java | 14 +-
.../org/apache/tajo/master/GlobalEngine.java | 70 +-
.../master/GreedyFragmentScheduleAlgorithm.java | 11 +-
.../apache/tajo/master/LazyTaskScheduler.java | 21 +-
.../master/NonForwardQueryResultScanner.java | 4 +-
.../java/org/apache/tajo/master/TajoMaster.java | 7 +-
.../tajo/master/TajoMasterClientService.java | 3 +-
.../master/event/FragmentScheduleEvent.java | 15 +-
.../apache/tajo/master/querymaster/Query.java | 6 +-
.../tajo/master/querymaster/QueryMaster.java | 9 -
.../master/querymaster/QueryMasterTask.java | 4 -
.../tajo/master/querymaster/QueryUnit.java | 14 +-
.../tajo/master/querymaster/Repartitioner.java | 70 +-
.../tajo/master/querymaster/SubQuery.java | 36 +-
.../java/org/apache/tajo/util/IndexUtil.java | 2 +-
.../org/apache/tajo/worker/TajoQueryEngine.java | 4 +-
.../main/java/org/apache/tajo/worker/Task.java | 2 +-
.../org/apache/tajo/BackendTestingUtil.java | 2 +-
.../planner/global/TestBroadcastJoinPlan.java | 2 +-
.../planner/physical/TestBNLJoinExec.java | 18 +-
.../planner/physical/TestBSTIndexExec.java | 16 +-
.../planner/physical/TestExternalSortExec.java | 13 +-
.../physical/TestFullOuterHashJoinExec.java | 36 +-
.../physical/TestFullOuterMergeJoinExec.java | 53 +-
.../planner/physical/TestHashAntiJoinExec.java | 12 +-
.../planner/physical/TestHashJoinExec.java | 18 +-
.../planner/physical/TestHashSemiJoinExec.java | 12 +-
.../physical/TestLeftOuterHashJoinExec.java | 40 +-
.../physical/TestLeftOuterNLJoinExec.java | 43 +-
.../planner/physical/TestMergeJoinExec.java | 14 +-
.../engine/planner/physical/TestNLJoinExec.java | 18 +-
.../planner/physical/TestPhysicalPlanner.java | 123 ++--
.../physical/TestProgressExternalSortExec.java | 11 +-
.../physical/TestRightOuterHashJoinExec.java | 26 +-
.../physical/TestRightOuterMergeJoinExec.java | 48 +-
.../engine/planner/physical/TestSortExec.java | 10 +-
.../tajo/engine/query/TestJoinBroadcast.java | 2 +-
.../apache/tajo/engine/query/TestJoinQuery.java | 71 +-
.../org/apache/tajo/jdbc/TestResultSet.java | 6 +-
.../tajo/master/TestExecutionBlockCursor.java | 2 +-
.../apache/tajo/storage/TestFileFragment.java | 4 +-
.../org/apache/tajo/storage/TestRowFile.java | 7 +-
.../tajo/worker/TestRangeRetrieverHandler.java | 14 +-
.../java/org/apache/tajo/storage/CSVFile.java | 8 +-
.../org/apache/tajo/storage/FileScanner.java | 7 +-
.../apache/tajo/storage/FileStorageManager.java | 702 +++++++++++++++++++
.../storage/HashShuffleAppenderManager.java | 4 +-
.../org/apache/tajo/storage/MergeScanner.java | 23 +-
.../org/apache/tajo/storage/NullScanner.java | 4 +-
.../java/org/apache/tajo/storage/RawFile.java | 24 +-
.../java/org/apache/tajo/storage/RowFile.java | 8 +-
.../org/apache/tajo/storage/StorageManager.java | 634 ++---------------
.../org/apache/tajo/storage/StorageUtil.java | 27 +-
.../apache/tajo/storage/avro/AvroScanner.java | 3 +-
.../tajo/storage/fragment/FileFragment.java | 17 +-
.../apache/tajo/storage/fragment/Fragment.java | 8 +
.../tajo/storage/parquet/ParquetScanner.java | 4 +-
.../org/apache/tajo/storage/rcfile/RCFile.java | 9 +-
.../sequencefile/SequenceFileScanner.java | 6 +-
.../tajo/storage/trevni/TrevniScanner.java | 6 +-
.../tajo/storage/TestCompressionStorages.java | 10 +-
.../apache/tajo/storage/TestFileSystems.java | 11 +-
.../apache/tajo/storage/TestMergeScanner.java | 11 +-
.../apache/tajo/storage/TestStorageManager.java | 23 +-
.../org/apache/tajo/storage/TestStorages.java | 58 +-
.../apache/tajo/storage/index/TestBSTIndex.java | 20 +-
.../index/TestSingleCSVFileBSTIndex.java | 4 +-
84 files changed, 1440 insertions(+), 1253 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index a9a3127..7d51e2c 100644
--- a/CHANGES
+++ b/CHANGES
@@ -11,6 +11,8 @@ Release 0.9.1 - unreleased
IMPROVEMENT
+ TAJO-1123: Use Fragment instead of FileFragment.(Hyoungjun Kim)
+
TAJO-1125: Separate logical plan and optimizer into a maven module.
(hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-client/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java b/tajo-client/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java
index 65954f1..aa6d155 100644
--- a/tajo-client/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java
+++ b/tajo-client/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java
@@ -34,6 +34,7 @@ import org.apache.tajo.storage.MergeScanner;
import org.apache.tajo.storage.Scanner;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.storage.fragment.Fragment;
import java.io.IOException;
import java.sql.SQLException;
@@ -91,7 +92,7 @@ public class TajoResultSet extends TajoResultSetBase {
totalRow = INFINITE_ROW_NUM;
}
- List<FileFragment> frags = getFragments(desc.getPath());
+ List<Fragment> frags = getFragments(desc.getPath());
scanner = new MergeScanner(conf, desc.getSchema(), desc.getMeta(), frags);
}
}
@@ -110,9 +111,9 @@ public class TajoResultSet extends TajoResultSetBase {
}
}
- private List<FileFragment> getFragments(Path tablePath)
+ private List<Fragment> getFragments(Path tablePath)
throws IOException {
- List<FileFragment> fragments = Lists.newArrayList();
+ List<Fragment> fragments = Lists.newArrayList();
FileStatus[] files = fs.listStatus(tablePath, new PathFilter() {
@Override
public boolean accept(Path path) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java b/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java
index be61783..08909b4 100644
--- a/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java
+++ b/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java
@@ -38,6 +38,7 @@ public class TajoConstants {
public static final String SYSTEM_HA_ACTIVE_DIR_NAME = "active";
public static final String SYSTEM_HA_BACKUP_DIR_NAME = "backup";
+ public static final int UNKNOWN_ROW_NUMBER = -1;
private TajoConstants() {}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
index f7f071f..ecc3616 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
@@ -47,10 +47,12 @@ import org.apache.tajo.ipc.TajoWorkerProtocol.DistinctGroupbyEnforcer.SortSpecAr
import org.apache.tajo.plan.LogicalPlan;
import org.apache.tajo.plan.util.PlannerUtil;
import org.apache.tajo.plan.logical.*;
+import org.apache.tajo.storage.FileStorageManager;
import org.apache.tajo.storage.StorageConstants;
import org.apache.tajo.storage.StorageManager;
import org.apache.tajo.storage.TupleComparator;
import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.storage.fragment.Fragment;
import org.apache.tajo.storage.fragment.FragmentConvertor;
import org.apache.tajo.util.FileUtil;
import org.apache.tajo.util.IndexUtil;
@@ -76,11 +78,9 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
private static final int UNGENERATED_PID = -1;
protected final TajoConf conf;
- protected final StorageManager sm;
- public PhysicalPlannerImpl(final TajoConf conf, final StorageManager sm) {
+ public PhysicalPlannerImpl(final TajoConf conf) {
this.conf = conf;
- this.sm = sm;
}
public PhysicalExec createPlan(final TaskAttemptContext context, final LogicalNode logicalPlan)
@@ -253,7 +253,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
List<FileFragment> fragments = FragmentConvertor.convert(ctx.getConf(), CatalogProtos.StoreType.CSV,
ctx.getTables(tableId));
for (FileFragment frag : fragments) {
- size += frag.getEndKey();
+ size += frag.getLength();
}
}
return size;
@@ -445,13 +445,13 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
leftSortNode.setSortSpecs(sortSpecs[0]);
leftSortNode.setInSchema(leftExec.getSchema());
leftSortNode.setOutSchema(leftExec.getSchema());
- ExternalSortExec outerSort = new ExternalSortExec(context, sm, leftSortNode, leftExec);
+ ExternalSortExec outerSort = new ExternalSortExec(context, leftSortNode, leftExec);
SortNode rightSortNode = LogicalPlan.createNodeWithoutPID(SortNode.class);
rightSortNode.setSortSpecs(sortSpecs[1]);
rightSortNode.setInSchema(rightExec.getSchema());
rightSortNode.setOutSchema(rightExec.getSchema());
- ExternalSortExec innerSort = new ExternalSortExec(context, sm, rightSortNode, rightExec);
+ ExternalSortExec innerSort = new ExternalSortExec(context, rightSortNode, rightExec);
LOG.info("Join (" + plan.getPID() +") chooses [Merge Join]");
return new MergeJoinExec(context, plan, outerSort, innerSort, sortSpecs[0], sortSpecs[1]);
@@ -542,13 +542,13 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
leftSortNode2.setSortSpecs(sortSpecs2[0]);
leftSortNode2.setInSchema(leftExec.getSchema());
leftSortNode2.setOutSchema(leftExec.getSchema());
- ExternalSortExec outerSort2 = new ExternalSortExec(context, sm, leftSortNode2, leftExec);
+ ExternalSortExec outerSort2 = new ExternalSortExec(context, leftSortNode2, leftExec);
SortNode rightSortNode2 = LogicalPlan.createNodeWithoutPID(SortNode.class);
rightSortNode2.setSortSpecs(sortSpecs2[1]);
rightSortNode2.setInSchema(rightExec.getSchema());
rightSortNode2.setOutSchema(rightExec.getSchema());
- ExternalSortExec innerSort2 = new ExternalSortExec(context, sm, rightSortNode2, rightExec);
+ ExternalSortExec innerSort2 = new ExternalSortExec(context, rightSortNode2, rightExec);
return new RightOuterMergeJoinExec(context, plan, outerSort2, innerSort2, sortSpecs2[0], sortSpecs2[1]);
}
@@ -633,13 +633,13 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
leftSortNode.setSortSpecs(sortSpecs3[0]);
leftSortNode.setInSchema(leftExec.getSchema());
leftSortNode.setOutSchema(leftExec.getSchema());
- ExternalSortExec outerSort3 = new ExternalSortExec(context, sm, leftSortNode, leftExec);
+ ExternalSortExec outerSort3 = new ExternalSortExec(context, leftSortNode, leftExec);
SortNode rightSortNode = LogicalPlan.createNodeWithoutPID(SortNode.class);
rightSortNode.setSortSpecs(sortSpecs3[1]);
rightSortNode.setInSchema(rightExec.getSchema());
rightSortNode.setOutSchema(rightExec.getSchema());
- ExternalSortExec innerSort3 = new ExternalSortExec(context, sm, rightSortNode, rightExec);
+ ExternalSortExec innerSort3 = new ExternalSortExec(context, rightSortNode, rightExec);
return new MergeFullOuterJoinExec(context, plan, outerSort3, innerSort3, sortSpecs3[0], sortSpecs3[1]);
}
@@ -767,7 +767,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
switch (plan.getShuffleType()) {
case HASH_SHUFFLE:
case SCATTERED_HASH_SHUFFLE:
- return new HashShuffleFileWriteExec(ctx, sm, plan, subOp);
+ return new HashShuffleFileWriteExec(ctx, plan, subOp);
case RANGE_SHUFFLE:
SortExec sortExec = PhysicalPlanUtil.findExecutor(subOp, SortExec.class);
@@ -782,7 +782,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
specs[i] = new SortSpec(columns[i]);
}
}
- return new RangeShuffleFileWriteExec(ctx, sm, subOp, plan.getInSchema(), plan.getInSchema(), sortSpecs);
+ return new RangeShuffleFileWriteExec(ctx, subOp, plan.getInSchema(), plan.getInSchema(), sortSpecs);
case NONE_SHUFFLE:
// if there is no given NULL CHAR property in the table property and the query is neither CTAS or INSERT,
@@ -868,7 +868,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
sortNode.setInSchema(child.getSchema());
sortNode.setOutSchema(child.getSchema());
- ExternalSortExec sortExec = new ExternalSortExec(context, sm, sortNode, child);
+ ExternalSortExec sortExec = new ExternalSortExec(context, sortNode, child);
LOG.info("The planner chooses [Sort-based Column Partitioned Store] algorithm");
return new SortBasedColPartitionStoreExec(context, storeTableNode, sortExec);
}
@@ -895,10 +895,10 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
// Since the default intermediate file format is raw file, it is not problem right now.
if (checkIfSortEquivalance(ctx, scanNode, node)) {
if (ctx.getTable(scanNode.getCanonicalName()) == null) {
- return new SeqScanExec(ctx, sm, scanNode, null);
+ return new SeqScanExec(ctx, scanNode, null);
}
FragmentProto [] fragments = ctx.getTables(scanNode.getCanonicalName());
- return new ExternalSortExec(ctx, sm, (SortNode) node.peek(), fragments);
+ return new ExternalSortExec(ctx, (SortNode) node.peek(), fragments);
} else {
Enforcer enforcer = ctx.getEnforcer();
@@ -918,25 +918,26 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
if (scanNode instanceof PartitionedTableScanNode) {
if (broadcastFlag) {
PartitionedTableScanNode partitionedTableScanNode = (PartitionedTableScanNode) scanNode;
- List<FileFragment> fileFragments = TUtil.newList();
+ List<Fragment> fileFragments = TUtil.newList();
+ FileStorageManager fileStorageManager = StorageManager.getFileStorageManager(ctx.getConf());
for (Path path : partitionedTableScanNode.getInputPaths()) {
- fileFragments.addAll(TUtil.newList(sm.split(scanNode.getCanonicalName(), path)));
+ fileFragments.addAll(TUtil.newList(fileStorageManager.split(scanNode.getCanonicalName(), path)));
}
FragmentProto[] fragments =
FragmentConvertor.toFragmentProtoArray(fileFragments.toArray(new FileFragment[fileFragments.size()]));
ctx.addFragments(scanNode.getCanonicalName(), fragments);
- return new PartitionMergeScanExec(ctx, sm, scanNode, fragments);
+ return new PartitionMergeScanExec(ctx, scanNode, fragments);
}
}
}
if (ctx.getTable(scanNode.getCanonicalName()) == null) {
- return new SeqScanExec(ctx, sm, scanNode, null);
+ return new SeqScanExec(ctx, scanNode, null);
}
FragmentProto [] fragments = ctx.getTables(scanNode.getCanonicalName());
- return new SeqScanExec(ctx, sm, scanNode, fragments);
+ return new SeqScanExec(ctx, scanNode, fragments);
}
}
@@ -996,7 +997,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
sortNode.setSortSpecs(sortSpecs);
sortNode.setInSchema(subOp.getSchema());
sortNode.setOutSchema(subOp.getSchema());
- ExternalSortExec sortExec = new ExternalSortExec(ctx, sm, sortNode, subOp);
+ ExternalSortExec sortExec = new ExternalSortExec(ctx, sortNode, subOp);
LOG.info("The planner chooses [Sort Aggregation] in (" + TUtil.arrayToString(sortSpecs) + ")");
return new SortAggregateExec(ctx, groupbyNode, sortExec);
}
@@ -1037,7 +1038,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
sortNode.setSortSpecs(sortSpecs);
sortNode.setInSchema(subOp.getSchema());
sortNode.setOutSchema(subOp.getSchema());
- child = new ExternalSortExec(context, sm, sortNode, subOp);
+ child = new ExternalSortExec(context, sortNode, subOp);
LOG.info("The planner chooses [Sort Aggregation] in (" + TUtil.arrayToString(sortSpecs) + ")");
}
@@ -1100,7 +1101,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
sortNode.setSortSpecs(sortSpecs.toArray(new SortSpec[]{}));
sortNode.setInSchema(distinctNode.getInSchema());
sortNode.setOutSchema(distinctNode.getInSchema());
- ExternalSortExec sortExec = new ExternalSortExec(context, sm, sortNode, subOp);
+ ExternalSortExec sortExec = new ExternalSortExec(context, sortNode, subOp);
return sortExec;
}
@@ -1131,7 +1132,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
sortNode.setSortSpecs(sortSpecs);
sortNode.setInSchema(subOp.getSchema());
sortNode.setOutSchema(eachGroupbyNode.getInSchema());
- ExternalSortExec sortExec = new ExternalSortExec(ctx, sm, sortNode, subOp);
+ ExternalSortExec sortExec = new ExternalSortExec(ctx, sortNode, subOp);
sortAggregateExec[index++] = new SortAggregateExec(ctx, eachGroupbyNode, sortExec);
}
@@ -1159,7 +1160,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
if (algorithm == SortEnforce.SortAlgorithm.IN_MEMORY_SORT) {
return new MemSortExec(context, sortNode, child);
} else {
- return new ExternalSortExec(context, sm, sortNode, child);
+ return new ExternalSortExec(context, sortNode, child);
}
}
@@ -1168,7 +1169,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
public SortExec createBestSortPlan(TaskAttemptContext context, SortNode sortNode,
PhysicalExec child) throws IOException {
- return new ExternalSortExec(context, sm, sortNode, child);
+ return new ExternalSortExec(context, sortNode, child);
}
public PhysicalExec createIndexScanExec(TaskAttemptContext ctx,
@@ -1183,11 +1184,12 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
FragmentConvertor.convert(ctx.getConf(), ctx.getDataChannel().getStoreType(), fragmentProtos);
String indexName = IndexUtil.getIndexNameOfFrag(fragments.get(0), annotation.getSortKeys());
- Path indexPath = new Path(sm.getTablePath(annotation.getTableName()), "index");
+ Path indexPath = new Path(
+ StorageManager.getFileStorageManager(ctx.getConf()).getTablePath(annotation.getTableName()), "index");
TupleComparator comp = new TupleComparator(annotation.getKeySchema(),
annotation.getSortKeys());
- return new BSTIndexScanExec(ctx, sm, annotation, fragments.get(0), new Path(indexPath, indexName),
+ return new BSTIndexScanExec(ctx, annotation, fragments.get(0), new Path(indexPath, indexName),
annotation.getKeySchema(), comp, annotation.getDatum());
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java
index f699607..6adc523 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java
@@ -18,6 +18,7 @@
package org.apache.tajo.engine.planner.physical;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.tajo.catalog.Schema;
@@ -47,8 +48,7 @@ public class BSTIndexScanExec extends PhysicalExec {
private float progress;
- public BSTIndexScanExec(TaskAttemptContext context,
- StorageManager sm , ScanNode scanNode ,
+ public BSTIndexScanExec(TaskAttemptContext context, ScanNode scanNode ,
FileFragment fragment, Path fileName , Schema keySchema,
TupleComparator comparator , Datum[] datum) throws IOException {
super(context, scanNode.getInSchema(), scanNode.getOutSchema());
@@ -61,7 +61,8 @@ public class BSTIndexScanExec extends PhysicalExec {
this.fileScanner.init();
this.projector = new Projector(context, inSchema, outSchema, scanNode.getTargets());
- this.reader = new BSTIndex(sm.getFileSystem().getConf()).
+ FileSystem fs = fileName.getFileSystem(context.getConf());
+ this.reader = new BSTIndex(fs.getConf()).
getIndexReader(fileName, keySchema, comparator);
this.reader.open();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java
index ec239de..7818fd7 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java
@@ -160,7 +160,8 @@ public abstract class ColPartitionStoreExec extends UnaryPhysicalExec {
actualFilePath = new Path(lastFileName + "_" + suffixId);
}
- appender = StorageManager.getStorageManager(context.getConf()).getAppender(meta, outSchema, actualFilePath);
+ appender = StorageManager.getFileStorageManager(context.getConf())
+ .getAppender(meta, outSchema, actualFilePath);
appender.enableStats();
appender.init();
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
index 4c0caea..105c374 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
@@ -106,7 +106,7 @@ public class ExternalSortExec extends SortExec {
/** total bytes of input data */
private long sortAndStoredBytes;
- private ExternalSortExec(final TaskAttemptContext context, final StorageManager sm, final SortNode plan)
+ private ExternalSortExec(final TaskAttemptContext context, final SortNode plan)
throws PhysicalPlanningException {
super(context, plan.getInSchema(), plan.getOutSchema(), null, plan.getSortKeys());
@@ -128,10 +128,9 @@ public class ExternalSortExec extends SortExec {
localFS = new RawLocalFileSystem();
}
- public ExternalSortExec(final TaskAttemptContext context,
- final StorageManager sm, final SortNode plan,
+ public ExternalSortExec(final TaskAttemptContext context,final SortNode plan,
final CatalogProtos.FragmentProto[] fragments) throws PhysicalPlanningException {
- this(context, sm, plan);
+ this(context, plan);
mergedInputFragments = TUtil.newList();
for (CatalogProtos.FragmentProto proto : fragments) {
@@ -140,10 +139,9 @@ public class ExternalSortExec extends SortExec {
}
}
- public ExternalSortExec(final TaskAttemptContext context,
- final StorageManager sm, final SortNode plan, final PhysicalExec child)
+ public ExternalSortExec(final TaskAttemptContext context, final SortNode plan, final PhysicalExec child)
throws IOException {
- this(context, sm, plan);
+ this(context, plan);
setChild(child);
}
@@ -777,7 +775,7 @@ public class ExternalSortExec extends SortExec {
if (finalOutputFiles != null) {
for (FileFragment frag : finalOutputFiles) {
File tmpFile = new File(localFS.makeQualified(frag.getPath()).toUri());
- if (frag.getStartKey() == 0 && frag.getEndKey() == tmpFile.length()) {
+ if (frag.getStartKey() == 0 && frag.getLength() == tmpFile.length()) {
localFS.delete(frag.getPath(), true);
LOG.info("Delete file: " + frag);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java
index 5bf80fd..d051fb6 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java
@@ -55,7 +55,7 @@ public final class HashShuffleFileWriteExec extends UnaryPhysicalExec {
private HashShuffleAppenderManager hashShuffleAppenderManager;
private int numHashShuffleBufferTuples;
- public HashShuffleFileWriteExec(TaskAttemptContext context, final StorageManager sm,
+ public HashShuffleFileWriteExec(TaskAttemptContext context,
final ShuffleFileWriteNode plan, final PhysicalExec child) throws IOException {
super(context, plan.getInSchema(), plan.getOutSchema(), child);
Preconditions.checkArgument(plan.hasShuffleKeys());
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PartitionMergeScanExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PartitionMergeScanExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PartitionMergeScanExec.java
index 0569c1b..5297e2c 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PartitionMergeScanExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PartitionMergeScanExec.java
@@ -44,25 +44,22 @@ public class PartitionMergeScanExec extends PhysicalExec {
private List<SeqScanExec> scanners = Lists.newArrayList();
private Iterator<SeqScanExec> iterator;
- private StorageManager sm;
-
private float progress;
protected TableStats inputStats;
- public PartitionMergeScanExec(TaskAttemptContext context, StorageManager sm,
+ public PartitionMergeScanExec(TaskAttemptContext context,
ScanNode plan, CatalogProtos.FragmentProto[] fragments) throws IOException {
super(context, plan.getInSchema(), plan.getOutSchema());
this.plan = plan;
this.fragments = fragments;
- this.sm = sm;
inputStats = new TableStats();
}
public void init() throws IOException {
for (CatalogProtos.FragmentProto fragment : fragments) {
- SeqScanExec scanExec = new SeqScanExec(context, sm, (ScanNode) PlannerUtil.clone(null, plan),
+ SeqScanExec scanExec = new SeqScanExec(context, (ScanNode) PlannerUtil.clone(null, plan),
new CatalogProtos.FragmentProto[] {fragment});
scanners.add(scanExec);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java
index 1b824b5..f2df2c8 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java
@@ -34,6 +34,7 @@ import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.plan.logical.NodeType;
import org.apache.tajo.plan.logical.PersistentStoreNode;
import org.apache.tajo.engine.query.QueryContext;
+import org.apache.tajo.storage.FileStorageManager;
import org.apache.tajo.storage.StorageConstants;
import org.apache.tajo.storage.StorageManager;
import org.apache.tajo.storage.TupleComparator;
@@ -122,7 +123,7 @@ public class PhysicalPlanUtil {
int startFileIndex, int numResultFiles,
AtomicInteger currentFileIndex) throws IOException {
if (fs.isDirectory(path)) {
- FileStatus[] files = fs.listStatus(path, StorageManager.hiddenFileFilter);
+ FileStatus[] files = fs.listStatus(path, FileStorageManager.hiddenFileFilter);
if (files != null && files.length > 0) {
for (FileStatus eachFile : files) {
if (result.size() >= numResultFiles) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java
index dd72910..ccdfc4d 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java
@@ -51,7 +51,7 @@ public class RangeShuffleFileWriteExec extends UnaryPhysicalExec {
private FileAppender appender;
private TableMeta meta;
- public RangeShuffleFileWriteExec(final TaskAttemptContext context, final StorageManager sm,
+ public RangeShuffleFileWriteExec(final TaskAttemptContext context,
final PhysicalExec child, final Schema inSchema, final Schema outSchema,
final SortSpec[] sortSpecs) throws IOException {
super(context, inSchema, outSchema, child);
@@ -70,7 +70,7 @@ public class RangeShuffleFileWriteExec extends UnaryPhysicalExec {
indexKeys[i] = inSchema.getColumnId(col.getQualifiedName());
}
- BSTIndex bst = new BSTIndex(new TajoConf());
+ BSTIndex bst = new BSTIndex(new TajoConf(context.getConf()));
this.comp = new TupleComparator(keySchema, sortSpecs);
Path storeTablePath = new Path(context.getWorkDir(), "output");
LOG.info("Output data directory: " + storeTablePath);
@@ -78,7 +78,7 @@ public class RangeShuffleFileWriteExec extends UnaryPhysicalExec {
context.getDataChannel().getStoreType() : CatalogProtos.StoreType.RAW);
FileSystem fs = new RawLocalFileSystem();
fs.mkdirs(storeTablePath);
- this.appender = (FileAppender) StorageManager.getStorageManager(context.getConf()).getAppender(meta,
+ this.appender = (FileAppender) StorageManager.getFileStorageManager(context.getConf()).getAppender(meta,
outSchema, new Path(storeTablePath, "output"));
this.appender.enableStats();
this.appender.init();
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
index c6a4f55..c909c21 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
@@ -40,6 +40,7 @@ import org.apache.tajo.plan.logical.ScanNode;
import org.apache.tajo.plan.rewrite.rules.PartitionedTableRewriter;
import org.apache.tajo.storage.*;
import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.storage.fragment.Fragment;
import org.apache.tajo.storage.fragment.FragmentConvertor;
import org.apache.tajo.worker.TaskAttemptContext;
@@ -67,7 +68,7 @@ public class SeqScanExec extends PhysicalExec {
private boolean cacheRead = false;
- public SeqScanExec(TaskAttemptContext context, StorageManager sm, ScanNode plan,
+ public SeqScanExec(TaskAttemptContext context, ScanNode plan,
CatalogProtos.FragmentProto [] fragments) throws IOException {
super(context, plan.getInSchema(), plan.getOutSchema());
@@ -79,9 +80,9 @@ public class SeqScanExec extends PhysicalExec {
String pathNameKey = "";
if (fragments != null) {
for (FragmentProto f : fragments) {
- FileFragment fileFragement = FragmentConvertor.convert(
+ Fragment fragement = FragmentConvertor.convert(
context.getConf(), plan.getTableDesc().getMeta().getStoreType(), f);
- pathNameKey += fileFragement.getPath();
+ pathNameKey += fragement.getKey();
}
}
@@ -215,13 +216,14 @@ public class SeqScanExec extends PhysicalExec {
if (fragments != null) {
if (fragments.length > 1) {
this.scanner = new MergeScanner(context.getConf(), plan.getPhysicalSchema(), plan.getTableDesc().getMeta(),
- FragmentConvertor.<FileFragment>convert(context.getConf(), plan.getTableDesc().getMeta().getStoreType(),
+ FragmentConvertor.convert(context.getConf(), plan.getTableDesc().getMeta().getStoreType(),
fragments), projected
);
} else {
- this.scanner = StorageManager.getStorageManager(
- context.getConf()).getScanner(plan.getTableDesc().getMeta(), plan.getPhysicalSchema(), fragments[0],
- projected);
+ StorageManager storageManager = StorageManager.getStorageManager(
+ context.getConf(), plan.getTableDesc().getMeta().getStoreType());
+ this.scanner = storageManager.getScanner(plan.getTableDesc().getMeta(),
+ plan.getPhysicalSchema(), fragments[0], projected);
}
scanner.init();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/StoreTableExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/StoreTableExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/StoreTableExec.java
index 0dc172c..725478f 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/StoreTableExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/StoreTableExec.java
@@ -89,10 +89,11 @@ public class StoreTableExec extends UnaryPhysicalExec {
if (plan instanceof InsertNode) {
InsertNode createTableNode = (InsertNode) plan;
- appender = StorageManager.getStorageManager(context.getConf()).getAppender(meta,
+ appender = StorageManager.getStorageManager(context.getConf(), meta.getStoreType()).getAppender(meta,
createTableNode.getTableSchema(), context.getOutputPath());
} else {
- appender = StorageManager.getStorageManager(context.getConf()).getAppender(meta, outSchema, lastFileName);
+ appender = StorageManager.getStorageManager(context.getConf(), meta.getStoreType()).getAppender(meta,
+ outSchema, lastFileName);
}
appender.enableStats();
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryUnitRequest.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryUnitRequest.java b/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryUnitRequest.java
index dc9a63d..731f64d 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryUnitRequest.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryUnitRequest.java
@@ -24,6 +24,7 @@ package org.apache.tajo.engine.query;
import org.apache.tajo.QueryUnitAttemptId;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.common.ProtoObject;
+import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.engine.planner.enforce.Enforcer;
import org.apache.tajo.engine.planner.global.DataChannel;
import org.apache.tajo.ipc.TajoWorkerProtocol;
@@ -44,7 +45,7 @@ public interface QueryUnitRequest extends ProtoObject<TajoWorkerProtocol.QueryUn
public List<FetchImpl> getFetches();
public boolean shouldDie();
public void setShouldDie();
- public QueryContext getQueryContext();
+ public QueryContext getQueryContext(TajoConf baseConf);
public DataChannel getDataChannel();
public Enforcer getEnforcer();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java b/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java
index ef82427..c874b70 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java
@@ -186,7 +186,7 @@ public class QueryUnitRequestImpl implements QueryUnitRequest {
fetches.add(fetch);
}
- public QueryContext getQueryContext() {
+ public QueryContext getQueryContext(TajoConf baseConf) {
QueryUnitRequestProtoOrBuilder p = viaProto ? proto : builder;
if (queryContext != null) {
return queryContext;
@@ -194,7 +194,7 @@ public class QueryUnitRequestImpl implements QueryUnitRequest {
if (!p.hasQueryContext()) {
return null;
}
- this.queryContext = new QueryContext(new TajoConf(), p.getQueryContext());
+ this.queryContext = new QueryContext(new TajoConf(baseConf), p.getQueryContext());
return this.queryContext;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/DefaultFragmentScheduleAlgorithm.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/DefaultFragmentScheduleAlgorithm.java b/tajo-core/src/main/java/org/apache/tajo/master/DefaultFragmentScheduleAlgorithm.java
index e4b98d4..6a2a705 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/DefaultFragmentScheduleAlgorithm.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/DefaultFragmentScheduleAlgorithm.java
@@ -21,6 +21,7 @@ package org.apache.tajo.master;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.yarn.util.RackResolver;
+import org.apache.tajo.storage.fragment.FileFragment;
import org.apache.tajo.util.NetUtils;
import java.util.*;
@@ -81,7 +82,10 @@ public class DefaultFragmentScheduleAlgorithm implements FragmentScheduleAlgorit
@Override
public void addFragment(FragmentPair fragmentPair) {
String[] hosts = fragmentPair.getLeftFragment().getHosts();
- int[] diskIds = fragmentPair.getLeftFragment().getDiskIds();
+ int[] diskIds = null;
+ if (fragmentPair.getLeftFragment() instanceof FileFragment) {
+ diskIds = ((FileFragment)fragmentPair.getLeftFragment()).getDiskIds();
+ }
for (int i = 0; i < hosts.length; i++) {
addFragment(hosts[i], diskIds[i], fragmentPair);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java b/tajo-core/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java
index 62d4892..b20c5d8 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java
@@ -43,6 +43,7 @@ import org.apache.tajo.master.querymaster.QueryUnitAttempt;
import org.apache.tajo.master.querymaster.SubQuery;
import org.apache.tajo.storage.DataLocation;
import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.storage.fragment.Fragment;
import org.apache.tajo.util.NetUtils;
import org.apache.tajo.worker.FetchImpl;
@@ -149,8 +150,8 @@ public class DefaultTaskScheduler extends AbstractTaskScheduler {
super.stop();
}
- private FileFragment[] fragmentsForNonLeafTask;
- private FileFragment[] broadcastFragmentsForNonLeafTask;
+ private Fragment[] fragmentsForNonLeafTask;
+ private Fragment[] broadcastFragmentsForNonLeafTask;
LinkedList<TaskRequestEvent> taskRequestEvents = new LinkedList<TaskRequestEvent>();
public void schedule() {
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/FragmentPair.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/FragmentPair.java b/tajo-core/src/main/java/org/apache/tajo/master/FragmentPair.java
index 598b1c5..827386b 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/FragmentPair.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/FragmentPair.java
@@ -19,7 +19,7 @@
package org.apache.tajo.master;
import com.google.common.base.Objects;
-import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.storage.fragment.Fragment;
/**
* FragmentPair consists of two fragments, a left fragment and a right fragment.
@@ -29,23 +29,23 @@ import org.apache.tajo.storage.fragment.FileFragment;
* For other queries, it is assumed to have only a left fragment.
*/
public class FragmentPair {
- private FileFragment leftFragment;
- private FileFragment rightFragment;
+ private Fragment leftFragment;
+ private Fragment rightFragment;
- public FragmentPair(FileFragment left) {
+ public FragmentPair(Fragment left) {
this.leftFragment = left;
}
- public FragmentPair(FileFragment left, FileFragment right) {
+ public FragmentPair(Fragment left, Fragment right) {
this.leftFragment = left;
this.rightFragment = right;
}
- public FileFragment getLeftFragment() {
+ public Fragment getLeftFragment() {
return leftFragment;
}
- public FileFragment getRightFragment() {
+ public Fragment getRightFragment() {
return rightFragment;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java b/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
index cd3be98..06acd83 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
@@ -40,6 +40,8 @@ import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.exception.*;
import org.apache.tajo.catalog.partition.PartitionMethodDesc;
import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.catalog.proto.CatalogProtos.FragmentProto;
+import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
import org.apache.tajo.catalog.statistics.TableStats;
import org.apache.tajo.client.TajoClient;
import org.apache.tajo.common.TajoDataTypes;
@@ -689,32 +691,18 @@ public class GlobalEngine extends AbstractService {
meta = CatalogUtil.newTableMeta(createTable.getStorageType());
}
- if(createTable.isExternal()){
+ if(StorageUtil.isFileStorageType(createTable.getStorageType()) && createTable.isExternal()){
Preconditions.checkState(createTable.hasPath(), "ERROR: LOCATION must be given.");
- } else {
- String databaseName;
- String tableName;
- if (CatalogUtil.isFQTableName(createTable.getTableName())) {
- databaseName = CatalogUtil.extractQualifier(createTable.getTableName());
- tableName = CatalogUtil.extractSimpleName(createTable.getTableName());
- } else {
- databaseName = queryContext.getCurrentDatabase();
- tableName = createTable.getTableName();
- }
-
- // create a table directory (i.e., ${WAREHOUSE_DIR}/${DATABASE_NAME}/${TABLE_NAME} )
- Path tablePath = StorageUtil.concatPath(sm.getWarehouseDir(), databaseName, tableName);
- createTable.setPath(tablePath);
}
- return createTableOnPath(queryContext, createTable.getTableName(), createTable.getTableSchema(),
- meta, createTable.getPath(), createTable.isExternal(), createTable.getPartitionMethod(), ifNotExists);
+ return createTable(queryContext, createTable.getTableName(), createTable.getStorageType(),
+ createTable.getTableSchema(), meta, createTable.getPath(), createTable.isExternal(),
+ createTable.getPartitionMethod(), ifNotExists);
}
- public TableDesc createTableOnPath(QueryContext queryContext, String tableName, Schema schema, TableMeta meta,
- Path path, boolean isExternal, PartitionMethodDesc partitionDesc,
- boolean ifNotExists)
- throws IOException {
+ public TableDesc createTable(QueryContext queryContext, String tableName, StoreType storeType,
+ Schema schema, TableMeta meta, Path location, boolean isExternal,
+ PartitionMethodDesc partitionDesc, boolean ifNotExists) throws IOException {
String databaseName;
String simpleTableName;
if (CatalogUtil.isFQTableName(tableName)) {
@@ -738,39 +726,15 @@ public class GlobalEngine extends AbstractService {
}
}
- FileSystem fs = path.getFileSystem(context.getConf());
-
- if (isExternal) {
- if(!fs.exists(path)) {
- LOG.error("ERROR: " + path.toUri() + " does not exist");
- throw new IOException("ERROR: " + path.toUri() + " does not exist");
- }
- } else {
- fs.mkdirs(path);
- }
-
- long totalSize = 0;
-
- try {
- totalSize = sm.calculateSize(path);
- } catch (IOException e) {
- LOG.warn("Cannot calculate the size of the relation", e);
- }
-
- TableStats stats = new TableStats();
- stats.setNumBytes(totalSize);
-
- if (isExternal) { // if it is an external table, there is no way to know the exact row number without processing.
- stats.setNumRows(TajoClient.UNKNOWN_ROW_NUMBER);
- }
-
TableDesc desc = new TableDesc(CatalogUtil.buildFQName(databaseName, simpleTableName),
- schema, meta, path, isExternal);
- desc.setStats(stats);
+ schema, meta, location, isExternal);
+
if (partitionDesc != null) {
desc.setPartitionMethod(partitionDesc);
}
+ StorageManager.getStorageManager(queryContext.getConf(), storeType).createTable(desc);
+
if (catalog.createTable(desc)) {
LOG.info("Table " + desc.getName() + " is created (" + desc.getStats().getNumBytes() + ")");
return desc;
@@ -864,15 +828,15 @@ public class GlobalEngine extends AbstractService {
}
}
- Path path = catalog.getTableDesc(qualifiedName).getPath();
+ TableDesc tableDesc = catalog.getTableDesc(qualifiedName);
catalog.dropTable(qualifiedName);
if (purge) {
try {
- FileSystem fs = path.getFileSystem(context.getConf());
- fs.delete(path, true);
+ StorageManager.getStorageManager(queryContext.getConf(), tableDesc.getMeta().getStoreType())
+ .purgeTable(tableDesc);
} catch (IOException e) {
- throw new InternalError(e.getMessage());
+ LOG.warn("Can't purge table: " + e.getMessage(), e);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/GreedyFragmentScheduleAlgorithm.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/GreedyFragmentScheduleAlgorithm.java b/tajo-core/src/main/java/org/apache/tajo/master/GreedyFragmentScheduleAlgorithm.java
index 82fd6fc..3798399 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/GreedyFragmentScheduleAlgorithm.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/GreedyFragmentScheduleAlgorithm.java
@@ -24,6 +24,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.yarn.util.RackResolver;
import org.apache.tajo.master.DefaultFragmentScheduleAlgorithm.FragmentsPerDisk;
+import org.apache.tajo.storage.fragment.FileFragment;
import org.apache.tajo.util.NetUtils;
import org.apache.tajo.util.TUtil;
@@ -101,7 +102,10 @@ public class GreedyFragmentScheduleAlgorithm implements FragmentScheduleAlgorith
@Override
public void addFragment(FragmentPair fragmentPair) {
String[] hosts = fragmentPair.getLeftFragment().getHosts();
- int[] diskIds = fragmentPair.getLeftFragment().getDiskIds();
+ int[] diskIds = null;
+ if (fragmentPair.getLeftFragment() instanceof FileFragment) {
+ diskIds = ((FileFragment)fragmentPair.getLeftFragment()).getDiskIds();
+ }
for (int i = 0; i < hosts.length; i++) {
addFragment(hosts[i], diskIds[i], fragmentPair);
}
@@ -276,7 +280,10 @@ public class GreedyFragmentScheduleAlgorithm implements FragmentScheduleAlgorith
public void removeFragment(FragmentPair fragmentPair) {
String [] hosts = fragmentPair.getLeftFragment().getHosts();
- int[] diskIds = fragmentPair.getLeftFragment().getDiskIds();
+ int[] diskIds = null;
+ if (fragmentPair.getLeftFragment() instanceof FileFragment) {
+ diskIds = ((FileFragment)fragmentPair.getLeftFragment()).getDiskIds();
+ }
for (int i = 0; i < hosts.length; i++) {
String normalizedHost = NetUtils.normalizeHost(hosts[i]);
Map<Integer, FragmentsPerDisk> diskFragmentMap = fragmentHostMapping.get(normalizedHost);
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/LazyTaskScheduler.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/LazyTaskScheduler.java b/tajo-core/src/main/java/org/apache/tajo/master/LazyTaskScheduler.java
index f7953e0..50a118e 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/LazyTaskScheduler.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/LazyTaskScheduler.java
@@ -39,6 +39,7 @@ import org.apache.tajo.master.querymaster.QueryUnit;
import org.apache.tajo.master.querymaster.QueryUnitAttempt;
import org.apache.tajo.master.querymaster.SubQuery;
import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.storage.fragment.Fragment;
import org.apache.tajo.util.NetUtils;
import org.apache.tajo.worker.FetchImpl;
@@ -197,15 +198,17 @@ public class LazyTaskScheduler extends AbstractTaskScheduler {
if (event.getType() == EventType.T_SCHEDULE) {
if (event instanceof FragmentScheduleEvent) {
FragmentScheduleEvent castEvent = (FragmentScheduleEvent) event;
- Collection<FileFragment> rightFragments = castEvent.getRightFragments();
+ Collection<Fragment> rightFragments = castEvent.getRightFragments();
if (rightFragments == null || rightFragments.isEmpty()) {
scheduledFragments.addFragment(new FragmentPair(castEvent.getLeftFragment(), null));
} else {
- for (FileFragment eachFragment: rightFragments) {
+ for (Fragment eachFragment: rightFragments) {
scheduledFragments.addFragment(new FragmentPair(castEvent.getLeftFragment(), eachFragment));
}
}
- initDiskBalancer(castEvent.getLeftFragment().getHosts(), castEvent.getLeftFragment().getDiskIds());
+ if (castEvent.getLeftFragment() instanceof FileFragment) {
+ initDiskBalancer(castEvent.getLeftFragment().getHosts(), ((FileFragment)castEvent.getLeftFragment()).getDiskIds());
+ }
} else if (event instanceof FetchScheduleEvent) {
FetchScheduleEvent castEvent = (FetchScheduleEvent) event;
scheduledFetches.addFetch(castEvent.getFetches());
@@ -375,13 +378,13 @@ public class LazyTaskScheduler extends AbstractTaskScheduler {
break;
}
- if (assignedFragmentSize + fragmentPair.getLeftFragment().getEndKey() > taskSize) {
+ if (assignedFragmentSize + fragmentPair.getLeftFragment().getLength() > taskSize) {
break;
} else {
fragmentPairs.add(fragmentPair);
- assignedFragmentSize += fragmentPair.getLeftFragment().getEndKey();
+ assignedFragmentSize += fragmentPair.getLeftFragment().getLength();
if (fragmentPair.getRightFragment() != null) {
- assignedFragmentSize += fragmentPair.getRightFragment().getEndKey();
+ assignedFragmentSize += fragmentPair.getRightFragment().getLength();
}
}
scheduledFragments.removeFragment(fragmentPair);
@@ -397,13 +400,13 @@ public class LazyTaskScheduler extends AbstractTaskScheduler {
break;
}
- if (assignedFragmentSize + fragmentPair.getLeftFragment().getEndKey() > taskSize) {
+ if (assignedFragmentSize + fragmentPair.getLeftFragment().getLength() > taskSize) {
break;
} else {
fragmentPairs.add(fragmentPair);
- assignedFragmentSize += fragmentPair.getLeftFragment().getEndKey();
+ assignedFragmentSize += fragmentPair.getLeftFragment().getLength();
if (fragmentPair.getRightFragment() != null) {
- assignedFragmentSize += fragmentPair.getRightFragment().getEndKey();
+ assignedFragmentSize += fragmentPair.getRightFragment().getLength();
}
}
scheduledFragments.removeFragment(fragmentPair);
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/NonForwardQueryResultScanner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/NonForwardQueryResultScanner.java b/tajo-core/src/main/java/org/apache/tajo/master/NonForwardQueryResultScanner.java
index 768528d..26cfb2e 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/NonForwardQueryResultScanner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/NonForwardQueryResultScanner.java
@@ -33,6 +33,7 @@ import org.apache.tajo.engine.query.QueryContext;
import org.apache.tajo.storage.RowStoreUtil;
import org.apache.tajo.storage.RowStoreUtil.RowStoreEncoder;
import org.apache.tajo.storage.StorageManager;
+import org.apache.tajo.storage.StorageUtil;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.worker.TaskAttemptContext;
@@ -86,8 +87,7 @@ public class NonForwardQueryResultScanner {
try {
// scanNode must be clone cause SeqScanExec change target in the case of a partitioned table.
- scanExec = new SeqScanExec(taskContext,
- StorageManager.getStorageManager(tajoConf), (ScanNode)scanNode.clone(), fragments);
+ scanExec = new SeqScanExec(taskContext, (ScanNode)scanNode.clone(), fragments);
} catch (CloneNotSupportedException e) {
throw new IOException(e.getMessage(), e);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java b/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
index 8d4a41f..e47eb60 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
@@ -49,6 +49,7 @@ import org.apache.tajo.master.rm.TajoWorkerResourceManager;
import org.apache.tajo.master.rm.WorkerResourceManager;
import org.apache.tajo.master.session.SessionManager;
import org.apache.tajo.rpc.RpcChannelFactory;
+import org.apache.tajo.storage.FileStorageManager;
import org.apache.tajo.storage.StorageManager;
import org.apache.tajo.util.*;
import org.apache.tajo.util.metrics.TajoSystemMetrics;
@@ -104,7 +105,7 @@ public class TajoMaster extends CompositeService {
private CatalogServer catalogServer;
private CatalogService catalog;
- private StorageManager storeManager;
+ private FileStorageManager storeManager;
private GlobalEngine globalEngine;
private AsyncDispatcher dispatcher;
private TajoMasterClientService tajoMasterClientService;
@@ -159,7 +160,7 @@ public class TajoMaster extends CompositeService {
// check the system directory and create if they are not created.
checkAndInitializeSystemDirectories();
- this.storeManager = StorageManager.getStorageManager(systemConf);
+ this.storeManager = StorageManager.getFileStorageManager(systemConf, null);
catalogServer = new CatalogServer(FunctionLoader.load());
addIfService(catalogServer);
@@ -394,7 +395,7 @@ public class TajoMaster extends CompositeService {
return this.catalogServer;
}
- public StorageManager getStorageManager() {
+ public FileStorageManager getStorageManager() {
return this.storeManager;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/TajoMasterClientService.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/TajoMasterClientService.java b/tajo-core/src/main/java/org/apache/tajo/master/TajoMasterClientService.java
index a4688d9..8968c6e 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/TajoMasterClientService.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/TajoMasterClientService.java
@@ -762,7 +762,8 @@ public class TajoMasterClientService extends AbstractService {
TableDesc desc;
try {
- desc = context.getGlobalEngine().createTableOnPath(queryContext, request.getName(), schema,
+ desc = context.getGlobalEngine().createTable(queryContext, request.getName(),
+ meta.getStoreType(), schema,
meta, path, true, partitionDesc, false);
} catch (Exception e) {
return TableResponse.newBuilder()
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/event/FragmentScheduleEvent.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/event/FragmentScheduleEvent.java b/tajo-core/src/main/java/org/apache/tajo/master/event/FragmentScheduleEvent.java
index 8cc17cb..c38f99f 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/event/FragmentScheduleEvent.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/event/FragmentScheduleEvent.java
@@ -20,22 +20,23 @@ package org.apache.tajo.master.event;
import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.storage.fragment.Fragment;
import java.util.Collection;
public class FragmentScheduleEvent extends TaskSchedulerEvent {
- private final FileFragment leftFragment;
- private final Collection<FileFragment> rightFragments;
+ private final Fragment leftFragment;
+ private final Collection<Fragment> rightFragments;
public FragmentScheduleEvent(final EventType eventType, final ExecutionBlockId blockId,
- final FileFragment fragment) {
+ final Fragment fragment) {
this(eventType, blockId, fragment, null);
}
public FragmentScheduleEvent(final EventType eventType,
final ExecutionBlockId blockId,
- final FileFragment leftFragment,
- final Collection<FileFragment> rightFragments) {
+ final Fragment leftFragment,
+ final Collection<Fragment> rightFragments) {
super(eventType, blockId);
this.leftFragment = leftFragment;
this.rightFragments = rightFragments;
@@ -45,11 +46,11 @@ public class FragmentScheduleEvent extends TaskSchedulerEvent {
return this.rightFragments != null && !this.rightFragments.isEmpty();
}
- public FileFragment getLeftFragment() {
+ public Fragment getLeftFragment() {
return leftFragment;
}
- public Collection<FileFragment> getRightFragments() { return rightFragments; }
+ public Collection<Fragment> getRightFragments() { return rightFragments; }
@Override
public String toString() {
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
index 1f445ec..e01b661 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
@@ -70,7 +70,6 @@ public class Query implements EventHandler<QueryEvent> {
private Map<ExecutionBlockId, SubQuery> subqueries;
private final EventHandler eventHandler;
private final MasterPlan plan;
- private final StorageManager sm;
QueryMasterTask.QueryMasterTaskContext context;
private ExecutionBlockCursor cursor;
@@ -212,7 +211,6 @@ public class Query implements EventHandler<QueryEvent> {
subqueries = Maps.newHashMap();
this.eventHandler = eventHandler;
this.plan = plan;
- this.sm = context.getStorageManager();
this.cursor = new ExecutionBlockCursor(plan, true);
StringBuilder sb = new StringBuilder("\n=======================================================");
@@ -358,7 +356,7 @@ public class Query implements EventHandler<QueryEvent> {
query.setStartTime();
SubQuery subQuery = new SubQuery(query.context, query.getPlan(),
- query.getExecutionBlockCursor().nextBlock(), query.sm);
+ query.getExecutionBlockCursor().nextBlock());
subQuery.setPriority(query.priority--);
query.addSubQuery(subQuery);
@@ -864,7 +862,7 @@ public class Query implements EventHandler<QueryEvent> {
private void executeNextBlock(Query query) {
ExecutionBlockCursor cursor = query.getExecutionBlockCursor();
ExecutionBlock nextBlock = cursor.nextBlock();
- SubQuery nextSubQuery = new SubQuery(query.context, query.getPlan(), nextBlock, query.sm);
+ SubQuery nextSubQuery = new SubQuery(query.context, query.getPlan(), nextBlock);
nextSubQuery.setPriority(query.priority--);
query.addSubQuery(nextSubQuery);
nextSubQuery.handle(new SubQueryEvent(nextSubQuery.getId(), SubQueryEventType.SQ_INIT));
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
index 7c3d799..235d827 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
@@ -41,7 +41,6 @@ import org.apache.tajo.rpc.NettyClientBase;
import org.apache.tajo.rpc.NullCallback;
import org.apache.tajo.rpc.RpcConnectionPool;
import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos;
-import org.apache.tajo.storage.StorageManager;
import org.apache.tajo.util.HAServiceUtil;
import org.apache.tajo.util.NetUtils;
import org.apache.tajo.worker.TajoWorker;
@@ -70,8 +69,6 @@ public class QueryMaster extends CompositeService implements EventHandler {
private GlobalPlanner globalPlanner;
- private StorageManager storageManager;
-
private TajoConf systemConf;
private Map<QueryId, QueryMasterTask> queryMasterTasks = Maps.newConcurrentMap();
@@ -115,8 +112,6 @@ public class QueryMaster extends CompositeService implements EventHandler {
this.dispatcher = new TajoAsyncDispatcher("querymaster_" + System.currentTimeMillis());
addIfService(dispatcher);
- this.storageManager = StorageManager.getStorageManager(systemConf);
-
globalPlanner = new GlobalPlanner(systemConf, workerContext);
dispatcher.register(QueryStartEvent.EventType.class, new QueryStartEventHandler());
@@ -372,10 +367,6 @@ public class QueryMaster extends CompositeService implements EventHandler {
return clock;
}
- public StorageManager getStorageManager() {
- return storageManager;
- }
-
public QueryMaster getQueryMaster() {
return QueryMaster.this;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
index 8ba9600..ef30135 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
@@ -556,10 +556,6 @@ public class QueryMasterTask extends CompositeService {
return queryId;
}
- public StorageManager getStorageManager() {
- return queryMasterContext.getStorageManager();
- }
-
public Path getStagingDir() {
return queryContext.getStagingDir();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
index fe2752f..c713660 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
@@ -42,6 +42,7 @@ import org.apache.tajo.master.event.QueryUnitAttemptScheduleEvent.QueryUnitAttem
import org.apache.tajo.plan.logical.*;
import org.apache.tajo.storage.DataLocation;
import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.storage.fragment.Fragment;
import org.apache.tajo.util.Pair;
import org.apache.tajo.util.TajoIdUtils;
import org.apache.tajo.worker.FetchImpl;
@@ -237,15 +238,18 @@ public class QueryUnit implements EventHandler<TaskEvent> {
}
}
- private void addDataLocation(FileFragment fragment) {
+ private void addDataLocation(Fragment fragment) {
String[] hosts = fragment.getHosts();
- int[] diskIds = fragment.getDiskIds();
+ int[] diskIds = null;
+ if (fragment instanceof FileFragment) {
+ diskIds = ((FileFragment)fragment).getDiskIds();
+ }
for (int i = 0; i < hosts.length; i++) {
dataLocations.add(new DataLocation(hosts[i], diskIds[i]));
}
}
- public void addFragment(FileFragment fragment, boolean useDataLocation) {
+ public void addFragment(Fragment fragment, boolean useDataLocation) {
Set<FragmentProto> fragmentProtos;
if (fragMap.containsKey(fragment.getTableName())) {
fragmentProtos = fragMap.get(fragment.getTableName());
@@ -260,8 +264,8 @@ public class QueryUnit implements EventHandler<TaskEvent> {
totalFragmentNum++;
}
- public void addFragments(Collection<FileFragment> fragments) {
- for (FileFragment eachFragment: fragments) {
+ public void addFragments(Collection<Fragment> fragments) {
+ for (Fragment eachFragment: fragments) {
addFragment(eachFragment, false);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
index 447c164..046b310 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
@@ -48,12 +48,14 @@ import org.apache.tajo.master.querymaster.QueryUnit.IntermediateEntry;
import org.apache.tajo.plan.util.PlannerUtil;
import org.apache.tajo.plan.PlanningException;
import org.apache.tajo.plan.logical.*;
-import org.apache.tajo.storage.RowStoreUtil;
+import org.apache.tajo.storage.FileStorageManager;
import org.apache.tajo.storage.StorageManager;
+import org.apache.tajo.storage.RowStoreUtil;
import org.apache.tajo.storage.TupleRange;
import org.apache.tajo.storage.fragment.FileFragment;
-import org.apache.tajo.unit.StorageUnit;
+import org.apache.tajo.storage.fragment.Fragment;
import org.apache.tajo.util.Pair;
+import org.apache.tajo.unit.StorageUnit;
import org.apache.tajo.util.TUtil;
import org.apache.tajo.util.TajoIdUtils;
import org.apache.tajo.worker.FetchImpl;
@@ -83,18 +85,19 @@ public class Repartitioner {
MasterPlan masterPlan = subQuery.getMasterPlan();
ExecutionBlock execBlock = subQuery.getBlock();
QueryMasterTask.QueryMasterTaskContext masterContext = subQuery.getContext();
- StorageManager storageManager = subQuery.getStorageManager();
ScanNode[] scans = execBlock.getScanNodes();
Path tablePath;
- FileFragment[] fragments = new FileFragment[scans.length];
+ Fragment[] fragments = new Fragment[scans.length];
long[] stats = new long[scans.length];
// initialize variables from the child operators
for (int i = 0; i < scans.length; i++) {
TableDesc tableDesc = masterContext.getTableDescMap().get(scans[i].getCanonicalName());
if (tableDesc == null) { // if it is a real table stored on storage
+ FileStorageManager storageManager = StorageManager.getFileStorageManager(subQuery.getContext().getConf());
+
tablePath = storageManager.getTablePath(scans[i].getTableName());
if (execBlock.getUnionScanMap() != null && !execBlock.getUnionScanMap().isEmpty()) {
for (Map.Entry<ExecutionBlockId, ExecutionBlockId> unionScanEntry: execBlock.getUnionScanMap().entrySet()) {
@@ -107,21 +110,23 @@ public class Repartitioner {
}
fragments[i] = new FileFragment(scans[i].getCanonicalName(), tablePath, 0, 0, new String[]{UNKNOWN_HOST});
} else {
- tablePath = tableDesc.getPath();
try {
stats[i] = GlobalPlanner.computeDescendentVolume(scans[i]);
} catch (PlanningException e) {
throw new IOException(e);
}
+ StorageManager storageManager =
+ StorageManager.getStorageManager(subQuery.getContext().getConf(), tableDesc.getMeta().getStoreType());
+
// if table has no data, storageManager will return empty FileFragment.
// So, we need to handle FileFragment by its size.
// If we don't check its size, it can cause IndexOutOfBoundsException.
- List<FileFragment> fileFragments = storageManager.getSplits(scans[i].getCanonicalName(), tableDesc.getMeta(), tableDesc.getSchema(), tablePath);
+ List<Fragment> fileFragments = storageManager.getSplits(scans[i].getCanonicalName(), tableDesc);
if (fileFragments.size() > 0) {
fragments[i] = fileFragments.get(0);
} else {
- fragments[i] = new FileFragment(scans[i].getCanonicalName(), tablePath, 0, 0, new String[]{UNKNOWN_HOST});
+ fragments[i] = new FileFragment(scans[i].getCanonicalName(), tableDesc.getPath(), 0, 0, new String[]{UNKNOWN_HOST});
}
}
}
@@ -268,14 +273,14 @@ public class Repartitioner {
//select intermediate scan and stats
ScanNode[] intermediateScans = new ScanNode[largeScanIndexList.size()];
long[] intermediateScanStats = new long[largeScanIndexList.size()];
- FileFragment[] intermediateFragments = new FileFragment[largeScanIndexList.size()];
+ Fragment[] intermediateFragments = new Fragment[largeScanIndexList.size()];
int index = 0;
for (Integer eachIdx : largeScanIndexList) {
intermediateScans[index] = scans[eachIdx];
intermediateScanStats[index] = stats[eachIdx];
intermediateFragments[index++] = fragments[eachIdx];
}
- FileFragment[] broadcastFragments = new FileFragment[broadcastIndexList.size()];
+ Fragment[] broadcastFragments = new Fragment[broadcastIndexList.size()];
ScanNode[] broadcastScans = new ScanNode[broadcastIndexList.size()];
index = 0;
for (Integer eachIdx : broadcastIndexList) {
@@ -309,9 +314,9 @@ public class Repartitioner {
SubQuery subQuery,
ScanNode[] scans,
long[] stats,
- FileFragment[] fragments,
+ Fragment[] fragments,
ScanNode[] broadcastScans,
- FileFragment[] broadcastFragments) throws IOException {
+ Fragment[] broadcastFragments) throws IOException {
MasterPlan masterPlan = subQuery.getMasterPlan();
ExecutionBlock execBlock = subQuery.getBlock();
// The hash map is modeling as follows:
@@ -394,7 +399,7 @@ public class Repartitioner {
int joinTaskNum = Math.min(maxTaskNum, hashEntries.size());
LOG.info("The determined number of join tasks is " + joinTaskNum);
- List<FileFragment> rightFragments = new ArrayList<FileFragment>();
+ List<Fragment> rightFragments = new ArrayList<Fragment>();
rightFragments.add(fragments[1]);
if (broadcastFragments != null) {
@@ -404,14 +409,17 @@ public class Repartitioner {
Path[] partitionScanPaths = null;
TableDesc tableDesc = masterContext.getTableDescMap().get(eachScan.getCanonicalName());
if (eachScan.getType() == NodeType.PARTITIONS_SCAN) {
+ FileStorageManager storageManager = StorageManager.getFileStorageManager(subQuery.getContext().getConf());
+
PartitionedTableScanNode partitionScan = (PartitionedTableScanNode)eachScan;
partitionScanPaths = partitionScan.getInputPaths();
// set null to inputPaths in getFragmentsFromPartitionedTable()
- getFragmentsFromPartitionedTable(subQuery.getStorageManager(), eachScan, tableDesc);
+ getFragmentsFromPartitionedTable(storageManager, eachScan, tableDesc);
partitionScan.setInputPaths(partitionScanPaths);
} else {
- Collection<FileFragment> scanFragments = subQuery.getStorageManager().getSplits(eachScan.getCanonicalName(),
- tableDesc.getMeta(), tableDesc.getSchema(), tableDesc.getPath());
+ StorageManager storageManager = StorageManager.getStorageManager(subQuery.getContext().getConf(),
+ tableDesc.getMeta().getStoreType());
+ Collection<Fragment> scanFragments = storageManager.getSplits(eachScan.getCanonicalName(), tableDesc);
if (scanFragments != null) {
rightFragments.addAll(scanFragments);
}
@@ -480,10 +488,10 @@ public class Repartitioner {
/**
* It creates a number of fragments for all partitions.
*/
- public static List<FileFragment> getFragmentsFromPartitionedTable(StorageManager sm,
+ public static List<Fragment> getFragmentsFromPartitionedTable(FileStorageManager sm,
ScanNode scan,
TableDesc table) throws IOException {
- List<FileFragment> fragments = Lists.newArrayList();
+ List<Fragment> fragments = Lists.newArrayList();
PartitionedTableScanNode partitionsScan = (PartitionedTableScanNode) scan;
fragments.addAll(sm.getSplits(
scan.getCanonicalName(), table.getMeta(), table.getSchema(), partitionsScan.getInputPaths()));
@@ -492,7 +500,7 @@ public class Repartitioner {
}
private static void scheduleLeafTasksWithBroadcastTable(TaskSchedulerContext schedulerContext, SubQuery subQuery,
- int baseScanId, FileFragment[] fragments) throws IOException {
+ int baseScanId, Fragment[] fragments) throws IOException {
ExecutionBlock execBlock = subQuery.getBlock();
ScanNode[] scans = execBlock.getScanNodes();
@@ -511,23 +519,27 @@ public class Repartitioner {
// . add all partition paths to node's inputPaths variable
// -> SCAN
// . add all fragments to broadcastFragments
- Collection<FileFragment> baseFragments = null;
- List<FileFragment> broadcastFragments = new ArrayList<FileFragment>();
+ Collection<Fragment> baseFragments = null;
+ List<Fragment> broadcastFragments = new ArrayList<Fragment>();
for (int i = 0; i < scans.length; i++) {
ScanNode scan = scans[i];
TableDesc desc = subQuery.getContext().getTableDescMap().get(scan.getCanonicalName());
TableMeta meta = desc.getMeta();
- Collection<FileFragment> scanFragments;
+ Collection<Fragment> scanFragments;
Path[] partitionScanPaths = null;
if (scan.getType() == NodeType.PARTITIONS_SCAN) {
PartitionedTableScanNode partitionScan = (PartitionedTableScanNode)scan;
partitionScanPaths = partitionScan.getInputPaths();
// set null to inputPaths in getFragmentsFromPartitionedTable()
- scanFragments = getFragmentsFromPartitionedTable(subQuery.getStorageManager(), scan, desc);
+ FileStorageManager storageManager =
+ StorageManager.getFileStorageManager(subQuery.getContext().getConf());
+ scanFragments = getFragmentsFromPartitionedTable(storageManager, scan, desc);
} else {
- scanFragments = subQuery.getStorageManager().getSplits(scan.getCanonicalName(), meta, desc.getSchema(),
- desc.getPath());
+ StorageManager storageManager =
+ StorageManager.getStorageManager(subQuery.getContext().getConf(), desc.getMeta().getStoreType());
+
+ scanFragments = storageManager.getSplits(scan.getCanonicalName(), desc);
}
if (scanFragments != null) {
@@ -630,7 +642,7 @@ public class Repartitioner {
ExecutionBlock execBlock = subQuery.getBlock();
ScanNode scan = execBlock.getScanNodes()[0];
Path tablePath;
- tablePath = subQuery.getContext().getStorageManager().getTablePath(scan.getTableName());
+ tablePath = StorageManager.getFileStorageManager(subQuery.getContext().getConf()).getTablePath(scan.getTableName());
ExecutionBlock sampleChildBlock = masterPlan.getChild(subQuery.getId(), 0);
SortNode sortNode = PlannerUtil.findTopNode(sampleChildBlock.getPlan(), NodeType.SORT);
@@ -772,14 +784,14 @@ public class Repartitioner {
public static void scheduleHashShuffledFetches(TaskSchedulerContext schedulerContext, MasterPlan masterPlan,
SubQuery subQuery, DataChannel channel,
- int maxNum) {
+ int maxNum) throws IOException {
ExecutionBlock execBlock = subQuery.getBlock();
ScanNode scan = execBlock.getScanNodes()[0];
Path tablePath;
- tablePath = subQuery.getContext().getStorageManager().getTablePath(scan.getTableName());
+ tablePath = StorageManager.getFileStorageManager(subQuery.getContext().getConf()).getTablePath(scan.getTableName());
- FileFragment frag = new FileFragment(scan.getCanonicalName(), tablePath, 0, 0, new String[]{UNKNOWN_HOST});
- List<FileFragment> fragments = new ArrayList<FileFragment>();
+ Fragment frag = new FileFragment(scan.getCanonicalName(), tablePath, 0, 0, new String[]{UNKNOWN_HOST});
+ List<Fragment> fragments = new ArrayList<Fragment>();
fragments.add(frag);
SubQuery.scheduleFragments(subQuery, fragments);
http://git-wip-us.apache.org/repos/asf/tajo/blob/cc46aeac/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
index 91fd22d..ec9afaa 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
@@ -58,10 +58,12 @@ import org.apache.tajo.master.TaskRunnerGroupEvent.EventType;
import org.apache.tajo.master.event.*;
import org.apache.tajo.master.event.QueryUnitAttemptScheduleEvent.QueryUnitAttemptScheduleContext;
import org.apache.tajo.master.querymaster.QueryUnit.IntermediateEntry;
+import org.apache.tajo.storage.FileStorageManager;
import org.apache.tajo.plan.util.PlannerUtil;
import org.apache.tajo.plan.logical.*;
import org.apache.tajo.storage.StorageManager;
import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.storage.fragment.Fragment;
import org.apache.tajo.unit.StorageUnit;
import org.apache.tajo.util.KeyValueSet;
import org.apache.tajo.worker.FetchImpl;
@@ -93,7 +95,6 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
private TableStats resultStatistics;
private TableStats inputStatistics;
private EventHandler<Event> eventHandler;
- private final StorageManager sm;
private AbstractTaskScheduler taskScheduler;
private QueryMasterTask.QueryMasterTaskContext context;
private final List<String> diagnostics = new ArrayList<String>();
@@ -281,12 +282,10 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
private List<IntermediateEntry> hashShuffleIntermediateEntries = new ArrayList<IntermediateEntry>();
private AtomicInteger completeReportReceived = new AtomicInteger(0);
- public SubQuery(QueryMasterTask.QueryMasterTaskContext context, MasterPlan masterPlan,
- ExecutionBlock block, StorageManager sm) {
+ public SubQuery(QueryMasterTask.QueryMasterTaskContext context, MasterPlan masterPlan, ExecutionBlock block) {
this.context = context;
this.masterPlan = masterPlan;
this.block = block;
- this.sm = sm;
this.eventHandler = context.getEventHandler();
ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
@@ -434,10 +433,6 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
return this.priority;
}
- public StorageManager getStorageManager() {
- return sm;
- }
-
public ExecutionBlockId getId() {
return block.getId();
}
@@ -975,7 +970,7 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
ScanNode scan = scans[0];
TableDesc table = subQuery.context.getTableDescMap().get(scan.getCanonicalName());
- Collection<FileFragment> fragments;
+ Collection<Fragment> fragments;
TableMeta meta = table.getMeta();
// Depending on scanner node's type, it creates fragments. If scan is for
@@ -984,10 +979,13 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
// span a number of blocks or possibly consists of a number of files.
if (scan.getType() == NodeType.PARTITIONS_SCAN) {
// After calling this method, partition paths are removed from the physical plan.
- fragments = Repartitioner.getFragmentsFromPartitionedTable(subQuery.getStorageManager(), scan, table);
+ FileStorageManager storageManager =
+ StorageManager.getFileStorageManager(subQuery.getContext().getConf());
+ fragments = Repartitioner.getFragmentsFromPartitionedTable(storageManager, scan, table);
} else {
- Path inputPath = table.getPath();
- fragments = subQuery.getStorageManager().getSplits(scan.getCanonicalName(), meta, table.getSchema(), inputPath);
+ StorageManager storageManager =
+ StorageManager.getStorageManager(subQuery.getContext().getConf(), meta.getStoreType());
+ fragments = storageManager.getSplits(scan.getCanonicalName(), table);
}
SubQuery.scheduleFragments(subQuery, fragments);
@@ -1005,27 +1003,27 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
}
}
- public static void scheduleFragment(SubQuery subQuery, FileFragment fragment) {
+ public static void scheduleFragment(SubQuery subQuery, Fragment fragment) {
subQuery.taskScheduler.handle(new FragmentScheduleEvent(TaskSchedulerEvent.EventType.T_SCHEDULE,
subQuery.getId(), fragment));
}
- public static void scheduleFragments(SubQuery subQuery, Collection<FileFragment> fragments) {
- for (FileFragment eachFragment : fragments) {
+ public static void scheduleFragments(SubQuery subQuery, Collection<Fragment> fragments) {
+ for (Fragment eachFragment : fragments) {
scheduleFragment(subQuery, eachFragment);
}
}
- public static void scheduleFragments(SubQuery subQuery, Collection<FileFragment> leftFragments,
- Collection<FileFragment> broadcastFragments) {
- for (FileFragment eachLeafFragment : leftFragments) {
+ public static void scheduleFragments(SubQuery subQuery, Collection<Fragment> leftFragments,
+ Collection<Fragment> broadcastFragments) {
+ for (Fragment eachLeafFragment : leftFragments) {
scheduleFragment(subQuery, eachLeafFragment, broadcastFragments);
}
}
public static void scheduleFragment(SubQuery subQuery,
- FileFragment leftFragment, Collection<FileFragment> rightFragments) {
+ Fragment leftFragment, Collection<Fragment> rightFragments) {
subQuery.taskScheduler.handle(new FragmentScheduleEvent(TaskSchedulerEvent.EventType.T_SCHEDULE,
subQuery.getId(), leftFragment, rightFragments));
}