You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by dh...@apache.org on 2018/06/20 20:13:14 UTC
[1/4] impala git commit: IMPALA-7140 (part 2). Create skeleton for
LocalFsTable
Repository: impala
Updated Branches:
refs/heads/master afe114682 -> 11aaa6caa
IMPALA-7140 (part 2). Create skeleton for LocalFsTable
This adds a skeleton implementation of FeFsTable for the LocalCatalog.
Most of the implementation in this patch is stubbed out -- only the bare
bones necessary to instantiate the subclass when a table is loaded and
the simplest table-level methods.
Change-Id: Id2b184104d92e128250df5a08ac7ffb3dde011a8
Reviewed-on: http://gerrit.cloudera.org:8080/10712
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/08cf0aaf
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/08cf0aaf
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/08cf0aaf
Branch: refs/heads/master
Commit: 08cf0aafae7d383afe40c195a6a86055e2a252d8
Parents: 0063e0c
Author: Todd Lipcon <to...@cloudera.com>
Authored: Tue Jun 12 18:43:43 2018 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Wed Jun 20 05:07:22 2018 +0000
----------------------------------------------------------------------
.../org/apache/impala/catalog/FeFsTable.java | 11 ++
.../org/apache/impala/catalog/HdfsTable.java | 43 +++--
.../catalog/local/DirectMetaProvider.java | 8 +
.../impala/catalog/local/LocalCatalog.java | 14 ++
.../catalog/local/LocalCatalogException.java | 4 +
.../impala/catalog/local/LocalFsTable.java | 176 +++++++++++++++++++
.../apache/impala/catalog/local/LocalTable.java | 14 +-
.../impala/catalog/local/MetaProvider.java | 3 +
.../org/apache/impala/util/MetaStoreUtil.java | 19 +-
.../impala/catalog/local/LocalCatalogTest.java | 10 ++
10 files changed, 278 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/08cf0aaf/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 02c6211..86bf2eb 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeFsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeFsTable.java
@@ -32,6 +32,9 @@ import org.apache.impala.util.ListMap;
/**
* Frontend interface for interacting with a filesystem-backed table.
+ *
+ * TODO(vercegovac): various method names and comments in this interface refer
+ * to HDFS where they should be more generically "Fs".
*/
public interface FeFsTable extends FeTable {
/** hive's default value for table property 'serialization.null.format' */
@@ -61,6 +64,9 @@ public interface FeFsTable extends FeTable {
/**
* @return the value Hive is configured to use for NULL partition key values.
+ *
+ * TODO(todd): this is an HMS-wide setting, rather than a per-table setting.
+ * Perhaps this should move to the FeCatalog interface?
*/
public String getNullPartitionKeyValue();
@@ -84,6 +90,11 @@ public interface FeFsTable extends FeTable {
long getTotalHdfsBytes();
/**
+ * @return the total number of files stored on HDFS for this table
+ */
+ long getTotalNumFiles();
+
+ /**
* @return true if this table is backed by the Avro file format
*/
boolean isAvroTable();
http://git-wip-us.apache.org/repos/asf/impala/blob/08cf0aaf/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 fea85f7..0106c5a 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -1255,8 +1255,7 @@ public class HdfsTable extends Table implements FeFsTable {
try {
if (loadTableSchema) {
// set nullPartitionKeyValue from the hive conf.
- nullPartitionKeyValue_ = client.getConfigValue(
- "hive.exec.default.partition.name", "__HIVE_DEFAULT_PARTITION__");
+ nullPartitionKeyValue_ = MetaStoreUtil.getNullPartitionKeyValue(client);
loadSchema(msTbl);
loadAllColumnStats(client);
}
@@ -1776,6 +1775,9 @@ public class HdfsTable extends Table implements FeFsTable {
public long getTotalHdfsBytes() { return fileMetadataStats_.totalFileBytes; }
@Override // FeFsTable
+ public long getTotalNumFiles() { return fileMetadataStats_.numFiles; }
+
+ @Override // FeFsTable
public String getHdfsBaseDir() { return hdfsBaseDir_; }
public Path getHdfsBaseDirPath() { return new Path(hdfsBaseDir_); }
@Override // FeFsTable
@@ -1989,18 +1991,24 @@ public class HdfsTable extends Table implements FeFsTable {
@Override // FeFsTable
public TResultSet getTableStats() {
+ return getTableStats(this);
+ }
+
+ // TODO(todd): move to FeCatalogUtils. Upon moving to Java 8, could be
+ // a default method of FeFsTable.
+ public static TResultSet getTableStats(FeFsTable table) {
TResultSet result = new TResultSet();
TResultSetMetadata resultSchema = new TResultSetMetadata();
result.setSchema(resultSchema);
- for (int i = 0; i < numClusteringCols_; ++i) {
+ for (int i = 0; i < table.getNumClusteringCols(); ++i) {
// Add the partition-key values as strings for simplicity.
- Column partCol = getColumns().get(i);
+ Column partCol = table.getColumns().get(i);
TColumn colDesc = new TColumn(partCol.getName(), Type.STRING.toThrift());
resultSchema.addToColumns(colDesc);
}
- boolean statsExtrap = isStatsExtrapolationEnabled();
+ boolean statsExtrap = table.isStatsExtrapolationEnabled();
resultSchema.addToColumns(new TColumn("#Rows", Type.BIGINT.toThrift()));
if (statsExtrap) {
@@ -2015,12 +2023,12 @@ public class HdfsTable extends Table implements FeFsTable {
resultSchema.addToColumns(new TColumn("Location", Type.STRING.toThrift()));
// Pretty print partitions and their stats.
- ArrayList<HdfsPartition> orderedPartitions =
- Lists.newArrayList(partitionMap_.values());
+ List<FeFsPartition> orderedPartitions = new ArrayList<>(
+ FeCatalogUtils.loadAllPartitions(table));
Collections.sort(orderedPartitions, HdfsPartition.KV_COMPARATOR);
long totalCachedBytes = 0L;
- for (HdfsPartition p: orderedPartitions) {
+ for (FeFsPartition p: orderedPartitions) {
TResultRowBuilder rowBuilder = new TResultRowBuilder();
// Add the partition-key values (as strings for simplicity).
@@ -2033,7 +2041,7 @@ public class HdfsTable extends Table implements FeFsTable {
// Compute and report the extrapolated row count because the set of files could
// have changed since we last computed stats for this partition. We also follow
// this policy during scan-cardinality estimation.
- if (statsExtrap) rowBuilder.add(getExtrapolatedNumRows(p.getSize()));
+ if (statsExtrap) rowBuilder.add(table.getExtrapolatedNumRows(p.getSize()));
rowBuilder.add(p.getFileDescriptors().size()).addBytes(p.getSize());
if (!p.isMarkedCached()) {
// Helps to differentiate partitions that have 0B cached versus partitions
@@ -2058,7 +2066,7 @@ public class HdfsTable extends Table implements FeFsTable {
// Extract cache replication factor from the parameters of the table
// if the table is not partitioned or directly from the partition.
Short rep = HdfsCachingUtil.getCachedCacheReplication(
- numClusteringCols_ == 0 ?
+ table.getNumClusteringCols() == 0 ?
p.getTable().getMetaStoreTable().getParameters() :
p.getParameters());
rowBuilder.add(rep.toString());
@@ -2071,9 +2079,9 @@ public class HdfsTable extends Table implements FeFsTable {
}
// For partitioned tables add a summary row at the bottom.
- if (numClusteringCols_ > 0) {
+ if (table.getNumClusteringCols() > 0) {
TResultRowBuilder rowBuilder = new TResultRowBuilder();
- int numEmptyCells = numClusteringCols_ - 1;
+ int numEmptyCells = table.getNumClusteringCols() - 1;
rowBuilder.add("Total");
for (int i = 0; i < numEmptyCells; ++i) {
rowBuilder.add("");
@@ -2081,15 +2089,15 @@ public class HdfsTable extends Table implements FeFsTable {
// Total rows, extrapolated rows, files, bytes, cache stats.
// Leave format empty.
- rowBuilder.add(getNumRows());
+ rowBuilder.add(table.getNumRows());
// Compute and report the extrapolated row count because the set of files could
// have changed since we last computed stats for this partition. We also follow
// this policy during scan-cardinality estimation.
if (statsExtrap) {
- rowBuilder.add(getExtrapolatedNumRows(fileMetadataStats_.totalFileBytes));
+ rowBuilder.add(table.getExtrapolatedNumRows(table.getTotalHdfsBytes()));
}
- rowBuilder.add(fileMetadataStats_.numFiles)
- .addBytes(fileMetadataStats_.totalFileBytes)
+ rowBuilder.add(table.getTotalNumFiles())
+ .addBytes(table.getTotalHdfsBytes())
.addBytes(totalCachedBytes).add("").add("").add("").add("");
result.addToRows(rowBuilder.get());
}
@@ -2279,7 +2287,8 @@ public class HdfsTable extends Table implements FeFsTable {
HiveConf hiveConf = new HiveConf(HdfsTable.class);
// set nullPartitionKeyValue from the hive conf.
tmpTable.nullPartitionKeyValue_ = hiveConf.get(
- "hive.exec.default.partition.name", "__HIVE_DEFAULT_PARTITION__");
+ MetaStoreUtil.NULL_PARTITION_KEY_VALUE_CONF_KEY,
+ MetaStoreUtil.DEFAULT_NULL_PARTITION_KEY_VALUE);
tmpTable.loadSchema(msTbl);
tmpTable.initializePartitionMetadata(msTbl);
tmpTable.setTableStats(msTbl);
http://git-wip-us.apache.org/repos/asf/impala/blob/08cf0aaf/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
----------------------------------------------------------------------
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 0867091..ac86bfd 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
@@ -26,6 +26,7 @@ import org.apache.impala.catalog.MetaStoreClientPool;
import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.thrift.TBackendGflags;
+import org.apache.impala.util.MetaStoreUtil;
import org.apache.thrift.TException;
import com.google.common.collect.ImmutableList;
@@ -80,4 +81,11 @@ class DirectMetaProvider implements MetaProvider {
return c.getHiveClient().getTable(dbName, tableName);
}
}
+
+ @Override
+ public String loadNullPartitionKeyValue() throws MetaException, TException {
+ try (MetaStoreClient c = msClientPool_.getClient()) {
+ return MetaStoreUtil.getNullPartitionKeyValue(c.getHiveClient());
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/impala/blob/08cf0aaf/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 2b246e6..007450d 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
@@ -66,6 +66,7 @@ import com.google.common.collect.Maps;
public class LocalCatalog implements FeCatalog {
private final MetaProvider metaProvider_;
private Map<String, FeDb> dbs_ = Maps.newHashMap();
+ private String nullPartitionKeyValue_;
private static final Db builtinsDb_ = new BuiltinsDb(ImpaladCatalog.BUILTINS_DB);
public static FeCatalog create(String defaultKuduMasterHosts) {
@@ -200,6 +201,19 @@ public class LocalCatalog implements FeCatalog {
throw new UnsupportedOperationException("TODO");
}
+ public String getNullPartitionKeyValue() {
+ if (nullPartitionKeyValue_ == null) {
+ try {
+ nullPartitionKeyValue_ =
+ metaProvider_.loadNullPartitionKeyValue();
+ } catch (TException e) {
+ throw new LocalCatalogException(
+ "Could not load null partition key value", e);
+ }
+ }
+ return nullPartitionKeyValue_;
+ }
+
@Override
public boolean isReady() {
// We are always ready.
http://git-wip-us.apache.org/repos/asf/impala/blob/08cf0aaf/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalogException.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalogException.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalogException.java
index 7b4624c..6875bd3 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalogException.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalogException.java
@@ -34,4 +34,8 @@ public class LocalCatalogException extends RuntimeException {
public LocalCatalogException(Throwable cause) {
super(cause);
}
+
+ public LocalCatalogException(String msg) {
+ super(msg);
+ }
}
http://git-wip-us.apache.org/repos/asf/impala/blob/08cf0aaf/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
new file mode 100644
index 0000000..b6ab559
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java
@@ -0,0 +1,176 @@
+// 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.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+
+import org.apache.impala.analysis.LiteralExpr;
+import org.apache.impala.catalog.CatalogException;
+import org.apache.impala.catalog.FeFsPartition;
+import org.apache.impala.catalog.FeFsTable;
+import org.apache.impala.catalog.HdfsTable;
+import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
+import org.apache.impala.catalog.PrunablePartition;
+import org.apache.impala.thrift.TNetworkAddress;
+import org.apache.impala.thrift.TPartitionKeyValue;
+import org.apache.impala.thrift.TResultSet;
+import org.apache.impala.util.ListMap;
+
+public class LocalFsTable extends LocalTable implements FeFsTable {
+ public LocalFsTable(LocalDb db, String tblName, SchemaInfo schemaInfo) {
+ super(db, tblName, schemaInfo);
+ }
+
+ @Override
+ public boolean isCacheable() {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ @Override
+ public boolean isLocationCacheable() {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ @Override
+ public boolean isMarkedCached() {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ @Override
+ public String getLocation() {
+ return getMetaStoreTable().getSd().getLocation();
+ }
+
+ @Override
+ public String getNullPartitionKeyValue() {
+ return db_.getCatalog().getNullPartitionKeyValue();
+ }
+
+ @Override
+ public TResultSet getFiles(List<List<TPartitionKeyValue>> partitionSet)
+ throws CatalogException {
+ // TODO(todd): implement fetching files from HDFS
+ return null;
+ }
+
+ @Override
+ public String getHdfsBaseDir() {
+ // TODO(todd): this is redundant with getLocation, it seems.
+ return getLocation();
+ }
+
+ @Override
+ public long getTotalHdfsBytes() {
+ // TODO Auto-generated method stub
+ return 0;
+ }
+
+ @Override
+ public long getTotalNumFiles() {
+ // TODO Auto-generated method stub
+ return 0;
+ }
+
+ @Override
+ public boolean isAvroTable() {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ @Override
+ public long getExtrapolatedNumRows(long totalBytes) {
+ // TODO Auto-generated method stub
+ return 0;
+ }
+
+ @Override
+ public boolean isStatsExtrapolationEnabled() {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ @Override
+ public TResultSet getTableStats() {
+ return HdfsTable.getTableStats(this);
+ }
+
+ @Override
+ public Collection<? extends PrunablePartition> getPartitions() {
+ // TODO(todd): implement me
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Set<Long> getPartitionIds() {
+ // TODO(todd): implement me
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Map<Long, ? extends PrunablePartition> getPartitionMap() {
+ // TODO(todd): implement me
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public TreeMap<LiteralExpr, HashSet<Long>> getPartitionValueMap(int col) {
+ // TODO(todd): implement me
+ return new TreeMap<>();
+ }
+
+ @Override
+ public Set<Long> getNullPartitionIds(int colIdx) {
+ // TODO(todd): implement me
+ return Collections.emptySet();
+ }
+
+ @Override
+ public List<? extends FeFsPartition> loadPartitions(Collection<Long> ids) {
+ // TODO(todd): implement me
+ return Collections.emptyList();
+ }
+
+ @Override
+ public int parseSkipHeaderLineCount(StringBuilder error) {
+ // TODO Auto-generated method stub
+ return 0;
+ }
+
+ @Override
+ public Map<Long, List<FileDescriptor>> getFilesSample(
+ Collection<? extends FeFsPartition> inputParts,
+ long percentBytes, long minSampleBytes, long randomSeed) {
+ // TODO(todd): implement me
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public ListMap<TNetworkAddress> getHostIndex() {
+ // TODO(todd): implement me
+ return new ListMap<>();
+ }
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/08cf0aaf/fe/src/main/java/org/apache/impala/catalog/local/LocalTable.java
----------------------------------------------------------------------
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 d6b0c88..191f248 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
@@ -32,6 +32,7 @@ import org.apache.impala.catalog.FeCatalogUtils;
import org.apache.impala.catalog.FeDb;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeTable;
+import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.StructField;
import org.apache.impala.catalog.StructType;
import org.apache.impala.catalog.TableLoadingException;
@@ -53,7 +54,7 @@ import com.google.common.collect.Lists;
* each catalog instance.
*/
class LocalTable implements FeTable {
- private final LocalDb db_;
+ protected final LocalDb db_;
/** The lower-case name of the table. */
private final String name_;
private final SchemaInfo schemaInfo_;
@@ -62,10 +63,13 @@ class LocalTable implements FeTable {
// In order to know which kind of table subclass to instantiate, we need
// to eagerly grab and parse the top-level Table object from the HMS.
SchemaInfo schemaInfo = SchemaInfo.load(db, tblName);
+ if (HdfsFileFormat.isHdfsInputFormatClass(
+ schemaInfo.msTable_.getSd().getInputFormat())) {
+ return new LocalFsTable(db, tblName, schemaInfo);
+ }
- // TODO: change this function to instantiate the appropriate
- // subclass based on the table type (eg view, hbase table, etc)
- return new LocalTable(db, tblName, schemaInfo);
+ throw new LocalCatalogException("Unknown table type for table " +
+ db.getName() + "." + tblName);
}
public LocalTable(LocalDb db, String tblName, SchemaInfo schemaInfo) {
@@ -191,7 +195,7 @@ class LocalTable implements FeTable {
* effort, could be refactored to avoid duplication.
*/
@Immutable
- private static class SchemaInfo {
+ protected static class SchemaInfo {
private final Table msTable_;
private final ArrayType type_;
http://git-wip-us.apache.org/repos/asf/impala/blob/08cf0aaf/fe/src/main/java/org/apache/impala/catalog/local/MetaProvider.java
----------------------------------------------------------------------
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 cd8d3b9..6b25257 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
@@ -45,4 +45,7 @@ interface MetaProvider {
Table loadTable(String dbName, String tableName)
throws NoSuchObjectException, MetaException, TException;
+
+ String loadNullPartitionKeyValue()
+ throws MetaException, TException;
}
http://git-wip-us.apache.org/repos/asf/impala/blob/08cf0aaf/fe/src/main/java/org/apache/impala/util/MetaStoreUtil.java
----------------------------------------------------------------------
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 34082f4..5a5654a 100644
--- a/fe/src/main/java/org/apache/impala/util/MetaStoreUtil.java
+++ b/fe/src/main/java/org/apache/impala/util/MetaStoreUtil.java
@@ -23,6 +23,7 @@ import java.util.Map;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.ConfigValSecurityException;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.impala.catalog.HdfsTable;
@@ -31,8 +32,6 @@ import org.apache.impala.compat.MetastoreShim;
import org.apache.log4j.Logger;
import org.apache.thrift.TException;
-import org.apache.impala.catalog.HdfsTable;
-import org.apache.impala.common.AnalysisException;
import org.apache.impala.thrift.TColumn;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
@@ -72,6 +71,13 @@ public class MetaStoreUtil {
// Hive configuration.
private static short maxPartitionsPerRpc_ = DEFAULT_MAX_PARTITIONS_PER_RPC;
+ // The configuration key that Hive uses to set the null partition key value.
+ public static final String NULL_PARTITION_KEY_VALUE_CONF_KEY =
+ "hive.exec.default.partition.name";
+ // The default value for the above configuration key.
+ public static final String DEFAULT_NULL_PARTITION_KEY_VALUE =
+ "__HIVE_DEFAULT_PARTITION__";
+
static {
// Get the value from the Hive configuration, if present.
HiveConf hiveConf = new HiveConf(HdfsTable.class);
@@ -92,6 +98,15 @@ public class MetaStoreUtil {
}
/**
+ * Return the value that Hive is configured to use for NULL partition key values.
+ */
+ public static String getNullPartitionKeyValue(IMetaStoreClient client)
+ throws ConfigValSecurityException, TException {
+ return client.getConfigValue(
+ NULL_PARTITION_KEY_VALUE_CONF_KEY, DEFAULT_NULL_PARTITION_KEY_VALUE);
+ }
+
+ /**
* Fetches all partitions for a table in batches, with each batch containing at most
* 'maxPartsPerRpc' partitions. Returns a List containing all fetched Partitions.
* Will throw a MetaException if existing partitions are dropped while a fetch is in
http://git-wip-us.apache.org/repos/asf/impala/blob/08cf0aaf/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java
----------------------------------------------------------------------
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 c8315df..a296399 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
@@ -23,8 +23,11 @@ import java.util.Set;
import org.apache.impala.catalog.CatalogTest;
import org.apache.impala.catalog.FeDb;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.Type;
+import org.apache.impala.thrift.TResultSet;
+import org.apache.impala.util.MetaStoreUtil;
import org.apache.impala.util.PatternMatcher;
import org.junit.Before;
import org.junit.Test;
@@ -91,5 +94,12 @@ public class LocalCatalogTest {
Type.TIMESTAMP});
FeTable t = functionalDb.getTable("alltypes");
assertEquals(7300, t.getNumRows());
+
+ assertTrue(t instanceof LocalFsTable);
+ FeFsTable fsTable = (FeFsTable) t;
+ assertEquals(MetaStoreUtil.DEFAULT_NULL_PARTITION_KEY_VALUE,
+ fsTable.getNullPartitionKeyValue());
+ TResultSet stats = fsTable.getTableStats();
+ assertEquals(1, stats.getRowsSize());
}
}
[2/4] impala git commit: IMPALA-7141 (part 2). Extract interfaces for
partition pruning without loading
Posted by dh...@apache.org.
IMPALA-7141 (part 2). Extract interfaces for partition pruning without loading
This adds a new PrunablePartition interface which HdfsPartition
implements. The interface includes only the partition ID and the
partition key/values.
For the case of the current catalog implementation, this provides no
benefit. However, for LocalCatalog, we want to defer loading partition
information until after pruning. With this interface, we can construct
PrunablePartition objects using just the partition names, and then load
the partitions once pruning is complete.
Change-Id: Ifdcfd6cffcd298bbf44531e1ec2f47c3a5b7d1fa
Reviewed-on: http://gerrit.cloudera.org:8080/10631
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/0063e0c5
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/0063e0c5
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/0063e0c5
Branch: refs/heads/master
Commit: 0063e0c5d7747aca39a8d9c53cc1c52eed5c884b
Parents: 747e0a9
Author: Todd Lipcon <to...@cloudera.com>
Authored: Wed Jun 6 19:56:16 2018 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Wed Jun 20 05:07:22 2018 +0000
----------------------------------------------------------------------
.../analysis/AlterTableSetCachedStmt.java | 2 +-
.../analysis/AlterTableSetLocationStmt.java | 2 +-
.../impala/analysis/ComputeStatsStmt.java | 12 ++++--
.../apache/impala/analysis/LoadDataStmt.java | 4 --
.../apache/impala/analysis/PartitionSet.java | 6 +--
.../apache/impala/catalog/FeCatalogUtils.java | 28 +++++++++++++
.../org/apache/impala/catalog/FeFsTable.java | 13 ++++--
.../apache/impala/catalog/HdfsPartition.java | 2 +-
.../org/apache/impala/catalog/HdfsTable.java | 19 +++++++--
.../impala/catalog/PrunablePartition.java | 44 ++++++++++++++++++++
.../impala/planner/HdfsPartitionFilter.java | 9 ++--
.../impala/planner/HdfsPartitionPruner.java | 27 +++++++-----
.../org/apache/impala/planner/HdfsScanNode.java | 4 +-
.../impala/planner/SingleNodePlanner.java | 2 +-
.../impala/service/CatalogOpExecutor.java | 31 ++++++++++----
.../catalog/CatalogObjectToFromThriftTest.java | 12 ++++--
.../org/apache/impala/catalog/CatalogTest.java | 3 +-
.../impala/testutil/BlockIdGenerator.java | 7 +++-
18 files changed, 175 insertions(+), 52 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/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 0ffca1e..178725d 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AlterTableSetCachedStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AlterTableSetCachedStmt.java
@@ -74,7 +74,7 @@ public class AlterTableSetCachedStmt extends AlterTableSetStmt {
HdfsTable hdfsTable = (HdfsTable)table;
StringBuilder nameSb = new StringBuilder();
if (partitionSet != null) {
- List<FeFsPartition> parts = partitionSet.getPartitions();
+ List<? extends FeFsPartition> parts = partitionSet.getPartitions();
nameSb.append("Partition(s) (");
for(FeFsPartition part: parts) {
isCacheable = isCacheable && part.isCacheable();
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/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 524f076..db8a804 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AlterTableSetLocationStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AlterTableSetLocationStmt.java
@@ -81,7 +81,7 @@ public class AlterTableSetLocationStmt extends AlterTableSetStmt {
HdfsTable hdfsTable = (HdfsTable) table;
if (getPartitionSet() != null) {
// Targeting a partition rather than a table.
- List<FeFsPartition> partitions = getPartitionSet().getPartitions();
+ List<? extends FeFsPartition> partitions = getPartitionSet().getPartitions();
if (partitions.isEmpty()) { return; }
if (partitions.size() != 1) {
// Sort the partitions to get a consistent error reporting.
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/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 9259e49..7743120 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
@@ -27,6 +27,7 @@ import java.util.Set;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.impala.authorization.Privilege;
import org.apache.impala.catalog.Column;
+import org.apache.impala.catalog.FeCatalogUtils;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeTable;
@@ -438,7 +439,7 @@ public class ComputeStatsStmt extends StatementBase {
}
Collection<? extends FeFsPartition> allPartitions =
- hdfsTable.getPartitions();
+ FeCatalogUtils.loadAllPartitions(hdfsTable);
for (FeFsPartition p: allPartitions) {
TPartitionStats partStats = p.getPartitionStats();
if (!p.hasIncrementalStats() || tableIsMissingColStats) {
@@ -476,7 +477,8 @@ public class ComputeStatsStmt extends StatementBase {
// TODO(todd) avoid loading all partitions.
HashSet<FeFsPartition> targetPartitions =
Sets.newHashSet(partitionSet_.getPartitions());
- Collection<? extends FeFsPartition> allPartitions = hdfsTable.getPartitions();
+ Collection<? extends FeFsPartition> allPartitions =
+ FeCatalogUtils.loadAllPartitions(hdfsTable);
for (FeFsPartition p : allPartitions) {
if (targetPartitions.contains(p)) continue;
TPartitionStats partStats = p.getPartitionStats();
@@ -610,7 +612,9 @@ public class ComputeStatsStmt extends StatementBase {
// Compute the sample of files and set 'sampleFileBytes_'.
long minSampleBytes = analyzer.getQueryOptions().compute_stats_min_sample_size;
long samplePerc = sampleParams_.getPercentBytes();
- Collection<? extends FeFsPartition> partitions = hdfsTable.getPartitions();
+ // TODO(todd): can we avoid loading all the partitions for this?
+ Collection<? extends FeFsPartition> partitions =
+ FeCatalogUtils.loadAllPartitions(hdfsTable);
Map<Long, List<FileDescriptor>> sample = hdfsTable.getFilesSample(
partitions, samplePerc, minSampleBytes, sampleSeed);
long sampleFileBytes = 0;
@@ -758,7 +762,7 @@ public class ComputeStatsStmt extends StatementBase {
affectedPartitions = partitionSet_.getPartitions();
} else {
FeFsTable hdfsTable = (FeFsTable)table_;
- affectedPartitions = hdfsTable.getPartitions();
+ affectedPartitions = FeCatalogUtils.loadAllPartitions(hdfsTable);
}
for (FeFsPartition partition: affectedPartitions) {
if (partition.getFileFormat() != HdfsFileFormat.PARQUET
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/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 e20a88c..43ca216 100644
--- a/fe/src/main/java/org/apache/impala/analysis/LoadDataStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/LoadDataStmt.java
@@ -21,7 +21,6 @@ import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.List;
-import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.adl.AdlFileSystem;
@@ -29,14 +28,11 @@ 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.FeFsPartition;
import org.apache.impala.catalog.FeTable;
-import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.HdfsPartition;
import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.common.FileSystemUtil;
-import org.apache.impala.thrift.ImpalaInternalServiceConstants;
import org.apache.impala.thrift.TLoadDataReq;
import org.apache.impala.thrift.TTableName;
import org.apache.impala.util.FsPermissionChecker;
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/fe/src/main/java/org/apache/impala/analysis/PartitionSet.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/PartitionSet.java b/fe/src/main/java/org/apache/impala/analysis/PartitionSet.java
index f04504c..46a5d7b 100644
--- a/fe/src/main/java/org/apache/impala/analysis/PartitionSet.java
+++ b/fe/src/main/java/org/apache/impala/analysis/PartitionSet.java
@@ -43,14 +43,14 @@ import com.google.common.collect.Sets;
public class PartitionSet extends PartitionSpecBase {
private final List<Expr> partitionExprs_;
- // Result of analysis
- private List<FeFsPartition> partitions_ = Lists.newArrayList();
+ // Result of analysis, null until analysis is complete.
+ private List<? extends FeFsPartition> partitions_;
public PartitionSet(List<Expr> partitionExprs) {
this.partitionExprs_ = ImmutableList.copyOf(partitionExprs);
}
- public List<FeFsPartition> getPartitions() { return partitions_; }
+ public List<? extends FeFsPartition> getPartitions() { return partitions_; }
@Override
public void analyze(Analyzer analyzer) throws AnalysisException {
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/fe/src/main/java/org/apache/impala/catalog/FeCatalogUtils.java
----------------------------------------------------------------------
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 15f139b..7d811c2 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeCatalogUtils.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeCatalogUtils.java
@@ -17,12 +17,15 @@
package org.apache.impala.catalog;
+import java.util.Collection;
+import java.util.Collections;
import java.util.Map;
import org.apache.hadoop.hive.common.StatsSetupConst;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
/**
* Static utility functions shared between FeCatalog implementations.
@@ -120,4 +123,29 @@ public abstract class FeCatalogUtils {
return -1;
}
+ /**
+ * Convenience method to load exactly one partition from a table.
+ *
+ * TODO(todd): upon moving to Java8 this could be a default method
+ * in FeFsTable.
+ */
+ public static FeFsPartition loadPartition(FeFsTable table,
+ long partitionId) {
+ Collection<? extends FeFsPartition> partCol = table.loadPartitions(
+ Collections.singleton(partitionId));
+ if (partCol.size() != 1) {
+ throw new AssertionError(String.format(
+ "expected exactly one result fetching partition ID %s from table %s " +
+ "(got %s)", partitionId, table.getFullName(), partCol.size()));
+ }
+ return Iterables.getOnlyElement(partCol);
+ }
+
+ /**
+ * Load all partitions from the given table.
+ */
+ public static Collection<? extends FeFsPartition> loadAllPartitions(
+ FeFsTable table) {
+ return table.loadPartitions(table.getPartitionIds());
+ }
}
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/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 c64b922..02c6211 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeFsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeFsTable.java
@@ -109,7 +109,7 @@ public interface FeFsTable extends FeTable {
/**
* @return all partitions of this table
*/
- Collection<? extends FeFsPartition> getPartitions();
+ Collection<? extends PrunablePartition> getPartitions();
/**
* @return identifiers for all partitions in this table
@@ -117,9 +117,9 @@ public interface FeFsTable extends FeTable {
public Set<Long> getPartitionIds();
/**
- * @return the map from partition identifier to partition object
+ * Returns the map from partition identifier to prunable partition.
*/
- Map<Long, ? extends FeFsPartition> getPartitionMap();
+ Map<Long, ? extends PrunablePartition> getPartitionMap();
/**
* @param the index of the target partitioning column
@@ -135,6 +135,13 @@ public interface FeFsTable extends FeTable {
Set<Long> getNullPartitionIds(int colIdx);
/**
+ * Returns the full partition objects for the given partition IDs, which must
+ * have been obtained by prior calls to the above methods.
+ * @throws IllegalArgumentException if any partition ID does not exist
+ */
+ List<? extends FeFsPartition> loadPartitions(Collection<Long> ids);
+
+ /**
* Parses and returns the value of the 'skip.header.line.count' table property. If the
* value is not set for the table, returns 0. If parsing fails or a value < 0 is found,
* the error parameter is updated to contain an error message.
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
----------------------------------------------------------------------
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 b265c6a..a6c128b 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
@@ -75,7 +75,7 @@ import com.google.flatbuffers.FlatBufferBuilder;
* order with NULLs sorting last. The ordering is useful for displaying partitions
* in SHOW statements.
*/
-public class HdfsPartition implements FeFsPartition {
+public class HdfsPartition implements FeFsPartition, PrunablePartition {
/**
* Metadata for a single file in this partition.
*/
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/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 a34eb0a..fea85f7 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -367,7 +367,7 @@ public class HdfsTable extends Table implements FeFsTable {
public boolean isCacheable() {
if (!isLocationCacheable()) return false;
if (!isMarkedCached() && numClusteringCols_ > 0) {
- for (FeFsPartition partition: getPartitions()) {
+ for (FeFsPartition partition: partitionMap_.values()) {
if (!partition.isCacheable()) {
return false;
}
@@ -559,16 +559,29 @@ public class HdfsTable extends Table implements FeFsTable {
public boolean isMarkedCached() { return isMarkedCached_; }
@Override // FeFsTable
- public Collection<? extends FeFsPartition> getPartitions() {
+ public Collection<? extends PrunablePartition> getPartitions() {
return partitionMap_.values();
}
@Override // FeFsTable
- public Map<Long, ? extends FeFsPartition> getPartitionMap() {
+ public Map<Long, ? extends PrunablePartition> getPartitionMap() {
return partitionMap_;
}
@Override // FeFsTable
+ public List<FeFsPartition> loadPartitions(Collection<Long> ids) {
+ List<FeFsPartition> partitions = Lists.newArrayListWithCapacity(ids.size());
+ for (Long id : ids) {
+ HdfsPartition partition = partitionMap_.get(id);
+ if (partition == null) {
+ throw new IllegalArgumentException("no such partition id " + id);
+ }
+ partitions.add(partition);
+ }
+ return partitions;
+ }
+
+ @Override // FeFsTable
public Set<Long> getNullPartitionIds(int i) { return nullPartitionIds_.get(i); }
public HdfsPartitionLocationCompressor getPartitionLocationCompressor() {
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/fe/src/main/java/org/apache/impala/catalog/PrunablePartition.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/PrunablePartition.java b/fe/src/main/java/org/apache/impala/catalog/PrunablePartition.java
new file mode 100644
index 0000000..bdcfcf5
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/catalog/PrunablePartition.java
@@ -0,0 +1,44 @@
+// 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;
+
+import java.util.List;
+
+import org.apache.impala.analysis.LiteralExpr;
+import org.apache.impala.planner.HdfsPartitionPruner;
+
+/**
+ * A specification of a partition that is part of a table, sufficient
+ * for pruning by {@link HdfsPartitionPruner}.
+ *
+ * This interface only exposes the partition ID and values, so it can
+ * be implemented by only fetching partition names and not the complete
+ * partition metadata.
+ */
+public interface PrunablePartition {
+ /**
+ * Returns the identifier of this partition, suitable for later passing
+ * to {@link FeFsTable#loadPartitions(java.util.Collection)}
+ */
+ public long getId();
+
+ /**
+ * Returns the values associated with this partition
+ */
+ List<LiteralExpr> getPartitionValues();
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/fe/src/main/java/org/apache/impala/planner/HdfsPartitionFilter.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsPartitionFilter.java b/fe/src/main/java/org/apache/impala/planner/HdfsPartitionFilter.java
index c8f1cc6..13c8281 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsPartitionFilter.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsPartitionFilter.java
@@ -28,9 +28,8 @@ import org.apache.impala.analysis.SlotDescriptor;
import org.apache.impala.analysis.SlotId;
import org.apache.impala.analysis.SlotRef;
import org.apache.impala.catalog.Column;
-import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
-import org.apache.impala.catalog.HdfsTable;
+import org.apache.impala.catalog.PrunablePartition;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.NotImplementedException;
import org.apache.impala.service.FeSupport;
@@ -84,14 +83,14 @@ public class HdfsPartitionFilter {
* Evaluate a filter against a batch of partitions and return the partition ids
* that pass the filter.
*/
- public HashSet<Long> getMatchingPartitionIds(ArrayList<FeFsPartition> partitions,
+ public HashSet<Long> getMatchingPartitionIds(ArrayList<PrunablePartition> partitions,
Analyzer analyzer) throws ImpalaException {
HashSet<Long> result = new HashSet<Long>();
// List of predicates to evaluate
ArrayList<Expr> predicates = new ArrayList<Expr>(partitions.size());
long[] partitionIds = new long[partitions.size()];
int indx = 0;
- for (FeFsPartition p: partitions) {
+ for (PrunablePartition p: partitions) {
predicates.add(buildPartitionPredicate(p, analyzer));
partitionIds[indx++] = p.getId();
}
@@ -111,7 +110,7 @@ public class HdfsPartitionFilter {
* Construct a predicate for a given partition by substituting the SlotRefs
* for the partition cols with the respective partition-key values.
*/
- private Expr buildPartitionPredicate(FeFsPartition p, Analyzer analyzer)
+ private Expr buildPartitionPredicate(PrunablePartition p, Analyzer analyzer)
throws ImpalaException {
// construct smap
ExprSubstitutionMap sMap = new ExprSubstitutionMap();
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/fe/src/main/java/org/apache/impala/planner/HdfsPartitionPruner.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsPartitionPruner.java b/fe/src/main/java/org/apache/impala/planner/HdfsPartitionPruner.java
index 1240e7f..07a2636 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsPartitionPruner.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsPartitionPruner.java
@@ -41,6 +41,7 @@ import org.apache.impala.analysis.SlotRef;
import org.apache.impala.analysis.TupleDescriptor;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
+import org.apache.impala.catalog.PrunablePartition;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.rewrite.BetweenToCompoundRule;
@@ -49,7 +50,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
+import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
+import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
@@ -97,7 +100,7 @@ public class HdfsPartitionPruner {
* that conjuncts used for filtering will be removed from the list 'conjuncts'.
* If 'allowEmpty' is False, empty partitions are not returned.
*/
- public List<FeFsPartition> prunePartitions(
+ public List<? extends FeFsPartition> prunePartitions(
Analyzer analyzer, List<Expr> conjuncts, boolean allowEmpty)
throws ImpalaException {
// Start with creating a collection of partition filters for the applicable conjuncts.
@@ -154,12 +157,16 @@ public class HdfsPartitionPruner {
evalPartitionFiltersInBe(partitionFilters, matchingPartitionIds, analyzer);
// Populate the list of valid, non-empty partitions to process
- List<FeFsPartition> results = Lists.newArrayList();
- Map<Long, ? extends FeFsPartition> partitionMap = tbl_.getPartitionMap();
- for (Long id: matchingPartitionIds) {
- FeFsPartition partition = partitionMap.get(id);
- Preconditions.checkNotNull(partition);
- if (partition.hasFileDescriptors() || allowEmpty) results.add(partition);
+ List<? extends FeFsPartition> results = tbl_.loadPartitions(
+ matchingPartitionIds);
+ if (!allowEmpty) {
+ results = Lists.newArrayList(Iterables.filter(results,
+ new Predicate<FeFsPartition>() {
+ @Override
+ public boolean apply(FeFsPartition partition) {
+ return partition.hasFileDescriptors();
+ }
+ }));
}
return results;
}
@@ -443,16 +450,16 @@ public class HdfsPartitionPruner {
*/
private void evalPartitionFiltersInBe(List<HdfsPartitionFilter> filters,
HashSet<Long> matchingPartitionIds, Analyzer analyzer) throws ImpalaException {
- Map<Long, ? extends FeFsPartition> partitionMap = tbl_.getPartitionMap();
+ Map<Long, ? extends PrunablePartition> partitionMap = tbl_.getPartitionMap();
// Set of partition ids that pass a filter
HashSet<Long> matchingIds = Sets.newHashSet();
// Batch of partitions
- ArrayList<FeFsPartition> partitionBatch = Lists.newArrayList();
+ ArrayList<PrunablePartition> partitionBatch = Lists.newArrayList();
// Identify the partitions that pass all filters.
for (HdfsPartitionFilter filter: filters) {
// Iterate through the currently valid partitions
for (Long id: matchingPartitionIds) {
- FeFsPartition p = partitionMap.get(id);
+ PrunablePartition p = partitionMap.get(id);
Preconditions.checkState(
p.getPartitionValues().size() == tbl_.getNumClusteringCols());
// Add the partition to the current batch
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
index 63428fd..1a54fb1 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
@@ -165,7 +165,7 @@ public class HdfsScanNode extends ScanNode {
private final HdfsTable tbl_;
// List of partitions to be scanned. Partitions have been pruned.
- private final List<FeFsPartition> partitions_;
+ private final List<? extends FeFsPartition> partitions_;
// Parameters for table sampling. Null if not sampling.
private final TableSampleClause sampleParams_;
@@ -278,7 +278,7 @@ public class HdfsScanNode extends ScanNode {
* class comments above for details.
*/
public HdfsScanNode(PlanNodeId id, TupleDescriptor desc, List<Expr> conjuncts,
- List<FeFsPartition> partitions, TableRef hdfsTblRef, AggregateInfo aggInfo) {
+ List<? extends FeFsPartition> partitions, TableRef hdfsTblRef, AggregateInfo aggInfo) {
super(id, desc, "SCAN HDFS");
Preconditions.checkState(desc.getTable() instanceof FeFsTable);
tbl_ = (HdfsTable)desc.getTable();
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
index 42236a3..9b7111f 100644
--- a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
@@ -1212,7 +1212,7 @@ public class SingleNodePlanner {
// Do partition pruning before deciding which slots to materialize because we might
// end up removing some predicates.
HdfsPartitionPruner pruner = new HdfsPartitionPruner(tupleDesc);
- List<FeFsPartition> partitions = pruner.prunePartitions(analyzer, conjuncts, false);
+ List<? extends FeFsPartition> partitions = pruner.prunePartitions(analyzer, conjuncts, false);
// Mark all slots referenced by the remaining conjuncts as materialized.
analyzer.materializeSlots(conjuncts);
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
----------------------------------------------------------------------
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 f0b0adb..23bd142 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -19,6 +19,7 @@ package org.apache.impala.service;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -61,6 +62,7 @@ import org.apache.impala.catalog.Column;
import org.apache.impala.catalog.ColumnNotFoundException;
import org.apache.impala.catalog.DataSource;
import org.apache.impala.catalog.Db;
+import org.apache.impala.catalog.FeCatalogUtils;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeTable;
@@ -818,7 +820,10 @@ public class CatalogOpExecutor {
HdfsTable table) throws ImpalaException {
Preconditions.checkState(params.isSetPartition_stats());
List<HdfsPartition> modifiedParts = Lists.newArrayList();
- for (FeFsPartition fePartition: table.getPartitions()) {
+ // TODO(todd) only load the partitions that were modified in 'params'.
+ Collection<? extends FeFsPartition> parts =
+ FeCatalogUtils.loadAllPartitions(table);
+ for (FeFsPartition fePartition: parts) {
// TODO(todd): avoid downcast to implementation class
HdfsPartition partition = (HdfsPartition)fePartition;
@@ -1265,7 +1270,9 @@ public class CatalogOpExecutor {
// List of partitions that were modified as part of this operation.
List<HdfsPartition> modifiedParts = Lists.newArrayList();
- for (FeFsPartition fePart: hdfsTable.getPartitions()) {
+ Collection<? extends FeFsPartition> parts =
+ FeCatalogUtils.loadAllPartitions(hdfsTable);
+ for (FeFsPartition fePart: parts) {
// TODO(todd): avoid downcast
HdfsPartition part = (HdfsPartition) fePart;
boolean isModified = false;
@@ -1484,7 +1491,9 @@ public class CatalogOpExecutor {
}
}
if (table.getNumClusteringCols() > 0) {
- for (FeFsPartition part: hdfsTable.getPartitions()) {
+ Collection<? extends FeFsPartition> parts =
+ FeCatalogUtils.loadAllPartitions(hdfsTable);
+ for (FeFsPartition part: parts) {
if (part.isMarkedCached()) {
try {
HdfsCachingUtil.removePartitionCacheDirective(part);
@@ -1530,7 +1539,9 @@ public class CatalogOpExecutor {
catalog_.getLock().writeLock().unlock();
try {
HdfsTable hdfsTable = (HdfsTable)table;
- for (FeFsPartition part: hdfsTable.getPartitions()) {
+ Collection<? extends FeFsPartition> parts =
+ FeCatalogUtils.loadAllPartitions(hdfsTable);
+ for (FeFsPartition part: parts) {
FileSystemUtil.deleteAllVisibleFiles(new Path(part.getLocation()));
}
@@ -2569,7 +2580,9 @@ public class CatalogOpExecutor {
if (tbl.getNumClusteringCols() > 0) {
// If this is a partitioned table, submit cache directives for all uncached
// partitions.
- for (FeFsPartition fePartition: hdfsTable.getPartitions()) {
+ Collection<? extends FeFsPartition> parts =
+ FeCatalogUtils.loadAllPartitions(hdfsTable);
+ for (FeFsPartition fePartition: parts) {
// TODO(todd): avoid downcast
HdfsPartition partition = (HdfsPartition) fePartition;
// Only issue cache directives if the data is uncached or the cache directive
@@ -2621,7 +2634,9 @@ public class CatalogOpExecutor {
if (cacheDirId != null) HdfsCachingUtil.removeTblCacheDirective(msTbl);
// Uncache all table partitions.
if (tbl.getNumClusteringCols() > 0) {
- for (FeFsPartition fePartition: hdfsTable.getPartitions()) {
+ Collection<? extends FeFsPartition> parts =
+ FeCatalogUtils.loadAllPartitions(hdfsTable);
+ for (FeFsPartition fePartition: parts) {
// TODO(todd): avoid downcast
HdfsPartition partition = (HdfsPartition) fePartition;
if (partition.isMarkedCached()) {
@@ -3287,7 +3302,9 @@ public class CatalogOpExecutor {
HashSet<String> partsToCreate =
Sets.newHashSet(update.getCreated_partitions());
partsToLoadMetadata = Sets.newHashSet(partsToCreate);
- for (FeFsPartition partition: ((HdfsTable) table).getPartitions()) {
+ Collection<? extends FeFsPartition> parts =
+ FeCatalogUtils.loadAllPartitions((HdfsTable)table);
+ for (FeFsPartition partition: parts) {
// TODO: In the BE we build partition names without a trailing char. In FE
// we build partition name with a trailing char. We should make this
// consistent.
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/fe/src/test/java/org/apache/impala/catalog/CatalogObjectToFromThriftTest.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/catalog/CatalogObjectToFromThriftTest.java b/fe/src/test/java/org/apache/impala/catalog/CatalogObjectToFromThriftTest.java
index bbb1993..bf182d6 100644
--- a/fe/src/test/java/org/apache/impala/catalog/CatalogObjectToFromThriftTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/CatalogObjectToFromThriftTest.java
@@ -19,6 +19,7 @@ package org.apache.impala.catalog;
import static org.junit.Assert.fail;
+import java.util.Collection;
import java.util.Map;
import org.apache.impala.analysis.LiteralExpr;
@@ -96,7 +97,9 @@ public class CatalogObjectToFromThriftTest {
HdfsTable newHdfsTable = (HdfsTable) newTable;
Assert.assertEquals(newHdfsTable.getPartitions().size(), 24);
Assert.assertEquals(newHdfsTable.getPartitionIds().size(), 24);
- for (FeFsPartition hdfsPart: newHdfsTable.getPartitions()) {
+ Collection<? extends FeFsPartition> parts =
+ FeCatalogUtils.loadAllPartitions(newHdfsTable);
+ for (FeFsPartition hdfsPart: parts) {
Assert.assertEquals(hdfsPart.getFileDescriptors().size(), 1);
Assert.assertTrue(
hdfsPart.getFileDescriptors().get(0).getNumFileBlocks() > 0);
@@ -215,9 +218,10 @@ public class CatalogObjectToFromThriftTest {
HdfsTable hdfsTable = (HdfsTable) table;
// Get any partition with valid HMS parameters to create a
// dummy partition.
- HdfsPartition part = (HdfsPartition)Iterables.getFirst(
- hdfsTable.getPartitions(), null);
- Assert.assertNotNull(part);
+ long id = Iterables.getFirst(hdfsTable.getPartitionIds(), -1L);
+ HdfsPartition part = (HdfsPartition) FeCatalogUtils.loadPartition(
+ hdfsTable, id);
+ Assert.assertNotNull(part);;
// Create a dummy partition with an invalid decimal type.
try {
HdfsPartition dummyPart = new HdfsPartition(hdfsTable, part.toHmsPartition(),
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
----------------------------------------------------------------------
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 4126b72..eeceb36 100644
--- a/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
@@ -287,7 +287,8 @@ public class CatalogTest {
public void TestPartitions() throws CatalogException {
HdfsTable table =
(HdfsTable) catalog_.getOrLoadTable("functional", "AllTypes");
- Collection<? extends FeFsPartition> partitions = table.getPartitions();
+ Collection<? extends FeFsPartition> partitions =
+ FeCatalogUtils.loadAllPartitions(table);
// check that partition keys cover the date range 1/1/2009-12/31/2010
// and that we have one file per partition.
http://git-wip-us.apache.org/repos/asf/impala/blob/0063e0c5/fe/src/test/java/org/apache/impala/testutil/BlockIdGenerator.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/testutil/BlockIdGenerator.java b/fe/src/test/java/org/apache/impala/testutil/BlockIdGenerator.java
index 07ed153..f04ce4a 100644
--- a/fe/src/test/java/org/apache/impala/testutil/BlockIdGenerator.java
+++ b/fe/src/test/java/org/apache/impala/testutil/BlockIdGenerator.java
@@ -19,6 +19,7 @@ package org.apache.impala.testutil;
import java.io.File;
import java.io.FileWriter;
+import java.util.Collection;
import java.util.List;
import org.apache.hadoop.fs.Path;
@@ -28,12 +29,12 @@ import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
import org.apache.impala.catalog.Catalog;
+import org.apache.impala.catalog.FeCatalogUtils;
import org.apache.impala.catalog.FeDb;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
import org.apache.impala.catalog.HdfsTable;
-import org.apache.impala.thrift.ImpalaInternalServiceConstants;
import org.apache.impala.util.PatternMatcher;
/**
@@ -72,7 +73,9 @@ public class BlockIdGenerator {
// Write the output as <tablename>: <blockid1> <blockid2> <etc>
writer.write(tableName + ":");
- for (FeFsPartition partition: hdfsTable.getPartitions()) {
+ Collection<? extends FeFsPartition> parts =
+ FeCatalogUtils.loadAllPartitions(hdfsTable);
+ for (FeFsPartition partition: parts) {
List<FileDescriptor> fileDescriptors = partition.getFileDescriptors();
for (FileDescriptor fd : fileDescriptors) {
Path p = new Path(partition.getLocation(), fd.getFileName());
[4/4] impala git commit: IMPALA-7046: introduce "global" debug_actions
Posted by dh...@apache.org.
IMPALA-7046: introduce "global" debug_actions
The motivation is to add jitter to backend startup in test_failpoints.
The race in IMPALA-7033 can be reproduced by adding jitter to the exec
rpcs when some backends fail. Let's add jitter to test_failpoints to get
better coverage of exec startup races.
This builds on top of the debug action extensions added in the async
admission control patch by allowing the new "global" debug actions
(i.e. actions that can be used in points outside of the ExecNodes).
See the code comments for details.
For now, we're only using the SLEEP and JITTER commands, but I've
included a FAIL command as well since I'll want to use that to write a
test for IMPALA-6788 to simulate exec rpc failure.
Note that I don't bother resolving the actions ahead of time (like we do
for ExecNode actions). It doesn't seem worth it since the resolution
only needs to occur after we've matched the label and I don't expect the
same label to be hit many times within a single thread. We can always
optimize this later if needed.
Testing:
- Verified that test_failpoints can reproduce the race in
IMPALA-7033 by reverting that fix and testing.
- Ran the modified tests and grepped the impalad log to see
that the sleeps are still occuring.
- Manually verify global FAIL command (in a build with another patch).
- Manually verified invalid debug_actions (both ExecNode and global)
Change-Id: I77663a539be18711a4f12c470ffd7474e3d69388
Reviewed-on: http://gerrit.cloudera.org:8080/10690
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/11aaa6ca
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/11aaa6ca
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/11aaa6ca
Branch: refs/heads/master
Commit: 11aaa6caa0818a30db662a4b6f07147faf1f52b2
Parents: 08cf0aa
Author: Dan Hecht <dh...@cloudera.com>
Authored: Mon Jun 11 16:30:06 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Wed Jun 20 19:37:51 2018 +0000
----------------------------------------------------------------------
be/src/runtime/coordinator.cc | 1 +
be/src/runtime/debug-options.cc | 66 +++++++-----
be/src/scheduling/admission-controller.cc | 6 +-
be/src/service/client-request-state.cc | 10 +-
be/src/util/debug-util.cc | 105 +++++++++++++++++--
be/src/util/debug-util.h | 49 ++++++---
common/thrift/ImpalaService.thrift | 21 ++--
.../custom_cluster/test_admission_controller.py | 8 +-
tests/failure/test_failpoints.py | 3 +
tests/query_test/test_observability.py | 4 +-
10 files changed, 202 insertions(+), 71 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/11aaa6ca/be/src/runtime/coordinator.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/coordinator.cc b/be/src/runtime/coordinator.cc
index 61fed4c..2f14825 100644
--- a/be/src/runtime/coordinator.cc
+++ b/be/src/runtime/coordinator.cc
@@ -334,6 +334,7 @@ void Coordinator::StartBackendExec() {
for (BackendState* backend_state: backend_states_) {
ExecEnv::GetInstance()->exec_rpc_thread_pool()->Offer(
[backend_state, this, &debug_options]() {
+ DebugActionNoFail(schedule_.query_options(), "COORD_BEFORE_EXEC_RPC");
backend_state->Exec(query_ctx(), debug_options, filter_routing_table_,
exec_rpcs_complete_barrier_.get());
});
http://git-wip-us.apache.org/repos/asf/impala/blob/11aaa6ca/be/src/runtime/debug-options.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/debug-options.cc b/be/src/runtime/debug-options.cc
index 193daee..6e5062d 100644
--- a/be/src/runtime/debug-options.cc
+++ b/be/src/runtime/debug-options.cc
@@ -16,10 +16,14 @@
// under the License.
#include <boost/algorithm/string.hpp>
+#include <gutil/strings/substitute.h>
#include "runtime/debug-options.h"
+#include "util/debug-util.h"
#include "common/logging.h"
+#include "common/names.h"
+
using namespace impala;
using namespace std;
using namespace boost;
@@ -33,33 +37,42 @@ DebugOptions::DebugOptions(const TQueryOptions& query_options)
// signal not set
return;
}
+ const DebugActionTokens& actions = TokenizeDebugActions(query_options.debug_action);
+ for (const vector<string>& components : actions) {
+ // This will filter out global debug actions which only have two components.
+ if (components.size() < 3 || components.size() > 4) continue;
- vector<string> components;
- split(components, query_options.debug_action, is_any_of(":"), token_compress_on);
- if (components.size() < 3 || components.size() > 4) return;
-
- const string* phase_str;
- const string* action_str;
- if (components.size() == 3) {
- instance_idx_ = -1;
- node_id_ = atoi(components[0].c_str());
- phase_str = &components[1];
- action_str = &components[2];
- } else {
- instance_idx_ = atoi(components[0].c_str());
- node_id_ = atoi(components[1].c_str());
- phase_str = &components[2];
- action_str = &components[3];
- }
- phase_ = GetExecNodePhase(*phase_str);
- if (!GetDebugAction(*action_str, &action_, &action_param_)) {
- LOG(WARNING) << "Invalid debug action " << *action_str;
- phase_ = TExecNodePhase::INVALID;
- return;
+ const string* phase_str;
+ const string* action_str;
+ if (components.size() == 3) {
+ instance_idx_ = -1;
+ node_id_ = atoi(components[0].c_str());
+ phase_str = &components[1];
+ action_str = &components[2];
+ } else {
+ instance_idx_ = atoi(components[0].c_str());
+ node_id_ = atoi(components[1].c_str());
+ phase_str = &components[2];
+ action_str = &components[3];
+ }
+ phase_ = GetExecNodePhase(*phase_str);
+ if (phase_ == TExecNodePhase::INVALID) {
+ LOG(WARNING) << Substitute("Invalid debug_action phase $0", *phase_str);
+ return;
+ }
+ if (!GetDebugAction(*action_str, &action_, &action_param_)) {
+ LOG(WARNING) << Substitute("Invalid debug_action command $0", *action_str);
+ phase_ = TExecNodePhase::INVALID;
+ return;
+ }
+ if (phase_ == TExecNodePhase::CLOSE && action_ == TDebugAction::WAIT) {
+ LOG(WARNING) << "Do not use CLOSE:WAIT debug actions because nodes cannot be "
+ "cancelled in Close()";
+ phase_ = TExecNodePhase::INVALID;
+ return;
+ }
+ return; // Found ExecNode debug action
}
- DCHECK(!(phase_ == TExecNodePhase::CLOSE && action_ == TDebugAction::WAIT))
- << "Do not use CLOSE:WAIT debug actions because nodes cannot be cancelled in "
- "Close()";
}
TExecNodePhase::type DebugOptions::GetExecNodePhase(const string& key) {
@@ -76,8 +89,7 @@ TExecNodePhase::type DebugOptions::GetExecNodePhase(const string& key) {
bool DebugOptions::GetDebugAction(
const string& action, TDebugAction::type* type, string* action_param) {
// Either "ACTION_TYPE" or "ACTION_TYPE@<integer value>".
- vector<string> tokens;
- split(tokens, action, is_any_of("@"), token_compress_on);
+ vector<string> tokens = TokenizeDebugActionParams(action);
if (tokens.size() < 1 || tokens.size() > 2) return false;
if (tokens.size() == 2) *action_param = tokens[1];
for (auto& entry : _TDebugAction_VALUES_TO_NAMES) {
http://git-wip-us.apache.org/repos/asf/impala/blob/11aaa6ca/be/src/scheduling/admission-controller.cc
----------------------------------------------------------------------
diff --git a/be/src/scheduling/admission-controller.cc b/be/src/scheduling/admission-controller.cc
index 8104a58..a1c6c77 100644
--- a/be/src/scheduling/admission-controller.cc
+++ b/be/src/scheduling/admission-controller.cc
@@ -41,9 +41,6 @@ DEFINE_int64(queue_wait_timeout_ms, 60 * 1000, "Maximum amount of time (in "
namespace impala {
-// Available 'sleep_label' string that can be used in the debug_action query option.
-static const string SLEEP_AFTER_ADMISSION_OUTCOME_MS = "SLEEP_AFTER_ADMISSION_OUTCOME_MS";
-
/// Convenience method.
std::string PrintBytes(int64_t value) {
return PrettyPrinter::Print(value, TUnit::BYTES);
@@ -628,7 +625,8 @@ Status AdmissionController::AdmitQuery(QuerySchedule* schedule,
Substitute(
PROFILE_INFO_VAL_INITIAL_QUEUE_REASON, wait_time_ms, not_admitted_reason));
- SleepIfSetInDebugOptions(schedule->query_options(), SLEEP_AFTER_ADMISSION_OUTCOME_MS);
+ // Disallow the FAIL action here. It would leave the queue in an inconsistent state.
+ DebugActionNoFail(schedule->query_options(), "AC_AFTER_ADMISSION_OUTCOME");
{
lock_guard<mutex> lock(admission_ctrl_lock_);
http://git-wip-us.apache.org/repos/asf/impala/blob/11aaa6ca/be/src/service/client-request-state.cc
----------------------------------------------------------------------
diff --git a/be/src/service/client-request-state.cc b/be/src/service/client-request-state.cc
index 0c3910d..9378ddb 100644
--- a/be/src/service/client-request-state.cc
+++ b/be/src/service/client-request-state.cc
@@ -64,11 +64,6 @@ static const string TABLES_MISSING_STATS_KEY = "Tables Missing Stats";
static const string TABLES_WITH_CORRUPT_STATS_KEY = "Tables With Corrupt Table Stats";
static const string TABLES_WITH_MISSING_DISK_IDS_KEY = "Tables With Missing Disk Ids";
-// Available 'sleep_label' strings that can be used in the debug_action query option.
-static const string SLEEP_BEFORE_ADMISSION_MS = "SLEEP_BEFORE_ADMISSION_MS";
-static const string SLEEP_AFTER_COORDINATOR_STARTS_MS =
- "SLEEP_AFTER_COORDINATOR_STARTS_MS";
-
ClientRequestState::ClientRequestState(
const TQueryCtx& query_ctx, ExecEnv* exec_env, Frontend* frontend,
ImpalaServer* server, shared_ptr<ImpalaServer::SessionState> session)
@@ -489,7 +484,8 @@ Status ClientRequestState::ExecAsyncQueryOrDmlRequest(
}
void ClientRequestState::FinishExecQueryOrDmlRequest() {
- SleepIfSetInDebugOptions(schedule_->query_options(), SLEEP_BEFORE_ADMISSION_MS);
+ DebugActionNoFail(schedule_->query_options(), "CRS_BEFORE_ADMISSION");
+
DCHECK(exec_env_->admission_controller() != nullptr);
Status admit_status = ExecEnv::GetInstance()->admission_controller()->AdmitQuery(
schedule_.get(), &admit_outcome_);
@@ -500,7 +496,7 @@ void ClientRequestState::FinishExecQueryOrDmlRequest() {
coord_.reset(new Coordinator(*schedule_, query_events_));
Status exec_status = coord_->Exec();
- SleepIfSetInDebugOptions(schedule_->query_options(), SLEEP_AFTER_COORDINATOR_STARTS_MS);
+ DebugActionNoFail(schedule_->query_options(), "CRS_AFTER_COORD_STARTS");
bool cancelled = false;
Status cancellation_status;
http://git-wip-us.apache.org/repos/asf/impala/blob/11aaa6ca/be/src/util/debug-util.cc
----------------------------------------------------------------------
diff --git a/be/src/util/debug-util.cc b/be/src/util/debug-util.cc
index 7a31748..f305943 100644
--- a/be/src/util/debug-util.cc
+++ b/be/src/util/debug-util.cc
@@ -18,6 +18,7 @@
#include "util/debug-util.h"
#include <iomanip>
+#include <random>
#include <sstream>
#include <boost/tokenizer.hpp>
@@ -42,6 +43,9 @@ extern void DumpStackTraceToString(std::string* s);
#include "common/names.h"
using boost::char_separator;
+using boost::is_any_of;
+using boost::split;
+using boost::token_compress_on;
using boost::tokenizer;
using namespace beeswax;
using namespace parquet;
@@ -286,13 +290,102 @@ string GetBackendString() {
return Substitute("$0:$1", FLAGS_hostname, FLAGS_be_port);
}
-void SleepIfSetInDebugOptionsImpl(
- const TQueryOptions& query_options, const string& sleep_label) {
- vector<string> components;
- boost::split(components, query_options.debug_action, boost::is_any_of(":"));
- if (components.size() == 2 && components[0].compare(sleep_label) == 0) {
- SleepForMs(atoi(components[1].c_str()));
+DebugActionTokens TokenizeDebugActions(const string& debug_actions) {
+ DebugActionTokens results;
+ list<string> actions;
+ split(actions, debug_actions, is_any_of("|"), token_compress_on);
+ for (const string& a : actions) {
+ vector<string> components;
+ split(components, a, is_any_of(":"), token_compress_on);
+ results.push_back(components);
}
+ return results;
+}
+
+vector<string> TokenizeDebugActionParams(const string& action) {
+ vector<string> tokens;
+ split(tokens, action, is_any_of("@"), token_compress_on);
+ return tokens;
+}
+
+/// Helper to DebugActionImpl(). Given a probability as a string (e.g. "0.3"),
+/// determine whether the action should be executed, as returned in 'should_execute'.
+/// Returns true if the parsing was successful.
+static bool ParseProbability(const string& prob_str, bool* should_execute) {
+ StringParser::ParseResult parse_result;
+ double probability = StringParser::StringToFloat<double>(
+ prob_str.c_str(), prob_str.size(), &parse_result);
+ if (parse_result != StringParser::PARSE_SUCCESS ||
+ probability < 0.0 || probability > 1.0) {
+ return false;
+ }
+ // +1L ensures probability of 0.0 and 1.0 work as expected.
+ *should_execute = rand() < probability * (RAND_MAX + 1L);
+ return true;
+}
+
+Status DebugActionImpl(
+ const TQueryOptions& query_options, const char* label) {
+ const DebugActionTokens& action_list = TokenizeDebugActions(
+ query_options.debug_action);
+ static const char ERROR_MSG[] = "Invalid debug_action $0:$1 ($2)";
+ for (const vector<string>& components : action_list) {
+ if (components.size() != 2 || components[0].compare(label) != 0) continue;
+ // 'tokens' becomes {command, param0, param1, ... }
+ vector<string> tokens = TokenizeDebugActionParams(components[1]);
+ DCHECK_GE(tokens.size(), 1);
+ const string& cmd = tokens[0];
+ int sleep_millis = 0;
+ if (cmd.compare("SLEEP") == 0) {
+ // SLEEP@<millis>
+ if (tokens.size() != 2) {
+ return Status(Substitute(ERROR_MSG, components[0], components[1],
+ "expected SLEEP@<ms>"));
+ }
+ sleep_millis = atoi(tokens[1].c_str());
+ } else if (cmd.compare("JITTER") == 0) {
+ // JITTER@<millis>[@<probability>}
+ if (tokens.size() < 2 || tokens.size() > 3) {
+ return Status(Substitute(ERROR_MSG, components[0], components[1],
+ "expected JITTER@<ms>[@<probability>]"));
+ }
+ int max_millis = atoi(tokens[1].c_str());
+ if (tokens.size() == 3) {
+ bool should_execute = true;
+ if (!ParseProbability(tokens[2], &should_execute)) {
+ return Status(Substitute(ERROR_MSG, components[0], components[1],
+ "invalid probability"));
+ }
+ if (!should_execute) continue;
+ }
+ sleep_millis = rand() % (max_millis + 1);
+ } else if (cmd.compare("FAIL") == 0) {
+ // FAIL[@<probability>]
+ if (tokens.size() > 2) {
+ return Status(Substitute(ERROR_MSG, components[0], components[1],
+ "expected FAIL[@<probability>]"));
+ }
+ if (tokens.size() == 2) {
+ bool should_execute = true;
+ if (!ParseProbability(tokens[1], &should_execute)) {
+ return Status(Substitute(ERROR_MSG, components[0], components[1],
+ "invalid probability"));
+ }
+ if (!should_execute) continue;
+ }
+ return Status(TErrorCode::INTERNAL_ERROR, Substitute("Debug Action: $0:$1",
+ components[0], components[1]));
+ } else {
+ return Status(Substitute(ERROR_MSG, components[0], components[1],
+ "invalid command"));
+ }
+ if (sleep_millis > 0) {
+ VLOG(1) << Substitute("Debug Action: $0:$1 sleeping for $2 ms",
+ components[0], components[1], sleep_millis);
+ SleepForMs(sleep_millis);
+ }
+ }
+ return Status::OK();
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/11aaa6ca/be/src/util/debug-util.h
----------------------------------------------------------------------
diff --git a/be/src/util/debug-util.h b/be/src/util/debug-util.h
index 0978e50..005cc23 100644
--- a/be/src/util/debug-util.h
+++ b/be/src/util/debug-util.h
@@ -108,21 +108,40 @@ std::string GetStackTrace();
/// Returns the backend name in "host:port" form suitable for human consumption.
std::string GetBackendString();
-/// Slow path implementing SleepIfSetInDebugOptions() for the case where 'query_options'
-/// is non-empty.
-void SleepIfSetInDebugOptionsImpl(
- const TQueryOptions& query_options, const string& sleep_label);
-
-/// If sleep time is specified in the debug_action query option in the format
-/// <sleep_label>:<sleep_time_ms>, where <sleep_label> is a string and <sleep_time_ms>
-/// is an integer, and 'sleep_label' matches the one specified in the query option, then
-/// this methods extracts the corresponding <sleep_time_ms> and initiates a sleep for that
-/// many milliseconds.
-static inline void SleepIfSetInDebugOptions(
- const TQueryOptions& query_options, const string& sleep_label) {
- // Make sure this is very cheap if debug actions are not in use.
- if (LIKELY(query_options.debug_action.empty())) return;
- return SleepIfSetInDebugOptionsImpl(query_options, sleep_label);
+/// Tokenize 'debug_actions' into a list of tokenized rows, where columns are separated
+/// by ':' and rows by '|'. i.e. if debug_actions="a:b:c|x:y", then the returned
+/// structure is {{"a", "b", "c"}, {"x", "y"}}
+typedef std::list<std::vector<string>> DebugActionTokens;
+DebugActionTokens TokenizeDebugActions(const string& debug_actions);
+
+/// Tokenize 'action' which has an optional parameter separated by '@'. i.e. "x@y"
+/// becomes {"x", "y"} and "x" becomes {"x"}.
+std::vector<std::string> TokenizeDebugActionParams(const string& action);
+
+/// Slow path implementing DebugAction() for the case where
+/// 'query_options.debug_action' is non-empty.
+Status DebugActionImpl(
+ const TQueryOptions& query_options, const char* label) WARN_UNUSED_RESULT;
+
+/// If debug_action query option has a "global action" (i.e. not exec-node specific)
+/// and matches the given 'label', apply the the action. See ImpalaService.thrift for
+/// details of the format and available global actions. For ExecNode code, use
+/// ExecNode::ExecDebugAction() instead.
+WARN_UNUSED_RESULT static inline Status DebugAction(
+ const TQueryOptions& query_options, const char* label) {
+ if (LIKELY(query_options.debug_action.empty())) return Status::OK();
+ return DebugActionImpl(query_options, label);
+}
+
+/// Like DebugAction() but for use in contexts that can't safely propagate an error
+/// status. Debug action FAIL should not be used in these contexts and will be logged
+/// and ignored.
+static inline void DebugActionNoFail(
+ const TQueryOptions& query_options, const char* label) {
+ Status status = DebugAction(query_options, label);
+ if (!status.ok()) {
+ LOG(ERROR) << "Ignoring debug action failure: " << status.GetDetail();
+ }
}
// FILE_CHECKs are conditions that we expect to be true but could fail due to a malformed
http://git-wip-us.apache.org/repos/asf/impala/blob/11aaa6ca/common/thrift/ImpalaService.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/ImpalaService.thrift b/common/thrift/ImpalaService.thrift
index 91b4de8..cf62ddf 100644
--- a/common/thrift/ImpalaService.thrift
+++ b/common/thrift/ImpalaService.thrift
@@ -79,7 +79,8 @@ enum TImpalaQueryOptions {
// DEBUG ONLY:
// Accepted formats:
- // 1. "[<backend number>:]<node id>:<TExecNodePhase>:<TDebugAction>",
+ // 1. ExecNode actions
+ // "[<instance idx>:]<node id>:<TExecNodePhase>:<TDebugAction>",
// the exec node with the given id will perform the specified action in the given
// phase. If the optional backend number (starting from 0) is specified, only that
// backend instance will perform the debug action, otherwise all backends will behave
@@ -87,11 +88,19 @@ enum TImpalaQueryOptions {
// If the string doesn't have the required format or if any of its components is
// invalid, the option is ignored.
//
- // 2. <sleep_label>:<sleep_time_ms>,
- // <sleep_label> is used as a string key that represents a location in the codebase
- // where sleep() is to be executed for <sleep_time_ms> milliseconds.
- // Available sleep_label : SLEEP_BEFORE_ADMISSION_MS,
- // SLEEP_AFTER_COORDINATOR_STARTS_MS, SLEEP_AFTER_ADMISSION_OUTCOME_MS
+ // 2. Global actions
+ // "<global label>:<command>@<param0>@<param1>@...<paramN>",
+ // global labels are marked in the code with DEBUG_ACTION*() macros.
+ // Available global actions:
+ // - SLEEP@<ms> will sleep for the 'ms' milliseconds.
+ // - JITTER@<ms>[@<probability>] will sleep for a random amount of time between 0
+ // and 'ms' milliseconds with the given probability. If <probability> is omitted,
+ // it is 1.0.
+ // - FAIL[@<probability>] returns an INTERNAL_ERROR status with the given
+ // probability. If <probability> is omitted, it is 1.0.
+ //
+ // Only a single ExecNode action is allowed, but multiple global actions can be
+ // specified. To specify multiple actions, separate them with "|".
DEBUG_ACTION,
ABORT_ON_DEFAULT_LIMIT_EXCEEDED, // Removed
http://git-wip-us.apache.org/repos/asf/impala/blob/11aaa6ca/tests/custom_cluster/test_admission_controller.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_admission_controller.py b/tests/custom_cluster/test_admission_controller.py
index a907a74..32fd971 100644
--- a/tests/custom_cluster/test_admission_controller.py
+++ b/tests/custom_cluster/test_admission_controller.py
@@ -528,7 +528,7 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
impalad = self.cluster.impalads[0]
client = impalad.service.create_beeswax_client()
try:
- client.set_configuration_option("debug_action", "SLEEP_BEFORE_ADMISSION_MS:2000")
+ client.set_configuration_option("debug_action", "CRS_BEFORE_ADMISSION:SLEEP@2000")
client.set_configuration_option("mem_limit", self.PROC_MEM_TEST_LIMIT + 1 )
handle = client.execute_async("select 1")
sleep(1)
@@ -537,7 +537,7 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
"Ready to be Rejected but already cancelled, query id=")
client.clear_configuration()
- client.set_configuration_option("debug_action", "SLEEP_BEFORE_ADMISSION_MS:2000")
+ client.set_configuration_option("debug_action", "CRS_BEFORE_ADMISSION:SLEEP@2000")
handle = client.execute_async("select 1")
sleep(1)
client.close_query(handle)
@@ -545,7 +545,7 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
"Ready to be Admitted immediately but already cancelled, query id=")
client.set_configuration_option("debug_action",
- "SLEEP_AFTER_COORDINATOR_STARTS_MS:2000")
+ "CRS_AFTER_COORD_STARTS:SLEEP@2000")
handle = client.execute_async("select 1")
sleep(1)
client.close_query(handle)
@@ -555,7 +555,7 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
client.clear_configuration()
handle = client.execute_async("select sleep(10000)")
client.set_configuration_option("debug_action",
- "SLEEP_AFTER_ADMISSION_OUTCOME_MS:2000")
+ "AC_AFTER_ADMISSION_OUTCOME:SLEEP@2000")
queued_query_handle = client.execute_async("select 1")
sleep(1)
assert client.get_state(queued_query_handle) == QueryState.COMPILED
http://git-wip-us.apache.org/repos/asf/impala/blob/11aaa6ca/tests/failure/test_failpoints.py
----------------------------------------------------------------------
diff --git a/tests/failure/test_failpoints.py b/tests/failure/test_failpoints.py
index 1f0c1c5..9f54257 100644
--- a/tests/failure/test_failpoints.py
+++ b/tests/failure/test_failpoints.py
@@ -103,6 +103,9 @@ class TestFailpoints(ImpalaTestSuite):
for node_id in plan_node_ids:
debug_action = '%d:%s:%s' % (node_id, location, FAILPOINT_ACTION_MAP[action])
+ # IMPALA-7046: add jitter to backend startup to exercise various failure paths.
+ debug_action += '|COORD_BEFORE_EXEC_RPC:JITTER@100@0.3'
+
LOG.info('Current debug action: SET DEBUG_ACTION=%s' % debug_action)
vector.get_value('exec_option')['debug_action'] = debug_action
http://git-wip-us.apache.org/repos/asf/impala/blob/11aaa6ca/tests/query_test/test_observability.py
----------------------------------------------------------------------
diff --git a/tests/query_test/test_observability.py b/tests/query_test/test_observability.py
index 4e4bc25..86e78cc 100644
--- a/tests/query_test/test_observability.py
+++ b/tests/query_test/test_observability.py
@@ -90,7 +90,7 @@ class TestObservability(ImpalaTestSuite):
"""Tests that the query profile shows expected query states."""
query = "select count(*) from functional.alltypes"
handle = self.execute_query_async(query,
- {"debug_action": "SLEEP_BEFORE_ADMISSION_MS:1000"})
+ {"debug_action": "CRS_BEFORE_ADMISSION:SLEEP@1000"})
# If ExecuteStatement() has completed and the query is paused in the admission control
# phase, then the query must be in COMPILED state.
profile = self.client.get_runtime_profile(handle)
@@ -125,7 +125,7 @@ class TestObservability(ImpalaTestSuite):
"""Test that the exec summary is populated correctly in every query state"""
query = "select count(*) from functional.alltypes"
handle = self.execute_query_async(query,
- {"debug_action": "SLEEP_BEFORE_ADMISSION_MS:1000"})
+ {"debug_action": "CRS_BEFORE_ADMISSION:SLEEP@1000"})
# If ExecuteStatement() has completed and the query is paused in the admission control
# phase, then the coordinator has not started yet and exec_summary should be empty.
exec_summary = self.client.get_exec_summary(handle)
[3/4] impala git commit: IMPALA-7141 (part 1): clean up handling of
default/dummy partition
Posted by dh...@apache.org.
IMPALA-7141 (part 1): clean up handling of default/dummy partition
Currently, HdfsTable inconsistently uses the term "default partition"
to refer to two different concepts:
1) For unpartitioned tables, a single partition with ID 1 and no partition
keys is created and added to the partition map.
2) All tables have an additional partition added with partition ID -1
which acts as a sort of prototype for partition creation: when new
partitions are created during an INSERT operation, the file format
and other related options are copied out of this special partition.
This partition is inconsistently referred to as either the "default
partition" or the "dummy partition".
The handling of this second case (the partition with id -1) was somewhat
messy:
- the partition shows up in the partitionMap_ member, but does not show
up in the partitionIds_ member.
- almost all of the call sites that iterate through the partitions of an
HdfsTable instance ended up skipping over the dummy partition.
- several call sites called getPartitions().size() but then had to
adjust the result by subtracting one in order to actually count the
number of partitions in a table.
- similarly, test assertions had to assert that tables with 24
partitions had an expected partition map size of 25.
In order to address the above, this patch makes the following changes:
- getPartitions() and getPartitionMap() no longer include the dummy
partition. This removes a bunch of special case checks to skip over
the dummy partition or to adjust partition counts based on it.
- to clarify the purpose of this partition, references to it are renamed
to "prototype partition" instead of "default partition".
- when converting the HdfsTable to/from Thrift, the prototype partition
is included in its own field in the struct, instead of being stuffed
into the same map with the true partitions of the table. This reflects
the fact that this partition is special (eg missing fields like
'location' which otherwise are required for real partitions)
This change should should be entirely internal with no functional
differences. As such, the only testing changes are some fixes for
assertions on the Thrift serialized structures and other internals.
Change-Id: I15e91b50eb7c2a5e0bac8c33d603d6cd8cbaca2e
Reviewed-on: http://gerrit.cloudera.org:8080/10711
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/747e0a94
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/747e0a94
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/747e0a94
Branch: refs/heads/master
Commit: 747e0a949316b8e14fa6ad123ffc34d9c450608b
Parents: afe1146
Author: Todd Lipcon <to...@cloudera.com>
Authored: Wed Jun 13 13:55:22 2018 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Wed Jun 20 05:07:22 2018 +0000
----------------------------------------------------------------------
be/src/exec/hdfs-table-sink.cc | 121 +++++++++----------
be/src/exec/hdfs-table-sink.h | 5 +-
be/src/runtime/descriptors.cc | 2 +
be/src/runtime/descriptors.h | 5 +
common/thrift/CatalogObjects.thrift | 11 +-
common/thrift/ImpalaInternalService.thrift | 3 -
.../impala/analysis/ComputeStatsStmt.java | 4 +-
.../apache/impala/analysis/LoadDataStmt.java | 15 +--
.../apache/impala/catalog/FeFsPartition.java | 6 -
.../apache/impala/catalog/HdfsPartition.java | 14 +--
.../org/apache/impala/catalog/HdfsTable.java | 62 +++++-----
.../org/apache/impala/planner/HdfsScanNode.java | 2 +-
.../impala/planner/SingleNodePlanner.java | 4 +-
.../impala/service/CatalogOpExecutor.java | 29 +----
.../catalog/CatalogObjectToFromThriftTest.java | 71 ++++++-----
.../org/apache/impala/catalog/CatalogTest.java | 9 +-
.../apache/impala/common/FrontendTestBase.java | 2 +-
.../impala/testutil/BlockIdGenerator.java | 5 -
18 files changed, 164 insertions(+), 206 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/be/src/exec/hdfs-table-sink.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-table-sink.cc b/be/src/exec/hdfs-table-sink.cc
index 4b718ec..3b7dc0f 100644
--- a/be/src/exec/hdfs-table-sink.cc
+++ b/be/src/exec/hdfs-table-sink.cc
@@ -61,7 +61,7 @@ HdfsTableSink::HdfsTableSink(const RowDescriptor* row_desc, const TDataSink& tsi
RuntimeState* state)
: DataSink(row_desc, "HdfsTableSink", state),
table_desc_(nullptr),
- default_partition_(nullptr),
+ prototype_partition_(nullptr),
table_id_(tsink.table_sink.target_table_id),
skip_header_line_count_(
tsink.table_sink.hdfs_table_sink.__isset.skip_header_line_count ?
@@ -150,74 +150,67 @@ Status HdfsTableSink::Open(RuntimeState* state) {
DCHECK_EQ(partition_key_exprs_.size(), partition_key_expr_evals_.size());
RETURN_IF_ERROR(ScalarExprEvaluator::Open(partition_key_expr_evals_, state));
- // Get file format for default partition in table descriptor, and build a map from
- // partition key values to partition descriptor for multiple output format support. The
- // map is keyed on the concatenation of the non-constant keys of the PARTITION clause of
- // the INSERT statement.
+ // Build a map from partition key values to partition descriptor for multiple output
+ // format support. The map is keyed on the concatenation of the non-constant keys of
+ // the PARTITION clause of the INSERT statement.
for (const HdfsTableDescriptor::PartitionIdToDescriptorMap::value_type& id_to_desc:
table_desc_->partition_descriptors()) {
- if (id_to_desc.first == g_ImpalaInternalService_constants.DEFAULT_PARTITION_ID) {
- default_partition_ = id_to_desc.second;
- } else {
- // Build a map whose key is computed from the value of dynamic partition keys for a
- // particular partition, and whose value is the descriptor for that partition.
-
- // True if this partition might be written to, false otherwise.
- // A partition may be written to iff:
- // For all partition key exprs e, either:
- // 1. e is not constant
- // 2. The value supplied by the query for this partition key is equal to e's
- // constant value.
- // Only relevant partitions are remembered in partition_descriptor_map_.
- bool relevant_partition = true;
- HdfsPartitionDescriptor* partition = id_to_desc.second;
- DCHECK_EQ(partition->partition_key_value_evals().size(),
- partition_key_expr_evals_.size());
- vector<ScalarExprEvaluator*> dynamic_partition_key_value_evals;
- for (size_t i = 0; i < partition_key_expr_evals_.size(); ++i) {
- // Remember non-constant partition key exprs for building hash table of Hdfs files
- DCHECK(&partition_key_expr_evals_[i]->root() == partition_key_exprs_[i]);
- if (!partition_key_exprs_[i]->is_constant()) {
- dynamic_partition_key_value_evals.push_back(
- partition->partition_key_value_evals()[i]);
- } else {
- // Deal with the following: one partition has (year=2009, month=3); another has
- // (year=2010, month=3).
- // A query like: INSERT INTO TABLE... PARTITION(year=2009) SELECT month FROM...
- // would lead to both partitions having the same key modulo ignored constant
- // partition keys. So only keep a reference to the partition which matches
- // partition_key_values for constant values, since only that is written to.
- void* table_partition_key_value =
- partition->partition_key_value_evals()[i]->GetValue(nullptr);
- void* target_partition_key_value =
- partition_key_expr_evals_[i]->GetValue(nullptr);
- if (table_partition_key_value == nullptr
- && target_partition_key_value == nullptr) {
- continue;
- }
- if (table_partition_key_value == nullptr
- || target_partition_key_value == nullptr
- || !RawValue::Eq(table_partition_key_value, target_partition_key_value,
- partition_key_expr_evals_[i]->root().type())) {
- relevant_partition = false;
- break;
- }
+ // Build a map whose key is computed from the value of dynamic partition keys for a
+ // particular partition, and whose value is the descriptor for that partition.
+
+ // True if this partition might be written to, false otherwise.
+ // A partition may be written to iff:
+ // For all partition key exprs e, either:
+ // 1. e is not constant
+ // 2. The value supplied by the query for this partition key is equal to e's
+ // constant value.
+ // Only relevant partitions are remembered in partition_descriptor_map_.
+ bool relevant_partition = true;
+ HdfsPartitionDescriptor* partition = id_to_desc.second;
+ DCHECK_EQ(partition->partition_key_value_evals().size(),
+ partition_key_expr_evals_.size());
+ vector<ScalarExprEvaluator*> dynamic_partition_key_value_evals;
+ for (size_t i = 0; i < partition_key_expr_evals_.size(); ++i) {
+ // Remember non-constant partition key exprs for building hash table of Hdfs files
+ DCHECK(&partition_key_expr_evals_[i]->root() == partition_key_exprs_[i]);
+ if (!partition_key_exprs_[i]->is_constant()) {
+ dynamic_partition_key_value_evals.push_back(
+ partition->partition_key_value_evals()[i]);
+ } else {
+ // Deal with the following: one partition has (year=2009, month=3); another has
+ // (year=2010, month=3).
+ // A query like: INSERT INTO TABLE... PARTITION(year=2009) SELECT month FROM...
+ // would lead to both partitions having the same key modulo ignored constant
+ // partition keys. So only keep a reference to the partition which matches
+ // partition_key_values for constant values, since only that is written to.
+ void* table_partition_key_value =
+ partition->partition_key_value_evals()[i]->GetValue(nullptr);
+ void* target_partition_key_value =
+ partition_key_expr_evals_[i]->GetValue(nullptr);
+ if (table_partition_key_value == nullptr
+ && target_partition_key_value == nullptr) {
+ continue;
+ }
+ if (table_partition_key_value == nullptr
+ || target_partition_key_value == nullptr
+ || !RawValue::Eq(table_partition_key_value, target_partition_key_value,
+ partition_key_expr_evals_[i]->root().type())) {
+ relevant_partition = false;
+ break;
}
- }
- if (relevant_partition) {
- string key;
- // Pass nullptr as row, since all of these expressions are constant, and can
- // therefore be evaluated without a valid row context.
- GetHashTblKey(nullptr, dynamic_partition_key_value_evals, &key);
- DCHECK(partition_descriptor_map_.find(key) == partition_descriptor_map_.end())
- << "Partitions with duplicate 'static' keys found during INSERT";
- partition_descriptor_map_[key] = partition;
}
}
+ if (relevant_partition) {
+ string key;
+ // Pass nullptr as row, since all of these expressions are constant, and can
+ // therefore be evaluated without a valid row context.
+ GetHashTblKey(nullptr, dynamic_partition_key_value_evals, &key);
+ DCHECK(partition_descriptor_map_.find(key) == partition_descriptor_map_.end())
+ << "Partitions with duplicate 'static' keys found during INSERT";
+ partition_descriptor_map_[key] = partition;
+ }
}
- if (default_partition_ == nullptr) {
- return Status("No default partition found for HdfsTextTableSink");
- }
+ prototype_partition_ = CHECK_NOTNULL(table_desc_->prototype_partition_descriptor());
return Status::OK();
}
@@ -570,7 +563,7 @@ inline Status HdfsTableSink::GetOutputPartition(RuntimeState* state, const Tuple
existing_partition = partition_keys_to_output_partitions_.find(key);
if (existing_partition == partition_keys_to_output_partitions_.end()) {
// Create a new OutputPartition, and add it to partition_keys_to_output_partitions.
- const HdfsPartitionDescriptor* partition_descriptor = default_partition_;
+ const HdfsPartitionDescriptor* partition_descriptor = prototype_partition_;
PartitionDescriptorMap::const_iterator it = partition_descriptor_map_.find(key);
if (it != partition_descriptor_map_.end()) {
partition_descriptor = it->second;
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/be/src/exec/hdfs-table-sink.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-table-sink.h b/be/src/exec/hdfs-table-sink.h
index ef8b7a6..002e866 100644
--- a/be/src/exec/hdfs-table-sink.h
+++ b/be/src/exec/hdfs-table-sink.h
@@ -242,8 +242,9 @@ class HdfsTableSink : public DataSink {
/// Descriptor of target table. Set in Prepare().
const HdfsTableDescriptor* table_desc_;
- /// Currently this is the default partition since we don't support multi-format sinks.
- const HdfsPartitionDescriptor* default_partition_;
+ /// The partition descriptor used when creating new partitions from this sink.
+ /// Currently we don't support multi-format sinks.
+ const HdfsPartitionDescriptor* prototype_partition_;
/// Table id resolved in Prepare() to set tuple_desc_;
TableId table_id_;
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/be/src/runtime/descriptors.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/descriptors.cc b/be/src/runtime/descriptors.cc
index bc983de..12de054 100644
--- a/be/src/runtime/descriptors.cc
+++ b/be/src/runtime/descriptors.cc
@@ -226,6 +226,8 @@ HdfsTableDescriptor::HdfsTableDescriptor(const TTableDescriptor& tdesc, ObjectPo
pool->Add(new HdfsPartitionDescriptor(tdesc.hdfsTable, entry.second));
partition_descriptors_[entry.first] = partition;
}
+ prototype_partition_descriptor_ = pool->Add(new HdfsPartitionDescriptor(
+ tdesc.hdfsTable, tdesc.hdfsTable.prototype_partition));
avro_schema_ = tdesc.hdfsTable.__isset.avroSchema ? tdesc.hdfsTable.avroSchema : "";
}
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/be/src/runtime/descriptors.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/descriptors.h b/be/src/runtime/descriptors.h
index acbce34..982ede3 100644
--- a/be/src/runtime/descriptors.h
+++ b/be/src/runtime/descriptors.h
@@ -328,6 +328,10 @@ class HdfsTableDescriptor : public TableDescriptor {
return partition_descriptors_;
}
+ const HdfsPartitionDescriptor* prototype_partition_descriptor() const {
+ return prototype_partition_descriptor_;
+ }
+
virtual std::string DebugString() const;
protected:
@@ -336,6 +340,7 @@ class HdfsTableDescriptor : public TableDescriptor {
/// Special string to indicate NULL values in text-encoded columns.
std::string null_column_value_;
PartitionIdToDescriptorMap partition_descriptors_;
+ HdfsPartitionDescriptor* prototype_partition_descriptor_;
/// Set to the table's Avro schema if this is an Avro table, empty string otherwise
std::string avro_schema_;
};
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/common/thrift/CatalogObjects.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/CatalogObjects.thrift b/common/thrift/CatalogObjects.thrift
index 0f71f5f..23da91f 100644
--- a/common/thrift/CatalogObjects.thrift
+++ b/common/thrift/CatalogObjects.thrift
@@ -281,6 +281,11 @@ struct THdfsPartition {
18: optional bool has_incremental_stats
}
+// Constant partition ID used for THdfsPartition.prototype_partition above.
+// Must be < 0 to avoid collisions
+const i64 PROTOTYPE_PARTITION_ID = -1;
+
+
struct THdfsTable {
1: required string hdfsBaseDir
@@ -296,9 +301,13 @@ struct THdfsTable {
// Set to the table's Avro schema if this is an Avro table
6: optional string avroSchema
- // map from partition id to partition metadata
+ // Map from partition id to partition metadata.
+ // Does not include the special prototype partition -1 (see below).
4: required map<i64, THdfsPartition> partitions
+ // Prototype partition, used when creating new partitions during insert.
+ 10: required THdfsPartition prototype_partition
+
// Each TNetworkAddress is a datanode which contains blocks of a file in the table.
// Used so that each THdfsFileBlock can just reference an index in this list rather
// than duplicate the list of network address, which helps reduce memory usage.
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/common/thrift/ImpalaInternalService.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/ImpalaInternalService.thrift b/common/thrift/ImpalaInternalService.thrift
index f3d8af3..aa19106 100644
--- a/common/thrift/ImpalaInternalService.thrift
+++ b/common/thrift/ImpalaInternalService.thrift
@@ -42,9 +42,6 @@ const i32 NUM_NODES_ALL_RACKS = -1
// constants for TPlanNodeId
const i32 INVALID_PLAN_NODE_ID = -1
-// Constant default partition ID, must be < 0 to avoid collisions
-const i64 DEFAULT_PARTITION_ID = -1;
-
enum TParquetFallbackSchemaResolution {
POSITION,
NAME
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/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 dec3f45..9259e49 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
@@ -440,7 +440,6 @@ public class ComputeStatsStmt extends StatementBase {
Collection<? extends FeFsPartition> allPartitions =
hdfsTable.getPartitions();
for (FeFsPartition p: allPartitions) {
- if (p.isDefaultPartition()) continue;
TPartitionStats partStats = p.getPartitionStats();
if (!p.hasIncrementalStats() || tableIsMissingColStats) {
if (partStats == null) {
@@ -458,7 +457,7 @@ public class ComputeStatsStmt extends StatementBase {
validPartStats_.add(partStats);
}
}
- if (expectedPartitions_.size() == hdfsTable.getPartitions().size() - 1) {
+ if (expectedPartitions_.size() == hdfsTable.getPartitions().size()) {
expectedPartitions_.clear();
expectAllPartitions_ = true;
}
@@ -479,7 +478,6 @@ public class ComputeStatsStmt extends StatementBase {
Sets.newHashSet(partitionSet_.getPartitions());
Collection<? extends FeFsPartition> allPartitions = hdfsTable.getPartitions();
for (FeFsPartition p : allPartitions) {
- if (p.isDefaultPartition()) continue;
if (targetPartitions.contains(p)) continue;
TPartitionStats partStats = p.getPartitionStats();
if (partStats != null) validPartStats_.add(partStats);
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/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 3116eb4..e20a88c 100644
--- a/fe/src/main/java/org/apache/impala/analysis/LoadDataStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/LoadDataStmt.java
@@ -32,6 +32,7 @@ import org.apache.impala.authorization.Privilege;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.HdfsFileFormat;
+import org.apache.impala.catalog.HdfsPartition;
import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.common.FileSystemUtil;
@@ -205,24 +206,20 @@ public class LoadDataStmt extends StatementBase {
"target table (%s) because Impala does not have WRITE access to HDFS " +
"location: ", hdfsTable.getFullName());
- FeFsPartition partition;
- String location;
if (partitionSpec_ != null) {
- partition = hdfsTable.getPartition(partitionSpec_.getPartitionSpecKeyValues());
- location = partition.getLocation();
+ HdfsPartition partition = hdfsTable.getPartition(
+ partitionSpec_.getPartitionSpecKeyValues());
+ String location = partition.getLocation();
if (!TAccessLevelUtil.impliesWriteAccess(partition.getAccessLevel())) {
throw new AnalysisException(noWriteAccessErrorMsg + location);
}
} else {
- // "default" partition
- partition = hdfsTable.getPartitionMap().get(
- ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID);
- location = hdfsTable.getLocation();
+ // 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());
}
}
- Preconditions.checkNotNull(partition);
} catch (FileNotFoundException e) {
throw new AnalysisException("File not found: " + e.getMessage(), e);
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/fe/src/main/java/org/apache/impala/catalog/FeFsPartition.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/FeFsPartition.java b/fe/src/main/java/org/apache/impala/catalog/FeFsPartition.java
index 9d0c40d..52edeb8 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeFsPartition.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeFsPartition.java
@@ -43,12 +43,6 @@ public interface FeFsPartition {
long getId();
/**
- * @return true if this partition represents the "default partition" of an
- * unpartitioned tabe
- */
- boolean isDefaultPartition();
-
- /**
* @return the table that contains this partition
*/
FeFsTable getTable();
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
----------------------------------------------------------------------
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 294d8a7..b265c6a 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
@@ -42,7 +42,7 @@ import org.apache.impala.common.Reference;
import org.apache.impala.fb.FbCompression;
import org.apache.impala.fb.FbFileBlock;
import org.apache.impala.fb.FbFileDesc;
-import org.apache.impala.thrift.ImpalaInternalServiceConstants;
+import org.apache.impala.thrift.CatalogObjectsConstants;
import org.apache.impala.thrift.TAccessLevel;
import org.apache.impala.thrift.TExpr;
import org.apache.impala.thrift.TExprNode;
@@ -67,7 +67,6 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
-import com.google.common.primitives.Ints;
import com.google.flatbuffers.FlatBufferBuilder;
/**
@@ -453,11 +452,6 @@ public class HdfsPartition implements FeFsPartition {
}
@Override // FeFsPartition
- public boolean isDefaultPartition() {
- return id_ == ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID;
- }
-
- @Override // FeFsPartition
public boolean isCacheable() {
return FileSystemUtil.isPathCacheable(new Path(getLocation()));
}
@@ -764,13 +758,13 @@ public class HdfsPartition implements FeFsPartition {
accessLevel);
}
- public static HdfsPartition defaultPartition(
+ public static HdfsPartition prototypePartition(
HdfsTable table, HdfsStorageDescriptor storageDescriptor) {
List<LiteralExpr> emptyExprList = Lists.newArrayList();
List<FileDescriptor> emptyFileDescriptorList = Lists.newArrayList();
return new HdfsPartition(table, null, emptyExprList,
storageDescriptor, emptyFileDescriptorList,
- ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID, null,
+ CatalogObjectsConstants.PROTOTYPE_PARTITION_ID, null,
TAccessLevel.READ_WRITE);
}
@@ -819,7 +813,7 @@ public class HdfsPartition implements FeFsPartition {
thriftPartition.blockSize);
List<LiteralExpr> literalExpr = Lists.newArrayList();
- if (id != ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID) {
+ if (id != CatalogObjectsConstants.PROTOTYPE_PARTITION_ID) {
List<Column> clusterCols = Lists.newArrayList();
for (int i = 0; i < table.getNumClusteringCols(); ++i) {
clusterCols.add(table.getColumns().get(i));
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/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 02ad3da..a34eb0a 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -66,7 +66,7 @@ import org.apache.impala.common.Reference;
import org.apache.impala.compat.HdfsShim;
import org.apache.impala.fb.FbFileBlock;
import org.apache.impala.service.BackendConfig;
-import org.apache.impala.thrift.ImpalaInternalServiceConstants;
+import org.apache.impala.thrift.CatalogObjectsConstants;
import org.apache.impala.thrift.TAccessLevel;
import org.apache.impala.thrift.TCatalogObjectType;
import org.apache.impala.thrift.TColumn;
@@ -94,9 +94,11 @@ import org.slf4j.LoggerFactory;
import com.codahale.metrics.Gauge;
import com.codahale.metrics.Timer;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
@@ -188,6 +190,12 @@ public class HdfsTable extends Table implements FeFsTable {
// Store all the partition ids of an HdfsTable.
private final HashSet<Long> partitionIds_ = Sets.newHashSet();
+ // The partition used as a prototype when creating new partitions during
+ // insertion. New partitions inherit file format and other settings from
+ // the prototype.
+ @VisibleForTesting
+ HdfsPartition prototypePartition_;
+
// Estimate (in bytes) of the incremental stats size per column per partition
public static final long STATS_SIZE_PER_COLUMN_BYTES = 400;
@@ -360,9 +368,6 @@ public class HdfsTable extends Table implements FeFsTable {
if (!isLocationCacheable()) return false;
if (!isMarkedCached() && numClusteringCols_ > 0) {
for (FeFsPartition partition: getPartitions()) {
- if (partition.getId() == ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID) {
- continue;
- }
if (!partition.isCacheable()) {
return false;
}
@@ -664,7 +669,6 @@ public class HdfsTable extends Table implements FeFsTable {
// Search through all the partitions and check if their partition key values
// match the values being searched for.
for (HdfsPartition partition: partitionMap_.values()) {
- if (partition.isDefaultPartition()) continue;
List<LiteralExpr> partitionValues = partition.getPartitionValues();
Preconditions.checkState(partitionValues.size() == targetValues.size());
boolean matchFound = true;
@@ -748,7 +752,7 @@ public class HdfsTable extends Table implements FeFsTable {
}
/**
- * Resets any partition metadata, creates the default partition and sets the base
+ * Resets any partition metadata, creates the prototype partition and sets the base
* table directory path as well as the caching info from the HMS table.
*/
private void initializePartitionMetadata(
@@ -756,10 +760,7 @@ public class HdfsTable extends Table implements FeFsTable {
Preconditions.checkNotNull(msTbl);
resetPartitions();
hdfsBaseDir_ = msTbl.getSd().getLocation();
- // INSERT statements need to refer to this if they try to write to new partitions
- // Scans don't refer to this because by definition all partitions they refer to
- // exist.
- addDefaultPartition(msTbl.getSd());
+ setPrototypePartition(msTbl.getSd());
// We silently ignore cache directives that no longer exist in HDFS, and remove
// non-existing cache directives from the parameters.
@@ -1188,16 +1189,15 @@ public class HdfsTable extends Table implements FeFsTable {
}
/**
- * Adds or replaces the default partition.
+ * Update the prototype partition used when creating new partitions for
+ * this table. New partitions will inherit storage properties from the
+ * provided descriptor.
*/
- public void addDefaultPartition(StorageDescriptor storageDescriptor)
+ public void setPrototypePartition(StorageDescriptor storageDescriptor)
throws CatalogException {
- // Default partition has no files and is not referred to by scan nodes. Data sinks
- // refer to this to understand how to create new partitions.
HdfsStorageDescriptor hdfsStorageDescriptor =
HdfsStorageDescriptor.fromStorageDescriptor(this.name_, storageDescriptor);
- HdfsPartition partition = HdfsPartition.defaultPartition(this, hdfsStorageDescriptor);
- partitionMap_.put(partition.getId(), partition);
+ prototypePartition_ = HdfsPartition.prototypePartition(this, hdfsStorageDescriptor);
}
@Override
@@ -1305,14 +1305,15 @@ public class HdfsTable extends Table implements FeFsTable {
/**
* Updates the file metadata of an unpartitioned HdfsTable.
*/
- private void updateUnpartitionedTableFileMd() throws Exception {
+ private void updateUnpartitionedTableFileMd() throws CatalogException {
+ Preconditions.checkState(getNumClusteringCols() == 0);
if (LOG.isTraceEnabled()) {
LOG.trace("update unpartitioned table: " + getFullName());
}
resetPartitions();
org.apache.hadoop.hive.metastore.api.Table msTbl = getMetaStoreTable();
Preconditions.checkNotNull(msTbl);
- addDefaultPartition(msTbl.getSd());
+ setPrototypePartition(msTbl.getSd());
HdfsPartition part = createPartition(msTbl.getSd(), null);
addPartition(part);
if (isMarkedCached_) part.markCached();
@@ -1355,8 +1356,6 @@ public class HdfsTable extends Table implements FeFsTable {
// Identify dirty partitions that need to be loaded from the Hive Metastore and
// partitions that no longer exist in the Hive Metastore.
for (HdfsPartition partition: partitionMap_.values()) {
- // Ignore the default partition
- if (partition.isDefaultPartition()) continue;
// Remove partitions that don't exist in the Hive Metastore. These are partitions
// that were removed from HMS using some external process, e.g. Hive.
if (!msPartitionNames.contains(partition.getPartitionName())) {
@@ -1440,12 +1439,11 @@ public class HdfsTable extends Table implements FeFsTable {
@Override
public void setTableStats(org.apache.hadoop.hive.metastore.api.Table msTbl) {
super.setTableStats(msTbl);
- // For unpartitioned tables set the numRows in its partitions
+ // For unpartitioned tables set the numRows in its single partition
// to the table's numRows.
if (numClusteringCols_ == 0 && !partitionMap_.isEmpty()) {
- // Unpartitioned tables have a 'dummy' partition and a default partition.
- // Temp tables used in CTAS statements have one partition.
- Preconditions.checkState(partitionMap_.size() == 2 || partitionMap_.size() == 1);
+ // Unpartitioned tables have a default partition.
+ Preconditions.checkState(partitionMap_.size() == 1);
for (HdfsPartition p: partitionMap_.values()) {
p.setNumRows(getNumRows());
}
@@ -1664,6 +1662,9 @@ public class HdfsTable extends Table implements FeFsTable {
HdfsPartition.fromThrift(this, part.getKey(), part.getValue());
addPartition(hdfsPart);
}
+ prototypePartition_ = HdfsPartition.fromThrift(this,
+ CatalogObjectsConstants.PROTOTYPE_PARTITION_ID,
+ hdfsTable.prototype_partition);
} catch (CatalogException e) {
throw new TableLoadingException(e.getMessage());
}
@@ -1740,11 +1741,13 @@ public class HdfsTable extends Table implements FeFsTable {
}
if (includeFileDesc) fileMetadataStats_.set(stats);
+ THdfsPartition prototypePartition = prototypePartition_.toThrift(false, false);
+
memUsageEstimate += fileMetadataStats_.numFiles * PER_FD_MEM_USAGE_BYTES +
fileMetadataStats_.numBlocks * PER_BLOCK_MEM_USAGE_BYTES;
setEstimatedMetadataSize(memUsageEstimate);
THdfsTable hdfsTable = new THdfsTable(hdfsBaseDir_, getColumnNames(),
- nullPartitionKeyValue_, nullColumnValue_, idToPartition);
+ nullPartitionKeyValue_, nullColumnValue_, idToPartition, prototypePartition);
hdfsTable.setAvroSchema(avroSchema_);
hdfsTable.setMultiple_filesystems(multipleFileSystems_);
if (includeFileDesc) {
@@ -1772,8 +1775,12 @@ public class HdfsTable extends Table implements FeFsTable {
* Returns the file format that the majority of partitions are stored in.
*/
public HdfsFileFormat getMajorityFormat() {
+ // In the case that we have no partitions added to the table yet, it's
+ // important to add the "prototype" partition as a fallback.
+ Iterable<HdfsPartition> partitionsToConsider = Iterables.concat(
+ partitionMap_.values(), Collections.singleton(prototypePartition_));
Map<HdfsFileFormat, Integer> numPartitionsByFormat = Maps.newHashMap();
- for (HdfsPartition partition: partitionMap_.values()) {
+ for (HdfsPartition partition: partitionsToConsider) {
HdfsFileFormat format = partition.getInputFormatDescriptor().getFileFormat();
Integer numPartitions = numPartitionsByFormat.get(format);
if (numPartitions == null) {
@@ -1805,7 +1812,6 @@ public class HdfsTable extends Table implements FeFsTable {
HashSet<List<LiteralExpr>> existingPartitions = new HashSet<List<LiteralExpr>>();
// Get the list of partition values of existing partitions in Hive Metastore.
for (HdfsPartition partition: partitionMap_.values()) {
- if (partition.isDefaultPartition()) continue;
existingPartitions.add(partition.getPartitionValues());
}
@@ -2002,8 +2008,6 @@ public class HdfsTable extends Table implements FeFsTable {
long totalCachedBytes = 0L;
for (HdfsPartition p: orderedPartitions) {
- // Ignore dummy default partition.
- if (p.isDefaultPartition()) continue;
TResultRowBuilder rowBuilder = new TResultRowBuilder();
// Add the partition-key values (as strings for simplicity).
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
index e84a6a0..63428fd 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
@@ -1204,7 +1204,7 @@ public class HdfsScanNode extends ScanNode {
if (detailLevel.ordinal() >= TExplainLevel.STANDARD.ordinal()) {
if (tbl_.getNumClusteringCols() == 0) numPartitions_ = 1;
output.append(String.format("%spartitions=%s/%s files=%s size=%s", detailPrefix,
- numPartitions_, table.getPartitions().size() - 1, totalFiles_,
+ numPartitions_, table.getPartitions().size(), totalFiles_,
PrintUtils.printBytes(totalBytes_)));
output.append("\n");
if (!conjuncts_.isEmpty()) {
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
index 331c910..42236a3 100644
--- a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
@@ -1231,9 +1231,7 @@ public class SingleNodePlanner {
for (FeFsPartition partition: partitions) {
// Ignore empty partitions to match the behavior of the scan based approach.
- if (partition.isDefaultPartition() || partition.getSize() == 0) {
- continue;
- }
+ if (partition.getSize() == 0) continue;
List<Expr> exprs = Lists.newArrayList();
for (SlotDescriptor slotDesc: tupleDesc.getSlots()) {
// UnionNode.init() will go through all the slots in the tuple descriptor so
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
----------------------------------------------------------------------
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 f846e60..f0b0adb 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -90,7 +90,6 @@ import org.apache.impala.common.InternalException;
import org.apache.impala.common.Pair;
import org.apache.impala.common.Reference;
import org.apache.impala.compat.MetastoreShim;
-import org.apache.impala.thrift.ImpalaInternalServiceConstants;
import org.apache.impala.thrift.JniCatalogConstants;
import org.apache.impala.thrift.TAlterDbParams;
import org.apache.impala.thrift.TAlterDbSetOwnerParams;
@@ -822,7 +821,6 @@ public class CatalogOpExecutor {
for (FeFsPartition fePartition: table.getPartitions()) {
// TODO(todd): avoid downcast to implementation class
HdfsPartition partition = (HdfsPartition)fePartition;
- if (partition.isDefaultPartition()) continue;
// NULL keys are returned as 'NULL' in the partition_stats map, so don't substitute
// this partition's keys with Hive's replacement value.
@@ -1271,11 +1269,6 @@ public class CatalogOpExecutor {
// TODO(todd): avoid downcast
HdfsPartition part = (HdfsPartition) fePart;
boolean isModified = false;
- // The default partition is an Impala-internal abstraction and is not
- // represented in the Hive Metastore.
- if (part.getId() == ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID) {
- continue;
- }
if (part.getPartitionStats() != null) {
PartitionStatsUtil.deletePartStats(part);
isModified = true;
@@ -1538,7 +1531,6 @@ public class CatalogOpExecutor {
try {
HdfsTable hdfsTable = (HdfsTable)table;
for (FeFsPartition part: hdfsTable.getPartitions()) {
- if (part.isDefaultPartition()) continue;
FileSystemUtil.deleteAllVisibleFiles(new Path(part.getLocation()));
}
@@ -2355,9 +2347,9 @@ public class CatalogOpExecutor {
org.apache.hadoop.hive.metastore.api.Table msTbl =
tbl.getMetaStoreTable().deepCopy();
setStorageDescriptorFileFormat(msTbl.getSd(), fileFormat);
- // The default partition must be updated if the file format is changed so that new
+ // The prototype partition must be updated if the file format is changed so that new
// partitions are created with the new file format.
- if (tbl instanceof HdfsTable) ((HdfsTable) tbl).addDefaultPartition(msTbl.getSd());
+ if (tbl instanceof HdfsTable) ((HdfsTable) tbl).setPrototypePartition(msTbl.getSd());
applyAlterTable(msTbl, true);
reloadFileMetadata = true;
} else {
@@ -2394,9 +2386,9 @@ public class CatalogOpExecutor {
tbl.getMetaStoreTable().deepCopy();
StorageDescriptor sd = msTbl.getSd();
HiveStorageDescriptorFactory.setSerdeInfo(rowFormat, sd.getSerdeInfo());
- // The default partition must be updated if the row format is changed so that new
+ // The prototype partition must be updated if the row format is changed so that new
// partitions are created with the new file format.
- ((HdfsTable) tbl).addDefaultPartition(msTbl.getSd());
+ ((HdfsTable) tbl).setPrototypePartition(msTbl.getSd());
applyAlterTable(msTbl, true);
reloadFileMetadata = true;
} else {
@@ -2580,11 +2572,6 @@ public class CatalogOpExecutor {
for (FeFsPartition fePartition: hdfsTable.getPartitions()) {
// TODO(todd): avoid downcast
HdfsPartition partition = (HdfsPartition) fePartition;
- // No need to cache the default partition because it contains no files and is
- // not referred to by scan nodes.
- if (partition.getId() == ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID) {
- continue;
- }
// Only issue cache directives if the data is uncached or the cache directive
// needs to be updated
if (!partition.isMarkedCached() ||
@@ -2637,9 +2624,6 @@ public class CatalogOpExecutor {
for (FeFsPartition fePartition: hdfsTable.getPartitions()) {
// TODO(todd): avoid downcast
HdfsPartition partition = (HdfsPartition) fePartition;
- if (partition.getId() == ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID) {
- continue;
- }
if (partition.isMarkedCached()) {
HdfsCachingUtil.removePartitionCacheDirective(partition);
try {
@@ -3304,11 +3288,6 @@ public class CatalogOpExecutor {
Sets.newHashSet(update.getCreated_partitions());
partsToLoadMetadata = Sets.newHashSet(partsToCreate);
for (FeFsPartition partition: ((HdfsTable) table).getPartitions()) {
- // Skip dummy default partition.
- long partitionId = partition.getId();
- if (partitionId == ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID) {
- continue;
- }
// TODO: In the BE we build partition names without a trailing char. In FE
// we build partition name with a trailing char. We should make this
// consistent.
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/fe/src/test/java/org/apache/impala/catalog/CatalogObjectToFromThriftTest.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/catalog/CatalogObjectToFromThriftTest.java b/fe/src/test/java/org/apache/impala/catalog/CatalogObjectToFromThriftTest.java
index 3750149..bbb1993 100644
--- a/fe/src/test/java/org/apache/impala/catalog/CatalogObjectToFromThriftTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/CatalogObjectToFromThriftTest.java
@@ -25,7 +25,7 @@ import org.apache.impala.analysis.LiteralExpr;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.testutil.CatalogServiceTestCatalog;
-import org.apache.impala.thrift.ImpalaInternalServiceConstants;
+import org.apache.impala.thrift.CatalogObjectsConstants;
import org.apache.impala.thrift.TAccessLevel;
import org.apache.impala.thrift.THBaseTable;
import org.apache.impala.thrift.THdfsPartition;
@@ -37,6 +37,7 @@ import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
+import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
/**
@@ -59,6 +60,9 @@ public class CatalogObjectToFromThriftTest {
"functional_seq"};
for (String dbName: dbNames) {
Table table = catalog_.getOrLoadTable(dbName, "alltypes");
+ Assert.assertEquals(24, ((HdfsTable)table).getPartitions().size());
+ Assert.assertEquals(24, ((HdfsTable)table).getPartitionIds().size());
+
TTable thriftTable = getThriftTable(table);
Assert.assertEquals(thriftTable.tbl_name, "alltypes");
Assert.assertEquals(thriftTable.db_name, dbName);
@@ -68,17 +72,17 @@ public class CatalogObjectToFromThriftTest {
THdfsTable hdfsTable = thriftTable.getHdfs_table();
Assert.assertTrue(hdfsTable.hdfsBaseDir != null);
- // The table has 24 partitions + the default partition
- Assert.assertEquals(hdfsTable.getPartitions().size(), 25);
- Assert.assertTrue(hdfsTable.getPartitions().containsKey(
- new Long(ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID)));
-
+ // The table has 24 partitions.
+ Assert.assertEquals(24, hdfsTable.getPartitions().size());
+ Assert.assertFalse(hdfsTable.getPartitions().containsKey(
+ CatalogObjectsConstants.PROTOTYPE_PARTITION_ID));
+ // The prototype partition should be included and set properly.
+ Assert.assertTrue(hdfsTable.isSetPrototype_partition());
+ Assert.assertEquals(CatalogObjectsConstants.PROTOTYPE_PARTITION_ID,
+ hdfsTable.getPrototype_partition().id);
+ Assert.assertNull(hdfsTable.getPrototype_partition().location);
for (Map.Entry<Long, THdfsPartition> kv: hdfsTable.getPartitions().entrySet()) {
- if (kv.getKey() == ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID) {
- Assert.assertEquals(kv.getValue().getPartitionKeyExprs().size(), 0);
- } else {
- Assert.assertEquals(kv.getValue().getPartitionKeyExprs().size(), 2);
- }
+ Assert.assertEquals(kv.getValue().getPartitionKeyExprs().size(), 2);
}
// Now try to load the thrift struct.
@@ -90,29 +94,26 @@ public class CatalogObjectToFromThriftTest {
if (dbName.equals("functional")) Assert.assertEquals(7300, newTable.getNumRows());
HdfsTable newHdfsTable = (HdfsTable) newTable;
- Assert.assertEquals(newHdfsTable.getPartitions().size(), 25);
- boolean foundDefaultPartition = false;
+ Assert.assertEquals(newHdfsTable.getPartitions().size(), 24);
+ Assert.assertEquals(newHdfsTable.getPartitionIds().size(), 24);
for (FeFsPartition hdfsPart: newHdfsTable.getPartitions()) {
- if (hdfsPart.getId() == ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID) {
- Assert.assertEquals(foundDefaultPartition, false);
- foundDefaultPartition = true;
+ Assert.assertEquals(hdfsPart.getFileDescriptors().size(), 1);
+ Assert.assertTrue(
+ hdfsPart.getFileDescriptors().get(0).getNumFileBlocks() > 0);
+
+ // Verify the partition access level is getting set properly. The alltypes_seq
+ // table has two partitions that are read_only.
+ if (dbName.equals("functional_seq") && (
+ hdfsPart.getPartitionName().equals("year=2009/month=1") ||
+ hdfsPart.getPartitionName().equals("year=2009/month=3"))) {
+ Assert.assertEquals(TAccessLevel.READ_ONLY, hdfsPart.getAccessLevel());
} else {
- Assert.assertEquals(hdfsPart.getFileDescriptors().size(), 1);
- Assert.assertTrue(
- hdfsPart.getFileDescriptors().get(0).getNumFileBlocks() > 0);
-
- // Verify the partition access level is getting set properly. The alltypes_seq
- // table has two partitions that are read_only.
- if (dbName.equals("functional_seq") && (
- hdfsPart.getPartitionName().equals("year=2009/month=1") ||
- hdfsPart.getPartitionName().equals("year=2009/month=3"))) {
- Assert.assertEquals(TAccessLevel.READ_ONLY, hdfsPart.getAccessLevel());
- } else {
- Assert.assertEquals(TAccessLevel.READ_WRITE, hdfsPart.getAccessLevel());
- }
+ Assert.assertEquals(TAccessLevel.READ_WRITE, hdfsPart.getAccessLevel());
}
}
- Assert.assertEquals(foundDefaultPartition, true);
+ Assert.assertNotNull(newHdfsTable.prototypePartition_);
+ Assert.assertEquals(((HdfsTable)table).prototypePartition_.getParameters(),
+ newHdfsTable.prototypePartition_.getParameters());
}
}
@@ -214,13 +215,9 @@ public class CatalogObjectToFromThriftTest {
HdfsTable hdfsTable = (HdfsTable) table;
// Get any partition with valid HMS parameters to create a
// dummy partition.
- HdfsPartition part = null;
- for (FeFsPartition partition: hdfsTable.getPartitions()) {
- if (!partition.isDefaultPartition()) {
- part = (HdfsPartition) partition;
- break;
- }
- }
+ HdfsPartition part = (HdfsPartition)Iterables.getFirst(
+ hdfsTable.getPartitions(), null);
+ Assert.assertNotNull(part);
// Create a dummy partition with an invalid decimal type.
try {
HdfsPartition dummyPart = new HdfsPartition(hdfsTable, part.toHmsPartition(),
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
----------------------------------------------------------------------
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 f8a2682..4126b72 100644
--- a/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
@@ -17,7 +17,6 @@
package org.apache.impala.catalog;
-import static org.apache.impala.thrift.ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
@@ -291,14 +290,10 @@ public class CatalogTest {
Collection<? extends FeFsPartition> partitions = table.getPartitions();
// check that partition keys cover the date range 1/1/2009-12/31/2010
- // and that we have one file per partition, plus the default partition
- assertEquals(25, partitions.size());
+ // and that we have one file per partition.
+ assertEquals(24, partitions.size());
Set<Long> months = Sets.newHashSet();
for (FeFsPartition p: partitions) {
- if (p.getId() == DEFAULT_PARTITION_ID) {
- continue;
- }
-
assertEquals(2, p.getPartitionValues().size());
LiteralExpr key1Expr = p.getPartitionValues().get(0);
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/fe/src/test/java/org/apache/impala/common/FrontendTestBase.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/common/FrontendTestBase.java b/fe/src/test/java/org/apache/impala/common/FrontendTestBase.java
index 913f424..64d743c 100644
--- a/fe/src/test/java/org/apache/impala/common/FrontendTestBase.java
+++ b/fe/src/test/java/org/apache/impala/common/FrontendTestBase.java
@@ -182,7 +182,7 @@ public class FrontendTestBase {
}
try {
HdfsTable hdfsTable = (HdfsTable) dummyTable;
- hdfsTable.addDefaultPartition(msTbl.getSd());
+ hdfsTable.setPrototypePartition(msTbl.getSd());
} catch (CatalogException e) {
e.printStackTrace();
fail("Failed to add test table:\n" + createTableSql);
http://git-wip-us.apache.org/repos/asf/impala/blob/747e0a94/fe/src/test/java/org/apache/impala/testutil/BlockIdGenerator.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/testutil/BlockIdGenerator.java b/fe/src/test/java/org/apache/impala/testutil/BlockIdGenerator.java
index 820b33a..07ed153 100644
--- a/fe/src/test/java/org/apache/impala/testutil/BlockIdGenerator.java
+++ b/fe/src/test/java/org/apache/impala/testutil/BlockIdGenerator.java
@@ -73,11 +73,6 @@ public class BlockIdGenerator {
// Write the output as <tablename>: <blockid1> <blockid2> <etc>
writer.write(tableName + ":");
for (FeFsPartition partition: hdfsTable.getPartitions()) {
- // Ignore the default partition.
- if (partition.getId() ==
- ImpalaInternalServiceConstants.DEFAULT_PARTITION_ID) {
- continue;
- }
List<FileDescriptor> fileDescriptors = partition.getFileDescriptors();
for (FileDescriptor fd : fileDescriptors) {
Path p = new Path(partition.getLocation(), fd.getFileName());