You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2021/06/05 00:22:35 UTC
[impala] branch master updated: IMPALA-7501: Slim down partition
metadata in LocalCatalog mode
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
The following commit(s) were added to refs/heads/master by this push:
new bb30621 IMPALA-7501: Slim down partition metadata in LocalCatalog mode
bb30621 is described below
commit bb3062197b134f33e2796fac603e3367ab8bef1a
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Mon May 24 15:41:30 2021 +0800
IMPALA-7501: Slim down partition metadata in LocalCatalog mode
In LocalCatalog mode, the coordinator caches HMS partition objects in
its local cache. HMS partition contains many fields that Impala won't
used and some fields that are suboptimal. Most of them are in the
StorageDescriptor:
- partition-level schema (list<FieldSchema>) which is never used.
- location string which can be prefix-compressed since the prefix is
usually the table location.
- input/outputFormat string which is always duplicated and can be
represented by an integer(enum).
An experiment on a large table with 478 columns and 87320 partitions
(one non-empty file per partition) shows that more than 90% of the
memory space (1.1GB) are occupied by these fields. The dominant part is
the partition-level schema which consumed 76% of the cache.
On the other hand, these unused or suboptimal fields are got in one
response from catalogd, wrapped in TPartialPartitionInfo which finally
belongs to a TGetPartialCatalogObjectResponse. They dramatically
increase the serialized thrift object size of the response, which has a
2GB array size limit in JVM. Fetching metadata of many partitions from
catalogd could cause it runs into OOM error that hits the 2GB limit
(e.g. IMPALA-9896).
This patch extracts the HMS partition object and replaces it with the
fields that Impala actually uses. In the LocalCatalog cache, the HMS
partition object is replaced with
- hms parameters
- write id
- HdfsStorageDescriptor which represents the input/output format and
some delimiters.
- prefix-compressed location
The hms_partition field of TPartialPartitionInfo is also extracted with
corresponding fields. However, CatalogHmsAPIHelper still requires the
whole hms partition object. So the hms_partition field is kept for its
usage. To distinguish the different requirements, we add a new field,
want_hms_partition in TTableInfoSelector. The existing
'want_partition_metadata' field means returning these extracted fields,
and the 'want_hms_partition' field means returning the whole HMS
partition object.
Improvement results in the above case:
- reduce the heap usage from 1.1GB to 113.2MB, objects from 41m to 2.3m
- reduce the response size from 1.7GB to 28.41MB.
Tests:
- Run local-catalog related tests locally
- Run CORE tests
Change-Id: I307e7a8193b54a7b3ab93d9ebd194766bbdbd977
Reviewed-on: http://gerrit.cloudera.org:8080/17505
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Vihang Karajgaonkar <vi...@cloudera.com>
---
be/src/runtime/descriptors.cc | 17 ++--
common/thrift/CatalogObjects.thrift | 22 ++++--
common/thrift/CatalogService.thrift | 25 +++++-
.../org/apache/impala/catalog/FeCatalogUtils.java | 8 +-
.../org/apache/impala/catalog/FeIcebergTable.java | 6 +-
.../GetPartialCatalogObjectRequestBuilder.java | 10 +--
.../org/apache/impala/catalog/HdfsPartition.java | 16 +++-
.../impala/catalog/HdfsStorageDescriptor.java | 18 +++--
.../java/org/apache/impala/catalog/HdfsTable.java | 15 +++-
.../impala/catalog/iceberg/IcebergCtasTarget.java | 12 ++-
.../impala/catalog/local/CatalogdMetaProvider.java | 91 ++++++++++++++++++----
.../impala/catalog/local/DirectMetaProvider.java | 31 +++++++-
.../impala/catalog/local/LocalFsPartition.java | 63 +++++++--------
.../apache/impala/catalog/local/LocalFsTable.java | 34 ++++----
.../impala/catalog/local/LocalIcebergTable.java | 5 +-
.../apache/impala/catalog/local/MetaProvider.java | 9 ++-
.../impala/catalog/PartialCatalogInfoTest.java | 15 ++--
.../catalog/PartialCatalogInfoWriteIdTest.java | 23 ++++--
.../catalog/local/CatalogdMetaProviderTest.java | 8 +-
19 files changed, 288 insertions(+), 140 deletions(-)
diff --git a/be/src/runtime/descriptors.cc b/be/src/runtime/descriptors.cc
index 052b822..f29423c 100644
--- a/be/src/runtime/descriptors.cc
+++ b/be/src/runtime/descriptors.cc
@@ -202,14 +202,15 @@ string TableDescriptor::DebugString() const {
HdfsPartitionDescriptor::HdfsPartitionDescriptor(
const THdfsTable& thrift_table, const THdfsPartition& thrift_partition)
- : line_delim_(thrift_partition.lineDelim),
- field_delim_(thrift_partition.fieldDelim),
- collection_delim_(thrift_partition.collectionDelim),
- escape_char_(thrift_partition.escapeChar),
- block_size_(thrift_partition.blockSize),
- id_(thrift_partition.id),
- thrift_partition_key_exprs_(thrift_partition.partitionKeyExprs),
- file_format_(thrift_partition.fileFormat) {
+ : id_(thrift_partition.id),
+ thrift_partition_key_exprs_(thrift_partition.partitionKeyExprs) {
+ THdfsStorageDescriptor sd = thrift_partition.hdfs_storage_descriptor;
+ line_delim_ = sd.lineDelim;
+ field_delim_ = sd.fieldDelim;
+ collection_delim_ = sd.collectionDelim;
+ escape_char_ = sd.escapeChar;
+ block_size_ = sd.blockSize;
+ file_format_ = sd.fileFormat;
DecompressLocation(thrift_table, thrift_partition, &location_);
}
diff --git a/common/thrift/CatalogObjects.thrift b/common/thrift/CatalogObjects.thrift
index 90d5a7b..a477d80 100644
--- a/common/thrift/CatalogObjects.thrift
+++ b/common/thrift/CatalogObjects.thrift
@@ -301,6 +301,18 @@ struct THdfsPartitionLocation {
2: required string suffix
}
+// Represents the file format metadata for files stored in a HDFS table or partition.
+struct THdfsStorageDescriptor {
+ 1: required byte lineDelim
+ 2: required byte fieldDelim
+ 3: required byte collectionDelim
+ 4: required byte mapKeyDelim
+ 5: required byte escapeChar
+ 6: required byte quoteChar
+ 7: required THdfsFileFormat fileFormat
+ 8: required i32 blockSize
+}
+
// Represents an HDFS partition
// TODO(vercegovac): rename to TFsPartition
struct THdfsPartition {
@@ -310,16 +322,8 @@ struct THdfsPartition {
// as part of query plans and fragments.
// ============================================================
- 1: optional byte lineDelim
- 2: optional byte fieldDelim
- 3: optional byte collectionDelim
- 4: optional byte mapKeyDelim
- 5: optional byte escapeChar
- 6: optional THdfsFileFormat fileFormat
-
// These are Literal expressions
7: list<Exprs.TExpr> partitionKeyExprs
- 8: optional i32 blockSize
10: optional THdfsPartitionLocation location
@@ -381,6 +385,8 @@ struct THdfsPartition {
23: optional string db_name
24: optional string tbl_name
25: optional string partition_name
+
+ 27: optional THdfsStorageDescriptor hdfs_storage_descriptor
}
// Constant partition ID used for THdfsPartition.prototype_partition below.
diff --git a/common/thrift/CatalogService.thrift b/common/thrift/CatalogService.thrift
index d70e4b6..058ab8a 100644
--- a/common/thrift/CatalogService.thrift
+++ b/common/thrift/CatalogService.thrift
@@ -350,7 +350,9 @@ struct TTableInfoSelector {
// ... each such partition should include its name.
3: bool want_partition_names
- // ... each such partition should include metadata (location, etc).
+ // ... each such partition should include metadata (location, etc). Note that the
+ // response won't contain hms partition structs (use want_hms_partition instead for
+ // this purpose).
4: bool want_partition_metadata
// ... each such partition should include its file info
@@ -380,6 +382,13 @@ struct TTableInfoSelector {
// to true, the list provided in want_stats_for_column_names will be ignored and
// stats for all columns will be returned.
11: optional bool want_stats_for_all_columns
+
+ // The response should contain the HMS partition struct. This indicates all fields
+ // requested by want_partition_metadata will be contained in the HMS partitions.
+ // So setting both want_partition_metadata and want_hms_partition to true is a waste.
+ // Note that want_hms_partition=true will consume more space (IMPALA-7501), so only use
+ // it in cases the clients do need HMS partition structs.
+ 12: bool want_hms_partition
}
// Returned information about a particular partition.
@@ -389,7 +398,7 @@ struct TPartialPartitionInfo {
// Set if 'want_partition_names' was set in TTableInfoSelector.
2: optional string name
- // Set if 'want_partition_metadata' was set in TTableInfoSelector.
+ // Set if 'want_hms_partition' was set in TTableInfoSelector.
3: optional hive_metastore.Partition hms_partition
// Set if 'want_partition_files' was set in TTableInfoSelector.
@@ -416,6 +425,14 @@ struct TPartialPartitionInfo {
// necessarily mean the data is cached. Set when 'want_partition_metadata' is true in
// TTableInfoSelector.
7: optional bool is_marked_cached
+
+ // Fields 10-13 are set if 'want_partition_metadata' was set in TTableInfoSelector.
+ // These fields are actual info of hms_partition that Impala needs, and are better
+ // compressed.
+ 10: optional map<string, string> hms_parameters
+ 11: optional i64 write_id
+ 12: optional CatalogObjects.THdfsStorageDescriptor hdfs_storage_descriptor
+ 13: optional CatalogObjects.THdfsPartitionLocation location
}
// Returned information about a Table, as selected by TTableInfoSelector.
@@ -453,6 +470,10 @@ struct TPartialTableInfo {
// Set if this table is marked as cached by hdfs caching. Does not necessarily mean the
// data is cached or that all/any partitions are cached. Only used in analyzing DDLs.
10: optional bool is_marked_cached
+
+ // The prefixes of locations of partitions in this table. See THdfsPartitionLocation for
+ // the description of how a prefix is computed.
+ 11: optional list<string> partition_prefixes
}
struct TBriefTableMeta {
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 b4f30e4..268440b 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeCatalogUtils.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeCatalogUtils.java
@@ -350,14 +350,8 @@ public abstract class FeCatalogUtils {
ThriftObjectType type) {
HdfsStorageDescriptor sd = part.getInputFormatDescriptor();
THdfsPartition thriftHdfsPart = new THdfsPartition();
- thriftHdfsPart.setLineDelim(sd.getLineDelim());
- thriftHdfsPart.setFieldDelim(sd.getFieldDelim());
- thriftHdfsPart.setCollectionDelim(sd.getCollectionDelim());
- thriftHdfsPart.setMapKeyDelim(sd.getMapKeyDelim());
- thriftHdfsPart.setEscapeChar(sd.getEscapeChar());
- thriftHdfsPart.setFileFormat(sd.getFileFormat().toThrift());
+ thriftHdfsPart.setHdfs_storage_descriptor(sd.toThrift());
thriftHdfsPart.setPartitionKeyExprs(Expr.treesToThrift(part.getPartitionValues()));
- thriftHdfsPart.setBlockSize(sd.getBlockSize());
thriftHdfsPart.setId(part.getId());
thriftHdfsPart.setLocation(part.getLocationAsThrift());
if (part.getWriteId() >= 0)
diff --git a/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java b/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
index 8ad21f0..9191f91 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
@@ -36,12 +36,10 @@ import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.TableMetadata;
-import org.apache.iceberg.types.Types;
import org.apache.impala.analysis.IcebergPartitionField;
import org.apache.impala.analysis.IcebergPartitionSpec;
import org.apache.impala.analysis.LiteralExpr;
import org.apache.impala.common.FileSystemUtil;
-import org.apache.impala.common.ImpalaRuntimeException;
import org.apache.impala.common.Reference;
import org.apache.impala.compat.HdfsShim;
import org.apache.impala.thrift.TColumn;
@@ -417,8 +415,8 @@ public interface FeIcebergTable extends FeFsTable {
THdfsTable hdfsTable) {
for (Map.Entry<Long, THdfsPartition> entry : hdfsTable.getPartitions().entrySet()) {
THdfsPartition partition = entry.getValue();
- partition.setFileFormat(IcebergUtil.toTHdfsFileFormat(
- icebergTable.getIcebergFileFormat()));
+ partition.getHdfs_storage_descriptor().setFileFormat(
+ IcebergUtil.toTHdfsFileFormat(icebergTable.getIcebergFileFormat()));
}
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/GetPartialCatalogObjectRequestBuilder.java b/fe/src/main/java/org/apache/impala/catalog/GetPartialCatalogObjectRequestBuilder.java
index c012eed..b62d304 100644
--- a/fe/src/main/java/org/apache/impala/catalog/GetPartialCatalogObjectRequestBuilder.java
+++ b/fe/src/main/java/org/apache/impala/catalog/GetPartialCatalogObjectRequestBuilder.java
@@ -34,7 +34,7 @@ import org.apache.impala.thrift.TTableInfoSelector;
*/
public class GetPartialCatalogObjectRequestBuilder {
private boolean wantFileMetadata_;
- private boolean wantPartitionMeta_;
+ private boolean wantHmsPartition_;
private boolean wantPartitionNames_;
private String tblName_, dbName_;
private boolean wantHmsTable_;
@@ -64,7 +64,7 @@ public class GetPartialCatalogObjectRequestBuilder {
*/
public GetPartialCatalogObjectRequestBuilder wantFiles() {
wantFileMetadata_ = true;
- wantPartitionMeta_ = true;
+ wantHmsPartition_ = true;
wantPartitionNames_ = true;
return this;
}
@@ -75,7 +75,7 @@ public class GetPartialCatalogObjectRequestBuilder {
*/
public GetPartialCatalogObjectRequestBuilder wantPartitions() {
wantPartitionNames_ = true;
- wantPartitionMeta_ = true;
+ wantHmsPartition_ = true;
return this;
}
@@ -133,8 +133,8 @@ public class GetPartialCatalogObjectRequestBuilder {
if (wantPartitionNames_) {
req.table_info_selector.want_partition_names = true;
}
- if (wantPartitionMeta_) {
- req.table_info_selector.want_partition_metadata = true;
+ if (wantHmsPartition_) {
+ req.table_info_selector.want_hms_partition = true;
}
if (wantFileMetadata_) {
req.table_info_selector.want_partition_files = true;
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 28f1f3d..b8c0a34 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
@@ -61,6 +61,7 @@ import org.apache.impala.thrift.THdfsFileDesc;
import org.apache.impala.thrift.THdfsPartition;
import org.apache.impala.thrift.THdfsPartitionLocation;
import org.apache.impala.thrift.TNetworkAddress;
+import org.apache.impala.thrift.TPartialPartitionInfo;
import org.apache.impala.thrift.TPartitionStats;
import org.apache.impala.util.HdfsCachingUtil;
import org.apache.impala.util.ListMap;
@@ -949,6 +950,17 @@ public class HdfsPartition extends CatalogObjectImpl
return cachedMsPartitionDescriptor_;
}
+ public void setPartitionMetadata(TPartialPartitionInfo tPart) {
+ // The special "prototype partition" or the only partition of an unpartitioned table
+ // have a null cachedMsPartitionDescriptor.
+ if (cachedMsPartitionDescriptor_ == null) return;
+ // Don't need to make a copy here since the caller should not modify the parameters.
+ tPart.hms_parameters = getParameters();
+ tPart.write_id = writeId_;
+ tPart.hdfs_storage_descriptor = fileFormatDescriptor_.toThrift();
+ tPart.location = getLocationAsThrift();
+ }
+
/**
* Returns a Hive-compatible partition object that may be used in calls to the
* metastore.
@@ -1529,8 +1541,8 @@ public class HdfsPartition extends CatalogObjectImpl
}
public Builder fromThrift(THdfsPartition thriftPartition) {
- fileFormatDescriptor_ = HdfsStorageDescriptor.fromThriftPartition(
- thriftPartition, table_.getName());
+ fileFormatDescriptor_ = HdfsStorageDescriptor.fromThrift(
+ thriftPartition.hdfs_storage_descriptor, table_.getName());
partitionKeyValues_ = new ArrayList<>();
if (id_ != CatalogObjectsConstants.PROTOTYPE_PARTITION_ID) {
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsStorageDescriptor.java b/fe/src/main/java/org/apache/impala/catalog/HdfsStorageDescriptor.java
index fd993fe..695f9fb 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsStorageDescriptor.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsStorageDescriptor.java
@@ -25,6 +25,7 @@ import org.apache.hadoop.hive.metastore.api.SerDeInfo;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.serde.serdeConstants;
import org.apache.impala.thrift.THdfsPartition;
+import org.apache.impala.thrift.THdfsStorageDescriptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -177,7 +178,6 @@ public class HdfsStorageDescriptor {
if (escapeChar == fieldDelim ||
escapeChar == lineDelim ||
escapeChar == collectionDelim) {
- // TODO: we should output the table name here but it's hard to get to now.
this.escapeChar_ = DEFAULT_ESCAPE_CHAR;
LOG.warn("Escape character for table, " + tblName + " is set to "
+ "the same character as one of the delimiters. Ignoring escape character.");
@@ -241,15 +241,17 @@ public class HdfsStorageDescriptor {
}
}
- public static HdfsStorageDescriptor fromThriftPartition(THdfsPartition thriftPartition,
+ public static HdfsStorageDescriptor fromThrift(THdfsStorageDescriptor tDesc,
String tableName) {
return INTERNER.intern(new HdfsStorageDescriptor(tableName,
- HdfsFileFormat.fromThrift(thriftPartition.getFileFormat()),
- thriftPartition.lineDelim, thriftPartition.fieldDelim,
- thriftPartition.collectionDelim, thriftPartition.mapKeyDelim,
- thriftPartition.escapeChar,
- (byte) '"', // TODO: We should probably add quoteChar to THdfsPartition.
- thriftPartition.blockSize));
+ HdfsFileFormat.fromThrift(tDesc.getFileFormat()), tDesc.lineDelim,
+ tDesc.fieldDelim, tDesc.collectionDelim, tDesc.mapKeyDelim, tDesc.escapeChar,
+ tDesc.quoteChar, tDesc.blockSize));
+ }
+
+ public THdfsStorageDescriptor toThrift() {
+ return new THdfsStorageDescriptor(lineDelim_, fieldDelim_, collectionDelim_,
+ mapKeyDelim_, escapeChar_, quoteChar_, fileFormat_.toThrift(), blockSize_);
}
public HdfsStorageDescriptor cloneWithChangedFileFormat(HdfsFileFormat newFormat) {
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 0539387..f8bd2f5 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -37,6 +37,7 @@ import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.apache.avro.Schema;
+import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -2056,6 +2057,12 @@ public class HdfsTable extends Table implements FeFsTable {
req.table_info_selector.want_partition_metadata ||
req.table_info_selector.want_partition_names ||
req.table_info_selector.want_partition_stats;
+ if (req.table_info_selector.want_partition_metadata
+ && req.table_info_selector.want_hms_partition) {
+ LOG.warn("Bad request that has both want_partition_metadata and " +
+ "want_hms_partition set to true. Duplicated data will be returned. {}",
+ StringUtils.abbreviate(req.toString(), 1000));
+ }
Collection<Long> partIds = req.table_info_selector.partition_ids;
if (partIds == null && wantPartitionInfo) {
@@ -2087,9 +2094,12 @@ public class HdfsTable extends Table implements FeFsTable {
}
if (req.table_info_selector.want_partition_metadata) {
- partInfo.hms_partition = part.toHmsPartition();
+ part.setPartitionMetadata(partInfo);
partInfo.setHas_incremental_stats(part.hasIncrementalStats());
}
+ if (req.table_info_selector.want_hms_partition) {
+ partInfo.hms_partition = part.toHmsPartition();
+ }
if (req.table_info_selector.want_partition_files) {
try {
@@ -2128,6 +2138,9 @@ public class HdfsTable extends Table implements FeFsTable {
resp.table_info.partitions.add(partInfo);
}
}
+ // In most of the cases, the prefix map only contains one item for the table location.
+ // Here we always send it since it's small.
+ resp.table_info.setPartition_prefixes(partitionLocationCompressor_.getPrefixes());
if (reqWriteIdList != null) {
LOG.debug("{} files filtered out of table {} for {}. Hit rate : {}",
diff --git a/fe/src/main/java/org/apache/impala/catalog/iceberg/IcebergCtasTarget.java b/fe/src/main/java/org/apache/impala/catalog/iceberg/IcebergCtasTarget.java
index cd9c8ec..205aac7 100644
--- a/fe/src/main/java/org/apache/impala/catalog/iceberg/IcebergCtasTarget.java
+++ b/fe/src/main/java/org/apache/impala/catalog/iceberg/IcebergCtasTarget.java
@@ -18,7 +18,6 @@
package org.apache.impala.catalog.iceberg;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -27,7 +26,6 @@ import java.util.Set;
import com.google.common.base.Preconditions;
-import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.catalog.Namespace;
@@ -42,7 +40,6 @@ import org.apache.impala.catalog.CtasTargetTable;
import org.apache.impala.catalog.Db;
import org.apache.impala.catalog.FeCatalogUtils;
import org.apache.impala.catalog.FeDb;
-import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.HdfsStorageDescriptor;
@@ -54,12 +51,11 @@ import org.apache.impala.catalog.local.LocalDb;
import org.apache.impala.catalog.local.LocalFsTable;
import org.apache.impala.common.ImpalaRuntimeException;
import org.apache.impala.catalog.StructType;
-import org.apache.impala.catalog.Table;
-import org.apache.impala.catalog.TableLoadingException;
import org.apache.impala.thrift.CatalogObjectsConstants;
import org.apache.impala.thrift.TCatalogObjectType;
import org.apache.impala.thrift.TColumn;
import org.apache.impala.thrift.THdfsPartition;
+import org.apache.impala.thrift.THdfsStorageDescriptor;
import org.apache.impala.thrift.THdfsTable;
import org.apache.impala.thrift.TIcebergCatalog;
import org.apache.impala.thrift.TIcebergFileFormat;
@@ -266,8 +262,10 @@ public class IcebergCtasTarget extends CtasTargetTable implements FeIcebergTable
private THdfsPartition createPrototypePartition() {
THdfsPartition prototypePart = new THdfsPartition();
- prototypePart.setFileFormat(IcebergUtil.toTHdfsFileFormat(icebergFileFormat_));
- prototypePart.setBlockSize(hdfsSd_.getBlockSize());
+ THdfsStorageDescriptor sd = new THdfsStorageDescriptor();
+ sd.setFileFormat(IcebergUtil.toTHdfsFileFormat(icebergFileFormat_));
+ sd.setBlockSize(hdfsSd_.getBlockSize());
+ prototypePart.setHdfs_storage_descriptor(sd);
prototypePart.setId(CatalogObjectsConstants.PROTOTYPE_PARTITION_ID);
return prototypePart;
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/CatalogdMetaProvider.java b/fe/src/main/java/org/apache/impala/catalog/local/CatalogdMetaProvider.java
index 1ee1041..af198ec 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/CatalogdMetaProvider.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/CatalogdMetaProvider.java
@@ -41,7 +41,6 @@ import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
-import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.SQLForeignKey;
import org.apache.hadoop.hive.metastore.api.SQLPrimaryKey;
import org.apache.hadoop.hive.metastore.api.Table;
@@ -56,6 +55,8 @@ import org.apache.impala.catalog.CatalogObjectCache;
import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.HdfsCachePool;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
+import org.apache.impala.catalog.HdfsPartitionLocationCompressor;
+import org.apache.impala.catalog.HdfsStorageDescriptor;
import org.apache.impala.catalog.ImpaladCatalog.ObjectUpdateSequencer;
import org.apache.impala.catalog.Principal;
import org.apache.impala.catalog.PrincipalPrivilege;
@@ -729,7 +730,8 @@ public class CatalogdMetaProvider implements MetaProvider {
return new TableMetaRefImpl(
dbName, tableName, resp.table_info.hms_table, resp.object_version_number,
new SqlConstraints(primaryKeys, foreignKeys),
- resp.table_info.valid_write_ids, resp.table_info.is_marked_cached);
+ resp.table_info.valid_write_ids, resp.table_info.is_marked_cached,
+ resp.table_info.partition_prefixes);
}
});
// The table list is populated based on tables in a given Db in catalogd. If a table
@@ -951,12 +953,28 @@ public class CatalogdMetaProvider implements MetaProvider {
for (int i = 0; i < ids.size(); i++) {
PartitionRef partRef = partRefs.get(i);
TPartialPartitionInfo part = resp.table_info.partitions.get(i);
- Partition msPart = part.getHms_partition();
- if (msPart == null) {
+ HdfsStorageDescriptor hdfsStorageDescriptor = null;
+ HdfsPartitionLocationCompressor.Location location;
+ if (part.isSetHdfs_storage_descriptor()) {
+ Preconditions.checkNotNull(part.location, "location should not be null");
+ hdfsStorageDescriptor = HdfsStorageDescriptor.fromThrift(
+ part.hdfs_storage_descriptor, table.tableName_);
+ location = table.getPartitionLocationCompressor().new Location(part.location);
+ } else {
checkResponse(table.msTable_.getPartitionKeysSize() == 0, req,
- "Should not return a partition with missing HMS partition unless " +
+ "Should not return a partition with missing partition meta unless " +
"the table is unpartitioned");
- msPart = DirectMetaProvider.msTableToPartition(table.msTable_);
+ // For the only partition of a nonpartitioned table, reuse table-level metadata.
+ try {
+ hdfsStorageDescriptor = HdfsStorageDescriptor.fromStorageDescriptor(
+ table.tableName_, table.msTable_.getSd());
+ } catch (HdfsStorageDescriptor.InvalidStorageDescriptorException e) {
+ Preconditions.checkState(false, "Failed to create HdfsStorageDescriptor " +
+ "using sd of table");
+ }
+ location = table.getPartitionLocationCompressor().new Location(
+ table.msTable_.getSd().getLocation());
+ part.setHms_parameters(table.msTable_.getParameters());
}
// Transform the file descriptors to the caller's index.
@@ -967,9 +985,10 @@ public class CatalogdMetaProvider implements MetaProvider {
part.insert_file_descriptors, resp.table_info.network_addresses, hostIndex);
ImmutableList<FileDescriptor> deleteFds = convertThriftFdList(
part.delete_file_descriptors, resp.table_info.network_addresses, hostIndex);
- PartitionMetadataImpl metaImpl = new PartitionMetadataImpl(msPart,
+ PartitionMetadataImpl metaImpl = new PartitionMetadataImpl(part.getHms_parameters(),
+ part.write_id, hdfsStorageDescriptor,
fds, insertFds, deleteFds, part.getPartition_stats(),
- part.has_incremental_stats, part.is_marked_cached);
+ part.has_incremental_stats, part.is_marked_cached, location);
checkResponse(partRef != null, req, "returned unexpected partition id %s", part.id);
@@ -1497,7 +1516,15 @@ public class CatalogdMetaProvider implements MetaProvider {
}
public static class PartitionMetadataImpl implements PartitionMetadata {
- private final Partition msPartition_;
+ private final Map<String, String> hmsParameters_;
+ private final long writeId_;
+ private final HdfsStorageDescriptor hdfsStorageDescriptor_;
+ // Prefix-compressed location. The corresponding prefix map, i.e. the
+ // HdfsPartitionLocationCompressor object is in the corresponding TableMetaRefImpl
+ // object. TableMetaRefImpl may be evicted. But when this partition is accessed,
+ // we can assure that the table meta is already loaded, so the prefix map won't be
+ // missing.
+ private final HdfsPartitionLocationCompressor.Location location_;
private final ImmutableList<FileDescriptor> fds_;
private final ImmutableList<FileDescriptor> insertFds_;
private final ImmutableList<FileDescriptor> deleteFds_;
@@ -1505,10 +1532,15 @@ public class CatalogdMetaProvider implements MetaProvider {
private final boolean hasIncrementalStats_;
private final boolean isMarkedCached_;
- public PartitionMetadataImpl(Partition msPartition, ImmutableList<FileDescriptor> fds,
+ public PartitionMetadataImpl(Map<String, String> hmsParameters, long writeId,
+ HdfsStorageDescriptor hdfsStorageDescriptor, ImmutableList<FileDescriptor> fds,
ImmutableList<FileDescriptor> insertFds, ImmutableList<FileDescriptor> deleteFds,
- byte[] partitionStats, boolean hasIncrementalStats, boolean isMarkedCached) {
- this.msPartition_ = Preconditions.checkNotNull(msPartition);
+ byte[] partitionStats, boolean hasIncrementalStats, boolean isMarkedCached,
+ HdfsPartitionLocationCompressor.Location location) {
+ this.hmsParameters_ = hmsParameters;
+ this.writeId_ = writeId;
+ this.hdfsStorageDescriptor_ = hdfsStorageDescriptor;
+ this.location_ = location;
this.fds_ = fds;
this.insertFds_ = insertFds;
this.deleteFds_ = deleteFds;
@@ -1530,8 +1562,9 @@ public class CatalogdMetaProvider implements MetaProvider {
insertFds_, origIndex, dstIndex);
ImmutableList<FileDescriptor> deleteFds = cloneFdsRelativeToHostIndex(
deleteFds_, origIndex, dstIndex);
- return new PartitionMetadataImpl(msPartition_, fds, insertFds, deleteFds,
- partitionStats_, hasIncrementalStats_, isMarkedCached_);
+ return new PartitionMetadataImpl(hmsParameters_, writeId_, hdfsStorageDescriptor_,
+ fds, insertFds, deleteFds, partitionStats_, hasIncrementalStats_,
+ isMarkedCached_, location_);
}
private static ImmutableList<FileDescriptor> cloneFdsRelativeToHostIndex(
@@ -1545,11 +1578,20 @@ public class CatalogdMetaProvider implements MetaProvider {
}
@Override
- public Partition getHmsPartition() {
- return msPartition_;
+ public Map<String, String> getHmsParameters() { return hmsParameters_; }
+
+ @Override
+ public long getWriteId() { return writeId_; }
+
+ @Override
+ public HdfsStorageDescriptor getInputFormatDescriptor() {
+ return hdfsStorageDescriptor_;
}
@Override
+ public HdfsPartitionLocationCompressor.Location getLocation() { return location_; }
+
+ @Override
public ImmutableList<FileDescriptor> getFileDescriptors() {
if (insertFds_.isEmpty()) return fds_;
List<FileDescriptor> ret = Lists.newArrayListWithCapacity(
@@ -1615,9 +1657,12 @@ public class CatalogdMetaProvider implements MetaProvider {
*/
private final boolean isMarkedCached_;
+ private final HdfsPartitionLocationCompressor partitionLocationCompressor_;
+
public TableMetaRefImpl(String dbName, String tableName,
Table msTable, long catalogVersion, SqlConstraints sqlConstraints,
- TValidWriteIdList validWriteIds, boolean isMarkedCached) {
+ TValidWriteIdList validWriteIds, boolean isMarkedCached,
+ List<String> locationPrefixes) {
this.dbName_ = dbName;
this.tableName_ = tableName;
this.msTable_ = msTable;
@@ -1625,6 +1670,10 @@ public class CatalogdMetaProvider implements MetaProvider {
this.sqlConstraints_ = sqlConstraints;
this.validWriteIds_ = validWriteIds;
this.isMarkedCached_ = isMarkedCached;
+ // Non-hdfs tables will have null locationPrefixes.
+ this.partitionLocationCompressor_ = (locationPrefixes == null) ? null :
+ new HdfsPartitionLocationCompressor(
+ msTable.getPartitionKeysSize(), locationPrefixes);
}
@Override
@@ -1633,6 +1682,14 @@ public class CatalogdMetaProvider implements MetaProvider {
}
public boolean isMarkedCached() { return isMarkedCached_; }
+
+ public HdfsPartitionLocationCompressor getPartitionLocationCompressor() {
+ return partitionLocationCompressor_;
+ }
+
+ public List<String> getPartitionPrefixes() {
+ return partitionLocationCompressor_.getPrefixes();
+ }
}
/**
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 44a677c..0745f56 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
@@ -41,6 +41,8 @@ import org.apache.impala.catalog.FileMetadataLoader;
import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.HdfsCachePool;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
+import org.apache.impala.catalog.HdfsPartitionLocationCompressor;
+import org.apache.impala.catalog.HdfsStorageDescriptor;
import org.apache.impala.catalog.MetaStoreClientPool;
import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
import org.apache.impala.catalog.SqlConstraints;
@@ -401,8 +403,28 @@ class DirectMetaProvider implements MetaProvider {
}
@Override
- public Partition getHmsPartition() {
- return msPartition_;
+ public Map<String, String> getHmsParameters() { return msPartition_.getParameters(); }
+
+ @Override
+ public long getWriteId() { return msPartition_.getWriteId(); }
+
+ @Override
+ public HdfsStorageDescriptor getInputFormatDescriptor() {
+ String tblName = msPartition_.getDbName() + "." + msPartition_.getTableName();
+ try {
+ return HdfsStorageDescriptor.fromStorageDescriptor(tblName, msPartition_.getSd());
+ } catch (HdfsStorageDescriptor.InvalidStorageDescriptorException e) {
+ throw new LocalCatalogException(String.format(
+ "Invalid input format descriptor for partition (values=%s) of table %s",
+ msPartition_.getValues(), tblName), e);
+ }
+ }
+
+ @Override
+ public HdfsPartitionLocationCompressor.Location getLocation() {
+ // Treat as no prefix compression.
+ return new HdfsPartitionLocationCompressor(0).new Location(
+ msPartition_.getSd().getLocation());
}
@Override
@@ -464,6 +486,11 @@ class DirectMetaProvider implements MetaProvider {
throw new UnsupportedOperationException("Hdfs caching not supported with " +
"DirectMetaProvider implementation");
}
+
+ @Override
+ public List<String> getPartitionPrefixes() {
+ return Collections.emptyList();
+ }
}
@Override
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java
index 575e01a..9ec78ec 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java
@@ -25,18 +25,18 @@ import javax.annotation.Nullable;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.impala.analysis.LiteralExpr;
import org.apache.impala.catalog.FeCatalogUtils;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
+import org.apache.impala.catalog.HdfsPartitionLocationCompressor;
import org.apache.impala.catalog.HdfsStorageDescriptor;
-import org.apache.impala.catalog.HdfsStorageDescriptor.InvalidStorageDescriptorException;
import org.apache.impala.catalog.IcebergTable;
import org.apache.impala.catalog.PartitionStatsUtil;
import org.apache.impala.common.FileSystemUtil;
-import org.apache.impala.compat.MetastoreShim;
import org.apache.impala.thrift.TAccessLevel;
import org.apache.impala.thrift.THdfsPartitionLocation;
import org.apache.impala.thrift.TNetworkAddress;
@@ -50,7 +50,12 @@ import org.apache.impala.util.IcebergUtil;
public class LocalFsPartition implements FeFsPartition {
private final LocalFsTable table_;
private final LocalPartitionSpec spec_;
- private final Partition msPartition_;
+ private final Map<String, String> hmsParameters_;
+ private final long writeId_;
+ private final HdfsStorageDescriptor hdfsStorageDescriptor_;
+ // Prefix-compressed location. The prefix map is carried in TableMetaRef of LocalTable.
+ // The special "prototype partition" has a null location.
+ private final HdfsPartitionLocationCompressor.Location location_;
/**
* Null in the case of a 'prototype partition'.
@@ -76,13 +81,19 @@ public class LocalFsPartition implements FeFsPartition {
private final boolean isMarkedCached_;
public LocalFsPartition(LocalFsTable table, LocalPartitionSpec spec,
- Partition msPartition, ImmutableList<FileDescriptor> fileDescriptors,
+ Map<String, String> hmsParameters, long writeId,
+ HdfsStorageDescriptor hdfsStorageDescriptor,
+ ImmutableList<FileDescriptor> fileDescriptors,
ImmutableList<FileDescriptor> insertFileDescriptors,
ImmutableList<FileDescriptor> deleteFileDescriptors,
- byte [] partitionStats, boolean hasIncrementalStats, boolean isMarkedCached) {
+ byte [] partitionStats, boolean hasIncrementalStats, boolean isMarkedCached,
+ HdfsPartitionLocationCompressor.Location location) {
table_ = Preconditions.checkNotNull(table);
spec_ = Preconditions.checkNotNull(spec);
- msPartition_ = Preconditions.checkNotNull(msPartition);
+ hmsParameters_ = hmsParameters;
+ writeId_ = writeId;
+ hdfsStorageDescriptor_ = hdfsStorageDescriptor;
+ location_ = location;
fileDescriptors_ = fileDescriptors;
insertFileDescriptors_ = insertFileDescriptors;
deleteFileDescriptors_ = deleteFileDescriptors;
@@ -151,22 +162,17 @@ public class LocalFsPartition implements FeFsPartition {
@Override
public String getLocation() {
- return msPartition_.getSd().getLocation();
+ return (location_ != null) ? location_.toString() : null;
}
@Override
public THdfsPartitionLocation getLocationAsThrift() {
- String loc = getLocation();
- // The special "prototype partition" has a null location.
- if (loc == null) return null;
- // TODO(todd): support prefix-compressed partition locations. For now,
- // using -1 indicates that the location is a full path string.
- return new THdfsPartitionLocation(/*prefix_index=*/-1, loc);
+ return (location_ != null) ? location_.toThrift() : null;
}
@Override
public Path getLocationPath() {
- Preconditions.checkNotNull(getLocation(),
+ Preconditions.checkNotNull(location_,
"LocalFsPartition location is null");
return new Path(getLocation());
}
@@ -189,17 +195,7 @@ public class LocalFsPartition implements FeFsPartition {
@Override
public HdfsStorageDescriptor getInputFormatDescriptor() {
- try {
- // TODO(todd): should we do this in the constructor to avoid having to worry
- // about throwing an exception from this getter? The downside is that then
- // the object would end up long-lived, whereas its actual usage is short-lived.
- return HdfsStorageDescriptor.fromStorageDescriptor(table_.getName(),
- msPartition_.getSd());
- } catch (InvalidStorageDescriptorException e) {
- throw new LocalCatalogException(String.format(
- "Invalid input format descriptor for partition %s of table %s",
- getPartitionName(), table_.getFullName()), e);
- }
+ return hdfsStorageDescriptor_;
}
@Override
@@ -238,7 +234,7 @@ public class LocalFsPartition implements FeFsPartition {
@Override
public long getNumRows() {
- return FeCatalogUtils.getRowCount(msPartition_.getParameters());
+ return FeCatalogUtils.getRowCount(hmsParameters_);
}
@Override
@@ -263,28 +259,29 @@ public class LocalFsPartition implements FeFsPartition {
@Override
public Map<String, String> getParameters() {
- return msPartition_.getParameters();
+ return hmsParameters_;
}
@Override
public long getWriteId() {
- return MetastoreShim.getWriteIdFromMSPartition(msPartition_);
+ return writeId_;
}
@Override
public LocalFsPartition genInsertDeltaPartition() {
ImmutableList<FileDescriptor> fds = insertFileDescriptors_.isEmpty() ?
fileDescriptors_ : insertFileDescriptors_;
- return new LocalFsPartition(table_, spec_, msPartition_, fds,
- ImmutableList.of(), ImmutableList.of(), partitionStats_,
- hasIncrementalStats_, isMarkedCached_);
+ return new LocalFsPartition(table_, spec_, hmsParameters_, writeId_,
+ hdfsStorageDescriptor_, fds, ImmutableList.of(), ImmutableList.of(),
+ partitionStats_, hasIncrementalStats_, isMarkedCached_, location_);
}
@Override
public LocalFsPartition genDeleteDeltaPartition() {
if (deleteFileDescriptors_.isEmpty()) return null;
- return new LocalFsPartition(table_, spec_, msPartition_, deleteFileDescriptors_,
+ return new LocalFsPartition(table_, spec_, hmsParameters_, writeId_,
+ hdfsStorageDescriptor_, deleteFileDescriptors_,
ImmutableList.of(), ImmutableList.of(), partitionStats_,
- hasIncrementalStats_, isMarkedCached_);
+ hasIncrementalStats_, isMarkedCached_, location_);
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java
index 956b964..8ce6f50 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java
@@ -30,7 +30,6 @@ import java.util.TreeMap;
import org.apache.avro.Schema;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
-import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.serde.serdeConstants;
@@ -45,6 +44,7 @@ import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
+import org.apache.impala.catalog.HdfsStorageDescriptor;
import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.catalog.PrunablePartition;
import org.apache.impala.catalog.SqlConstraints;
@@ -352,6 +352,7 @@ public class LocalFsTable extends LocalTable implements FeFsTable {
TValidWriteIdList validWriteIdList =
db_.getCatalog().getMetaProvider().getValidWriteIdList(ref_);
if (validWriteIdList != null) hdfsTable.setValid_write_ids(validWriteIdList);
+ hdfsTable.setPartition_prefixes(ref_.getPartitionPrefixes());
}
tableDesc.setHdfsTable(hdfsTable);
return tableDesc;
@@ -370,23 +371,24 @@ public class LocalFsTable extends LocalTable implements FeFsTable {
}
public LocalFsPartition createPrototypePartition() {
- Partition protoMsPartition = new Partition();
-
// The prototype partition should not have a location set in its storage
// descriptor, or else all inserted files will end up written into the
// table directory instead of the new partition directories.
StorageDescriptor sd = getMetaStoreTable().getSd().deepCopy();
sd.unsetLocation();
- protoMsPartition.setSd(sd);
-
- protoMsPartition.setParameters(Collections.<String, String>emptyMap());
+ HdfsStorageDescriptor hdfsStorageDescriptor = null;
+ try {
+ hdfsStorageDescriptor = HdfsStorageDescriptor.fromStorageDescriptor(name_, sd);
+ } catch (HdfsStorageDescriptor.InvalidStorageDescriptorException e) {
+ Preconditions.checkState(false, "Failed to create prototype partition " +
+ "HdfsStorageDescriptor using sd of table");
+ }
LocalPartitionSpec spec = new LocalPartitionSpec(
this, CatalogObjectsConstants.PROTOTYPE_PARTITION_ID);
- LocalFsPartition prototypePartition = new LocalFsPartition(
- this, spec, protoMsPartition, /*fileDescriptors=*/null,
- /*insertFileDescriptors=*/null, /*deleteFileDescriptors=*/null,
- /*partitionStats=*/null, /*hasIncrementalStats=*/false, /*isMarkedCached=*/false);
- return prototypePartition;
+ return new LocalFsPartition(this, spec, Collections.emptyMap(), /*writeId=*/-1,
+ hdfsStorageDescriptor, /*fileDescriptors=*/null, /*insertFileDescriptors=*/null,
+ /*deleteFileDescriptors=*/null, /*partitionStats=*/null,
+ /*hasIncrementalStats=*/false, /*isMarkedCached=*/false, /*location*/null);
}
@Override
@@ -461,10 +463,10 @@ public class LocalFsTable extends LocalTable implements FeFsTable {
ImmutableList<FileDescriptor> fds = p.getInsertFileDescriptors().isEmpty() ?
p.getFileDescriptors() : ImmutableList.of();
- LocalFsPartition part = new LocalFsPartition(this, spec, p.getHmsPartition(),
- fds, p.getInsertFileDescriptors(),
+ LocalFsPartition part = new LocalFsPartition(this, spec, p.getHmsParameters(),
+ p.getWriteId(), p.getInputFormatDescriptor(), fds, p.getInsertFileDescriptors(),
p.getDeleteFileDescriptors(), p.getPartitionStats(), p.hasIncrementalStats(),
- p.isMarkedCached());
+ p.isMarkedCached(), p.getLocation());
ret.add(part);
}
return ret;
@@ -581,4 +583,8 @@ public class LocalFsTable extends LocalTable implements FeFsTable {
}
return sqlConstraints_;
}
+
+ public List<String> getPartitionPrefixes() {
+ return ref_ == null ? Collections.emptyList() : ref_.getPartitionPrefixes();
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalIcebergTable.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalIcebergTable.java
index ff3b372..463deab 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalIcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalIcebergTable.java
@@ -18,7 +18,6 @@
package org.apache.impala.catalog.local;
import java.io.IOException;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
@@ -29,7 +28,6 @@ import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.iceberg.Schema;
import org.apache.iceberg.TableMetadata;
-import org.apache.iceberg.types.Types;
import org.apache.impala.analysis.IcebergPartitionSpec;
import org.apache.impala.catalog.CatalogObject;
import org.apache.impala.catalog.Column;
@@ -37,10 +35,8 @@ import org.apache.impala.catalog.FeCatalogUtils;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeIcebergTable;
-import org.apache.impala.catalog.IcebergColumn;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
import org.apache.impala.catalog.TableLoadingException;
-import org.apache.impala.catalog.Type;
import org.apache.impala.thrift.THdfsPartition;
import org.apache.impala.thrift.THdfsTable;
import org.apache.impala.thrift.TIcebergCatalog;
@@ -206,6 +202,7 @@ public class LocalIcebergTable extends LocalTable implements FeIcebergTable {
getColumnNames(), localFsTable_.getNullPartitionKeyValue(),
FeFsTable.DEFAULT_NULL_COLUMN_VALUE, idToPartition, tPrototypePartition);
Utils.updateIcebergPartitionFileFormat(this, hdfsTable);
+ hdfsTable.setPartition_prefixes(localFsTable_.getPartitionPrefixes());
return hdfsTable;
}
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 3d9a402..fe94f51 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
@@ -24,7 +24,6 @@ import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
-import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.api.UnknownDBException;
import org.apache.impala.authorization.AuthorizationPolicy;
@@ -32,6 +31,8 @@ import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.HdfsCachePool;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
+import org.apache.impala.catalog.HdfsPartitionLocationCompressor;
+import org.apache.impala.catalog.HdfsStorageDescriptor;
import org.apache.impala.catalog.SqlConstraints;
import org.apache.impala.common.Pair;
import org.apache.impala.thrift.TBriefTableMeta;
@@ -130,6 +131,7 @@ public interface MetaProvider {
*/
interface TableMetaRef {
boolean isMarkedCached();
+ List<String> getPartitionPrefixes();
}
/**
@@ -145,7 +147,10 @@ public interface MetaProvider {
* Partition metadata as returned by loadPartitionsByRefs().
*/
interface PartitionMetadata {
- Partition getHmsPartition();
+ Map<String, String> getHmsParameters();
+ long getWriteId();
+ HdfsStorageDescriptor getInputFormatDescriptor();
+ HdfsPartitionLocationCompressor.Location getLocation();
ImmutableList<FileDescriptor> getFileDescriptors();
ImmutableList<FileDescriptor> getInsertFileDescriptors();
ImmutableList<FileDescriptor> getDeleteFileDescriptors();
diff --git a/fe/src/test/java/org/apache/impala/catalog/PartialCatalogInfoTest.java b/fe/src/test/java/org/apache/impala/catalog/PartialCatalogInfoTest.java
index 810a370..7629084 100644
--- a/fe/src/test/java/org/apache/impala/catalog/PartialCatalogInfoTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/PartialCatalogInfoTest.java
@@ -35,6 +35,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.SQLForeignKey;
import org.apache.hadoop.hive.metastore.api.SQLPrimaryKey;
+import org.apache.hadoop.hive.serde.serdeConstants;
import org.apache.impala.common.InternalException;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.testutil.CatalogServiceTestCatalog;
@@ -47,6 +48,7 @@ import org.apache.impala.thrift.TDatabase;
import org.apache.impala.thrift.TDbInfoSelector;
import org.apache.impala.thrift.TGetPartialCatalogObjectRequest;
import org.apache.impala.thrift.TGetPartialCatalogObjectResponse;
+import org.apache.impala.thrift.THdfsFileFormat;
import org.apache.impala.thrift.TPartialPartitionInfo;
import org.apache.impala.thrift.TTable;
import org.apache.impala.thrift.TTableInfoSelector;
@@ -166,14 +168,17 @@ public class PartialCatalogInfoTest {
resp = sendRequest(req);
assertNull(resp.table_info.hms_table);
assertEquals(2, resp.table_info.partitions.size());
+ assertEquals(1, resp.table_info.getPartition_prefixesSize());
+ assertEquals("hdfs://localhost:20500/test-warehouse/alltypes/",
+ resp.table_info.partition_prefixes.get(0));
partInfo = resp.table_info.partitions.get(0);
assertNull(partInfo.name);
assertEquals(req.table_info_selector.partition_ids.get(0), (Long)partInfo.id);
- assertTrue(partInfo.hms_partition.getSd().getLocation().startsWith(
- "hdfs://localhost:20500/test-warehouse/alltypes/year="));
- // TODO(todd): we should probably transfer a compressed descriptor instead
- // and refactor the MetaProvider interface to expose those since there is
- // a lot of redundant info in partition descriptors.
+ assertEquals(0, partInfo.location.prefix_index);
+ assertTrue("Bad suffix " + partInfo.location.suffix,
+ partInfo.location.suffix.matches("year=\\d+/month=\\d+"));
+ assertNotNull(partInfo.hdfs_storage_descriptor);
+ assertEquals(THdfsFileFormat.TEXT, partInfo.hdfs_storage_descriptor.fileFormat);
}
@Test
diff --git a/fe/src/test/java/org/apache/impala/catalog/PartialCatalogInfoWriteIdTest.java b/fe/src/test/java/org/apache/impala/catalog/PartialCatalogInfoWriteIdTest.java
index 288eca3..ff5038a 100644
--- a/fe/src/test/java/org/apache/impala/catalog/PartialCatalogInfoWriteIdTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/PartialCatalogInfoWriteIdTest.java
@@ -17,9 +17,6 @@
package org.apache.impala.catalog;
-import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertTrue;
-
import java.sql.SQLException;
import java.util.List;
import java.util.concurrent.TimeUnit;
@@ -37,7 +34,6 @@ import org.apache.impala.thrift.TCatalogObjectType;
import org.apache.impala.thrift.TGetPartialCatalogObjectRequest;
import org.apache.impala.thrift.TGetPartialCatalogObjectResponse;
import org.apache.impala.thrift.THdfsFileDesc;
-import org.apache.impala.thrift.THdfsPartition;
import org.apache.impala.thrift.THdfsTable;
import org.apache.impala.thrift.TPartialPartitionInfo;
import org.apache.impala.thrift.TTable;
@@ -261,10 +257,17 @@ public class PartialCatalogInfoWriteIdTest {
.build();
TGetPartialCatalogObjectResponse response = sendRequest(request);
Assert.assertEquals(1, response.getTable_info().getPartitionsSize());
+ Assert.assertEquals(1, response.getTable_info().getPartition_prefixesSize());
+ Assert.assertNotNull(
+ response.getTable_info().getPartitions().get(0).getFile_descriptors());
+ Assert.assertNotNull(
+ response.getTable_info().getPartitions().get(0).getLocation());
+ Assert.assertEquals(0,
+ response.getTable_info().getPartitions().get(0).getLocation().getPrefix_index());
Assert.assertNotNull(
- response.getTable_info().getPartitions().get(0).getFile_descriptors());
+ response.getTable_info().getPartitions().get(0).getHdfs_storage_descriptor());
Assert.assertNotNull(
- response.getTable_info().getPartitions().get(0).getHms_partition());
+ response.getTable_info().getPartitions().get(0).getHms_parameters());
// skipping request for file-metadata should not affect the result
request = new RequestBuilder()
@@ -277,7 +280,8 @@ public class PartialCatalogInfoWriteIdTest {
Assert.assertEquals(1, response.getTable_info().getPartitionsSize());
for (TPartialPartitionInfo partInfo : response.getTable_info().getPartitions()) {
Assert.assertNull(partInfo.getFile_descriptors());
- Assert.assertNull(partInfo.getHms_partition());
+ Assert.assertNull(partInfo.getHdfs_storage_descriptor());
+ Assert.assertNull(partInfo.getLocation());
}
// we request a newer WriteIdList now, and catalog needs to reload
@@ -289,10 +293,13 @@ public class PartialCatalogInfoWriteIdTest {
.build();
response = sendRequest(request);
Assert.assertEquals(2, response.getTable_info().getPartitionsSize());
+ Assert.assertEquals(1, response.getTable_info().getPartition_prefixesSize());
// we expect both the partitions to have the file-metadata in the response
for (TPartialPartitionInfo partInfo : response.getTable_info().getPartitions()) {
Assert.assertNotNull(partInfo.getFile_descriptors());
- Assert.assertNotNull(partInfo.getHms_partition());
+ Assert.assertNotNull(partInfo.getHdfs_storage_descriptor());
+ Assert.assertNotNull(partInfo.getLocation());
+ Assert.assertEquals(0, partInfo.getLocation().getPrefix_index());
}
// table must be reloaded now
long newerVersion = catalog_.getTable(testDbName, testPartitionedTbl)
diff --git a/fe/src/test/java/org/apache/impala/catalog/local/CatalogdMetaProviderTest.java b/fe/src/test/java/org/apache/impala/catalog/local/CatalogdMetaProviderTest.java
index b378ce0..20e5d32 100644
--- a/fe/src/test/java/org/apache/impala/catalog/local/CatalogdMetaProviderTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/local/CatalogdMetaProviderTest.java
@@ -177,10 +177,12 @@ public class CatalogdMetaProviderTest {
// Unfortunately Guava doesn't provide a statistic on the total weight of cached
// elements. So, we'll just instantiate the weigher directly and sanity check
- // the size loosely.
+ // the size loosely. The size will grow if we add more fields into
+ // TPartialPartitionInfo in future.
SizeOfWeigher weigher = new SizeOfWeigher();
- assertTrue(weigher.weigh(refs, null) > 3000);
- assertTrue(weigher.weigh(refs, null) < 4000);
+ int weigh = weigher.weigh(refs, null);
+ assertTrue("Actual weigh: " + weigh, weigh > 4000);
+ assertTrue("Actual weigh: " + weigh, weigh < 5000);
}
@Test