You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by jn...@apache.org on 2017/01/24 06:15:22 UTC
[2/5] drill git commit: DRILL-4956: Temporary tables support
DRILL-4956: Temporary tables support
close apache/drill#666
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/bb29f19f
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/bb29f19f
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/bb29f19f
Branch: refs/heads/master
Commit: bb29f19ff8807fd07cdaa9e7110c1a003b3da15b
Parents: 8a4d7a9
Author: Arina Ielchiieva <ar...@gmail.com>
Authored: Thu Nov 3 16:55:38 2016 +0000
Committer: Jinfeng Ni <jn...@apache.org>
Committed: Mon Jan 23 17:08:06 2017 -0800
----------------------------------------------------------------------
.../src/resources/drill-override-example.conf | 8 +-
.../src/main/codegen/includes/parserImpls.ftl | 9 +-
.../org/apache/drill/exec/ExecConstants.java | 7 +-
.../exec/physical/base/AbstractWriter.java | 17 +-
.../exec/physical/impl/WriterRecordBatch.java | 22 +-
.../logical/FileSystemCreateTableEntry.java | 18 +-
.../drill/exec/planner/sql/DrillSqlWorker.java | 9 +-
.../drill/exec/planner/sql/SchemaUtilites.java | 42 +-
.../drill/exec/planner/sql/SqlConverter.java | 141 +++++--
.../sql/handlers/CreateTableHandler.java | 111 ++++-
.../planner/sql/handlers/DropTableHandler.java | 65 ++-
.../planner/sql/handlers/SqlHandlerUtil.java | 49 ++-
.../exec/planner/sql/handlers/ViewHandler.java | 63 +--
.../sql/parser/CompoundIdentifierConverter.java | 4 +-
.../exec/planner/sql/parser/SqlCreateTable.java | 28 +-
.../apache/drill/exec/rpc/user/UserServer.java | 13 +-
.../apache/drill/exec/rpc/user/UserSession.java | 170 +++++++-
.../org/apache/drill/exec/server/Drillbit.java | 23 +-
.../apache/drill/exec/store/AbstractSchema.java | 21 +-
.../drill/exec/store/SchemaTreeProvider.java | 32 +-
.../drill/exec/store/StorageStrategy.java | 194 +++++++++
.../drill/exec/store/SubSchemaWrapper.java | 6 +-
.../drill/exec/store/dfs/FileSelection.java | 5 +-
.../exec/store/dfs/FileSystemSchemaFactory.java | 10 +-
.../exec/store/dfs/WorkspaceSchemaFactory.java | 10 +-
.../drill/exec/store/dfs/easy/EasyWriter.java | 10 +-
.../exec/store/easy/json/JSONFormatPlugin.java | 4 +-
.../exec/store/easy/json/JsonRecordWriter.java | 24 +-
.../exec/store/easy/text/TextFormatPlugin.java | 4 +-
.../exec/store/parquet/ParquetRecordWriter.java | 64 ++-
.../drill/exec/store/parquet/ParquetWriter.java | 10 +-
.../exec/store/text/DrillTextRecordWriter.java | 31 +-
.../src/main/resources/drill-module.conf | 8 +-
.../java/org/apache/drill/BaseTestQuery.java | 3 +-
.../java/org/apache/drill/TestDropTable.java | 10 +-
.../user/TemporaryTablesAutomaticDropTest.java | 95 +++++
.../drill/exec/sql/TestBaseViewSupport.java | 4 +-
.../org/apache/drill/exec/sql/TestCTTAS.java | 422 +++++++++++++++++++
.../drill/exec/store/StorageStrategyTest.java | 222 ++++++++++
.../resources/bootstrap-storage-plugins.json | 2 +-
.../apache/drill/exec/rpc/RemoteConnection.java | 9 +-
.../java/org/apache/drill/exec/rpc/RpcBus.java | 8 +-
42 files changed, 1797 insertions(+), 210 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/distribution/src/resources/drill-override-example.conf
----------------------------------------------------------------------
diff --git a/distribution/src/resources/drill-override-example.conf b/distribution/src/resources/drill-override-example.conf
index f9d27b3..3baac5e 100644
--- a/distribution/src/resources/drill-override-example.conf
+++ b/distribution/src/resources/drill-override-example.conf
@@ -184,7 +184,13 @@ drill.exec: {
# Set this property if custom absolute root should be used for remote directories
root: "/app/drill"
}
- }
+ },
+ # Settings for Temporary Tables.
+ # See https://gist.github.com/arina-ielchiieva/50158175867a18eee964b5ba36455fbf#file-temporarytablessupport-md.
+ # Temporary table can be created ONLY in default temporary workspace.
+ # Full workspace name should be indicated (including schema and workspace separated by dot).
+ # Workspace MUST be file-based and writable. Workspace name is case-sensitive.
+ default_temporary_workspace: "dfs.tmp"
}
# Below SSL parameters need to be set for custom transport layer settings.
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/codegen/includes/parserImpls.ftl
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/codegen/includes/parserImpls.ftl b/exec/java-exec/src/main/codegen/includes/parserImpls.ftl
index 0017446..d9ceed9 100644
--- a/exec/java-exec/src/main/codegen/includes/parserImpls.ftl
+++ b/exec/java-exec/src/main/codegen/includes/parserImpls.ftl
@@ -214,8 +214,8 @@ SqlNode SqlDropView() :
}
/**
- * Parses a CTAS statement.
- * CREATE TABLE tblname [ (field1, field2, ...) ] AS select_statement.
+ * Parses a CTAS or CTTAS statement.
+ * CREATE [TEMPORARY] TABLE tblname [ (field1, field2, ...) ] AS select_statement.
*/
SqlNode SqlCreateTable() :
{
@@ -224,12 +224,14 @@ SqlNode SqlCreateTable() :
SqlNodeList fieldList;
SqlNodeList partitionFieldList;
SqlNode query;
+ boolean isTemporary = false;
}
{
{
partitionFieldList = SqlNodeList.EMPTY;
}
<CREATE> { pos = getPos(); }
+ ( <TEMPORARY> { isTemporary = true; } )?
<TABLE>
tblName = CompoundIdentifier()
fieldList = ParseOptionalFieldList("Table")
@@ -239,7 +241,8 @@ SqlNode SqlCreateTable() :
<AS>
query = OrderedQueryOrExpr(ExprContext.ACCEPT_QUERY)
{
- return new SqlCreateTable(pos, tblName, fieldList, partitionFieldList, query);
+ return new SqlCreateTable(pos,tblName, fieldList, partitionFieldList, query,
+ SqlLiteral.createBoolean(isTemporary, getPos()));
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
index 740eb4b..e8cc75c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -128,6 +128,11 @@ public interface ExecConstants {
*/
String DRILL_TMP_DIR = "drill.tmp-dir";
+ /**
+ * Temporary tables can be created ONLY in default temporary workspace.
+ */
+ String DEFAULT_TEMPORARY_WORKSPACE = "drill.exec.default_temporary_workspace";
+
String OUTPUT_FORMAT_OPTION = "store.format";
OptionValidator OUTPUT_FORMAT_VALIDATOR = new StringValidator(OUTPUT_FORMAT_OPTION, "parquet");
String PARQUET_BLOCK_SIZE = "store.parquet.block-size";
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractWriter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractWriter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractWriter.java
index af23d5f..6ba570b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractWriter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractWriter.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -17,7 +17,12 @@
*/
package org.apache.drill.exec.physical.base;
-public abstract class AbstractWriter extends AbstractSingle implements Writer{
+import org.apache.drill.exec.store.StorageStrategy;
+
+public abstract class AbstractWriter extends AbstractSingle implements Writer {
+
+ /** Storage strategy is used during table folder and files creation*/
+ private StorageStrategy storageStrategy;
public AbstractWriter(PhysicalOperator child) {
super(child);
@@ -27,4 +32,12 @@ public abstract class AbstractWriter extends AbstractSingle implements Writer{
public <T, X, E extends Throwable> T accept(PhysicalVisitor<T, X, E> physicalVisitor, X value) throws E {
return physicalVisitor.visitWriter(this, value);
}
+
+ public void setStorageStrategy(StorageStrategy storageStrategy) {
+ this.storageStrategy = storageStrategy;
+ }
+
+ public StorageStrategy getStorageStrategy() {
+ return storageStrategy;
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WriterRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WriterRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WriterRecordBatch.java
index e6c946c..939832b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WriterRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WriterRecordBatch.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -174,13 +174,25 @@ public class WriterRecordBatch extends AbstractRecordBatch<Writer> {
schema = container.getSchema();
}
+ /** Clean up needs to be performed before closing writer. Partially written data will be removed. */
private void closeWriter() {
- if (recordWriter != null) {
+ if (recordWriter == null) {
+ return;
+ }
+
+ try {
+ recordWriter.cleanup();
+ } catch(IOException ex) {
+ context.fail(ex);
+ } finally {
try {
- recordWriter.cleanup();
+ if (!processed) {
+ recordWriter.abort();
+ }
+ } catch (IOException e) {
+ logger.error("Abort failed. There could be leftover output files.", e);
+ } finally {
recordWriter = null;
- } catch(IOException ex) {
- context.fail(ex);
}
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/FileSystemCreateTableEntry.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/FileSystemCreateTableEntry.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/FileSystemCreateTableEntry.java
index 90eb05c..23ea23f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/FileSystemCreateTableEntry.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/FileSystemCreateTableEntry.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -22,9 +22,10 @@ import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.common.exceptions.UserException;
-import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.logical.FormatPluginConfig;
+import org.apache.drill.exec.physical.base.AbstractWriter;
import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.store.StorageStrategy;
import org.apache.drill.exec.physical.base.Writer;
import org.apache.drill.exec.store.StoragePluginRegistry;
import org.apache.drill.exec.store.dfs.FileSystemConfig;
@@ -34,7 +35,6 @@ import com.fasterxml.jackson.annotation.JacksonInject;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonTypeName;
-import org.apache.drill.exec.store.ischema.Records;
/**
* Implements <code>CreateTableEntry</code> interface to create new tables in FileSystem storage.
@@ -47,28 +47,33 @@ public class FileSystemCreateTableEntry implements CreateTableEntry {
private FormatPlugin formatPlugin;
private String location;
private final List<String> partitionColumns;
+ private final StorageStrategy storageStrategy;
@JsonCreator
public FileSystemCreateTableEntry(@JsonProperty("storageConfig") FileSystemConfig storageConfig,
@JsonProperty("formatConfig") FormatPluginConfig formatConfig,
@JsonProperty("location") String location,
@JsonProperty("partitionColumn") List<String> partitionColumns,
+ @JsonProperty("storageStrategy") StorageStrategy storageStrategy,
@JacksonInject StoragePluginRegistry engineRegistry)
throws ExecutionSetupException {
this.storageConfig = storageConfig;
this.formatPlugin = engineRegistry.getFormatPlugin(storageConfig, formatConfig);
this.location = location;
this.partitionColumns = partitionColumns;
+ this.storageStrategy = storageStrategy;
}
public FileSystemCreateTableEntry(FileSystemConfig storageConfig,
FormatPlugin formatPlugin,
String location,
- List<String> partitionColumns) {
+ List<String> partitionColumns,
+ StorageStrategy storageStrategy) {
this.storageConfig = storageConfig;
this.formatPlugin = formatPlugin;
this.location = location;
this.partitionColumns = partitionColumns;
+ this.storageStrategy = storageStrategy;
}
@JsonProperty("storageConfig")
@@ -89,11 +94,14 @@ public class FileSystemCreateTableEntry implements CreateTableEntry {
formatPlugin.getName())).build(logger);
}
- return formatPlugin.getWriter(child, location, partitionColumns);
+ AbstractWriter writer = formatPlugin.getWriter(child, location, partitionColumns);
+ writer.setStorageStrategy(storageStrategy);
+ return writer;
}
@Override
public List<String> getPartitionColumns() {
return partitionColumns;
}
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java
index 76529d4..0ad3944 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -58,12 +58,7 @@ public class DrillSqlWorker {
public static PhysicalPlan getPlan(QueryContext context, String sql, Pointer<String> textPlan)
throws ForemanSetupException {
- final SqlConverter parser = new SqlConverter(
- context.getPlannerSettings(),
- context.getNewDefaultSchema(),
- context.getDrillOperatorTable(),
- (UdfUtilities) context,
- context.getFunctionRegistry());
+ final SqlConverter parser = new SqlConverter(context);
injector.injectChecked(context.getExecutionControls(), "sql-parsing", ForemanSetupException.class);
final SqlNode sqlNode = parser.parse(sql);
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SchemaUtilites.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SchemaUtilites.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SchemaUtilites.java
index 085f808..20c92c7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SchemaUtilites.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SchemaUtilites.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -22,8 +22,12 @@ import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.tools.ValidationException;
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.exceptions.DrillRuntimeException;
import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.store.AbstractSchema;
+import org.apache.drill.exec.store.dfs.WorkspaceSchemaFactory;
import java.util.Collections;
import java.util.List;
@@ -116,6 +120,11 @@ public class SchemaUtilites {
return SCHEMA_PATH_JOINER.join(getSchemaPathAsList(schema));
}
+ /** Utility method to get the schema path for given list of schema path. */
+ public static String getSchemaPath(List<String> schemaPath) {
+ return SCHEMA_PATH_JOINER.join(schemaPath);
+ }
+
/** Utility method to get the schema path as list for given schema instance. */
public static List<String> getSchemaPathAsList(SchemaPlus schema) {
if (isRootSchema(schema)) {
@@ -177,4 +186,35 @@ public class SchemaUtilites {
return drillSchema;
}
+
+ /**
+ * Looks in schema tree for default temporary workspace instance.
+ * Makes sure that temporary workspace is mutable and file-based
+ * (instance of {@link WorkspaceSchemaFactory.WorkspaceSchema}).
+ *
+ * @param defaultSchema default schema
+ * @param config drill config
+ * @return default temporary workspace
+ */
+ public static AbstractSchema getTemporaryWorkspace(SchemaPlus defaultSchema, DrillConfig config) {
+ List<String> temporarySchemaPath = Lists.newArrayList(config.getString(ExecConstants.DEFAULT_TEMPORARY_WORKSPACE));
+ AbstractSchema temporarySchema = resolveToMutableDrillSchema(defaultSchema, temporarySchemaPath);
+ if (!(temporarySchema instanceof WorkspaceSchemaFactory.WorkspaceSchema)) {
+ DrillRuntimeException.format("Temporary workspace [%s] must be file-based, instance of " +
+ "WorkspaceSchemaFactory.WorkspaceSchema", temporarySchemaPath);
+ }
+ return temporarySchema;
+ }
+
+ /**
+ * Checks that passed schema path is the same as temporary workspace path.
+ * Check is case-sensitive.
+ *
+ * @param schemaPath schema path
+ * @param config drill config
+ * @return true is schema path corresponds to temporary workspace, false otherwise
+ */
+ public static boolean isTemporaryWorkspace(String schemaPath, DrillConfig config) {
+ return schemaPath.equals(config.getString(ExecConstants.DEFAULT_TEMPORARY_WORKSPACE));
+ }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SqlConverter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SqlConverter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SqlConverter.java
index 0c3c6a0..28196c1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SqlConverter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SqlConverter.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -21,10 +21,12 @@ import java.util.Arrays;
import java.util.List;
import java.util.Set;
+import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import org.apache.calcite.adapter.java.JavaTypeFactory;
import org.apache.calcite.avatica.util.Casing;
import org.apache.calcite.avatica.util.Quoting;
+import org.apache.calcite.jdbc.CalciteSchema;
import org.apache.calcite.jdbc.CalciteSchemaImpl;
import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
import org.apache.calcite.plan.ConventionTraitDef;
@@ -33,6 +35,7 @@ import org.apache.calcite.plan.RelOptCostFactory;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.plan.volcano.VolcanoPlanner;
import org.apache.calcite.prepare.CalciteCatalogReader;
+import org.apache.calcite.prepare.RelOptTableImpl;
import org.apache.calcite.rel.RelCollationTraitDef;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.type.RelDataType;
@@ -41,17 +44,14 @@ import org.apache.calcite.rel.type.RelDataTypeSystemImpl;
import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.schema.SchemaPlus;
-import org.apache.calcite.sql.SqlCall;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.SqlOperatorTable;
-import org.apache.calcite.sql.SqlSelect;
import org.apache.calcite.sql.parser.SqlParseException;
import org.apache.calcite.sql.parser.SqlParser;
import org.apache.calcite.sql.parser.SqlParserImplFactory;
import org.apache.calcite.sql.parser.SqlParserPos;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.sql.util.ChainedSqlOperatorTable;
-import org.apache.calcite.sql.validate.AggregatingSelectScope;
import org.apache.calcite.sql.validate.SqlConformance;
import org.apache.calcite.sql.validate.SqlValidatorCatalogReader;
import org.apache.calcite.sql.validate.SqlValidatorException;
@@ -61,18 +61,21 @@ import org.apache.calcite.sql2rel.RelDecorrelator;
import org.apache.calcite.sql2rel.SqlToRelConverter;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.exception.FunctionNotFoundException;
+import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
+import org.apache.drill.exec.ops.QueryContext;
import org.apache.drill.exec.ops.UdfUtilities;
import org.apache.drill.exec.planner.cost.DrillCostBase;
import org.apache.drill.exec.planner.logical.DrillConstExecutor;
import org.apache.drill.exec.planner.physical.DrillDistributionTraitDef;
import org.apache.drill.exec.planner.physical.PlannerSettings;
-import org.apache.drill.exec.planner.physical.PrelUtil;
import org.apache.drill.exec.planner.sql.parser.impl.DrillParserWithCompoundIdConverter;
import com.google.common.base.Joiner;
+import org.apache.drill.exec.rpc.user.UserSession;
/**
* Class responsible for managing parsing, validation and toRel conversion for sql statements.
@@ -86,7 +89,7 @@ public class SqlConverter {
private final SqlParser.Config parserConfig;
// Allow the default config to be modified using immutable configs
private SqlToRelConverter.Config sqlToRelConverterConfig;
- private final CalciteCatalogReader catalog;
+ private final DrillCalciteCatalogReader catalog;
private final PlannerSettings settings;
private final SchemaPlus rootSchema;
private final SchemaPlus defaultSchema;
@@ -96,35 +99,42 @@ public class SqlConverter {
private final boolean isInnerQuery;
private final UdfUtilities util;
private final FunctionImplementationRegistry functions;
+ private final String temporarySchema;
+ private final UserSession session;
+ private final DrillConfig drillConfig;
private String sql;
private VolcanoPlanner planner;
- public SqlConverter(PlannerSettings settings, SchemaPlus defaultSchema,
- final SqlOperatorTable operatorTable, UdfUtilities util, FunctionImplementationRegistry functions) {
- this.settings = settings;
- this.util = util;
- this.functions = functions;
+ public SqlConverter(QueryContext context) {
+ this.settings = context.getPlannerSettings();
+ this.util = (UdfUtilities) context;
+ this.functions = context.getFunctionRegistry();
this.parserConfig = new ParserConfig();
this.sqlToRelConverterConfig = new SqlToRelConverterConfig();
this.isInnerQuery = false;
this.typeFactory = new JavaTypeFactoryImpl(DRILL_TYPE_SYSTEM);
- this.defaultSchema = defaultSchema;
+ this.defaultSchema = context.getNewDefaultSchema();
this.rootSchema = rootSchema(defaultSchema);
- this.catalog = new CalciteCatalogReader(
+ this.temporarySchema = context.getConfig().getString(ExecConstants.DEFAULT_TEMPORARY_WORKSPACE);
+ this.session = context.getSession();
+ this.drillConfig = context.getConfig();
+ this.catalog = new DrillCalciteCatalogReader(
CalciteSchemaImpl.from(rootSchema),
parserConfig.caseSensitive(),
CalciteSchemaImpl.from(defaultSchema).path(null),
- typeFactory);
- this.opTab = new ChainedSqlOperatorTable(Arrays.asList(operatorTable, catalog));
+ typeFactory,
+ drillConfig,
+ session);
+ this.opTab = new ChainedSqlOperatorTable(Arrays.asList(context.getDrillOperatorTable(), catalog));
this.costFactory = (settings.useDefaultCosting()) ? null : new DrillCostBase.DrillCostFactory();
this.validator = new DrillValidator(opTab, catalog, typeFactory, SqlConformance.DEFAULT);
validator.setIdentifierExpansion(true);
}
private SqlConverter(SqlConverter parent, SchemaPlus defaultSchema, SchemaPlus rootSchema,
- CalciteCatalogReader catalog) {
+ DrillCalciteCatalogReader catalog) {
this.parserConfig = parent.parserConfig;
this.sqlToRelConverterConfig = parent.sqlToRelConverterConfig;
this.defaultSchema = defaultSchema;
@@ -139,6 +149,9 @@ public class SqlConverter {
this.opTab = parent.opTab;
this.planner = parent.planner;
this.validator = new DrillValidator(opTab, catalog, typeFactory, SqlConformance.DEFAULT);
+ this.temporarySchema = parent.temporarySchema;
+ this.session = parent.session;
+ this.drillConfig = parent.drillConfig;
validator.setIdentifierExpansion(true);
}
@@ -203,6 +216,11 @@ public class SqlConverter {
return defaultSchema;
}
+ /** Disallow temporary tables presence in sql statement (ex: in view definitions) */
+ public void disallowTemporaryTables() {
+ catalog.disallowTemporaryTables();
+ }
+
private class DrillValidator extends SqlValidatorImpl {
private final Set<SqlValidatorScope> identitySet = Sets.newIdentityHashSet();
@@ -272,26 +290,27 @@ public class SqlConverter {
public Expander() {
}
- public RelNode expandView(
- RelDataType rowType,
- String queryString,
- List<String> schemaPath) {
- SqlConverter parser = new SqlConverter(SqlConverter.this, defaultSchema, rootSchema,
- catalog.withSchemaPath(schemaPath));
+ public RelNode expandView(RelDataType rowType, String queryString, List<String> schemaPath) {
+ final DrillCalciteCatalogReader catalogReader = new DrillCalciteCatalogReader(
+ CalciteSchemaImpl.from(rootSchema),
+ parserConfig.caseSensitive(),
+ schemaPath,
+ typeFactory,
+ drillConfig,
+ session);
+ final SqlConverter parser = new SqlConverter(SqlConverter.this, defaultSchema, rootSchema, catalogReader);
return expandView(queryString, parser);
}
@Override
- public RelNode expandView(
- RelDataType rowType,
- String queryString,
- SchemaPlus rootSchema, // new root schema
- List<String> schemaPath) {
- final CalciteCatalogReader catalogReader = new CalciteCatalogReader(
- CalciteSchemaImpl.from(rootSchema),
+ public RelNode expandView(RelDataType rowType, String queryString, SchemaPlus rootSchema, List<String> schemaPath) {
+ final DrillCalciteCatalogReader catalogReader = new DrillCalciteCatalogReader(
+ CalciteSchemaImpl.from(rootSchema), // new root schema
parserConfig.caseSensitive(),
schemaPath,
- typeFactory);
+ typeFactory,
+ drillConfig,
+ session);
SchemaPlus schema = rootSchema;
for (String s : schemaPath) {
SchemaPlus newSchema = schema.getSubSchema(s);
@@ -447,4 +466,66 @@ public class SqlConverter {
return node;
}
}
+
+ /**
+ * Extension of {@link CalciteCatalogReader} to add ability to check for temporary tables first
+ * if schema is not indicated near table name during query parsing
+ * or indicated workspace is default temporary workspace.
+ */
+ private class DrillCalciteCatalogReader extends CalciteCatalogReader {
+
+ private final DrillConfig drillConfig;
+ private final UserSession session;
+ private boolean allowTemporaryTables;
+
+ DrillCalciteCatalogReader(CalciteSchema rootSchema,
+ boolean caseSensitive,
+ List<String> defaultSchema,
+ JavaTypeFactory typeFactory,
+ DrillConfig drillConfig,
+ UserSession session) {
+ super(rootSchema, caseSensitive, defaultSchema, typeFactory);
+ this.drillConfig = drillConfig;
+ this.session = session;
+ this.allowTemporaryTables = true;
+ }
+
+ /** Disallow temporary tables presence in sql statement (ex: in view definitions) */
+ public void disallowTemporaryTables() {
+ this.allowTemporaryTables = false;
+ }
+
+ /**
+ * If schema is not indicated (only one element in the list) or schema is default temporary workspace,
+ * we need to check among session temporary tables first in default temporary workspace.
+ * If temporary table is found and temporary tables usage is allowed, its table instance will be returned,
+ * otherwise search will be conducted in original workspace.
+ *
+ * @param names list of schema and table names, table name is always the last element
+ * @return table instance, null otherwise
+ * @throws UserException if temporary tables usage is disallowed
+ */
+ @Override
+ public RelOptTableImpl getTable(final List<String> names) {
+ RelOptTableImpl temporaryTable = null;
+ String schemaPath = SchemaUtilites.getSchemaPath(names.subList(0, names.size() - 1));
+ if (names.size() == 1 || SchemaUtilites.isTemporaryWorkspace(schemaPath, drillConfig)) {
+ String temporaryTableName = session.resolveTemporaryTableName(names.get(names.size() - 1));
+ if (temporaryTableName != null) {
+ List<String> temporaryNames = Lists.newArrayList(temporarySchema, temporaryTableName);
+ temporaryTable = super.getTable(temporaryNames);
+ }
+ }
+ if (temporaryTable != null) {
+ if (allowTemporaryTables) {
+ return temporaryTable;
+ }
+ throw UserException
+ .validationError()
+ .message("Temporary tables usage is disallowed. Used temporary table name: %s.", names)
+ .build(logger);
+ }
+ return super.getTable(names);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/CreateTableHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/CreateTableHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/CreateTableHandler.java
index b6ffde6..12c72c4 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/CreateTableHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/CreateTableHandler.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -31,13 +31,17 @@ import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.tools.RelConversionException;
import org.apache.calcite.tools.ValidationException;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.exceptions.UserException;
-import org.apache.drill.common.types.TypeProtos.MajorType;
+import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.rpc.user.UserSession;
+import org.apache.drill.exec.store.StorageStrategy;
import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.logical.DrillScreenRel;
import org.apache.drill.exec.planner.logical.DrillWriterRel;
@@ -67,43 +71,54 @@ public class CreateTableHandler extends DefaultSqlHandler {
@Override
public PhysicalPlan getPlan(SqlNode sqlNode) throws ValidationException, RelConversionException, IOException, ForemanSetupException {
SqlCreateTable sqlCreateTable = unwrap(sqlNode, SqlCreateTable.class);
- final String newTblName = sqlCreateTable.getName();
+ String originalTableName = sqlCreateTable.getName();
final ConvertedRelNode convertedRelNode = validateAndConvert(sqlCreateTable.getQuery());
final RelDataType validatedRowType = convertedRelNode.getValidatedRowType();
final RelNode queryRelNode = convertedRelNode.getConvertedNode();
-
final RelNode newTblRelNode =
SqlHandlerUtil.resolveNewTableRel(false, sqlCreateTable.getFieldNames(), validatedRowType, queryRelNode);
- final AbstractSchema drillSchema =
- SchemaUtilites.resolveToMutableDrillSchema(config.getConverter().getDefaultSchema(),
- sqlCreateTable.getSchemaPath());
- final String schemaPath = drillSchema.getFullSchemaName();
+ final DrillConfig drillConfig = context.getConfig();
+ final AbstractSchema drillSchema = resolveSchema(sqlCreateTable, config.getConverter().getDefaultSchema(), drillConfig);
- if (SqlHandlerUtil.getTableFromSchema(drillSchema, newTblName) != null) {
- throw UserException.validationError()
- .message("A table or view with given name [%s] already exists in schema [%s]", newTblName, schemaPath)
- .build(logger);
- }
+ checkDuplicatedObjectExistence(drillSchema, originalTableName, drillConfig, context.getSession());
- final RelNode newTblRelNodeWithPCol = SqlHandlerUtil.qualifyPartitionCol(newTblRelNode, sqlCreateTable.getPartitionColumns());
+ final RelNode newTblRelNodeWithPCol = SqlHandlerUtil.qualifyPartitionCol(newTblRelNode,
+ sqlCreateTable.getPartitionColumns());
log("Calcite", newTblRelNodeWithPCol, logger, null);
-
// Convert the query to Drill Logical plan and insert a writer operator on top.
- DrillRel drel = convertToDrel(newTblRelNodeWithPCol, drillSchema, newTblName, sqlCreateTable.getPartitionColumns(), newTblRelNode.getRowType());
+ StorageStrategy storageStrategy = sqlCreateTable.isTemporary() ?
+ StorageStrategy.TEMPORARY : StorageStrategy.PERSISTENT;
+
+ // If we are creating temporary table, initial table name will be replaced with generated table name.
+ // Generated table name is unique, UUID.randomUUID() is used for its generation.
+ // Original table name is stored in temporary tables cache, so it can be substituted to generated one during querying.
+ String newTableName = sqlCreateTable.isTemporary() ?
+ context.getSession().registerTemporaryTable(drillSchema, originalTableName) : originalTableName;
+
+ DrillRel drel = convertToDrel(newTblRelNodeWithPCol, drillSchema, newTableName,
+ sqlCreateTable.getPartitionColumns(), newTblRelNode.getRowType(), storageStrategy);
Prel prel = convertToPrel(drel, newTblRelNode.getRowType(), sqlCreateTable.getPartitionColumns());
logAndSetTextPlan("Drill Physical", prel, logger);
PhysicalOperator pop = convertToPop(prel);
PhysicalPlan plan = convertToPlan(pop);
log("Drill Plan", plan, logger);
+ String message = String.format("Creating %s table [%s].",
+ sqlCreateTable.isTemporary() ? "temporary" : "persistent", originalTableName);
+ logger.info(message);
return plan;
}
- private DrillRel convertToDrel(RelNode relNode, AbstractSchema schema, String tableName, List<String> partitionColumns, RelDataType queryRowType)
+ private DrillRel convertToDrel(RelNode relNode,
+ AbstractSchema schema,
+ String tableName,
+ List<String> partitionColumns,
+ RelDataType queryRowType,
+ StorageStrategy storageStrategy)
throws RelConversionException, SqlUnsupportedException {
final DrillRel convertedRelNode = convertToDrel(relNode);
@@ -114,7 +129,7 @@ public class CreateTableHandler extends DefaultSqlHandler {
final RelTraitSet traits = convertedRelNode.getCluster().traitSet().plus(DrillRel.DRILL_LOGICAL);
final DrillWriterRel writerRel = new DrillWriterRel(convertedRelNode.getCluster(),
- traits, topPreservedNameProj, schema.createNewTable(tableName, partitionColumns));
+ traits, topPreservedNameProj, schema.createNewTable(tableName, partitionColumns, storageStrategy));
return new DrillScreenRel(writerRel.getCluster(), writerRel.getTraitSet(), writerRel);
}
@@ -186,7 +201,7 @@ public class CreateTableHandler extends DefaultSqlHandler {
return (Prel) prel.copy(projectUnderWriter.getTraitSet(),
Collections.singletonList( (RelNode) projectUnderWriter));
} else {
- // find list of partiiton columns.
+ // find list of partition columns.
final List<RexNode> partitionColumnExprs = Lists.newArrayListWithExpectedSize(partitionColumns.size());
for (final String colName : partitionColumns) {
final RelDataTypeField field = childRowType.getField(colName, false, false);
@@ -242,4 +257,62 @@ public class CreateTableHandler extends DefaultSqlHandler {
return node;
}
+ /**
+ * Resolves schema taking into account type of table being created.
+ * If schema path wasn't indicated in sql call and table type to be created is temporary
+ * returns temporary workspace.
+ *
+ * If schema path is indicated, resolves to mutable drill schema.
+ * Though if table to be created is temporary table, checks if resolved schema is temporary,
+ * since temporary table are allowed to be created only in temporary workspace.
+ *
+ * @param sqlCreateTable create table call
+ * @param defaultSchema default schema
+ * @param config drill config
+ * @return resolved schema
+ * @throws UserException if attempted to create temporary table outside of temporary workspace
+ */
+ private AbstractSchema resolveSchema(SqlCreateTable sqlCreateTable, SchemaPlus defaultSchema, DrillConfig config) {
+ if (sqlCreateTable.isTemporary() && sqlCreateTable.getSchemaPath().size() == 0) {
+ return SchemaUtilites.getTemporaryWorkspace(defaultSchema, config);
+ } else {
+ AbstractSchema resolvedSchema = SchemaUtilites.resolveToMutableDrillSchema(defaultSchema, sqlCreateTable.getSchemaPath());
+ boolean isTemporaryWorkspace = SchemaUtilites.isTemporaryWorkspace(resolvedSchema.getFullSchemaName(), config);
+
+ if (sqlCreateTable.isTemporary() && !isTemporaryWorkspace) {
+ throw UserException
+ .validationError()
+ .message(String.format("Temporary tables are not allowed to be created " +
+ "outside of default temporary workspace [%s].", config.getString(ExecConstants.DEFAULT_TEMPORARY_WORKSPACE)))
+ .build(logger);
+ }
+ return resolvedSchema;
+ }
+ }
+
+ /**
+ * Checks if any object (persistent table / temporary table / view)
+ * with the same name as table to be created exists in indicated schema.
+ *
+ * @param drillSchema schema where table will be created
+ * @param tableName table name
+ * @param config drill config
+ * @param userSession current user session
+ * @throws UserException if duplicate is found
+ */
+ private void checkDuplicatedObjectExistence(AbstractSchema drillSchema,
+ String tableName,
+ DrillConfig config,
+ UserSession userSession) {
+ String schemaPath = drillSchema.getFullSchemaName();
+ boolean isTemporaryTable = userSession.isTemporaryTable(drillSchema, config, tableName);
+
+ if (isTemporaryTable || SqlHandlerUtil.getTableFromSchema(drillSchema, tableName) != null) {
+ throw UserException
+ .validationError()
+ .message("A table or view with given name [%s] already exists in schema [%s]",
+ tableName, schemaPath)
+ .build(logger);
+ }
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DropTableHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DropTableHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DropTableHandler.java
index 517c183..a9895db 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DropTableHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DropTableHandler.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -18,19 +18,21 @@
package org.apache.drill.exec.planner.sql.handlers;
import java.io.IOException;
+import java.util.List;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.schema.Table;
-import org.apache.calcite.sql.SqlIdentifier;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.tools.RelConversionException;
import org.apache.calcite.tools.ValidationException;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.planner.sql.DirectPlan;
import org.apache.drill.exec.planner.sql.SchemaUtilites;
import org.apache.drill.exec.planner.sql.parser.SqlDropTable;
+import org.apache.drill.exec.rpc.user.UserSession;
import org.apache.drill.exec.store.AbstractSchema;
// SqlHandler for dropping a table.
@@ -46,6 +48,7 @@ public class DropTableHandler extends DefaultSqlHandler {
* Function resolves the schema and invokes the drop method
* (while IF EXISTS statement is used function invokes the drop method only if table exists).
* Raises an exception if the schema is immutable.
+ *
* @param sqlNode - SqlDropTable (SQL parse tree of drop table [if exists] query)
* @return - Single row indicating drop succeeded or table is not found while IF EXISTS statement is used,
* raise exception otherwise
@@ -55,35 +58,51 @@ public class DropTableHandler extends DefaultSqlHandler {
*/
@Override
public PhysicalPlan getPlan(SqlNode sqlNode) throws ValidationException, RelConversionException, IOException {
-
SqlDropTable dropTableNode = ((SqlDropTable) sqlNode);
- SqlIdentifier tableIdentifier = dropTableNode.getTableIdentifier();
-
+ String originalTableName = dropTableNode.getName();
SchemaPlus defaultSchema = config.getConverter().getDefaultSchema();
- AbstractSchema drillSchema = null;
+ List<String> tableSchema = dropTableNode.getSchema();
+ DrillConfig drillConfig = context.getConfig();
+ UserSession session = context.getSession();
- if (tableIdentifier != null) {
- drillSchema = SchemaUtilites.resolveToMutableDrillSchema(defaultSchema, dropTableNode.getSchema());
- }
-
- String tableName = dropTableNode.getName();
- if (drillSchema == null) {
- throw UserException.validationError()
- .message("Invalid table_name [%s]", tableName)
- .build(logger);
- }
+ AbstractSchema temporarySchema = resolveToTemporarySchema(tableSchema, defaultSchema, drillConfig);
+ boolean isTemporaryTable = session.isTemporaryTable(temporarySchema, drillConfig, originalTableName);
- if (dropTableNode.checkTableExistence()) {
- final Table tableToDrop = SqlHandlerUtil.getTableFromSchema(drillSchema, tableName);
+ if (isTemporaryTable) {
+ session.removeTemporaryTable(temporarySchema, originalTableName);
+ } else {
+ AbstractSchema drillSchema = SchemaUtilites.resolveToMutableDrillSchema(defaultSchema, tableSchema);
+ Table tableToDrop = SqlHandlerUtil.getTableFromSchema(drillSchema, originalTableName);
if (tableToDrop == null || tableToDrop.getJdbcTableType() != Schema.TableType.TABLE) {
- return DirectPlan.createDirectPlan(context, true,
- String.format("Table [%s] not found", tableName));
+ if (dropTableNode.checkTableExistence()) {
+ return DirectPlan.createDirectPlan(context, false, String.format("Table [%s] not found", originalTableName));
+ } else {
+ throw UserException.validationError().message("Table [%s] not found", originalTableName).build(logger);
+ }
}
+ SqlHandlerUtil.dropTableFromSchema(drillSchema, originalTableName);
}
- drillSchema.dropTable(tableName);
+ String message = String.format("%s [%s] dropped", isTemporaryTable ? "Temporary table" : "Table", originalTableName);
+ logger.info(message);
+ return DirectPlan.createDirectPlan(context, true, message);
+ }
- return DirectPlan.createDirectPlan(context, true,
- String.format("Table [%s] %s", tableName, "dropped"));
+ /**
+ * If table schema is not indicated in sql call, returns temporary workspace.
+ * If schema is indicated, resolves to mutable table schema.
+ *
+ * @param tableSchema table schema
+ * @param defaultSchema default schema
+ * @param config drill config
+ * @return resolved schema
+ */
+ private AbstractSchema resolveToTemporarySchema(List<String> tableSchema, SchemaPlus defaultSchema, DrillConfig config) {
+ if (tableSchema.size() == 0) {
+ return SchemaUtilites.getTemporaryWorkspace(defaultSchema, config);
+ } else {
+ return SchemaUtilites.resolveToMutableDrillSchema(defaultSchema, tableSchema);
+ }
}
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SqlHandlerUtil.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SqlHandlerUtil.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SqlHandlerUtil.java
index ca7a510..04930a8 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SqlHandlerUtil.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SqlHandlerUtil.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -22,31 +22,24 @@ import com.google.common.collect.Sets;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.rex.RexInputRef;
-import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.schema.Table;
import org.apache.calcite.sql.SqlNodeList;
import org.apache.calcite.sql.SqlWriter;
-import org.apache.calcite.sql.TypedSqlNode;
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
-import org.apache.calcite.tools.Planner;
import org.apache.calcite.tools.RelConversionException;
-import org.apache.drill.common.exceptions.DrillException;
import org.apache.drill.common.exceptions.DrillRuntimeException;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.planner.StarColumnHelper;
import org.apache.drill.exec.planner.common.DrillRelOptUtil;
-import org.apache.drill.exec.planner.sql.DirectPlan;
-import org.apache.drill.exec.planner.types.DrillFixedRelDataTypeImpl;
import org.apache.drill.exec.store.AbstractSchema;
import org.apache.calcite.tools.ValidationException;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.sql.SqlNode;
-import org.apache.drill.exec.store.ischema.Records;
+import java.io.IOException;
import java.util.AbstractList;
import java.util.HashSet;
import java.util.List;
@@ -235,4 +228,42 @@ public class SqlHandlerUtil {
writer.keyword(")");
}
+ /**
+ * Drops table from schema.
+ * If drop has failed makes concurrency check: checks if table still exists.
+ * If table exists, throws {@link @UserException} since drop was unsuccessful,
+ * otherwise assumes that other user had dropped the view and exists without error.
+ *
+ * @param drillSchema drill schema
+ * @param tableName table name
+ */
+ public static void dropTableFromSchema(AbstractSchema drillSchema, String tableName) {
+ try {
+ drillSchema.dropTable(tableName);
+ } catch (Exception e) {
+ if (SqlHandlerUtil.getTableFromSchema(drillSchema, tableName) != null) {
+ throw e;
+ }
+ }
+ }
+
+ /**
+ * Drops view from schema.
+ * If drop has failed makes concurrency check: checks if view still exists.
+ * If view exists, throws {@link @UserException} since drop was unsuccessful,
+ * otherwise assumes that other user had dropped the view and exists without error.
+ *
+ * @param drillSchema drill schema
+ * @param viewName view name
+ */
+ public static void dropViewFromSchema(AbstractSchema drillSchema, String viewName) throws IOException {
+ try {
+ drillSchema.dropView(viewName);
+ } catch (Exception e) {
+ if (SqlHandlerUtil.getTableFromSchema(drillSchema, viewName) != null) {
+ throw e;
+ }
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ViewHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ViewHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ViewHandler.java
index b8396e6..495e8b5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ViewHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ViewHandler.java
@@ -62,9 +62,10 @@ public abstract class ViewHandler extends DefaultSqlHandler {
final String newViewName = createView.getName();
+ // Disallow temporary tables usage in view definition
+ config.getConverter().disallowTemporaryTables();
// Store the viewSql as view def SqlNode is modified as part of the resolving the new table definition below.
final String viewSql = createView.getQuery().toString();
-
final ConvertedRelNode convertedRelNode = validateAndConvert(createView.getQuery());
final RelDataType validatedRowType = convertedRelNode.getValidatedRowType();
final RelNode queryRelNode = convertedRelNode.getConvertedNode();
@@ -74,36 +75,50 @@ public abstract class ViewHandler extends DefaultSqlHandler {
final SchemaPlus defaultSchema = context.getNewDefaultSchema();
final AbstractSchema drillSchema = SchemaUtilites.resolveToMutableDrillSchema(defaultSchema, createView.getSchemaPath());
- final String schemaPath = drillSchema.getFullSchemaName();
final View view = new View(newViewName, viewSql, newViewRelNode.getRowType(),
SchemaUtilites.getSchemaPathAsList(defaultSchema));
- final Table existingTable = SqlHandlerUtil.getTableFromSchema(drillSchema, newViewName);
-
- if (existingTable != null) {
- if (existingTable.getJdbcTableType() != Schema.TableType.VIEW) {
- // existing table is not a view
- throw UserException.validationError()
- .message("A non-view table with given name [%s] already exists in schema [%s]",
- newViewName, schemaPath)
- .build(logger);
- }
-
- if (existingTable.getJdbcTableType() == Schema.TableType.VIEW && !createView.getReplace()) {
- // existing table is a view and create view has no "REPLACE" clause
- throw UserException.validationError()
- .message("A view with given name [%s] already exists in schema [%s]",
- newViewName, schemaPath)
- .build(logger);
- }
- }
+ validateViewCreationPossibility(drillSchema, createView, context);
final boolean replaced = drillSchema.createView(view);
final String summary = String.format("View '%s' %s successfully in '%s' schema",
- createView.getName(), replaced ? "replaced" : "created", schemaPath);
+ createView.getName(), replaced ? "replaced" : "created", drillSchema.getFullSchemaName());
return DirectPlan.createDirectPlan(context, true, summary);
}
+
+ /**
+ * Validates if view can be created in indicated schema:
+ * checks if object (persistent / temporary table) with the same name exists
+ * or if view with the same name exists but replace flag is not set.
+ *
+ * @param drillSchema schema where views will be created
+ * @param view create view call
+ * @param context query context
+ * @throws UserException if views can be created in indicated schema
+ */
+ private void validateViewCreationPossibility(AbstractSchema drillSchema, SqlCreateView view, QueryContext context) {
+ final String schemaPath = drillSchema.getFullSchemaName();
+ final String viewName = view.getName();
+ final Table existingTable = SqlHandlerUtil.getTableFromSchema(drillSchema, viewName);
+
+ if ((existingTable != null && existingTable.getJdbcTableType() != Schema.TableType.VIEW) ||
+ context.getSession().isTemporaryTable(drillSchema, context.getConfig(), viewName)) {
+ // existing table is not a view
+ throw UserException
+ .validationError()
+ .message("A non-view table with given name [%s] already exists in schema [%s]", viewName, schemaPath)
+ .build(logger);
+ }
+
+ if ((existingTable != null && existingTable.getJdbcTableType() == Schema.TableType.VIEW) && !view.getReplace()) {
+ // existing table is a view and create view has no "REPLACE" clause
+ throw UserException
+ .validationError()
+ .message("A view with given name [%s] already exists in schema [%s]", viewName, schemaPath)
+ .build(logger);
+ }
+ }
}
/** Handler for Drop View [If Exists] DDL command. */
@@ -124,7 +139,7 @@ public abstract class ViewHandler extends DefaultSqlHandler {
final Table viewToDrop = SqlHandlerUtil.getTableFromSchema(drillSchema, viewName);
if (dropView.checkViewExistence()) {
if (viewToDrop == null || viewToDrop.getJdbcTableType() != Schema.TableType.VIEW){
- return DirectPlan.createDirectPlan(context, true,
+ return DirectPlan.createDirectPlan(context, false,
String.format("View [%s] not found in schema [%s].", viewName, schemaPath));
}
} else {
@@ -139,7 +154,7 @@ public abstract class ViewHandler extends DefaultSqlHandler {
}
}
- drillSchema.dropView(viewName);
+ SqlHandlerUtil.dropViewFromSchema(drillSchema, viewName);
return DirectPlan.createDirectPlan(context, true,
String.format("View [%s] deleted successfully from schema [%s].", viewName, schemaPath));
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/CompoundIdentifierConverter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/CompoundIdentifierConverter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/CompoundIdentifierConverter.java
index 53e3cd5..db934e2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/CompoundIdentifierConverter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/CompoundIdentifierConverter.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -161,7 +161,7 @@ public class CompoundIdentifierConverter extends SqlShuttle {
//SqlNode offset,
//SqlNode fetch,
rules.put(SqlSelect.class, R(D, E, D, E, E, E, E, E, D, D));
- rules.put(SqlCreateTable.class, R(D, D, D, E));
+ rules.put(SqlCreateTable.class, R(D, D, D, E, D));
rules.put(SqlCreateView.class, R(D, E, E, D));
rules.put(SqlDescribeTable.class, R(D, D, E));
rules.put(SqlDropView.class, R(D, D));
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlCreateTable.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlCreateTable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlCreateTable.java
index 5835b10..bba60b2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlCreateTable.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlCreateTable.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -48,8 +48,13 @@ public class SqlCreateTable extends DrillSqlCall {
public static final SqlSpecialOperator OPERATOR = new SqlSpecialOperator("CREATE_TABLE", SqlKind.OTHER) {
@Override
public SqlCall createCall(SqlLiteral functionQualifier, SqlParserPos pos, SqlNode... operands) {
- Preconditions.checkArgument(operands.length == 4, "SqlCreateTable.createCall() has to get 4 operands!");
- return new SqlCreateTable(pos, (SqlIdentifier) operands[0], (SqlNodeList) operands[1], (SqlNodeList) operands[2], operands[3]);
+ Preconditions.checkArgument(operands.length == 5, "SqlCreateTable.createCall() has to get 5 operands!");
+ return new SqlCreateTable(pos,
+ (SqlIdentifier) operands[0],
+ (SqlNodeList) operands[1],
+ (SqlNodeList) operands[2],
+ operands[3],
+ (SqlLiteral) operands[4]);
}
};
@@ -57,13 +62,20 @@ public class SqlCreateTable extends DrillSqlCall {
private final SqlNodeList fieldList;
private final SqlNodeList partitionColumns;
private final SqlNode query;
-
- public SqlCreateTable(SqlParserPos pos, SqlIdentifier tblName, SqlNodeList fieldList, SqlNodeList partitionColumns, SqlNode query) {
+ private final SqlLiteral isTemporary;
+
+ public SqlCreateTable(SqlParserPos pos,
+ SqlIdentifier tblName,
+ SqlNodeList fieldList,
+ SqlNodeList partitionColumns,
+ SqlNode query,
+ SqlLiteral isTemporary) {
super(pos);
this.tblName = tblName;
this.fieldList = fieldList;
this.partitionColumns = partitionColumns;
this.query = query;
+ this.isTemporary = isTemporary;
}
@Override
@@ -78,12 +90,16 @@ public class SqlCreateTable extends DrillSqlCall {
ops.add(fieldList);
ops.add(partitionColumns);
ops.add(query);
+ ops.add(isTemporary);
return ops;
}
@Override
public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
writer.keyword("CREATE");
+ if (isTemporary.booleanValue()) {
+ writer.keyword("TEMPORARY");
+ }
writer.keyword("TABLE");
tblName.unparse(writer, leftPrec, rightPrec);
if (fieldList.size() > 0) {
@@ -142,4 +158,6 @@ public class SqlCreateTable extends DrillSqlCall {
public SqlNode getQuery() { return query; }
+ public boolean isTemporary() { return isTemporary.booleanValue(); }
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
index 4e17249..281b124 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -299,6 +299,17 @@ public class UserServer extends BasicServer<RpcType, UserClientConnectionImpl> {
public SocketAddress getRemoteAddress() {
return getChannel().remoteAddress();
}
+
+ @Override
+ public void closeSession() {
+ session.close();
+ }
+
+ @Override
+ public void close() {
+ closeSession();
+ super.close();
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java
index 3bf9051..c3639d2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -17,9 +17,14 @@
*/
package org.apache.drill.exec.rpc.user;
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.file.Paths;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.base.Preconditions;
@@ -27,9 +32,13 @@ import com.google.common.base.Strings;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
+import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.schema.Table;
import org.apache.calcite.tools.ValidationException;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.planner.sql.SchemaUtilites;
+import org.apache.drill.exec.planner.sql.handlers.SqlHandlerUtil;
import org.apache.drill.exec.proto.UserBitShared.UserCredentials;
import org.apache.drill.exec.proto.UserProtos.Property;
import org.apache.drill.exec.proto.UserProtos.UserProperties;
@@ -37,8 +46,14 @@ import org.apache.drill.exec.server.options.OptionManager;
import org.apache.drill.exec.server.options.SessionOptionManager;
import com.google.common.collect.Maps;
-
-public class UserSession {
+import org.apache.drill.exec.store.AbstractSchema;
+import org.apache.drill.exec.store.StorageStrategy;
+import org.apache.drill.exec.store.dfs.DrillFileSystem;
+import org.apache.drill.exec.store.dfs.WorkspaceSchemaFactory;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+public class UserSession implements Closeable {
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UserSession.class);
public static final String SCHEMA = "schema";
@@ -54,18 +69,43 @@ public class UserSession {
private Map<String, String> properties;
private OptionManager sessionOptions;
private final AtomicInteger queryCount;
+ private final String sessionId;
+
+ /** Stores list of temporary tables, key is original table name converted to lower case to achieve case-insensitivity,
+ * value is generated table name. **/
+ private final ConcurrentMap<String, String> temporaryTables;
+ /** Stores list of session temporary locations, key is path to location, value is file system associated with location. **/
+ private final ConcurrentMap<Path, FileSystem> temporaryLocations;
+
+ /** On session close deletes all session temporary locations recursively and clears temporary locations list. */
+ @Override
+ public void close() {
+ for (Map.Entry<Path, FileSystem> entry : temporaryLocations.entrySet()) {
+ Path path = entry.getKey();
+ FileSystem fs = entry.getValue();
+ try {
+ fs.delete(path, true);
+ logger.info("Deleted session temporary location [{}] from file system [{}]",
+ path.toUri().getPath(), fs.getUri());
+ } catch (Exception e) {
+ logger.warn("Error during session temporary location [{}] deletion from file system [{}]: [{}]",
+ path.toUri().getPath(), fs.getUri(), e.getMessage());
+ }
+ }
+ temporaryLocations.clear();
+ }
/**
* Implementations of this interface are allowed to increment queryCount.
* {@link org.apache.drill.exec.work.user.UserWorker} should have a member that implements the interface.
* No other core class should implement this interface. Test classes may implement (see ControlsInjectionUtil).
*/
- public static interface QueryCountIncrementer {
- public void increment(final UserSession session);
+ public interface QueryCountIncrementer {
+ void increment(final UserSession session);
}
public static class Builder {
- UserSession userSession;
+ private UserSession userSession;
public static Builder newBuilder() {
return new Builder();
@@ -115,6 +155,9 @@ public class UserSession {
private UserSession() {
queryCount = new AtomicInteger(0);
+ sessionId = UUID.randomUUID().toString();
+ temporaryTables = Maps.newConcurrentMap();
+ temporaryLocations = Maps.newConcurrentMap();
}
public boolean isSupportComplexTypes() {
@@ -197,7 +240,7 @@ public class UserSession {
/**
* Get default schema from current default schema path and given schema tree.
- * @param rootSchema
+ * @param rootSchema root schema
* @return A {@link org.apache.calcite.schema.SchemaPlus} object.
*/
public SchemaPlus getDefaultSchema(SchemaPlus rootSchema) {
@@ -207,18 +250,117 @@ public class UserSession {
return null;
}
- final SchemaPlus defaultSchema = SchemaUtilites.findSchema(rootSchema, defaultSchemaPath);
+ return SchemaUtilites.findSchema(rootSchema, defaultSchemaPath);
+ }
+
+ public boolean setSessionOption(String name, String value) {
+ return true;
+ }
+
+ /**
+ * @return unique session identifier
+ */
+ public String getSessionId() { return sessionId; }
+
+ /**
+ * Creates and adds session temporary location if absent using schema configuration.
+ * Generates temporary table name and stores it's original name as key
+ * and generated name as value in session temporary tables cache.
+ * Original temporary name is converted to lower case to achieve case-insensitivity.
+ * If original table name already exists, new name is not regenerated and is reused.
+ * This can happen if default temporary workspace was changed (file system or location) or
+ * orphan temporary table name has remained (name was registered but table creation did not succeed).
+ *
+ * @param schema table schema
+ * @param tableName original table name
+ * @return generated temporary table name
+ * @throws IOException if error during session temporary location creation
+ */
+ public String registerTemporaryTable(AbstractSchema schema, String tableName) throws IOException {
+ addTemporaryLocation((WorkspaceSchemaFactory.WorkspaceSchema) schema);
+ String temporaryTableName = Paths.get(sessionId, UUID.randomUUID().toString()).toString();
+ String oldTemporaryTableName = temporaryTables.putIfAbsent(tableName.toLowerCase(), temporaryTableName);
+ return oldTemporaryTableName == null ? temporaryTableName : oldTemporaryTableName;
+ }
+
+ /**
+ * Returns generated temporary table name from the list of session temporary tables, null otherwise.
+ * Original temporary name is converted to lower case to achieve case-insensitivity.
+ *
+ * @param tableName original table name
+ * @return generated temporary table name
+ */
+ public String resolveTemporaryTableName(String tableName) {
+ return temporaryTables.get(tableName.toLowerCase());
+ }
- if (defaultSchema == null) {
- // If the current schema resolves to null, return root schema as the current default schema.
- return defaultSchema;
+ /**
+ * Checks if passed table is temporary, table name is case-insensitive.
+ * Before looking for table checks if passed schema is temporary and returns false if not
+ * since temporary tables are allowed to be created in temporary workspace only.
+ * If passed workspace is temporary, looks for temporary table.
+ * First checks if table name is among temporary tables, if not returns false.
+ * If temporary table named was resolved, checks that temporary table exists on disk,
+ * to ensure that temporary table actually exists and resolved table name is not orphan
+ * (for example, in result of unsuccessful temporary table creation).
+ *
+ * @param drillSchema table schema
+ * @param config drill config
+ * @param tableName original table name
+ * @return true if temporary table exists in schema, false otherwise
+ */
+ public boolean isTemporaryTable(AbstractSchema drillSchema, DrillConfig config, String tableName) {
+ if (!SchemaUtilites.isTemporaryWorkspace(drillSchema.getFullSchemaName(), config)) {
+ return false;
}
+ String temporaryTableName = resolveTemporaryTableName(tableName);
+ if (temporaryTableName != null) {
+ Table temporaryTable = SqlHandlerUtil.getTableFromSchema(drillSchema, temporaryTableName);
+ if (temporaryTable != null && temporaryTable.getJdbcTableType() == Schema.TableType.TABLE) {
+ return true;
+ }
+ }
+ return false;
+ }
- return defaultSchema;
+ /**
+ * Removes temporary table name from the list of session temporary tables.
+ * Original temporary name is converted to lower case to achieve case-insensitivity.
+ *
+ * @param tableName original table name
+ */
+ public void removeTemporaryTable(AbstractSchema drillSchema, String tableName) {
+ String temporaryTable = resolveTemporaryTableName(tableName);
+ if (temporaryTable == null) {
+ return;
+ }
+ SqlHandlerUtil.dropTableFromSchema(drillSchema, temporaryTable);
+ temporaryTables.remove(tableName.toLowerCase());
}
- public boolean setSessionOption(String name, String value) {
- return true;
+ /**
+ * Session temporary tables are stored under temporary workspace location in session folder
+ * defined by unique session id. These session temporary locations are deleted on session close.
+ * If default temporary workspace file system or location is changed at runtime,
+ * new session temporary location will be added with corresponding file system
+ * to the list of session temporary locations. If location does not exist it will be created and
+ * {@link StorageStrategy#TEMPORARY} storage rules will be applied to it.
+ *
+ * @param temporaryWorkspace temporary workspace
+ * @throws IOException in case of error during temporary location creation
+ */
+ private void addTemporaryLocation(WorkspaceSchemaFactory.WorkspaceSchema temporaryWorkspace) throws IOException {
+ DrillFileSystem fs = temporaryWorkspace.getFS();
+ Path temporaryLocation = new Path(Paths.get(fs.getUri().toString(),
+ temporaryWorkspace.getDefaultLocation(), sessionId).toString());
+
+ FileSystem fileSystem = temporaryLocations.putIfAbsent(temporaryLocation, fs);
+
+ if (fileSystem == null) {
+ StorageStrategy.TEMPORARY.createPathAndApply(fs, temporaryLocation);
+ Preconditions.checkArgument(fs.exists(temporaryLocation),
+ String.format("Temporary location should exist [%s]", temporaryLocation.toUri().getPath()));
+ }
}
private String getProp(String key) {
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
index 547915e..25776ad 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -20,6 +20,7 @@ package org.apache.drill.exec.server;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.calcite.schema.SchemaPlus;
import org.apache.drill.common.AutoCloseables;
import org.apache.drill.common.StackTrace;
import org.apache.drill.common.config.DrillConfig;
@@ -30,13 +31,17 @@ import org.apache.drill.exec.coord.ClusterCoordinator;
import org.apache.drill.exec.coord.ClusterCoordinator.RegistrationHandle;
import org.apache.drill.exec.coord.zk.ZKClusterCoordinator;
import org.apache.drill.exec.exception.DrillbitStartupException;
+import org.apache.drill.exec.planner.sql.SchemaUtilites;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
import org.apache.drill.exec.server.options.OptionManager;
import org.apache.drill.exec.server.options.OptionValue;
import org.apache.drill.exec.server.options.OptionValue.OptionType;
import org.apache.drill.exec.server.rest.WebServer;
import org.apache.drill.exec.service.ServiceEngine;
+import org.apache.drill.exec.store.AbstractSchema;
+import org.apache.drill.exec.store.SchemaTreeProvider;
import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.dfs.WorkspaceSchemaFactory;
import org.apache.drill.exec.store.sys.store.provider.CachingPersistentStoreProvider;
import org.apache.drill.exec.store.sys.PersistentStoreProvider;
import org.apache.drill.exec.store.sys.PersistentStoreRegistry;
@@ -123,6 +128,7 @@ public class Drillbit implements AutoCloseable {
storageRegistry.init();
drillbitContext.getOptionManager().init();
javaPropertiesToSystemOptions();
+ validateTemporaryWorkspace(manager.getContext());
manager.getContext().getRemoteFunctionRegistry().init(context.getConfig(), storeProvider, coord);
registrationHandle = coord.register(md);
webServer.start();
@@ -215,6 +221,21 @@ public class Drillbit implements AutoCloseable {
}
/**
+ * Validates that temporary workspace indicated in configuration is
+ * mutable and file-based (instance of {@link WorkspaceSchemaFactory.WorkspaceSchema}).
+ *
+ * @param context drillbit context
+ * @throws Exception in case when temporary table schema is not mutable or
+ * not file-based (instance of WorkspaceSchemaFactory.WorkspaceSchema)
+ */
+ private void validateTemporaryWorkspace(DrillbitContext context) throws Exception {
+ try (SchemaTreeProvider schemaTreeProvider = new SchemaTreeProvider(context)) {
+ final SchemaPlus rootSchema = schemaTreeProvider.createRootSchema(context.getOptionManager());
+ SchemaUtilites.getTemporaryWorkspace(rootSchema, context.getConfig());
+ }
+ }
+
+ /**
* Shutdown hook for Drillbit. Closes the drillbit, and reports on errors that
* occur during closure, as well as the location the drillbit was started from.
*/
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/store/AbstractSchema.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/AbstractSchema.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/AbstractSchema.java
index 7a16d0a..618841b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/AbstractSchema.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/AbstractSchema.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -117,18 +117,33 @@ public abstract class AbstractSchema implements Schema, SchemaPartitionExplorer,
}
/**
+ * Creates table entry using table name, list of partition columns
+ * and storage strategy used to create table folder and files
*
* @param tableName : new table name.
* @param partitionColumns : list of partition columns. Empty list if there is no partition columns.
- * @return
+ * @param storageStrategy : storage strategy used to create table folder and files
+ * @return create table entry
*/
- public CreateTableEntry createNewTable(String tableName, List<String> partitionColumns) {
+ public CreateTableEntry createNewTable(String tableName, List<String> partitionColumns, StorageStrategy storageStrategy) {
throw UserException.unsupportedError()
.message("Creating new tables is not supported in schema [%s]", getSchemaPath())
.build(logger);
}
/**
+ * Creates table entry using table name and list of partition columns if any.
+ * Table folder and files will be created using persistent storage strategy.
+ *
+ * @param tableName : new table name.
+ * @param partitionColumns : list of partition columns. Empty list if there is no partition columns.
+ * @return create table entry
+ */
+ public CreateTableEntry createNewTable(String tableName, List<String> partitionColumns) {
+ return createNewTable(tableName, partitionColumns, StorageStrategy.PERSISTENT);
+ }
+
+ /**
* Reports whether to show items from this schema in INFORMATION_SCHEMA
* tables.
* (Controls ... TODO: Doc.: Mention what this typically controls or
http://git-wip-us.apache.org/repos/asf/drill/blob/bb29f19f/exec/java-exec/src/main/java/org/apache/drill/exec/store/SchemaTreeProvider.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/SchemaTreeProvider.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/SchemaTreeProvider.java
index d05cc43..4f426bb 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/SchemaTreeProvider.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/SchemaTreeProvider.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -22,7 +22,10 @@ import org.apache.calcite.schema.SchemaPlus;
import org.apache.drill.common.AutoCloseables;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.ops.ViewExpansionContext;
import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.exec.server.options.OptionValue;
import org.apache.drill.exec.store.SchemaConfig.SchemaConfigInfoProvider;
import org.apache.drill.exec.util.ImpersonationUtil;
@@ -49,6 +52,33 @@ public class SchemaTreeProvider implements AutoCloseable {
}
/**
+ * Return root schema for process user.
+ *
+ * @param options list of options
+ * @return root of the schema tree
+ */
+ public SchemaPlus createRootSchema(final OptionManager options) {
+ SchemaConfigInfoProvider schemaConfigInfoProvider = new SchemaConfigInfoProvider() {
+
+ @Override
+ public ViewExpansionContext getViewExpansionContext() {
+ throw new UnsupportedOperationException("View expansion context is not supported");
+ }
+
+ @Override
+ public OptionValue getOption(String optionKey) {
+ return options.getOption(optionKey);
+ }
+ };
+
+ final SchemaConfig schemaConfig = SchemaConfig.newBuilder(
+ ImpersonationUtil.getProcessUserName(), schemaConfigInfoProvider)
+ .build();
+
+ return createRootSchema(schemaConfig);
+ }
+
+ /**
* Return root schema with schema owner as the given user.
*
* @param userName Name of the user who is accessing the storage sources.