| /** |
| * 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.master; |
| |
| import com.google.common.base.Preconditions; |
| import com.google.protobuf.ByteString; |
| 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.service.AbstractService; |
| import org.apache.hadoop.util.StringUtils; |
| import org.apache.tajo.QueryId; |
| import org.apache.tajo.QueryIdFactory; |
| import org.apache.tajo.TajoConstants; |
| import org.apache.tajo.algebra.AlterTablespaceSetType; |
| import org.apache.tajo.algebra.Expr; |
| import org.apache.tajo.algebra.JsonHelper; |
| import org.apache.tajo.annotation.Nullable; |
| 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.statistics.TableStats; |
| import org.apache.tajo.common.TajoDataTypes; |
| import org.apache.tajo.conf.TajoConf; |
| import org.apache.tajo.datum.DatumFactory; |
| import org.apache.tajo.engine.eval.EvalNode; |
| import org.apache.tajo.engine.exception.IllegalQueryStatusException; |
| import org.apache.tajo.engine.exception.VerifyException; |
| import org.apache.tajo.engine.parser.SQLAnalyzer; |
| import org.apache.tajo.engine.planner.*; |
| import org.apache.tajo.engine.planner.logical.*; |
| import org.apache.tajo.engine.planner.physical.EvalExprExec; |
| import org.apache.tajo.engine.planner.physical.StoreTableExec; |
| import org.apache.tajo.engine.query.QueryContext; |
| import org.apache.tajo.ipc.ClientProtos; |
| import org.apache.tajo.master.TajoMaster.MasterContext; |
| import org.apache.tajo.master.querymaster.Query; |
| import org.apache.tajo.master.querymaster.QueryInfo; |
| import org.apache.tajo.master.querymaster.QueryJobManager; |
| import org.apache.tajo.master.querymaster.QueryMasterTask; |
| import org.apache.tajo.master.session.Session; |
| import org.apache.tajo.storage.*; |
| import org.apache.tajo.worker.TaskAttemptContext; |
| |
| import java.io.IOException; |
| import java.sql.SQLException; |
| import java.util.ArrayList; |
| import java.util.List; |
| |
| import static org.apache.tajo.TajoConstants.DEFAULT_TABLESPACE_NAME; |
| import static org.apache.tajo.catalog.proto.CatalogProtos.AlterTablespaceProto; |
| import static org.apache.tajo.ipc.ClientProtos.SubmitQueryResponse; |
| import static org.apache.tajo.ipc.ClientProtos.SubmitQueryResponse.SerializedResultSet; |
| |
| public class GlobalEngine extends AbstractService { |
| /** Class Logger */ |
| private final static Log LOG = LogFactory.getLog(GlobalEngine.class); |
| |
| private final MasterContext context; |
| private final AbstractStorageManager sm; |
| |
| private SQLAnalyzer analyzer; |
| private CatalogService catalog; |
| private PreLogicalPlanVerifier preVerifier; |
| private LogicalPlanner planner; |
| private LogicalOptimizer optimizer; |
| private LogicalPlanVerifier annotatedPlanVerifier; |
| private DistributedQueryHookManager hookManager; |
| |
| public GlobalEngine(final MasterContext context) { |
| super(GlobalEngine.class.getName()); |
| this.context = context; |
| this.catalog = context.getCatalog(); |
| this.sm = context.getStorageManager(); |
| } |
| |
| public void start() { |
| try { |
| analyzer = new SQLAnalyzer(); |
| preVerifier = new PreLogicalPlanVerifier(context.getCatalog()); |
| planner = new LogicalPlanner(context.getCatalog()); |
| optimizer = new LogicalOptimizer(context.getConf()); |
| annotatedPlanVerifier = new LogicalPlanVerifier(context.getConf(), context.getCatalog()); |
| |
| hookManager = new DistributedQueryHookManager(); |
| hookManager.addHook(new CreateTableHook()); |
| hookManager.addHook(new InsertHook()); |
| } catch (Throwable t) { |
| LOG.error(t.getMessage(), t); |
| } |
| super.start(); |
| } |
| |
| public void stop() { |
| super.stop(); |
| } |
| |
| public SubmitQueryResponse executeQuery(Session session, String query, boolean isJson) { |
| LOG.info("Query: " + query); |
| QueryContext queryContext = new QueryContext(); |
| queryContext.putAll(session.getAllVariables()); |
| Expr planningContext; |
| |
| try { |
| if (isJson) { |
| planningContext = buildExpressionFromJson(query); |
| } else { |
| // setting environment variables |
| String [] cmds = query.split(" "); |
| if(cmds != null) { |
| if(cmds[0].equalsIgnoreCase("set")) { |
| String[] params = cmds[1].split("="); |
| context.getConf().set(params[0], params[1]); |
| SubmitQueryResponse.Builder responseBuilder = SubmitQueryResponse.newBuilder(); |
| responseBuilder.setUserName(context.getConf().getVar(TajoConf.ConfVars.USERNAME)); |
| responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto()); |
| responseBuilder.setResultCode(ClientProtos.ResultCode.OK); |
| return responseBuilder.build(); |
| } |
| } |
| |
| planningContext = buildExpressionFromSql(queryContext, query); |
| } |
| |
| String jsonExpr = planningContext.toJson(); |
| LogicalPlan plan = createLogicalPlan(session, planningContext); |
| SubmitQueryResponse response = executeQueryInternal(queryContext, session, plan, query, jsonExpr); |
| return response; |
| } catch (Throwable t) { |
| context.getSystemMetrics().counter("Query", "errorQuery").inc(); |
| LOG.error("\nStack Trace:\n" + StringUtils.stringifyException(t)); |
| SubmitQueryResponse.Builder responseBuilder = SubmitQueryResponse.newBuilder(); |
| responseBuilder.setUserName(context.getConf().getVar(TajoConf.ConfVars.USERNAME)); |
| responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto()); |
| responseBuilder.setIsForwarded(true); |
| responseBuilder.setResultCode(ClientProtos.ResultCode.ERROR); |
| String errorMessage = t.getMessage(); |
| if (t.getMessage() == null) { |
| errorMessage = t.getClass().getName(); |
| } |
| responseBuilder.setErrorMessage(errorMessage); |
| responseBuilder.setErrorTrace(StringUtils.stringifyException(t)); |
| return responseBuilder.build(); |
| } |
| } |
| |
| public Expr buildExpressionFromJson(String json) { |
| return JsonHelper.fromJson(json, Expr.class); |
| } |
| |
| public Expr buildExpressionFromSql(QueryContext queryContext, String sql) |
| throws InterruptedException, IOException, IllegalQueryStatusException { |
| context.getSystemMetrics().counter("Query", "totalQuery").inc(); |
| return analyzer.parse(sql); |
| } |
| |
| private SubmitQueryResponse executeQueryInternal(QueryContext queryContext, |
| Session session, |
| LogicalPlan plan, |
| String sql, |
| String jsonExpr) throws Exception { |
| |
| LogicalRootNode rootNode = plan.getRootBlock().getRoot(); |
| |
| SubmitQueryResponse.Builder responseBuilder = SubmitQueryResponse.newBuilder(); |
| responseBuilder.setIsForwarded(false); |
| responseBuilder.setUserName(context.getConf().getVar(TajoConf.ConfVars.USERNAME)); |
| |
| if (PlannerUtil.checkIfDDLPlan(rootNode)) { |
| context.getSystemMetrics().counter("Query", "numDDLQuery").inc(); |
| updateQuery(session, rootNode.getChild()); |
| responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto()); |
| responseBuilder.setResultCode(ClientProtos.ResultCode.OK); |
| |
| } else if (plan.isExplain()) { // explain query |
| String explainStr = PlannerUtil.buildExplainString(plan.getRootBlock().getRoot()); |
| Schema schema = new Schema(); |
| schema.addColumn("explain", TajoDataTypes.Type.TEXT); |
| RowStoreUtil.RowStoreEncoder encoder = RowStoreUtil.createEncoder(schema); |
| |
| SerializedResultSet.Builder serializedResBuilder = SerializedResultSet.newBuilder(); |
| |
| VTuple tuple = new VTuple(1); |
| String[] lines = explainStr.split("\n"); |
| int bytesNum = 0; |
| for (String line : lines) { |
| tuple.put(0, DatumFactory.createText(line)); |
| byte [] encodedData = encoder.toBytes(tuple); |
| bytesNum += encodedData.length; |
| serializedResBuilder.addSerializedTuples(ByteString.copyFrom(encodedData)); |
| } |
| serializedResBuilder.setSchema(schema.getProto()); |
| serializedResBuilder.setBytesNum(bytesNum); |
| |
| responseBuilder.setResultSet(serializedResBuilder.build()); |
| responseBuilder.setMaxRowNum(lines.length); |
| responseBuilder.setResultCode(ClientProtos.ResultCode.OK); |
| responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto()); |
| |
| // Simple query indicates a form of 'select * from tb_name [LIMIT X];'. |
| } else if (PlannerUtil.checkIfSimpleQuery(plan)) { |
| ScanNode scanNode = plan.getRootBlock().getNode(NodeType.SCAN); |
| TableDesc desc = scanNode.getTableDesc(); |
| if (plan.getRootBlock().hasNode(NodeType.LIMIT)) { |
| LimitNode limitNode = plan.getRootBlock().getNode(NodeType.LIMIT); |
| responseBuilder.setMaxRowNum((int) limitNode.getFetchFirstNum()); |
| } else { |
| if (desc.getStats().getNumBytes() > 0 && desc.getStats().getNumRows() == 0) { |
| responseBuilder.setMaxRowNum(Integer.MAX_VALUE); |
| } |
| } |
| responseBuilder.setTableDesc(desc.getProto()); |
| responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto()); |
| responseBuilder.setResultCode(ClientProtos.ResultCode.OK); |
| |
| // NonFromQuery indicates a form of 'select a, x+y;' |
| } else if (PlannerUtil.checkIfNonFromQuery(plan)) { |
| Target [] targets = plan.getRootBlock().getRawTargets(); |
| if (targets == null) { |
| throw new PlanningException("No targets"); |
| } |
| final Tuple outTuple = new VTuple(targets.length); |
| for (int i = 0; i < targets.length; i++) { |
| EvalNode eval = targets[i].getEvalTree(); |
| outTuple.put(i, eval.eval(null, null)); |
| } |
| boolean isInsert = rootNode.getChild() != null && rootNode.getChild().getType() == NodeType.INSERT; |
| if (isInsert) { |
| InsertNode insertNode = rootNode.getChild(); |
| insertNonFromQuery(queryContext, insertNode, responseBuilder); |
| } else { |
| Schema schema = PlannerUtil.targetToSchema(targets); |
| RowStoreUtil.RowStoreEncoder encoder = RowStoreUtil.createEncoder(schema); |
| byte[] serializedBytes = encoder.toBytes(outTuple); |
| SerializedResultSet.Builder serializedResBuilder = SerializedResultSet.newBuilder(); |
| serializedResBuilder.addSerializedTuples(ByteString.copyFrom(serializedBytes)); |
| serializedResBuilder.setSchema(schema.getProto()); |
| serializedResBuilder.setBytesNum(serializedBytes.length); |
| |
| responseBuilder.setResultSet(serializedResBuilder); |
| responseBuilder.setMaxRowNum(1); |
| responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto()); |
| responseBuilder.setResultCode(ClientProtos.ResultCode.OK); |
| } |
| } else { // it requires distributed execution. So, the query is forwarded to a query master. |
| context.getSystemMetrics().counter("Query", "numDMLQuery").inc(); |
| hookManager.doHooks(queryContext, plan); |
| |
| QueryJobManager queryJobManager = this.context.getQueryJobManager(); |
| QueryInfo queryInfo; |
| |
| queryInfo = queryJobManager.createNewQueryJob(session, queryContext, sql, jsonExpr, rootNode); |
| |
| if(queryInfo == null) { |
| responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto()); |
| responseBuilder.setResultCode(ClientProtos.ResultCode.ERROR); |
| responseBuilder.setErrorMessage("Fail starting QueryMaster."); |
| } else { |
| responseBuilder.setIsForwarded(true); |
| responseBuilder.setQueryId(queryInfo.getQueryId().getProto()); |
| responseBuilder.setResultCode(ClientProtos.ResultCode.OK); |
| if(queryInfo.getQueryMasterHost() != null) { |
| responseBuilder.setQueryMasterHost(queryInfo.getQueryMasterHost()); |
| } |
| responseBuilder.setQueryMasterPort(queryInfo.getQueryMasterClientPort()); |
| LOG.info("Query is forwarded to " + queryInfo.getQueryMasterHost() + ":" + queryInfo.getQueryMasterPort()); |
| } |
| } |
| SubmitQueryResponse response = responseBuilder.build(); |
| return response; |
| } |
| |
| private void insertNonFromQuery(QueryContext queryContext, InsertNode insertNode, SubmitQueryResponse.Builder responseBuilder) |
| throws Exception { |
| String nodeUniqName = insertNode.getTableName() == null ? insertNode.getPath().getName() : insertNode.getTableName(); |
| String queryId = nodeUniqName + "_" + System.currentTimeMillis(); |
| |
| FileSystem fs = TajoConf.getWarehouseDir(context.getConf()).getFileSystem(context.getConf()); |
| Path stagingDir = QueryMasterTask.initStagingDir(context.getConf(), fs, queryId.toString()); |
| |
| Path stagingResultDir = new Path(stagingDir, TajoConstants.RESULT_DIR_NAME); |
| fs.mkdirs(stagingResultDir); |
| |
| TableDesc tableDesc = null; |
| Path finalOutputDir = null; |
| if (insertNode.getTableName() != null) { |
| tableDesc = this.catalog.getTableDesc(insertNode.getTableName()); |
| finalOutputDir = tableDesc.getPath(); |
| } else { |
| finalOutputDir = insertNode.getPath(); |
| } |
| |
| TaskAttemptContext taskAttemptContext = |
| new TaskAttemptContext(context.getConf(), queryContext, null, (CatalogProtos.FragmentProto[]) null, stagingDir); |
| taskAttemptContext.setOutputPath(new Path(stagingResultDir, "part-01-000000")); |
| |
| EvalExprExec evalExprExec = new EvalExprExec(taskAttemptContext, (EvalExprNode) insertNode.getChild()); |
| StoreTableExec exec = new StoreTableExec(taskAttemptContext, insertNode, evalExprExec); |
| try { |
| exec.init(); |
| exec.next(); |
| } finally { |
| exec.close(); |
| } |
| |
| if (insertNode.isOverwrite()) { // INSERT OVERWRITE INTO |
| // it moves the original table into the temporary location. |
| // Then it moves the new result table into the original table location. |
| // Upon failed, it recovers the original table if possible. |
| boolean movedToOldTable = false; |
| boolean committed = false; |
| Path oldTableDir = new Path(stagingDir, TajoConstants.INSERT_OVERWIRTE_OLD_TABLE_NAME); |
| try { |
| if (fs.exists(finalOutputDir)) { |
| fs.rename(finalOutputDir, oldTableDir); |
| movedToOldTable = fs.exists(oldTableDir); |
| } else { // if the parent does not exist, make its parent directory. |
| fs.mkdirs(finalOutputDir.getParent()); |
| } |
| fs.rename(stagingResultDir, finalOutputDir); |
| committed = fs.exists(finalOutputDir); |
| } catch (IOException ioe) { |
| // recover the old table |
| if (movedToOldTable && !committed) { |
| fs.rename(oldTableDir, finalOutputDir); |
| } |
| } |
| } else { |
| FileStatus[] files = fs.listStatus(stagingResultDir); |
| for (FileStatus eachFile: files) { |
| Path targetFilePath = new Path(finalOutputDir, eachFile.getPath().getName()); |
| if (fs.exists(targetFilePath)) { |
| targetFilePath = new Path(finalOutputDir, eachFile.getPath().getName() + "_" + System.currentTimeMillis()); |
| } |
| fs.rename(eachFile.getPath(), targetFilePath); |
| } |
| } |
| |
| if (insertNode.hasTargetTable()) { |
| TableStats stats = tableDesc.getStats(); |
| long volume = Query.getTableVolume(context.getConf(), finalOutputDir); |
| stats.setNumBytes(volume); |
| stats.setNumRows(1); |
| |
| catalog.dropTable(insertNode.getTableName()); |
| catalog.createTable(tableDesc); |
| |
| responseBuilder.setTableDesc(tableDesc.getProto()); |
| } else { |
| TableStats stats = new TableStats(); |
| long volume = Query.getTableVolume(context.getConf(), finalOutputDir); |
| stats.setNumBytes(volume); |
| stats.setNumRows(1); |
| |
| // Empty TableDesc |
| List<CatalogProtos.ColumnProto> columns = new ArrayList<CatalogProtos.ColumnProto>(); |
| CatalogProtos.TableDescProto tableDescProto = CatalogProtos.TableDescProto.newBuilder() |
| .setTableName(nodeUniqName) |
| .setMeta(CatalogProtos.TableProto.newBuilder().setStoreType(CatalogProtos.StoreType.CSV).build()) |
| .setSchema(CatalogProtos.SchemaProto.newBuilder().addAllFields(columns).build()) |
| .setStats(stats.getProto()) |
| .build(); |
| |
| responseBuilder.setTableDesc(tableDescProto); |
| } |
| |
| // If queryId == NULL_QUERY_ID and MaxRowNum == -1, TajoCli prints only number of inserted rows. |
| responseBuilder.setMaxRowNum(-1); |
| responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto()); |
| responseBuilder.setResultCode(ClientProtos.ResultCode.OK); |
| } |
| |
| |
| public QueryId updateQuery(Session session, String sql, boolean isJson) throws IOException, SQLException, PlanningException { |
| try { |
| LOG.info("SQL: " + sql); |
| |
| Expr expr; |
| if (isJson) { |
| expr = JsonHelper.fromJson(sql, Expr.class); |
| } else { |
| // parse the query |
| expr = analyzer.parse(sql); |
| } |
| |
| LogicalPlan plan = createLogicalPlan(session, expr); |
| LogicalRootNode rootNode = plan.getRootBlock().getRoot(); |
| |
| if (!PlannerUtil.checkIfDDLPlan(rootNode)) { |
| throw new SQLException("This is not update query:\n" + sql); |
| } else { |
| updateQuery(session, rootNode.getChild()); |
| return QueryIdFactory.NULL_QUERY_ID; |
| } |
| } catch (Exception e) { |
| LOG.error(e.getMessage(), e); |
| throw new IOException(e.getMessage(), e); |
| } |
| } |
| |
| private boolean updateQuery(Session session, LogicalNode root) throws IOException { |
| |
| switch (root.getType()) { |
| case CREATE_DATABASE: |
| CreateDatabaseNode createDatabase = (CreateDatabaseNode) root; |
| createDatabase(session, createDatabase.getDatabaseName(), null, createDatabase.isIfNotExists()); |
| return true; |
| case DROP_DATABASE: |
| DropDatabaseNode dropDatabaseNode = (DropDatabaseNode) root; |
| dropDatabase(session, dropDatabaseNode.getDatabaseName(), dropDatabaseNode.isIfExists()); |
| return true; |
| case CREATE_TABLE: |
| CreateTableNode createTable = (CreateTableNode) root; |
| createTable(session, createTable, createTable.isIfNotExists()); |
| return true; |
| case DROP_TABLE: |
| DropTableNode dropTable = (DropTableNode) root; |
| dropTable(session, dropTable.getTableName(), dropTable.isIfExists(), dropTable.isPurge()); |
| return true; |
| case ALTER_TABLESPACE: |
| AlterTablespaceNode alterTablespace = (AlterTablespaceNode) root; |
| alterTablespace(session, alterTablespace); |
| return true; |
| case ALTER_TABLE: |
| AlterTableNode alterTable = (AlterTableNode) root; |
| alterTable(session,alterTable); |
| return true; |
| case TRUNCATE_TABLE: |
| TruncateTableNode truncateTable = (TruncateTableNode) root; |
| truncateTable(session, truncateTable); |
| return true; |
| default: |
| throw new InternalError("updateQuery cannot handle such query: \n" + root.toJson()); |
| } |
| } |
| |
| private LogicalPlan createLogicalPlan(Session session, Expr expression) throws PlanningException { |
| |
| VerificationState state = new VerificationState(); |
| preVerifier.verify(session, state, expression); |
| if (!state.verified()) { |
| StringBuilder sb = new StringBuilder(); |
| for (String error : state.getErrorMessages()) { |
| sb.append(error).append("\n"); |
| } |
| throw new VerifyException(sb.toString()); |
| } |
| |
| LogicalPlan plan = planner.createPlan(session, expression); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("============================================="); |
| LOG.debug("Non Optimized Query: \n" + plan.toString()); |
| LOG.debug("============================================="); |
| } |
| LOG.info("Non Optimized Query: \n" + plan.toString()); |
| optimizer.optimize(session, plan); |
| LOG.info("============================================="); |
| LOG.info("Optimized Query: \n" + plan.toString()); |
| LOG.info("============================================="); |
| |
| annotatedPlanVerifier.verify(session, state, plan); |
| |
| if (!state.verified()) { |
| StringBuilder sb = new StringBuilder(); |
| for (String error : state.getErrorMessages()) { |
| sb.append(error).append("\n"); |
| } |
| throw new VerifyException(sb.toString()); |
| } |
| |
| return plan; |
| } |
| |
| /** |
| * Alter a given table |
| */ |
| public void alterTablespace(final Session session, final AlterTablespaceNode alterTablespace) { |
| |
| final CatalogService catalog = context.getCatalog(); |
| final String spaceName = alterTablespace.getTablespaceName(); |
| |
| AlterTablespaceProto.Builder builder = AlterTablespaceProto.newBuilder(); |
| builder.setSpaceName(spaceName); |
| if (alterTablespace.getSetType() == AlterTablespaceSetType.LOCATION) { |
| AlterTablespaceProto.AlterTablespaceCommand.Builder commandBuilder = |
| AlterTablespaceProto.AlterTablespaceCommand.newBuilder(); |
| commandBuilder.setType(AlterTablespaceProto.AlterTablespaceType.LOCATION); |
| commandBuilder.setLocation(AlterTablespaceProto.SetLocation.newBuilder().setUri(alterTablespace.getLocation())); |
| commandBuilder.build(); |
| builder.addCommand(commandBuilder); |
| } else { |
| throw new RuntimeException("This 'ALTER TABLESPACE' is not supported yet."); |
| } |
| |
| catalog.alterTablespace(builder.build()); |
| } |
| |
| /** |
| * Alter a given table |
| */ |
| public void alterTable(final Session session, final AlterTableNode alterTable) throws IOException { |
| |
| final CatalogService catalog = context.getCatalog(); |
| final String tableName = alterTable.getTableName(); |
| |
| String databaseName; |
| String simpleTableName; |
| if (CatalogUtil.isFQTableName(tableName)) { |
| String[] split = CatalogUtil.splitFQTableName(tableName); |
| databaseName = split[0]; |
| simpleTableName = split[1]; |
| } else { |
| databaseName = session.getCurrentDatabase(); |
| simpleTableName = tableName; |
| } |
| final String qualifiedName = CatalogUtil.buildFQName(databaseName, simpleTableName); |
| |
| if (!catalog.existsTable(databaseName, simpleTableName)) { |
| throw new NoSuchTableException(qualifiedName); |
| } |
| |
| switch (alterTable.getAlterTableOpType()) { |
| case RENAME_TABLE: |
| if (!catalog.existsTable(databaseName, simpleTableName)) { |
| throw new NoSuchTableException(alterTable.getTableName()); |
| } |
| if (catalog.existsTable(databaseName, alterTable.getNewTableName())) { |
| throw new AlreadyExistsTableException(alterTable.getNewTableName()); |
| } |
| |
| TableDesc desc = catalog.getTableDesc(databaseName, simpleTableName); |
| |
| if (!desc.isExternal()) { // if the table is the managed table |
| Path oldPath = StorageUtil.concatPath(context.getConf().getVar(TajoConf.ConfVars.WAREHOUSE_DIR), |
| databaseName, simpleTableName); |
| Path newPath = StorageUtil.concatPath(context.getConf().getVar(TajoConf.ConfVars.WAREHOUSE_DIR), |
| databaseName, alterTable.getNewTableName()); |
| FileSystem fs = oldPath.getFileSystem(context.getConf()); |
| |
| if (!fs.exists(oldPath)) { |
| throw new IOException("No such a table directory: " + oldPath); |
| } |
| if (fs.exists(newPath)) { |
| throw new IOException("Already table directory exists: " + newPath); |
| } |
| |
| fs.rename(oldPath, newPath); |
| } |
| catalog.alterTable(CatalogUtil.renameTable(qualifiedName, alterTable.getNewTableName(), |
| AlterTableType.RENAME_TABLE)); |
| break; |
| case RENAME_COLUMN: |
| if (existColumnName(qualifiedName, alterTable.getNewColumnName())) { |
| throw new ColumnNameAlreadyExistException(alterTable.getNewColumnName()); |
| } |
| catalog.alterTable(CatalogUtil.renameColumn(qualifiedName, alterTable.getColumnName(), alterTable.getNewColumnName(), AlterTableType.RENAME_COLUMN)); |
| break; |
| case ADD_COLUMN: |
| if (existColumnName(qualifiedName, alterTable.getAddNewColumn().getSimpleName())) { |
| throw new ColumnNameAlreadyExistException(alterTable.getAddNewColumn().getSimpleName()); |
| } |
| catalog.alterTable(CatalogUtil.addNewColumn(qualifiedName, alterTable.getAddNewColumn(), AlterTableType.ADD_COLUMN)); |
| break; |
| default: |
| //TODO |
| } |
| } |
| |
| /** |
| * Truncate table a given table |
| */ |
| public void truncateTable(final Session session, final TruncateTableNode truncateTableNode) throws IOException { |
| List<String> tableNames = truncateTableNode.getTableNames(); |
| final CatalogService catalog = context.getCatalog(); |
| |
| String databaseName; |
| String simpleTableName; |
| |
| List<TableDesc> tableDescList = new ArrayList<TableDesc>(); |
| for (String eachTableName: tableNames) { |
| if (CatalogUtil.isFQTableName(eachTableName)) { |
| String[] split = CatalogUtil.splitFQTableName(eachTableName); |
| databaseName = split[0]; |
| simpleTableName = split[1]; |
| } else { |
| databaseName = session.getCurrentDatabase(); |
| simpleTableName = eachTableName; |
| } |
| final String qualifiedName = CatalogUtil.buildFQName(databaseName, simpleTableName); |
| |
| if (!catalog.existsTable(databaseName, simpleTableName)) { |
| throw new NoSuchTableException(qualifiedName); |
| } |
| |
| Path warehousePath = new Path(TajoConf.getWarehouseDir(context.getConf()), databaseName); |
| TableDesc tableDesc = catalog.getTableDesc(databaseName, simpleTableName); |
| Path tablePath = tableDesc.getPath(); |
| if (tablePath.getParent() == null || |
| !tablePath.getParent().toUri().getPath().equals(warehousePath.toUri().getPath())) { |
| throw new IOException("Can't truncate external table:" + eachTableName + ", data dir=" + tablePath + |
| ", warehouse dir=" + warehousePath); |
| } |
| tableDescList.add(tableDesc); |
| } |
| |
| for (TableDesc eachTable: tableDescList) { |
| Path path = eachTable.getPath(); |
| LOG.info("Truncate table: " + eachTable.getName() + ", delete all data files in " + path); |
| FileSystem fs = path.getFileSystem(context.getConf()); |
| |
| FileStatus[] files = fs.listStatus(path); |
| if (files != null) { |
| for (FileStatus eachFile: files) { |
| fs.delete(eachFile.getPath(), true); |
| } |
| } |
| } |
| } |
| |
| private boolean existColumnName(String tableName, String columnName) { |
| final TableDesc tableDesc = catalog.getTableDesc(tableName); |
| return tableDesc.getSchema().containsByName(columnName) ? true : false; |
| } |
| |
| private TableDesc createTable(Session session, CreateTableNode createTable, boolean ifNotExists) throws IOException { |
| TableMeta meta; |
| |
| if (createTable.hasOptions()) { |
| meta = CatalogUtil.newTableMeta(createTable.getStorageType(), createTable.getOptions()); |
| } else { |
| meta = CatalogUtil.newTableMeta(createTable.getStorageType()); |
| } |
| |
| if(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 = session.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(session, createTable.getTableName(), createTable.getTableSchema(), |
| meta, createTable.getPath(), createTable.isExternal(), createTable.getPartitionMethod(), ifNotExists); |
| } |
| |
| public TableDesc createTableOnPath(Session session, String tableName, Schema schema, TableMeta meta, |
| Path path, boolean isExternal, PartitionMethodDesc partitionDesc, |
| boolean ifNotExists) |
| throws IOException { |
| String databaseName; |
| String simpleTableName; |
| if (CatalogUtil.isFQTableName(tableName)) { |
| String [] splitted = CatalogUtil.splitFQTableName(tableName); |
| databaseName = splitted[0]; |
| simpleTableName = splitted[1]; |
| } else { |
| databaseName = session.getCurrentDatabase(); |
| simpleTableName = tableName; |
| } |
| String qualifiedName = CatalogUtil.buildFQName(databaseName, simpleTableName); |
| |
| boolean exists = catalog.existsTable(databaseName, simpleTableName); |
| |
| if (exists) { |
| if (ifNotExists) { |
| LOG.info("relation \"" + qualifiedName + "\" is already exists." ); |
| return catalog.getTableDesc(databaseName, simpleTableName); |
| } else { |
| throw new AlreadyExistsTableException(CatalogUtil.buildFQName(databaseName, tableName)); |
| } |
| } |
| |
| 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); |
| TableDesc desc = new TableDesc(CatalogUtil.buildFQName(databaseName, simpleTableName), |
| schema, meta, path, isExternal); |
| desc.setStats(stats); |
| if (partitionDesc != null) { |
| desc.setPartitionMethod(partitionDesc); |
| } |
| |
| if (catalog.createTable(desc)) { |
| LOG.info("Table " + desc.getName() + " is created (" + desc.getStats().getNumBytes() + ")"); |
| return desc; |
| } else { |
| LOG.info("Table creation " + tableName + " is failed."); |
| throw new CatalogException("Cannot create table \"" + tableName + "\"."); |
| } |
| } |
| |
| public boolean createDatabase(@Nullable Session session, String databaseName, |
| @Nullable String tablespace, |
| boolean ifNotExists) throws IOException { |
| |
| String tablespaceName; |
| if (tablespace == null) { |
| tablespaceName = DEFAULT_TABLESPACE_NAME; |
| } else { |
| tablespaceName = tablespace; |
| } |
| |
| // CREATE DATABASE IF NOT EXISTS |
| boolean exists = catalog.existDatabase(databaseName); |
| if (exists) { |
| if (ifNotExists) { |
| LOG.info("database \"" + databaseName + "\" is already exists." ); |
| return true; |
| } else { |
| throw new AlreadyExistsDatabaseException(databaseName); |
| } |
| } |
| |
| if (catalog.createDatabase(databaseName, tablespaceName)) { |
| String normalized = databaseName; |
| Path databaseDir = StorageUtil.concatPath(context.getConf().getVar(TajoConf.ConfVars.WAREHOUSE_DIR), normalized); |
| FileSystem fs = databaseDir.getFileSystem(context.getConf()); |
| fs.mkdirs(databaseDir); |
| } |
| |
| return true; |
| } |
| |
| public boolean dropDatabase(Session session, String databaseName, boolean ifExists) { |
| |
| boolean exists = catalog.existDatabase(databaseName); |
| if(!exists) { |
| if (ifExists) { // DROP DATABASE IF EXISTS |
| LOG.info("database \"" + databaseName + "\" does not exists." ); |
| return true; |
| } else { // Otherwise, it causes an exception. |
| throw new NoSuchDatabaseException(databaseName); |
| } |
| } |
| |
| if (session.getCurrentDatabase().equals(databaseName)) { |
| throw new RuntimeException("ERROR: Cannot drop the current open database"); |
| } |
| |
| boolean result = catalog.dropDatabase(databaseName); |
| LOG.info("database " + databaseName + " is dropped."); |
| return result; |
| } |
| |
| /** |
| * Drop a given named table |
| * |
| * @param tableName to be dropped |
| * @param purge Remove all data if purge is true. |
| */ |
| public boolean dropTable(Session session, String tableName, boolean ifExists, boolean purge) { |
| CatalogService catalog = context.getCatalog(); |
| |
| String databaseName; |
| String simpleTableName; |
| if (CatalogUtil.isFQTableName(tableName)) { |
| String [] splitted = CatalogUtil.splitFQTableName(tableName); |
| databaseName = splitted[0]; |
| simpleTableName = splitted[1]; |
| } else { |
| databaseName = session.getCurrentDatabase(); |
| simpleTableName = tableName; |
| } |
| String qualifiedName = CatalogUtil.buildFQName(databaseName, simpleTableName); |
| |
| boolean exists = catalog.existsTable(qualifiedName); |
| if(!exists) { |
| if (ifExists) { // DROP TABLE IF EXISTS |
| LOG.info("relation \"" + qualifiedName + "\" is already exists." ); |
| return true; |
| } else { // Otherwise, it causes an exception. |
| throw new NoSuchTableException(qualifiedName); |
| } |
| } |
| |
| Path path = catalog.getTableDesc(qualifiedName).getPath(); |
| catalog.dropTable(qualifiedName); |
| |
| if (purge) { |
| try { |
| FileSystem fs = path.getFileSystem(context.getConf()); |
| fs.delete(path, true); |
| } catch (IOException e) { |
| throw new InternalError(e.getMessage()); |
| } |
| } |
| |
| LOG.info(String.format("relation \"%s\" is " + (purge ? " purged." : " dropped."), qualifiedName)); |
| return true; |
| } |
| |
| public interface DistributedQueryHook { |
| boolean isEligible(QueryContext queryContext, LogicalPlan plan); |
| void hook(QueryContext queryContext, LogicalPlan plan) throws Exception; |
| } |
| |
| public static class DistributedQueryHookManager { |
| private List<DistributedQueryHook> hooks = new ArrayList<DistributedQueryHook>(); |
| public void addHook(DistributedQueryHook hook) { |
| hooks.add(hook); |
| } |
| |
| public void doHooks(QueryContext queryContext, LogicalPlan plan) { |
| for (DistributedQueryHook hook : hooks) { |
| if (hook.isEligible(queryContext, plan)) { |
| try { |
| hook.hook(queryContext, plan); |
| } catch (Throwable t) { |
| t.printStackTrace(); |
| } |
| } |
| } |
| } |
| } |
| |
| public class CreateTableHook implements DistributedQueryHook { |
| |
| @Override |
| public boolean isEligible(QueryContext queryContext, LogicalPlan plan) { |
| LogicalRootNode rootNode = plan.getRootBlock().getRoot(); |
| return rootNode.getChild().getType() == NodeType.CREATE_TABLE; |
| } |
| |
| @Override |
| public void hook(QueryContext queryContext, LogicalPlan plan) throws Exception { |
| LogicalRootNode rootNode = plan.getRootBlock().getRoot(); |
| CreateTableNode createTableNode = rootNode.getChild(); |
| String [] splitted = CatalogUtil.splitFQTableName(createTableNode.getTableName()); |
| String databaseName = splitted[0]; |
| String tableName = splitted[1]; |
| queryContext.setOutputTable(tableName); |
| queryContext.setOutputPath( |
| StorageUtil.concatPath(TajoConf.getWarehouseDir(context.getConf()), databaseName, tableName)); |
| if(createTableNode.getPartitionMethod() != null) { |
| queryContext.setPartitionMethod(createTableNode.getPartitionMethod()); |
| } |
| queryContext.setCreateTable(); |
| } |
| } |
| |
| public static class InsertHook implements DistributedQueryHook { |
| |
| @Override |
| public boolean isEligible(QueryContext queryContext, LogicalPlan plan) { |
| return plan.getRootBlock().getRootType() == NodeType.INSERT; |
| } |
| |
| @Override |
| public void hook(QueryContext queryContext, LogicalPlan plan) throws Exception { |
| queryContext.setInsert(); |
| |
| InsertNode insertNode = plan.getRootBlock().getNode(NodeType.INSERT); |
| |
| // Set QueryContext settings, such as output table name and output path. |
| // It also remove data files if overwrite is true. |
| Path outputPath; |
| if (insertNode.hasTargetTable()) { // INSERT INTO [TB_NAME] |
| queryContext.setOutputTable(insertNode.getTableName()); |
| queryContext.setOutputPath(insertNode.getPath()); |
| } else { // INSERT INTO LOCATION ... |
| // When INSERT INTO LOCATION, must not set output table. |
| outputPath = insertNode.getPath(); |
| queryContext.setFileOutput(); |
| queryContext.setOutputPath(outputPath); |
| } |
| |
| if (insertNode.isOverwrite()) { |
| queryContext.setOutputOverwrite(); |
| } |
| } |
| } |
| } |
