Skip to content
This repository was archived by the owner on May 12, 2021. It is now read-only.

TAJO-1801: Table name is not unique of tableDescMap in QueryMasterTask #710

Closed
jihoonson wants to merge 3 commits into from
Closed

TAJO-1801: Table name is not unique of tableDescMap in QueryMasterTask #710

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
01db433
TAJO-1801
jihoonson Aug 25, 2015
42ba6e7
Merge branch 'master' of https://git-wip-us.apache.org/repos/asf/tajo…
jihoonson Aug 26, 2015
972e93c
TAJO-1801
jihoonson Aug 26, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 4 additions & 8 deletions tajo-core/src/main/java/org/apache/tajo/querymaster/Query.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,26 +35,22 @@
import org.apache.tajo.SessionVars;
import org.apache.tajo.TajoProtos.QueryState;
import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.proto.CatalogProtos.UpdateTableStatsProto;
import org.apache.tajo.catalog.proto.CatalogProtos.PartitionDescProto;
import org.apache.tajo.catalog.CatalogService;
import org.apache.tajo.catalog.TableDesc;
import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.catalog.proto.CatalogProtos.UpdateTableStatsProto;
import org.apache.tajo.catalog.statistics.TableStats;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.engine.planner.global.ExecutionBlock;
import org.apache.tajo.engine.planner.global.ExecutionBlockCursor;
import org.apache.tajo.engine.planner.global.ExecutionQueue;
import org.apache.tajo.engine.planner.global.MasterPlan;
import org.apache.tajo.exception.TajoException;
import org.apache.tajo.exception.TajoInternalError;
import org.apache.tajo.plan.logical.*;
import org.apache.tajo.engine.query.QueryContext;
import org.apache.tajo.exception.TajoInternalError;
import org.apache.tajo.master.event.*;
import org.apache.tajo.plan.logical.*;
import org.apache.tajo.plan.util.PlannerUtil;
import org.apache.tajo.storage.StorageConstants;
import org.apache.tajo.storage.TablespaceManager;
import org.apache.tajo.storage.Tablespace;
import org.apache.tajo.storage.TablespaceManager;
import org.apache.tajo.util.TUtil;
import org.apache.tajo.util.history.QueryHistory;
import org.apache.tajo.util.history.StageHistory;
Expand Down
12 changes: 6 additions & 6 deletions tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMasterTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ public class QueryMasterTask extends CompositeService {

private final long querySubmitTime;

private Map<String, TableDesc> tableDescMap = new HashMap<String, TableDesc>();
private final Map<Integer, TableDesc> tableDescMap = new HashMap<>();

private TajoConf systemConf;

Expand Down Expand Up @@ -332,23 +332,23 @@ public synchronized void startQuery() {
if (scanNodes != null) {
for (LogicalNode eachScanNode : scanNodes) {
ScanNode scanNode = (ScanNode) eachScanNode;
tableDescMap.put(scanNode.getCanonicalName(), scanNode.getTableDesc());
tableDescMap.put(scanNode.getPID(), scanNode.getTableDesc());
}
}

scanNodes = PlannerUtil.findAllNodes(block.getRoot(), NodeType.PARTITIONS_SCAN);
if (scanNodes != null) {
for (LogicalNode eachScanNode : scanNodes) {
ScanNode scanNode = (ScanNode) eachScanNode;
tableDescMap.put(scanNode.getCanonicalName(), scanNode.getTableDesc());
tableDescMap.put(scanNode.getPID(), scanNode.getTableDesc());
}
}

scanNodes = PlannerUtil.findAllNodes(block.getRoot(), NodeType.INDEX_SCAN);
if (scanNodes != null) {
for (LogicalNode eachScanNode : scanNodes) {
ScanNode scanNode = (ScanNode) eachScanNode;
tableDescMap.put(scanNode.getCanonicalName(), scanNode.getTableDesc());
tableDescMap.put(scanNode.getPID(), scanNode.getTableDesc());
}
}
}
Expand Down Expand Up @@ -533,8 +533,8 @@ public Stage getStage(ExecutionBlockId id) {
return query.getStage(id);
}

public Map<String, TableDesc> getTableDescMap() {
return tableDescMap;
public TableDesc getTableDesc(ScanNode scanNode) {
return tableDescMap.get(scanNode.getPID());
}

public float getProgress() {
Expand Down
6 changes: 3 additions & 3 deletions tajo-core/src/main/java/org/apache/tajo/querymaster/Repartitioner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,7 @@ public static void scheduleFragmentsForJoinQuery(TaskSchedulerContext schedulerC

// initialize variables from the child operators
for (int i = 0; i < scans.length; i++) {
TableDesc tableDesc = masterContext.getTableDescMap().get(scans[i].getCanonicalName());
TableDesc tableDesc = masterContext.getTableDesc(scans[i]);

if (tableDesc == null) { // if it is a real table stored on storage
if (execBlock.getUnionScanMap() != null && !execBlock.getUnionScanMap().isEmpty()) {
Expand Down Expand Up @@ -376,7 +376,7 @@ private static void scheduleSymmetricRepartitionJoin(QueryMasterTask.QueryMaster
for (ScanNode eachScan: broadcastScans) {

Path[] partitionScanPaths = null;
TableDesc tableDesc = masterContext.getTableDescMap().get(eachScan.getCanonicalName());
TableDesc tableDesc = masterContext.getTableDesc(eachScan);
Tablespace space = TablespaceManager.get(tableDesc.getUri()).get();

if (eachScan.getType() == NodeType.PARTITIONS_SCAN) {
Expand Down Expand Up @@ -498,7 +498,7 @@ private static void scheduleLeafTasksWithBroadcastTable(TaskSchedulerContext sch
List<Fragment> broadcastFragments = new ArrayList<Fragment>();
for (int i = 0; i < scans.length; i++) {
ScanNode scan = scans[i];
TableDesc desc = stage.getContext().getTableDescMap().get(scan.getCanonicalName());
TableDesc desc = stage.getContext().getTableDesc(scan);
TableMeta meta = desc.getMeta();

Collection<Fragment> scanFragments;
Expand Down
14 changes: 6 additions & 8 deletions tajo-core/src/main/java/org/apache/tajo/querymaster/Stage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.TableDesc;
import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.proto.CatalogProtos.PartitionDescProto;
import org.apache.tajo.catalog.statistics.ColumnStats;
import org.apache.tajo.catalog.statistics.StatisticsUtil;
Expand All @@ -44,21 +43,21 @@
import org.apache.tajo.engine.planner.global.MasterPlan;
import org.apache.tajo.exception.TajoException;
import org.apache.tajo.ipc.TajoWorkerProtocol;
import org.apache.tajo.plan.serder.PlanProto.DistinctGroupbyEnforcer.MultipleAggregationStage;
import org.apache.tajo.plan.serder.PlanProto.EnforceProperty;
import org.apache.tajo.master.TaskState;
import org.apache.tajo.master.event.*;
import org.apache.tajo.master.event.TaskAttemptToSchedulerEvent.TaskAttemptScheduleContext;
import org.apache.tajo.plan.logical.*;
import org.apache.tajo.plan.serder.PlanProto.DistinctGroupbyEnforcer.MultipleAggregationStage;
import org.apache.tajo.plan.serder.PlanProto.EnforceProperty;
import org.apache.tajo.plan.util.PlannerUtil;
import org.apache.tajo.querymaster.Task.IntermediateEntry;
import org.apache.tajo.rpc.AsyncRpcClient;
import org.apache.tajo.rpc.NullCallback;
import org.apache.tajo.rpc.RpcClientManager;
import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos;
import org.apache.tajo.storage.FileTablespace;
import org.apache.tajo.storage.TablespaceManager;
import org.apache.tajo.storage.Tablespace;
import org.apache.tajo.storage.TablespaceManager;
import org.apache.tajo.storage.fragment.Fragment;
import org.apache.tajo.unit.StorageUnit;
import org.apache.tajo.util.KeyValueSet;
Expand All @@ -76,9 +75,9 @@
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import static org.apache.tajo.ResourceProtos.*;
import static org.apache.tajo.conf.TajoConf.ConfVars;
import static org.apache.tajo.plan.serder.PlanProto.ShuffleType;
import static org.apache.tajo.ResourceProtos.*;


/**
Expand Down Expand Up @@ -1058,12 +1057,11 @@ public static int getNonLeafTaskNum(Stage stage) {

public static long getInputVolume(MasterPlan masterPlan, QueryMasterTask.QueryMasterTaskContext context,
ExecutionBlock execBlock) {
Map<String, TableDesc> tableMap = context.getTableDescMap();
if (masterPlan.isLeaf(execBlock)) {
ScanNode[] outerScans = execBlock.getScanNodes();
long maxVolume = 0;
for (ScanNode eachScanNode: outerScans) {
TableStats stat = tableMap.get(eachScanNode.getCanonicalName()).getStats();
TableStats stat = context.getTableDesc(eachScanNode).getStats();
if (stat.getNumBytes() > maxVolume) {
maxVolume = stat.getNumBytes();
}
Expand All @@ -1089,7 +1087,7 @@ private static void scheduleFragmentsForLeafQuery(Stage stage) throws IOExceptio
ScanNode[] scans = execBlock.getScanNodes();
Preconditions.checkArgument(scans.length == 1, "Must be Scan Query");
ScanNode scan = scans[0];
TableDesc table = stage.context.getTableDescMap().get(scan.getCanonicalName());
TableDesc table = stage.context.getTableDesc(scan);

Collection<Fragment> fragments;
Tablespace tablespace = TablespaceManager.get(scan.getTableDesc().getUri()).get();
Expand Down