You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2019/02/22 17:34:53 UTC
[impala] 10/13: IMPALA-7140 (part 3): load partitions for FS tables
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 4a279e1bc430b5204e59e73b3b8fb68a0b5e64ab
Author: Todd Lipcon <to...@cloudera.com>
AuthorDate: Tue Jun 12 20:16:07 2018 -0700
IMPALA-7140 (part 3): load partitions for FS tables
This implements the loading of partitions from the HMS for FS-based
tables. The LocalPartitionSpec class implements PrunablePartition based
on parsing the partition names returned from the HMS. After partitions
are pruned, the remaining partitions can be loaded by name.
With this patch, I am able to connect to an impalad running with
--use_local_catalog and issue 'show partitions functional.alltypes' with
the expected results.
A new unit test is added which shares some code with CatalogTest to
ensure that partitions can be loaded and parsed properly.
Testing partition pruning via end-to-end planning is not quite
supported yet: we need to implement 'toThriftDescriptor()' before we can
run those end-to-end tests.
Change-Id: Iddf2edbd6bdc0684560b2ecca9c4c6b6819ef1d3
Reviewed-on: http://gerrit.cloudera.org:8080/10713
Reviewed-by: Vuk Ercegovac <ve...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../org/apache/impala/catalog/FeCatalogUtils.java | 100 +++++++++++
.../org/apache/impala/catalog/HdfsPartition.java | 60 +------
.../java/org/apache/impala/catalog/HdfsTable.java | 24 +--
.../apache/impala/catalog/PartitionStatsUtil.java | 19 ++-
.../impala/catalog/local/DirectMetaProvider.java | 64 +++++++
.../impala/catalog/local/LocalFsPartition.java | 185 +++++++++++++++++++++
.../apache/impala/catalog/local/LocalFsTable.java | 185 +++++++++++++++++++--
.../impala/catalog/local/LocalPartitionSpec.java | 71 ++++++++
.../apache/impala/catalog/local/LocalTable.java | 4 +-
.../apache/impala/catalog/local/MetaProvider.java | 17 ++
.../apache/impala/service/CatalogOpExecutor.java | 21 +--
.../java/org/apache/impala/util/MetaStoreUtil.java | 21 +++
.../org/apache/impala/catalog/CatalogTest.java | 13 +-
.../impala/catalog/local/LocalCatalogTest.java | 31 +++-
14 files changed, 707 insertions(+), 108 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/catalog/FeCatalogUtils.java b/fe/src/main/java/org/apache/impala/catalog/FeCatalogUtils.java
index 7d811c2..80ab246 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeCatalogUtils.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeCatalogUtils.java
@@ -19,18 +19,32 @@ package org.apache.impala.catalog;
import java.util.Collection;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
+import org.apache.hadoop.hive.common.FileUtils;
import org.apache.hadoop.hive.common.StatsSetupConst;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.impala.analysis.Expr;
+import org.apache.impala.analysis.LiteralExpr;
+import org.apache.impala.analysis.NullLiteral;
+import org.apache.impala.analysis.PartitionKeyValue;
+import org.apache.impala.analysis.ToSqlUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
/**
* Static utility functions shared between FeCatalog implementations.
*/
public abstract class FeCatalogUtils {
+ private final static Logger LOG = LoggerFactory.getLogger(FeCatalogUtils.class);
+
/**
* Gets the ColumnType from the given FieldSchema by using Impala's SqlParser.
*
@@ -148,4 +162,90 @@ public abstract class FeCatalogUtils {
FeFsTable table) {
return table.loadPartitions(table.getPartitionIds());
}
+
+ /**
+ * Parse the partition key values out of their stringified format used by HMS.
+ */
+ public static List<LiteralExpr> parsePartitionKeyValues(FeFsTable table,
+ List<String> hmsPartitionValues) throws CatalogException {
+ Preconditions.checkArgument(
+ hmsPartitionValues.size() == table.getNumClusteringCols(),
+ "Cannot parse partition values %s for table %s: " +
+ "expected %s values but got %s",
+ hmsPartitionValues, table.getFullName(),
+ table.getNumClusteringCols(), hmsPartitionValues.size());
+ List<LiteralExpr> keyValues = Lists.newArrayList();
+ for (String partitionKey: hmsPartitionValues) {
+ Type type = table.getColumns().get(keyValues.size()).getType();
+ // Deal with Hive's special NULL partition key.
+ if (partitionKey.equals(table.getNullPartitionKeyValue())) {
+ keyValues.add(NullLiteral.create(type));
+ } else {
+ try {
+ keyValues.add(LiteralExpr.create(partitionKey, type));
+ } catch (Exception ex) {
+ LOG.warn("Failed to create literal expression of type: " + type, ex);
+ throw new CatalogException("Invalid partition key value of type: " + type,
+ ex);
+ }
+ }
+ }
+ for (Expr v: keyValues) v.analyzeNoThrow(null);
+ return keyValues;
+ }
+
+ /**
+ * Return a partition name formed by concatenating partition keys and their values,
+ * compatible with the way Hive names partitions. Reuses Hive's
+ * org.apache.hadoop.hive.common.FileUtils.makePartName() function to build the name
+ * string because there are a number of special cases for how partition names are URL
+ * escaped.
+ *
+ * TODO: this could be a default method in FeFsPartition in Java 8.
+ */
+ public static String getPartitionName(FeFsPartition partition) {
+ FeFsTable table = partition.getTable();
+ List<String> partitionCols = Lists.newArrayList();
+ for (int i = 0; i < table.getNumClusteringCols(); ++i) {
+ partitionCols.add(table.getColumns().get(i).getName());
+ }
+
+ return FileUtils.makePartName(
+ partitionCols, getPartitionValuesAsStrings(partition, true));
+ }
+
+ // TODO: this could be a default method in FeFsPartition in Java 8.
+ public static List<String> getPartitionValuesAsStrings(
+ FeFsPartition partition, boolean mapNullsToHiveKey) {
+ List<String> ret = Lists.newArrayList();
+ for (LiteralExpr partValue: partition.getPartitionValues()) {
+ if (mapNullsToHiveKey) {
+ ret.add(PartitionKeyValue.getPartitionKeyValueString(
+ partValue, partition.getTable().getNullPartitionKeyValue()));
+ } else {
+ ret.add(partValue.getStringValue());
+ }
+ }
+ return ret;
+ }
+
+ // TODO: this could be a default method in FeFsPartition in Java 8.
+ public static String getConjunctSqlForPartition(FeFsPartition part) {
+ List<String> partColSql = Lists.newArrayList();
+ for (Column partCol: part.getTable().getClusteringColumns()) {
+ partColSql.add(ToSqlUtils.getIdentSql(partCol.getName()));
+ }
+
+ List<String> conjuncts = Lists.newArrayList();
+ for (int i = 0; i < partColSql.size(); ++i) {
+ LiteralExpr partVal = part.getPartitionValues().get(i);
+ String partValSql = partVal.toSql();
+ if (partVal instanceof NullLiteral || partValSql.isEmpty()) {
+ conjuncts.add(partColSql.get(i) + " IS NULL");
+ } else {
+ conjuncts.add(partColSql.get(i) + "=" + partValSql);
+ }
+ }
+ return "(" + Joiner.on(" AND " ).join(conjuncts) + ")";
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java b/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
index 12b4a59..26f8c30 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
@@ -36,7 +36,6 @@ import org.apache.impala.analysis.NullLiteral;
import org.apache.impala.analysis.PartitionKeyValue;
import org.apache.impala.analysis.ToSqlUtils;
import org.apache.impala.common.FileSystemUtil;
-import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.Pair;
import org.apache.impala.common.Reference;
import org.apache.impala.fb.FbCompression;
@@ -447,60 +446,23 @@ public class HdfsPartition implements FeFsPartition, PrunablePartition {
return FileSystemUtil.isPathCacheable(new Path(getLocation()));
}
- /**
- * Return a partition name formed by concatenating partition keys and their values,
- * compatible with the way Hive names partitions. Reuses Hive's
- * org.apache.hadoop.hive.common.FileUtils.makePartName() function to build the name
- * string because there are a number of special cases for how partition names are URL
- * escaped.
- * TODO: Consider storing the PartitionKeyValue in HdfsPartition. It would simplify
- * this code would be useful in other places, such as fromThrift().
- */
@Override // FeFsPartition
public String getPartitionName() {
- List<String> partitionCols = Lists.newArrayList();
- for (int i = 0; i < getTable().getNumClusteringCols(); ++i) {
- partitionCols.add(getTable().getColumns().get(i).getName());
- }
-
- return org.apache.hadoop.hive.common.FileUtils.makePartName(
- partitionCols, getPartitionValuesAsStrings(true));
+ // TODO: Consider storing the PartitionKeyValue in HdfsPartition. It would simplify
+ // this code would be useful in other places, such as fromThrift().
+ return FeCatalogUtils.getPartitionName(this);
}
@Override
public List<String> getPartitionValuesAsStrings(boolean mapNullsToHiveKey) {
- List<String> ret = Lists.newArrayList();
- for (LiteralExpr partValue: getPartitionValues()) {
- if (mapNullsToHiveKey) {
- ret.add(PartitionKeyValue.getPartitionKeyValueString(
- partValue, getTable().getNullPartitionKeyValue()));
- } else {
- ret.add(partValue.getStringValue());
- }
- }
- return ret;
+ return FeCatalogUtils.getPartitionValuesAsStrings(this, mapNullsToHiveKey);
}
@Override // FeFsPartition
public String getConjunctSql() {
// TODO: Remove this when the TODO elsewhere in this file to save and expose the
// list of TPartitionKeyValues has been resolved.
- List<String> partColSql = Lists.newArrayList();
- for (Column partCol: getTable().getClusteringColumns()) {
- partColSql.add(ToSqlUtils.getIdentSql(partCol.getName()));
- }
-
- List<String> conjuncts = Lists.newArrayList();
- for (int i = 0; i < partColSql.size(); ++i) {
- LiteralExpr partVal = getPartitionValues().get(i);
- String partValSql = partVal.toSql();
- if (partVal instanceof NullLiteral || partValSql.isEmpty()) {
- conjuncts.add(partColSql.get(i) + " IS NULL");
- } else {
- conjuncts.add(partColSql.get(i) + "=" + partValSql);
- }
- }
- return "(" + Joiner.on(" AND " ).join(conjuncts) + ")";
+ return FeCatalogUtils.getConjunctSqlForPartition(this);
}
/**
@@ -565,18 +527,14 @@ public class HdfsPartition implements FeFsPartition, PrunablePartition {
@Override // FeFsPartition
public TPartitionStats getPartitionStats() {
- try {
- return PartitionStatsUtil.partStatsFromParameters(hmsParameters_);
- } catch (ImpalaException e) {
- LOG.warn("Could not deserialise incremental stats state for " + getPartitionName() +
- ", consider DROP INCREMENTAL STATS ... PARTITION ... and recomputing " +
- "incremental stats for this table.");
- return null;
- }
+ return PartitionStatsUtil.getPartStatsOrWarn(this);
}
@Override // FeFsPartition
public boolean hasIncrementalStats() {
+ // TODO: performance of this could improve substantially, since
+ // getPartitionStats() performs a bunch of expensive deserialization,
+ // only to end up throwing away the result.
TPartitionStats partStats = getPartitionStats();
return partStats != null && partStats.intermediate_col_stats != null;
}
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 b0b4387..e6c34bc 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -52,7 +52,6 @@ import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.serde.serdeConstants;
import org.apache.impala.analysis.ColumnDef;
-import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.LiteralExpr;
import org.apache.impala.analysis.NullLiteral;
import org.apache.impala.analysis.NumericLiteral;
@@ -1047,27 +1046,12 @@ public class HdfsTable extends Table implements FeFsTable {
throws CatalogException {
HdfsStorageDescriptor fileFormatDescriptor =
HdfsStorageDescriptor.fromStorageDescriptor(this.name_, storageDescriptor);
- List<LiteralExpr> keyValues = Lists.newArrayList();
+ List<LiteralExpr> keyValues;
if (msPartition != null) {
- // Load key values
- for (String partitionKey: msPartition.getValues()) {
- Type type = getColumns().get(keyValues.size()).getType();
- // Deal with Hive's special NULL partition key.
- if (partitionKey.equals(nullPartitionKeyValue_)) {
- keyValues.add(NullLiteral.create(type));
- } else {
- try {
- keyValues.add(LiteralExpr.create(partitionKey, type));
- } catch (Exception ex) {
- LOG.warn("Failed to create literal expression of type: " + type, ex);
- throw new CatalogException("Invalid partition key value of type: " + type,
- ex);
- }
- }
- }
- for (Expr v: keyValues) v.analyzeNoThrow(null);
+ keyValues = FeCatalogUtils.parsePartitionKeyValues(this, msPartition.getValues());
+ } else {
+ keyValues = Collections.emptyList();
}
-
Path partDirPath = new Path(storageDescriptor.getLocation());
try {
FileSystem fs = partDirPath.getFileSystem(CONF);
diff --git a/fe/src/main/java/org/apache/impala/catalog/PartitionStatsUtil.java b/fe/src/main/java/org/apache/impala/catalog/PartitionStatsUtil.java
index 4444de6..c77ac91 100644
--- a/fe/src/main/java/org/apache/impala/catalog/PartitionStatsUtil.java
+++ b/fe/src/main/java/org/apache/impala/catalog/PartitionStatsUtil.java
@@ -50,11 +50,28 @@ public class PartitionStatsUtil {
private final static Logger LOG = LoggerFactory.getLogger(PartitionStatsUtil.class);
/**
+ * Get the partition stats from the given partition, or null if no stats
+ * are available. If stats are present but cannot be parsed, logs a warning
+ * and returns null.
+ */
+ public static TPartitionStats getPartStatsOrWarn(FeFsPartition part) {
+ try {
+ return partStatsFromParameters(part.getParameters());
+ } catch (ImpalaException e) {
+ LOG.warn("Could not deserialise incremental stats state for " +
+ part.getPartitionName() +
+ ", consider DROP INCREMENTAL STATS ... PARTITION ... and recomputing " +
+ "incremental stats for this table.");
+ return null;
+ }
+ }
+
+ /**
* Reconstructs a TPartitionStats object from its serialised form in the given parameter
* map. Returns null if no stats are serialised, and throws an exception if there was an
* error during deserialisation.
*/
- public static TPartitionStats partStatsFromParameters(
+ private static TPartitionStats partStatsFromParameters(
Map<String, String> hmsParameters) throws ImpalaException {
if (hmsParameters == null) return null;
String numChunksStr = hmsParameters.get(INCREMENTAL_STATS_NUM_CHUNKS);
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java b/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
index ac86bfd..e09f031 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
@@ -17,9 +17,15 @@
package org.apache.impala.catalog.local;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.hadoop.hive.common.FileUtils;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.api.UnknownDBException;
import org.apache.impala.catalog.MetaStoreClientPool;
@@ -29,7 +35,10 @@ import org.apache.impala.thrift.TBackendGflags;
import org.apache.impala.util.MetaStoreUtil;
import org.apache.thrift.TException;
+import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Maps;
/**
* Metadata provider which calls out directly to the source systems
@@ -88,4 +97,59 @@ class DirectMetaProvider implements MetaProvider {
return MetaStoreUtil.getNullPartitionKeyValue(c.getHiveClient());
}
}
+
+ @Override
+ public List<String> loadPartitionNames(String dbName, String tableName)
+ throws MetaException, TException {
+ try (MetaStoreClient c = msClientPool_.getClient()) {
+ return c.getHiveClient().listPartitionNames(dbName, tableName,
+ /*max_parts=*/(short)-1);
+ }
+ }
+
+ @Override
+ public Map<String, Partition> loadPartitionsByNames(
+ String dbName, String tableName, List<String> partitionColumnNames,
+ List<String> partitionNames) throws MetaException, TException {
+ Preconditions.checkNotNull(dbName);
+ Preconditions.checkNotNull(tableName);
+ Preconditions.checkArgument(!partitionColumnNames.isEmpty());
+ Preconditions.checkNotNull(partitionNames);
+
+ Map<String, Partition> ret = Maps.newHashMapWithExpectedSize(
+ partitionNames.size());
+ if (partitionNames.isEmpty()) return ret;
+
+ // Fetch the partitions.
+ List<Partition> parts;
+ try (MetaStoreClient c = msClientPool_.getClient()) {
+ parts = MetaStoreUtil.fetchPartitionsByName(
+ c.getHiveClient(), partitionNames, dbName, tableName);
+ }
+
+ // HMS may return fewer partition objects than requested, and the
+ // returned partition objects don't carry enough information to get their
+ // names. So, we map the returned partitions back to the requested names
+ // using the passed-in partition column names.
+ Set<String> namesSet = ImmutableSet.copyOf(partitionNames);
+ for (Partition p: parts) {
+ List<String> vals = p.getValues();
+ if (vals.size() != partitionColumnNames.size()) {
+ throw new MetaException("Unexpected number of partition values for " +
+ "partition " + vals + " (expected " + partitionColumnNames.size() + ")");
+ }
+ String partName = FileUtils.makePartName(partitionColumnNames, p.getValues());
+ if (!namesSet.contains(partName)) {
+ throw new MetaException("HMS returned unexpected partition " + partName +
+ " which was not requested. Requested: " + namesSet);
+ }
+ Partition existing = ret.put(partName, p);
+ if (existing != null) {
+ throw new MetaException("HMS returned multiple partitions with name " +
+ partName);
+ }
+ }
+
+ return ret;
+ }
}
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
new file mode 100644
index 0000000..c33b8a2
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java
@@ -0,0 +1,185 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.catalog.local;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.impala.analysis.LiteralExpr;
+import org.apache.impala.catalog.FeCatalogUtils;
+import org.apache.impala.catalog.FeFsPartition;
+import org.apache.impala.catalog.FeFsTable;
+import org.apache.impala.catalog.HdfsFileFormat;
+import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
+import org.apache.impala.catalog.HdfsStorageDescriptor;
+import org.apache.impala.catalog.HdfsStorageDescriptor.InvalidStorageDescriptorException;
+import org.apache.impala.catalog.PartitionStatsUtil;
+import org.apache.impala.thrift.TAccessLevel;
+import org.apache.impala.thrift.TPartitionStats;
+
+import com.google.common.base.Preconditions;
+
+public class LocalFsPartition implements FeFsPartition {
+ private final LocalFsTable table_;
+ private final LocalPartitionSpec spec_;
+ private final Partition msPartition_;
+
+ public LocalFsPartition(LocalFsTable table, LocalPartitionSpec spec,
+ Partition msPartition) {
+ table_ = Preconditions.checkNotNull(table);
+ spec_ = Preconditions.checkNotNull(spec);
+ msPartition_ = Preconditions.checkNotNull(msPartition);
+ }
+
+ @Override
+ public String getPartitionName() {
+ return FeCatalogUtils.getPartitionName(this);
+ }
+
+ @Override
+ public long getId() {
+ return spec_.getId();
+ }
+
+ @Override
+ public FeFsTable getTable() {
+ return table_;
+ }
+
+ @Override
+ public List<FileDescriptor> getFileDescriptors() {
+ // TODO(todd): implement me
+ return Collections.emptyList();
+ }
+
+ @Override
+ public boolean hasFileDescriptors() {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ @Override
+ public int getNumFileDescriptors() {
+ // TODO Auto-generated method stub
+ return 0;
+ }
+
+ @Override
+ public String getLocation() {
+ return msPartition_.getSd().getLocation();
+ }
+
+ @Override
+ public Path getLocationPath() {
+ return new Path(getLocation());
+ }
+
+ @Override
+ public TAccessLevel getAccessLevel() {
+ // TODO(todd): implement me
+ return TAccessLevel.READ_ONLY;
+ }
+
+ @Override
+ public boolean isCacheable() {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ @Override
+ public boolean isMarkedCached() {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ @Override
+ public HdfsStorageDescriptor getInputFormatDescriptor() {
+ try {
+ // TODO(todd): should we do this in the constructor to avoid having to worry
+ // about throwing an exception from this getter? The downside is that then
+ // the object would end up long-lived, whereas its actual usage is short-lived.
+ return HdfsStorageDescriptor.fromStorageDescriptor(table_.getName(),
+ msPartition_.getSd());
+ } catch (InvalidStorageDescriptorException e) {
+ throw new LocalCatalogException(String.format(
+ "Invalid input format descriptor for partition %s of table %s",
+ getPartitionName(), table_.getFullName()), e);
+ }
+ }
+
+ @Override
+ public HdfsFileFormat getFileFormat() {
+ return getInputFormatDescriptor().getFileFormat();
+ }
+
+ @Override
+ public TPartitionStats getPartitionStats() {
+ // TODO(todd): after HIVE-19715 is complete, we should defer loading these
+ // stats until requested. Even prior to that, we should consider caching them
+ // separately to increase effective cache capacity.
+ return PartitionStatsUtil.getPartStatsOrWarn(this);
+ }
+
+ @Override
+ public boolean hasIncrementalStats() {
+ // TODO(todd): copy-paste from HdfsPartition
+ // TODO(todd): as in the equivalent method in HdfsPartition, this is
+ // expensive because it deserializes the stats completely.
+ TPartitionStats partStats = getPartitionStats();
+ return partStats != null && partStats.intermediate_col_stats != null;
+ }
+
+ @Override
+ public long getSize() {
+ // TODO Auto-generated method stub
+ return 0;
+ }
+
+ @Override
+ public long getNumRows() {
+ return FeCatalogUtils.getRowCount(msPartition_.getParameters());
+ }
+
+ @Override
+ public String getConjunctSql() {
+ return FeCatalogUtils.getConjunctSqlForPartition(this);
+ }
+
+ @Override
+ public List<String> getPartitionValuesAsStrings(boolean mapNullsToHiveKey) {
+ return FeCatalogUtils.getPartitionValuesAsStrings(this, mapNullsToHiveKey);
+ }
+
+ @Override
+ public List<LiteralExpr> getPartitionValues() {
+ return spec_.getPartitionValues();
+ }
+
+ @Override
+ public LiteralExpr getPartitionValue(int pos) {
+ return spec_.getPartitionValues().get(pos);
+ }
+
+ @Override
+ public Map<String, String> getParameters() {
+ return msPartition_.getParameters();
+ }
+}
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 b6ab559..352655c 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
@@ -17,6 +17,7 @@
package org.apache.impala.catalog.local;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
@@ -25,8 +26,11 @@ import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
+import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.impala.analysis.LiteralExpr;
+import org.apache.impala.analysis.NullLiteral;
import org.apache.impala.catalog.CatalogException;
+import org.apache.impala.catalog.Column;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.HdfsTable;
@@ -36,8 +40,36 @@ import org.apache.impala.thrift.TNetworkAddress;
import org.apache.impala.thrift.TPartitionKeyValue;
import org.apache.impala.thrift.TResultSet;
import org.apache.impala.util.ListMap;
+import org.apache.thrift.TException;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
public class LocalFsTable extends LocalTable implements FeFsTable {
+ /**
+ * Map from partition ID to partition spec.
+ *
+ * Set by loadPartitionSpecs().
+ */
+ ImmutableMap<Long, LocalPartitionSpec> partitionSpecs_;
+
+ /**
+ * For each partition column, a map from value to the set of partition IDs
+ * having that value.
+ *
+ * Set by loadPartitionValueMap().
+ */
+ private ArrayList<TreeMap<LiteralExpr, HashSet<Long>>> partitionValueMap_;
+
+ /**
+ * For each partition column, the set of partition IDs having a NULL value
+ * for that column.
+ *
+ * Set by loadPartitionValueMap().
+ */
+ private ArrayList<HashSet<Long>> nullPartitionIds_;
+
public LocalFsTable(LocalDb db, String tblName, SchemaInfo schemaInfo) {
super(db, tblName, schemaInfo);
}
@@ -120,38 +152,167 @@ public class LocalFsTable extends LocalTable implements FeFsTable {
@Override
public Collection<? extends PrunablePartition> getPartitions() {
- // TODO(todd): implement me
- return Collections.emptyList();
+ loadPartitionSpecs();
+ return partitionSpecs_.values();
}
@Override
public Set<Long> getPartitionIds() {
- // TODO(todd): implement me
- return Collections.emptySet();
+ loadPartitionSpecs();
+ return partitionSpecs_.keySet();
}
@Override
public Map<Long, ? extends PrunablePartition> getPartitionMap() {
- // TODO(todd): implement me
- return Collections.emptyMap();
+ loadPartitionSpecs();
+ return partitionSpecs_;
}
@Override
public TreeMap<LiteralExpr, HashSet<Long>> getPartitionValueMap(int col) {
- // TODO(todd): implement me
- return new TreeMap<>();
+ loadPartitionValueMap();
+ return partitionValueMap_.get(col);
}
@Override
public Set<Long> getNullPartitionIds(int colIdx) {
- // TODO(todd): implement me
- return Collections.emptySet();
+ loadPartitionValueMap();
+ return nullPartitionIds_.get(colIdx);
}
@Override
public List<? extends FeFsPartition> loadPartitions(Collection<Long> ids) {
- // TODO(todd): implement me
- return Collections.emptyList();
+ Preconditions.checkState(partitionSpecs_ != null,
+ "Cannot load partitions without having fetched partition IDs " +
+ "from the same LocalFsTable instance");
+ List<String> names = Lists.newArrayList();
+ for (Long id : ids) {
+ LocalPartitionSpec spec = partitionSpecs_.get(id);
+ Preconditions.checkArgument(spec != null, "Invalid partition ID for table %s: %s",
+ getFullName(), id);
+ String name = spec.getName();
+ if (name.isEmpty()) {
+ // Unpartitioned tables don't need to fetch partitions from the metadata
+ // provider. Rather, we just create a partition on the fly.
+ Preconditions.checkState(getNumClusteringCols() == 0,
+ "Cannot fetch empty partition name from a partitioned table");
+ Preconditions.checkArgument(ids.size() == 1,
+ "Expected to only fetch one partition for unpartitioned table %s",
+ getFullName());
+ return Lists.newArrayList(createUnpartitionedPartition(spec));
+ } else {
+ names.add(name);
+ }
+ }
+ Map<String, Partition> partsByName;
+ try {
+ partsByName = db_.getCatalog().getMetaProvider().loadPartitionsByNames(
+ db_.getName(), name_, getClusteringColumnNames(), names);
+ } catch (TException e) {
+ throw new LocalCatalogException(
+ "Could not load partitions for table " + getFullName(), e);
+ }
+ List<FeFsPartition> ret = Lists.newArrayListWithCapacity(ids.size());
+ for (Long id : ids) {
+ LocalPartitionSpec spec = partitionSpecs_.get(id);
+ Partition p = partsByName.get(spec.getName());
+ if (p == null) {
+ // TODO(todd): concurrent drop partition could result in this error.
+ // Should we recover in a more graceful way from such an unexpected event?
+ throw new LocalCatalogException(
+ "Could not load expected partitions for table " + getFullName() +
+ ": missing expected partition with name '" + spec.getName() +
+ "' (perhaps it was concurrently dropped by another process)");
+ }
+ ret.add(new LocalFsPartition(this, spec, p));
+ }
+ return ret;
+ }
+
+ private List<String> getClusteringColumnNames() {
+ List<String> names = Lists.newArrayListWithCapacity(getNumClusteringCols());
+ for (Column c : getClusteringColumns()) {
+ names.add(c.getName());
+ }
+ return names;
+ }
+
+ /**
+ * Create a partition which represents the main partition of an unpartitioned
+ * table.
+ */
+ private LocalFsPartition createUnpartitionedPartition(LocalPartitionSpec spec) {
+ Preconditions.checkArgument(spec.getName().isEmpty());
+ Partition msp = new Partition();
+ msp.setSd(getMetaStoreTable().getSd());
+ msp.setParameters(getMetaStoreTable().getParameters());
+ msp.setValues(Collections.<String>emptyList());
+ return new LocalFsPartition(this, spec, msp);
+ }
+
+ private LocalPartitionSpec createUnpartitionedPartitionSpec() {
+ return new LocalPartitionSpec(this, "", /*id=*/0);
+ }
+
+ private void loadPartitionValueMap() {
+ if (partitionValueMap_ != null) return;
+
+ loadPartitionSpecs();
+ ArrayList<TreeMap<LiteralExpr, HashSet<Long>>> valMapByCol = new ArrayList<>();
+ ArrayList<HashSet<Long>> nullParts = new ArrayList<>();
+
+ for (int i = 0; i < getNumClusteringCols(); i++) {
+ valMapByCol.add(new TreeMap<LiteralExpr, HashSet<Long>>());
+ nullParts.add(new HashSet<Long>());
+ }
+ for (LocalPartitionSpec partition : partitionSpecs_.values()) {
+ List<LiteralExpr> vals = partition.getPartitionValues();
+ for (int i = 0; i < getNumClusteringCols(); i++) {
+ LiteralExpr val = vals.get(i);
+ if (val instanceof NullLiteral) {
+ nullParts.get(i).add(partition.getId());
+ continue;
+ }
+
+ HashSet<Long> ids = valMapByCol.get(i).get(val);
+ if (ids == null) {
+ ids = new HashSet<>();
+ valMapByCol.get(i).put(val, ids);
+ }
+ ids.add(partition.getId());
+ }
+ }
+ partitionValueMap_ = valMapByCol;
+ nullPartitionIds_ = nullParts;
+ }
+
+ private void loadPartitionSpecs() {
+ if (partitionSpecs_ != null) return;
+
+ if (getNumClusteringCols() == 0) {
+ // Unpartitioned table.
+ // This table has no partition key, which means it has no declared partitions.
+ // We model partitions slightly differently to Hive - every file must exist in a
+ // partition, so add a single partition with no keys which will get all the
+ // files in the table's root directory.
+ partitionSpecs_ = ImmutableMap.of(0L, createUnpartitionedPartitionSpec());
+ return;
+ }
+ List<String> partNames;
+ try {
+ partNames = db_.getCatalog().getMetaProvider().loadPartitionNames(
+ db_.getName(), name_);
+ } catch (TException e) {
+ throw new LocalCatalogException("Could not load partition names for table " +
+ getFullName(), e);
+ }
+ ImmutableMap.Builder<Long, LocalPartitionSpec> b = new ImmutableMap.Builder<>();
+ long id = 0;
+ for (String partName : partNames) {
+ b.put(id, new LocalPartitionSpec(this, partName, id));
+ id++;
+ }
+ partitionSpecs_ = b.build();
}
@Override
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalPartitionSpec.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalPartitionSpec.java
new file mode 100644
index 0000000..f48f54e
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalPartitionSpec.java
@@ -0,0 +1,71 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.catalog.local;
+
+import java.util.List;
+
+import javax.annotation.concurrent.Immutable;
+
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.impala.analysis.LiteralExpr;
+import org.apache.impala.catalog.CatalogException;
+import org.apache.impala.catalog.FeCatalogUtils;
+import org.apache.impala.catalog.PrunablePartition;
+import org.apache.impala.util.MetaStoreUtil;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+/**
+ * Specification of a partition of a {@link LocalFsTable} containing
+ * everything necessary for partition pruning.
+ */
+@Immutable
+class LocalPartitionSpec implements PrunablePartition {
+ private final long id_;
+ private final String name_;
+ private final ImmutableList<LiteralExpr> partitionValues_;
+
+ LocalPartitionSpec(LocalFsTable table, String partName, long id) {
+ id_ = id;
+ name_ = Preconditions.checkNotNull(partName);
+ if (!partName.isEmpty()) {
+ try {
+ List<String> partValues = MetaStoreUtil.getPartValsFromName(
+ table.getMetaStoreTable(), partName);
+ partitionValues_ = ImmutableList.copyOf(FeCatalogUtils.parsePartitionKeyValues(
+ table, partValues));
+ } catch (CatalogException | MetaException e) {
+ throw new LocalCatalogException(String.format(
+ "Failed to parse partition name '%s' for table %s",
+ partName, table.getFullName()), e);
+ }
+ } else {
+ // Unpartitioned tables have a single partition with empty name.
+ partitionValues_= ImmutableList.of();
+ }
+ }
+
+ @Override
+ public long getId() { return id_; }
+
+ @Override
+ public List<LiteralExpr> getPartitionValues() { return partitionValues_; }
+
+ String getName() { return name_; }
+}
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalTable.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalTable.java
index 191f248..246b691 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalTable.java
@@ -56,8 +56,8 @@ import com.google.common.collect.Lists;
class LocalTable implements FeTable {
protected final LocalDb db_;
/** The lower-case name of the table. */
- private final String name_;
- private final SchemaInfo schemaInfo_;
+ protected final String name_;
+ protected final SchemaInfo schemaInfo_;
public static LocalTable load(LocalDb db, String tblName) {
// In order to know which kind of table subclass to instantiate, we need
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/MetaProvider.java b/fe/src/main/java/org/apache/impala/catalog/local/MetaProvider.java
index 6b25257..42bf40e 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/MetaProvider.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/MetaProvider.java
@@ -17,9 +17,13 @@
package org.apache.impala.catalog.local;
+import java.util.List;
+import java.util.Map;
+
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.api.UnknownDBException;
import org.apache.thrift.TException;
@@ -48,4 +52,17 @@ interface MetaProvider {
String loadNullPartitionKeyValue()
throws MetaException, TException;
+
+ List<String> loadPartitionNames(String dbName, String tableName)
+ throws MetaException, TException;
+
+ /**
+ * Load the given partitions from the specified table.
+ *
+ * If a requested partition does not exist, no exception will be thrown.
+ * Instead, the resulting map will contain no entry for that partition.
+ */
+ Map<String, Partition> loadPartitionsByNames(String dbName, String tableName,
+ List<String> partitionColumnNames, List<String> partitionNames)
+ throws MetaException, TException;
}
diff --git a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
index 23bd142..03258e4 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -23,7 +23,6 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -45,7 +44,6 @@ import org.apache.hadoop.hive.metastore.api.DecimalColumnStatsData;
import org.apache.hadoop.hive.metastore.api.DoubleColumnStatsData;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.LongColumnStatsData;
-import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.PrincipalType;
@@ -3338,7 +3336,7 @@ public class CatalogOpExecutor {
partition.setDbName(tblName.getDb());
partition.setTableName(tblName.getTbl());
- partition.setValues(getPartValsFromName(msTbl, partName));
+ partition.setValues(MetaStoreUtil.getPartValsFromName(msTbl, partName));
partition.setParameters(new HashMap<String, String>());
partition.setSd(msTbl.getSd().deepCopy());
partition.getSd().setSerdeInfo(msTbl.getSd().getSerdeInfo().deepCopy());
@@ -3438,23 +3436,6 @@ public class CatalogOpExecutor {
return response;
}
- private List<String> getPartValsFromName(org.apache.hadoop.hive.metastore.api.Table
- msTbl, String partName) throws MetaException, CatalogException {
- Preconditions.checkNotNull(msTbl);
- LinkedHashMap<String, String> hm =
- org.apache.hadoop.hive.metastore.Warehouse.makeSpecFromName(partName);
- List<String> partVals = Lists.newArrayList();
- for (FieldSchema field: msTbl.getPartitionKeys()) {
- String key = field.getName();
- String val = hm.get(key);
- if (val == null) {
- throw new CatalogException("Incomplete partition name - missing " + key);
- }
- partVals.add(val);
- }
- return partVals;
- }
-
/**
* Returns an existing, loaded table from the Catalog. Throws an exception if any
* of the following are true:
diff --git a/fe/src/main/java/org/apache/impala/util/MetaStoreUtil.java b/fe/src/main/java/org/apache/impala/util/MetaStoreUtil.java
index 5a5654a..bffd46a 100644
--- a/fe/src/main/java/org/apache/impala/util/MetaStoreUtil.java
+++ b/fe/src/main/java/org/apache/impala/util/MetaStoreUtil.java
@@ -18,14 +18,19 @@
package org.apache.impala.util;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.Warehouse;
import org.apache.hadoop.hive.metastore.api.ConfigValSecurityException;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.compat.MetastoreShim;
@@ -260,4 +265,20 @@ public class MetaStoreUtil {
}
return Joiner.on(",").join(outputList);
}
+
+ public static List<String> getPartValsFromName(Table msTbl, String partName)
+ throws MetaException, CatalogException {
+ Preconditions.checkNotNull(msTbl);
+ LinkedHashMap<String, String> hm = Warehouse.makeSpecFromName(partName);
+ List<String> partVals = Lists.newArrayList();
+ for (FieldSchema field: msTbl.getPartitionKeys()) {
+ String key = field.getName();
+ String val = hm.get(key);
+ if (val == null) {
+ throw new CatalogException("Incomplete partition name - missing " + key);
+ }
+ partVals.add(val);
+ }
+ return partVals;
+ }
}
diff --git a/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java b/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
index eeceb36..856fc95 100644
--- a/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
@@ -287,6 +287,13 @@ public class CatalogTest {
public void TestPartitions() throws CatalogException {
HdfsTable table =
(HdfsTable) catalog_.getOrLoadTable("functional", "AllTypes");
+ checkAllTypesPartitioning(table, true);
+ }
+
+ public static void checkAllTypesPartitioning(FeFsTable table,
+ boolean checkFileDescriptors) {
+ assertEquals(24, table.getPartitionIds().size());
+ assertEquals(24, table.getPartitions().size());
Collection<? extends FeFsPartition> partitions =
FeCatalogUtils.loadAllPartitions(table);
@@ -309,7 +316,11 @@ public class CatalogTest {
months.add(key1 * 100 + key2);
- assertEquals(p.getFileDescriptors().size(), 1);
+ if (checkFileDescriptors) {
+ // TODO(todd): once LocalCatalog supports file descriptors,
+ // no need for this boolean anymore.
+ assertEquals(p.getFileDescriptors().size(), 1);
+ }
}
assertEquals(months.size(), 24);
}
diff --git a/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java b/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java
index a296399..aabb873 100644
--- a/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java
@@ -22,7 +22,9 @@ import static org.junit.Assert.*;
import java.util.Set;
import org.apache.impala.catalog.CatalogTest;
+import org.apache.impala.catalog.FeCatalogUtils;
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.Type;
@@ -33,6 +35,7 @@ import org.junit.Before;
import org.junit.Test;
import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
public class LocalCatalogTest {
@@ -99,7 +102,33 @@ public class LocalCatalogTest {
FeFsTable fsTable = (FeFsTable) t;
assertEquals(MetaStoreUtil.DEFAULT_NULL_PARTITION_KEY_VALUE,
fsTable.getNullPartitionKeyValue());
+
+ // Stats should have one row per partition, plus a "total" row.
TResultSet stats = fsTable.getTableStats();
- assertEquals(1, stats.getRowsSize());
+ assertEquals(25, stats.getRowsSize());
+ }
+
+ @Test
+ public void testPartitioning() throws Exception {
+ FeFsTable t = (FeFsTable) catalog_.getTable("functional", "alltypes");
+ // TODO(todd): once we support file descriptors in LocalCatalog,
+ // run the full test.
+ CatalogTest.checkAllTypesPartitioning(t, /*checkFileDescriptors=*/false);
+ }
+
+ /**
+ * Test that partitions with a NULL value can be properly loaded.
+ */
+ @Test
+ public void testEmptyPartitionValue() throws Exception {
+ FeFsTable t = (FeFsTable) catalog_.getTable("functional", "alltypesagg");
+ // This table has one partition with a NULL value for the 'day'
+ // clustering column.
+ int dayCol = t.getColumn("day").getPosition();
+ Set<Long> ids = t.getNullPartitionIds(dayCol);
+ assertEquals(1, ids.size());
+ FeFsPartition partition = FeCatalogUtils.loadPartition(
+ t, Iterables.getOnlyElement(ids));
+ assertTrue(partition.getPartitionValue(dayCol).isNullLiteral());
}
}