You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ar...@apache.org on 2018/07/25 19:28:21 UTC
[10/10] impala git commit: IMPALA-7277. Support INSERT and LOAD DATA
statements in LocalCatalog
IMPALA-7277. Support INSERT and LOAD DATA statements in LocalCatalog
This adds support for INSERT and LOAD DATA statements when LocalCatalog
is enabled by fixing the following items:
* Remove some downcasts to HdfsTable in various Stmt classes and replace
them with casts to FeFsTable.
* Stub out the "write access" checks to fake always being writable. Left
a TODO to properly implement these.
* Implemented various 'getPartition()' calls which take a user-provided
partition specification, used by INSERT and LOAD statements that
specify an explicit target partition.
* Fixed the "prototype partition" created by LocalFsTable to not include
a location field. This fixed insertion into dynamic partitions.
Additionally fixed a couple other issues which were exercised by the
e2e test coverage for load/insert:
* The LocalCatalog getDb() and getTable() calls were previously assuming
that all callers passed pre-canonicalized table names, but that wasn't
the case. This adds the necessary toLower() calls so that statements
referencing capitalized table names work.
With this patch, most of the associated e2e tests pass, with the
exception of those that check permissions levels of the target
directories. Those calls are still stubbed out.
Overall, running 'run-tests.py -k "not hbase and not avro"' results in
about a 90% pass rate after this patch:
=====================================================
186 failed, 1657 passed, 56 skipped, 33 xfailed, 1 xpassed in 512.75 seconds
=====================================================
Remaining test failures seem mostly unrelated to the above changes and
will be addressed in continuing patches.
Change-Id: I4ae47a0b58022ed77abe51d2596c2b1d0111fae3
Reviewed-on: http://gerrit.cloudera.org:8080/10914
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/cec33fa0
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/cec33fa0
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/cec33fa0
Branch: refs/heads/master
Commit: cec33fa0ae75392668273d40b5a1bc4bbd7e9e2e
Parents: ba81386
Author: Todd Lipcon <to...@cloudera.com>
Authored: Tue Jul 10 16:24:41 2018 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Wed Jul 25 19:27:26 2018 +0000
----------------------------------------------------------------------
.../AlterTableRecoverPartitionsStmt.java | 6 ++--
.../analysis/AlterTableSetCachedStmt.java | 7 ++---
.../analysis/AlterTableSetLocationStmt.java | 6 ++--
.../analysis/AlterTableSetRowFormatStmt.java | 6 ++--
.../apache/impala/analysis/BaseTableRef.java | 9 +++---
.../impala/analysis/ComputeStatsStmt.java | 7 ++---
.../org/apache/impala/analysis/InsertStmt.java | 20 ++++++------
.../apache/impala/analysis/LoadDataStmt.java | 21 +++++++------
.../apache/impala/analysis/PartitionSpec.java | 4 +--
.../impala/analysis/PartitionSpecBase.java | 8 ++---
.../org/apache/impala/analysis/TableRef.java | 6 ++--
.../org/apache/impala/analysis/ToSqlUtils.java | 4 +--
.../apache/impala/analysis/TupleDescriptor.java | 6 ++--
.../org/apache/impala/catalog/FeFsTable.java | 17 +++++++++++
.../org/apache/impala/catalog/HdfsTable.java | 29 +++++++++++++-----
.../impala/catalog/local/LocalCatalog.java | 32 +++++++++++++++++---
.../apache/impala/catalog/local/LocalDb.java | 2 +-
.../impala/catalog/local/LocalFsPartition.java | 7 +++--
.../impala/catalog/local/LocalFsTable.java | 22 +++++++++++++-
19 files changed, 148 insertions(+), 71 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/analysis/AlterTableRecoverPartitionsStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/AlterTableRecoverPartitionsStmt.java b/fe/src/main/java/org/apache/impala/analysis/AlterTableRecoverPartitionsStmt.java
index efb8a70..d04f042 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AlterTableRecoverPartitionsStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AlterTableRecoverPartitionsStmt.java
@@ -17,7 +17,7 @@
package org.apache.impala.analysis;
-import org.apache.impala.catalog.HdfsTable;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.thrift.TAlterTableParams;
import org.apache.impala.thrift.TAlterTableType;
@@ -42,8 +42,8 @@ public class AlterTableRecoverPartitionsStmt extends AlterTableStmt {
public void analyze(Analyzer analyzer) throws AnalysisException {
super.analyze(analyzer);
- // Make sure the target table is HdfsTable.
- if (!(table_ instanceof HdfsTable)) {
+ // Make sure the target table is an FS-backed Table.
+ if (!(table_ instanceof FeFsTable)) {
throw new AnalysisException("ALTER TABLE RECOVER PARTITIONS " +
"must target an HDFS table: " + tableName_);
}
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/analysis/AlterTableSetCachedStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/AlterTableSetCachedStmt.java b/fe/src/main/java/org/apache/impala/analysis/AlterTableSetCachedStmt.java
index 178725d..0bcd2bf 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AlterTableSetCachedStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AlterTableSetCachedStmt.java
@@ -19,10 +19,9 @@ package org.apache.impala.analysis;
import java.util.List;
-import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.catalog.FeFsPartition;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeTable;
-import org.apache.impala.catalog.HdfsPartition;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.thrift.TAlterTableParams;
import org.apache.impala.thrift.TAlterTableSetCachedParams;
@@ -63,7 +62,7 @@ public class AlterTableSetCachedStmt extends AlterTableSetStmt {
FeTable table = getTargetTable();
Preconditions.checkNotNull(table);
- if (!(table instanceof HdfsTable)) {
+ if (!(table instanceof FeFsTable)) {
throw new AnalysisException("ALTER TABLE SET [CACHED|UNCACHED] must target an " +
"HDFS table: " + table.getFullName());
}
@@ -71,7 +70,7 @@ public class AlterTableSetCachedStmt extends AlterTableSetStmt {
if (cacheOp_.shouldCache()) {
boolean isCacheable = true;
PartitionSet partitionSet = getPartitionSet();
- HdfsTable hdfsTable = (HdfsTable)table;
+ FeFsTable hdfsTable = (FeFsTable)table;
StringBuilder nameSb = new StringBuilder();
if (partitionSet != null) {
List<? extends FeFsPartition> parts = partitionSet.getPartitions();
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/analysis/AlterTableSetLocationStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/AlterTableSetLocationStmt.java b/fe/src/main/java/org/apache/impala/analysis/AlterTableSetLocationStmt.java
index b034ab5..cb46493 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AlterTableSetLocationStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AlterTableSetLocationStmt.java
@@ -23,10 +23,10 @@ import java.util.List;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.impala.authorization.Privilege;
import org.apache.impala.catalog.FeFsPartition;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.HdfsPartition;
-import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.thrift.TAlterTableParams;
import org.apache.impala.thrift.TAlterTableSetLocationParams;
@@ -77,8 +77,8 @@ public class AlterTableSetLocationStmt extends AlterTableSetStmt {
FeTable table = getTargetTable();
Preconditions.checkNotNull(table);
- if (table instanceof HdfsTable) {
- HdfsTable hdfsTable = (HdfsTable) table;
+ if (table instanceof FeFsTable) {
+ FeFsTable hdfsTable = (FeFsTable) table;
if (getPartitionSet() != null) {
// Targeting a partition rather than a table.
List<? extends FeFsPartition> partitions = getPartitionSet().getPartitions();
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/analysis/AlterTableSetRowFormatStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/AlterTableSetRowFormatStmt.java b/fe/src/main/java/org/apache/impala/analysis/AlterTableSetRowFormatStmt.java
index cdc71b3..a8c9fea 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AlterTableSetRowFormatStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AlterTableSetRowFormatStmt.java
@@ -18,9 +18,9 @@
package org.apache.impala.analysis;
import org.apache.impala.catalog.FeFsPartition;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.HdfsFileFormat;
-import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.catalog.RowFormat;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.thrift.TAlterTableParams;
@@ -58,7 +58,7 @@ public class AlterTableSetRowFormatStmt extends AlterTableSetStmt {
public void analyze(Analyzer analyzer) throws AnalysisException {
super.analyze(analyzer);
FeTable tbl = getTargetTable();
- if (!(tbl instanceof HdfsTable)) {
+ if (!(tbl instanceof FeFsTable)) {
throw new AnalysisException(String.format("ALTER TABLE SET ROW FORMAT is only " +
"supported on HDFS tables. Conflicting table: %1$s", tbl.getFullName()));
}
@@ -75,7 +75,7 @@ public class AlterTableSetRowFormatStmt extends AlterTableSetStmt {
}
} else {
HdfsFileFormat format = HdfsFileFormat.fromHdfsInputFormatClass(
- ((HdfsTable) tbl).getMetaStoreTable().getSd().getInputFormat());
+ ((FeFsTable) tbl).getMetaStoreTable().getSd().getInputFormat());
if (format != HdfsFileFormat.TEXT &&
format != HdfsFileFormat.SEQUENCE_FILE) {
throw new AnalysisException(String.format("ALTER TABLE SET ROW FORMAT is " +
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/analysis/BaseTableRef.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/BaseTableRef.java b/fe/src/main/java/org/apache/impala/analysis/BaseTableRef.java
index 3fbc612..5b80712 100644
--- a/fe/src/main/java/org/apache/impala/analysis/BaseTableRef.java
+++ b/fe/src/main/java/org/apache/impala/analysis/BaseTableRef.java
@@ -17,9 +17,10 @@
package org.apache.impala.analysis;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeTable;
-import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.common.AnalysisException;
+
import com.google.common.base.Preconditions;
/**
@@ -90,11 +91,11 @@ public class BaseTableRef extends TableRef {
*/
private void analyzeSkipHeaderLineCount() throws AnalysisException {
FeTable table = getTable();
- if (!(table instanceof HdfsTable)) return;
- HdfsTable hdfsTable = (HdfsTable)table;
+ if (!(table instanceof FeFsTable)) return;
+ FeFsTable fsTable = (FeFsTable)table;
StringBuilder error = new StringBuilder();
- hdfsTable.parseSkipHeaderLineCount(error);
+ fsTable.parseSkipHeaderLineCount(error);
if (error.length() > 0) throw new AnalysisException(error.toString());
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java b/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
index 4f1173e..9ce7b0d 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
@@ -33,7 +33,6 @@ import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.HBaseTable;
import org.apache.impala.catalog.HdfsFileFormat;
-import org.apache.impala.catalog.HdfsPartition;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.catalog.Type;
@@ -496,8 +495,8 @@ public class ComputeStatsStmt extends StatementBase {
} else {
// Not computing incremental stats.
expectAllPartitions_ = true;
- if (table_ instanceof HdfsTable) {
- expectAllPartitions_ = !((HdfsTable) table_).isStatsExtrapolationEnabled();
+ if (table_ instanceof FeFsTable) {
+ expectAllPartitions_ = !((FeFsTable) table_).isStatsExtrapolationEnabled();
}
}
@@ -623,7 +622,7 @@ public class ComputeStatsStmt extends StatementBase {
}
// Compute effective sampling percent.
- long totalFileBytes = ((HdfsTable)table_).getTotalHdfsBytes();
+ long totalFileBytes = ((FeFsTable)table_).getTotalHdfsBytes();
if (totalFileBytes > 0) {
effectiveSamplePerc_ = (double) sampleFileBytes / (double) totalFileBytes;
} else {
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java b/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java
index 72ec828..55da237 100644
--- a/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java
@@ -26,11 +26,11 @@ import java.util.Set;
import org.apache.impala.authorization.Privilege;
import org.apache.impala.authorization.PrivilegeRequestBuilder;
import org.apache.impala.catalog.Column;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.FeView;
import org.apache.impala.catalog.HBaseTable;
-import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.catalog.KuduColumn;
import org.apache.impala.catalog.Type;
import org.apache.impala.catalog.View;
@@ -474,28 +474,28 @@ public class InsertStmt extends StatementBase {
}
}
- if (table_ instanceof HdfsTable) {
- HdfsTable hdfsTable = (HdfsTable) table_;
- if (!hdfsTable.hasWriteAccess()) {
+ if (table_ instanceof FeFsTable) {
+ FeFsTable fsTable = (FeFsTable) table_;
+ if (!fsTable.hasWriteAccess()) {
throw new AnalysisException(String.format("Unable to INSERT into target table " +
"(%s) because Impala does not have WRITE access to at least one HDFS path" +
- ": %s", targetTableName_, hdfsTable.getFirstLocationWithoutWriteAccess()));
+ ": %s", targetTableName_, fsTable.getFirstLocationWithoutWriteAccess()));
}
StringBuilder error = new StringBuilder();
- hdfsTable.parseSkipHeaderLineCount(error);
+ fsTable.parseSkipHeaderLineCount(error);
if (error.length() > 0) throw new AnalysisException(error.toString());
try {
- if (!FileSystemUtil.isImpalaWritableFilesystem(hdfsTable.getLocation())) {
+ if (!FileSystemUtil.isImpalaWritableFilesystem(fsTable.getLocation())) {
throw new AnalysisException(String.format("Unable to INSERT into target " +
"table (%s) because %s is not a supported filesystem.", targetTableName_,
- hdfsTable.getLocation()));
+ fsTable.getLocation()));
}
} catch (IOException e) {
throw new AnalysisException(String.format("Unable to INSERT into target " +
"table (%s): %s.", targetTableName_, e.getMessage()), e);
}
for (int colIdx = 0; colIdx < numClusteringCols; ++colIdx) {
- Column col = hdfsTable.getColumns().get(colIdx);
+ Column col = fsTable.getColumns().get(colIdx);
// Hive 1.x has a number of issues handling BOOLEAN partition columns (see HIVE-6590).
// Instead of working around the Hive bugs, INSERT is disabled for BOOLEAN
// partitions in Impala when built against Hive 1. HIVE-6590 is currently resolved,
@@ -798,7 +798,7 @@ public class InsertStmt extends StatementBase {
* an AnalysisException.
*/
private void analyzeSortColumns() throws AnalysisException {
- if (!(table_ instanceof HdfsTable)) return;
+ if (!(table_ instanceof FeFsTable)) return;
sortColumns_ = AlterTableSetTblProperties.analyzeSortColumns(table_,
table_.getMetaStoreTable().getParameters());
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/analysis/LoadDataStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/LoadDataStmt.java b/fe/src/main/java/org/apache/impala/analysis/LoadDataStmt.java
index 43ca216..6732f1f 100644
--- a/fe/src/main/java/org/apache/impala/analysis/LoadDataStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/LoadDataStmt.java
@@ -28,8 +28,10 @@ import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.s3a.S3AFileSystem;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.impala.authorization.Privilege;
+import org.apache.impala.catalog.FeCatalogUtils;
+import org.apache.impala.catalog.FeFsPartition;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeTable;
-import org.apache.impala.catalog.HdfsPartition;
import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.common.FileSystemUtil;
@@ -105,7 +107,7 @@ public class LoadDataStmt extends StatementBase {
public void analyze(Analyzer analyzer) throws AnalysisException {
dbName_ = analyzer.getTargetDbName(tableName_);
FeTable table = analyzer.getTable(tableName_, Privilege.INSERT);
- if (!(table instanceof HdfsTable)) {
+ if (!(table instanceof FeFsTable)) {
throw new AnalysisException("LOAD DATA only supported for HDFS tables: " +
dbName_ + "." + getTbl());
}
@@ -122,7 +124,7 @@ public class LoadDataStmt extends StatementBase {
"specified: " + dbName_ + "." + getTbl());
}
}
- analyzePaths(analyzer, (HdfsTable) table);
+ analyzePaths(analyzer, (FeFsTable) table);
}
/**
@@ -134,7 +136,7 @@ public class LoadDataStmt extends StatementBase {
* We don't check permissions for the S3AFileSystem and the AdlFileSystem due to
* limitations with thier getAclStatus() API. (see HADOOP-13892 and HADOOP-14437)
*/
- private void analyzePaths(Analyzer analyzer, HdfsTable hdfsTable)
+ private void analyzePaths(Analyzer analyzer, FeFsTable table)
throws AnalysisException {
// The user must have permission to access the source location. Since the files will
// be moved from this location, the user needs to have all permission.
@@ -200,11 +202,12 @@ public class LoadDataStmt extends StatementBase {
String noWriteAccessErrorMsg = String.format("Unable to LOAD DATA into " +
"target table (%s) because Impala does not have WRITE access to HDFS " +
- "location: ", hdfsTable.getFullName());
+ "location: ", table.getFullName());
if (partitionSpec_ != null) {
- HdfsPartition partition = hdfsTable.getPartition(
- partitionSpec_.getPartitionSpecKeyValues());
+ long partId = HdfsTable.getPartition(table,
+ partitionSpec_.getPartitionSpecKeyValues()).getId();
+ FeFsPartition partition = FeCatalogUtils.loadPartition(table, partId);
String location = partition.getLocation();
if (!TAccessLevelUtil.impliesWriteAccess(partition.getAccessLevel())) {
throw new AnalysisException(noWriteAccessErrorMsg + location);
@@ -212,8 +215,8 @@ public class LoadDataStmt extends StatementBase {
} else {
// No specific partition specified, so we need to check write access
// on the table as a whole.
- if (!hdfsTable.hasWriteAccess()) {
- throw new AnalysisException(noWriteAccessErrorMsg + hdfsTable.getLocation());
+ if (!table.hasWriteAccess()) {
+ throw new AnalysisException(noWriteAccessErrorMsg + table.getLocation());
}
}
} catch (FileNotFoundException e) {
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/analysis/PartitionSpec.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/PartitionSpec.java b/fe/src/main/java/org/apache/impala/analysis/PartitionSpec.java
index 1d76df6..1a82dca 100644
--- a/fe/src/main/java/org/apache/impala/analysis/PartitionSpec.java
+++ b/fe/src/main/java/org/apache/impala/analysis/PartitionSpec.java
@@ -18,13 +18,13 @@
package org.apache.impala.analysis;
import com.google.common.base.Joiner;
-import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.impala.catalog.Column;
+import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.catalog.Type;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.thrift.TPartitionKeyValue;
@@ -110,7 +110,7 @@ public class PartitionSpec extends PartitionSpecBase {
pk.getValue().toSql(), colType.toString(), pk.getColName()));
}
}
- partitionExists_ = table_.getPartition(partitionSpec_) != null;
+ partitionExists_ = HdfsTable.getPartition(table_, partitionSpec_) != null;
if (partitionShouldExist_ != null) {
if (partitionShouldExist_ && !partitionExists_) {
throw new AnalysisException("Partition spec does not exist: (" +
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/analysis/PartitionSpecBase.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/PartitionSpecBase.java b/fe/src/main/java/org/apache/impala/analysis/PartitionSpecBase.java
index ede438b..ccdb266 100644
--- a/fe/src/main/java/org/apache/impala/analysis/PartitionSpecBase.java
+++ b/fe/src/main/java/org/apache/impala/analysis/PartitionSpecBase.java
@@ -19,8 +19,8 @@ package org.apache.impala.analysis;
import org.apache.impala.authorization.Privilege;
import org.apache.impala.catalog.TableLoadingException;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeTable;
-import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.common.AnalysisException;
import com.google.common.base.Preconditions;
@@ -29,7 +29,7 @@ import com.google.common.base.Preconditions;
* specifications of related DDL operations.
*/
public abstract class PartitionSpecBase implements ParseNode {
- protected HdfsTable table_;
+ protected FeFsTable table_;
protected TableName tableName_;
protected Boolean partitionShouldExist_;
protected Privilege privilegeRequirement_;
@@ -85,8 +85,8 @@ public abstract class PartitionSpecBase implements ParseNode {
}
// Only HDFS tables are partitioned.
- Preconditions.checkState(table instanceof HdfsTable);
- table_ = (HdfsTable) table;
+ Preconditions.checkState(table instanceof FeFsTable);
+ table_ = (FeFsTable) table;
nullPartitionKeyValue_ = table_.getNullPartitionKeyValue();
}
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/analysis/TableRef.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/TableRef.java b/fe/src/main/java/org/apache/impala/analysis/TableRef.java
index 98861d1..8ccc9ab 100644
--- a/fe/src/main/java/org/apache/impala/analysis/TableRef.java
+++ b/fe/src/main/java/org/apache/impala/analysis/TableRef.java
@@ -22,8 +22,8 @@ import java.util.List;
import java.util.Set;
import org.apache.impala.authorization.Privilege;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeTable;
-import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.planner.JoinNode.DistributionMode;
import org.apache.impala.rewrite.ExprRewriter;
@@ -353,7 +353,7 @@ public class TableRef implements ParseNode {
if (sampleParams_ == null) return;
sampleParams_.analyze(analyzer);
if (!(this instanceof BaseTableRef)
- || !(resolvedPath_.destTable() instanceof HdfsTable)) {
+ || !(resolvedPath_.destTable() instanceof FeFsTable)) {
throw new AnalysisException(
"TABLESAMPLE is only supported on HDFS tables: " + getUniqueAlias());
}
@@ -376,7 +376,7 @@ public class TableRef implements ParseNode {
// BaseTableRef will always have their path resolved at this point.
Preconditions.checkState(getResolvedPath() != null);
if (getResolvedPath().destTable() != null &&
- !(getResolvedPath().destTable() instanceof HdfsTable)) {
+ !(getResolvedPath().destTable() instanceof FeFsTable)) {
analyzer.addWarning("Table hints only supported for Hdfs tables");
}
for (PlanHint hint: tableHints_) {
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java b/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java
index a669caf..dca6b94 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java
@@ -41,6 +41,7 @@ import org.apache.hadoop.hive.metastore.TableType;
import org.apache.hadoop.hive.ql.parse.HiveLexer;
import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.Column;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.FeView;
@@ -48,7 +49,6 @@ import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.HBaseTable;
import org.apache.impala.catalog.HdfsCompression;
import org.apache.impala.catalog.HdfsFileFormat;
-import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.catalog.KuduColumn;
import org.apache.impala.catalog.KuduTable;
import org.apache.impala.catalog.RowFormat;
@@ -286,7 +286,7 @@ public class ToSqlUtils {
// We shouldn't output the columns for external tables
colsSql = null;
}
- } else if (table instanceof HdfsTable) {
+ } else if (table instanceof FeFsTable) {
String inputFormat = msTable.getSd().getInputFormat();
format = HdfsFileFormat.fromHdfsInputFormatClass(inputFormat);
compression = HdfsCompression.fromHdfsInputFormatClass(inputFormat);
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/analysis/TupleDescriptor.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/TupleDescriptor.java b/fe/src/main/java/org/apache/impala/analysis/TupleDescriptor.java
index 22693ad..bf2d310 100644
--- a/fe/src/main/java/org/apache/impala/analysis/TupleDescriptor.java
+++ b/fe/src/main/java/org/apache/impala/analysis/TupleDescriptor.java
@@ -26,9 +26,9 @@ import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.apache.impala.catalog.ColumnStats;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.catalog.FeTable;
-import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.catalog.StructType;
import org.apache.impala.thrift.TTupleDescriptor;
@@ -335,9 +335,9 @@ public class TupleDescriptor {
*/
public boolean hasClusteringColsOnly() {
FeTable table = getTable();
- if (!(table instanceof HdfsTable) || table.getNumClusteringCols() == 0) return false;
+ if (!(table instanceof FeFsTable) || table.getNumClusteringCols() == 0) return false;
- HdfsTable hdfsTable = (HdfsTable)table;
+ FeFsTable hdfsTable = (FeFsTable)table;
for (SlotDescriptor slotDesc: getSlots()) {
if (!slotDesc.isMaterialized()) continue;
if (slotDesc.getColumn() == null ||
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/catalog/FeFsTable.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/FeFsTable.java b/fe/src/main/java/org/apache/impala/catalog/FeFsTable.java
index 73a7ffe..86b41f0 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeFsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeFsTable.java
@@ -103,6 +103,22 @@ public interface FeFsTable extends FeTable {
public HdfsFileFormat getMajorityFormat();
/**
+ * Return true if the table may be written to.
+ */
+ public boolean hasWriteAccess();
+
+ /**
+ * Return some location found without write access for this table, useful
+ * in error messages about insufficient permissions to insert into a table.
+ *
+ * In case multiple locations are missing write access, the particular
+ * location returned is implementation-defined.
+ *
+ * Returns null if all partitions have write access.
+ */
+ public String getFirstLocationWithoutWriteAccess();
+
+ /**
* @param totalBytes_ the known number of bytes in the table
* @return Returns an estimated row count for the given number of file bytes
*/
@@ -181,4 +197,5 @@ public interface FeFsTable extends FeTable {
* @return the index of hosts that store replicas of blocks of this table.
*/
ListMap<TNetworkAddress> getHostIndex();
+
}
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
index b4bd707..0995086 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -631,6 +631,7 @@ public class HdfsTable extends Table implements FeFsTable {
// True if Impala has HDFS write permissions on the hdfsBaseDir (for an unpartitioned
// table) or if Impala has write permissions on all partition directories (for
// a partitioned table).
+ @Override
public boolean hasWriteAccess() {
return TAccessLevelUtil.impliesWriteAccess(accessLevel_);
}
@@ -640,6 +641,7 @@ public class HdfsTable extends Table implements FeFsTable {
* to, or an null if none is found. For an unpartitioned table, this just
* checks the hdfsBaseDir. For a partitioned table it checks all partition directories.
*/
+ @Override
public String getFirstLocationWithoutWriteAccess() {
if (getMetaStoreTable() == null) return null;
@@ -662,13 +664,18 @@ public class HdfsTable extends Table implements FeFsTable {
* was found.
*/
public HdfsPartition getPartition(List<PartitionKeyValue> partitionSpec) {
+ return (HdfsPartition)getPartition(this, partitionSpec);
+ }
+
+ public static PrunablePartition getPartition(FeFsTable table,
+ List<PartitionKeyValue> partitionSpec) {
List<TPartitionKeyValue> partitionKeyValues = Lists.newArrayList();
for (PartitionKeyValue kv: partitionSpec) {
String value = PartitionKeyValue.getPartitionKeyValueString(
- kv.getLiteralValue(), getNullPartitionKeyValue());
+ kv.getLiteralValue(), table.getNullPartitionKeyValue());
partitionKeyValues.add(new TPartitionKeyValue(kv.getColName(), value));
}
- return getPartitionFromThriftPartitionSpec(partitionKeyValues);
+ return getPartitionFromThriftPartitionSpec(table, partitionKeyValues);
}
/**
@@ -677,11 +684,17 @@ public class HdfsTable extends Table implements FeFsTable {
*/
public HdfsPartition getPartitionFromThriftPartitionSpec(
List<TPartitionKeyValue> partitionSpec) {
- // First, build a list of the partition values to search for in the same order they
+ return (HdfsPartition)getPartitionFromThriftPartitionSpec(this, partitionSpec);
+ }
+
+ public static PrunablePartition getPartitionFromThriftPartitionSpec(
+ FeFsTable table,
+ List<TPartitionKeyValue> partitionSpec) {
+ // First, build a list of the partition values to search for in the same order they
// are defined in the table.
List<String> targetValues = Lists.newArrayList();
Set<String> keys = Sets.newHashSet();
- for (FieldSchema fs: getMetaStoreTable().getPartitionKeys()) {
+ for (FieldSchema fs: table.getMetaStoreTable().getPartitionKeys()) {
for (TPartitionKeyValue kv: partitionSpec) {
if (fs.getName().toLowerCase().equals(kv.getName().toLowerCase())) {
targetValues.add(kv.getValue());
@@ -695,27 +708,27 @@ public class HdfsTable extends Table implements FeFsTable {
// Make sure the number of values match up and that some values were found.
if (targetValues.size() == 0 ||
- (targetValues.size() != getMetaStoreTable().getPartitionKeysSize())) {
+ (targetValues.size() != table.getMetaStoreTable().getPartitionKeysSize())) {
return null;
}
// Search through all the partitions and check if their partition key values
// match the values being searched for.
- for (HdfsPartition partition: partitionMap_.values()) {
+ for (PrunablePartition partition: table.getPartitions()) {
List<LiteralExpr> partitionValues = partition.getPartitionValues();
Preconditions.checkState(partitionValues.size() == targetValues.size());
boolean matchFound = true;
for (int i = 0; i < targetValues.size(); ++i) {
String value;
if (partitionValues.get(i) instanceof NullLiteral) {
- value = getNullPartitionKeyValue();
+ value = table.getNullPartitionKeyValue();
} else {
value = partitionValues.get(i).getStringValue();
Preconditions.checkNotNull(value);
// See IMPALA-252: we deliberately map empty strings on to
// NULL when they're in partition columns. This is for
// backwards compatibility with Hive, and is clearly broken.
- if (value.isEmpty()) value = getNullPartitionKeyValue();
+ if (value.isEmpty()) value = table.getNullPartitionKeyValue();
}
if (!targetValues.get(i).equals(value)) {
matchFound = false;
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java
index 0d376b3..c630b5c 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java
@@ -31,20 +31,25 @@ import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.DatabaseNotFoundException;
import org.apache.impala.catalog.Db;
import org.apache.impala.catalog.FeCatalog;
+import org.apache.impala.catalog.FeCatalogUtils;
import org.apache.impala.catalog.FeDataSource;
import org.apache.impala.catalog.FeDb;
import org.apache.impala.catalog.FeFsPartition;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.Function.CompareMode;
-import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
import org.apache.impala.catalog.HdfsCachePool;
+import org.apache.impala.catalog.HdfsTable;
+import org.apache.impala.catalog.PartitionNotFoundException;
+import org.apache.impala.catalog.PrunablePartition;
import org.apache.impala.thrift.TCatalogObject;
import org.apache.impala.thrift.TPartitionKeyValue;
import org.apache.impala.thrift.TUniqueId;
import org.apache.impala.util.PatternMatcher;
import org.apache.thrift.TException;
+import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
@@ -131,7 +136,7 @@ public class LocalCatalog implements FeCatalog {
@Override
public FeDb getDb(String db) {
loadDbs();
- return dbs_.get(db);
+ return dbs_.get(db.toLowerCase());
}
private FeDb getDbOrThrow(String dbName) throws DatabaseNotFoundException {
@@ -143,11 +148,28 @@ public class LocalCatalog implements FeCatalog {
return db;
}
+ private void throwPartitionNotFound(List<TPartitionKeyValue> partitionSpec)
+ throws PartitionNotFoundException {
+ throw new PartitionNotFoundException(
+ "Partition not found: " + Joiner.on(", ").join(partitionSpec));
+ }
+
@Override
public FeFsPartition getHdfsPartition(
- String db, String tbl, List<TPartitionKeyValue> partition_spec)
+ String db, String tbl, List<TPartitionKeyValue> partitionSpec)
throws CatalogException {
- throw new UnsupportedOperationException("TODO");
+ // TODO(todd): somewhat copy-pasted from Catalog.getHdfsPartition
+
+ FeTable table = getTable(db, tbl);
+ // This is not an FS table, throw an error.
+ if (!(table instanceof FeFsTable)) {
+ throwPartitionNotFound(partitionSpec);
+ }
+ // Get the FeFsPartition object for the given partition spec.
+ PrunablePartition partition = HdfsTable.getPartitionFromThriftPartitionSpec(
+ (FeFsTable)table, partitionSpec);
+ if (partition == null) throwPartitionNotFound(partitionSpec);
+ return FeCatalogUtils.loadPartition((FeFsTable)table, partition.getId());
}
@Override
@@ -237,4 +259,4 @@ public class LocalCatalog implements FeCatalog {
MetaProvider getMetaProvider() {
return metaProvider_;
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/catalog/local/LocalDb.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalDb.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalDb.java
index 6b9209f..136093a 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalDb.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalDb.java
@@ -94,7 +94,7 @@ class LocalDb implements FeDb {
@Override
public FeTable getTable(String tblName) {
Preconditions.checkNotNull(tblName);
- Preconditions.checkArgument(tblName.toLowerCase().equals(tblName));
+ tblName = tblName.toLowerCase();
loadTableNames();
if (!tables_.containsKey(tblName)) {
// Table doesn't exist.
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java
index c5510d5..2e35ce8 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java
@@ -102,9 +102,12 @@ public class LocalFsPartition implements FeFsPartition {
@Override
public THdfsPartitionLocation getLocationAsThrift() {
+ String loc = getLocation();
+ // The special "prototype partition" has a null location.
+ if (loc == null) return null;
// TODO(todd): support prefix-compressed partition locations. For now,
// using -1 indicates that the location is a full path string.
- return new THdfsPartitionLocation(/*prefix_index=*/-1, getLocation());
+ return new THdfsPartitionLocation(/*prefix_index=*/-1, loc);
}
@Override
@@ -115,7 +118,7 @@ public class LocalFsPartition implements FeFsPartition {
@Override
public TAccessLevel getAccessLevel() {
// TODO(todd): implement me
- return TAccessLevel.READ_ONLY;
+ return TAccessLevel.READ_WRITE;
}
@Override
http://git-wip-us.apache.org/repos/asf/impala/blob/cec33fa0/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java
index 81dcae1..35044e3 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java
@@ -27,6 +27,7 @@ import java.util.Set;
import java.util.TreeMap;
import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.serde.serdeConstants;
import org.apache.impala.analysis.LiteralExpr;
@@ -188,6 +189,18 @@ public class LocalFsTable extends LocalTable implements FeFsTable {
}
@Override
+ public boolean hasWriteAccess() {
+ // TODO(todd): implement me properly
+ return true;
+ }
+
+ @Override
+ public String getFirstLocationWithoutWriteAccess() {
+ // TODO(todd): implement me properly
+ return null;
+ }
+
+ @Override
public long getExtrapolatedNumRows(long totalBytes) {
// TODO Auto-generated method stub
return -1;
@@ -239,7 +252,14 @@ public class LocalFsTable extends LocalTable implements FeFsTable {
private LocalFsPartition createPrototypePartition() {
Partition protoMsPartition = new Partition();
- protoMsPartition.setSd(getMetaStoreTable().getSd());
+
+ // The prototype partition should not have a location set in its storage
+ // descriptor, or else all inserted files will end up written into the
+ // table directory instead of the new partition directories.
+ StorageDescriptor sd = getMetaStoreTable().getSd().deepCopy();
+ sd.unsetLocation();
+ protoMsPartition.setSd(sd);
+
protoMsPartition.setParameters(Collections.<String, String>emptyMap());
LocalPartitionSpec spec = new LocalPartitionSpec(
this, "", CatalogObjectsConstants.PROTOTYPE_PARTITION_ID);