You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by mi...@apache.org on 2026/06/08 20:54:32 UTC
(impala) branch master updated: IMPALA-14564: Remove redundant partition info from Iceberg file descriptors
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits 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 f0f4fec03 IMPALA-14564: Remove redundant partition info from Iceberg file descriptors
f0f4fec03 is described below
commit f0f4fec03717c03cde782155641e46b9f18620b2
Author: Noemi Pap-Takacs <np...@cloudera.com>
AuthorDate: Fri Dec 5 14:46:06 2025 +0100
IMPALA-14564: Remove redundant partition info from Iceberg file descriptors
Iceberg file descriptors used to contain information about the
partition they belong to: the spec id and the partition values.
These fields uniquely identify the partition the file belongs
to and are only dependent on the partition not the file itself.
It means that it is redundant to store these fields in each
file descriptor in the Catalog.
This change normalizes partition-related metadata out of the
file descriptors in the Catalog and the Frontend.
The partition information is stored separately in the
IcebergContentFileStore in FlatBuffers binary format. File
descriptors only store a unique id of the partition they belong to.
Since the scan nodes need the partition information for execution,
we denormalize the file and partition metadata: we look up the
relevant partitions by their ids in the Planner, put them in the
file descriptors that are sent to the executors.
Metadata handling during scheduling and execution remains the same.
This change introduces a separate FlatBuffer schema to serialize
necessary Iceberg file metadata from the Catalog to the Frontend
and from the Frontend to the executors (in scan ranges) containing
only the relevant fields for maximum memory efficiency.
Testing:
- ran existing e2e tests
- extended FileMetadataLoaderTest to check the partitions
Assisted by Copilot (Claude Sonnet 4.6)
Change-Id: I57c2fd6f1ebb636aa9e7ca925413ca51858cbc2a
Reviewed-on: http://gerrit.cloudera.org:8080/23838
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/file-metadata-utils.cc | 14 +--
be/src/exec/file-metadata-utils.h | 8 +-
be/src/exec/hdfs-scan-node-base.cc | 12 ++-
be/src/exec/hdfs-scan-node-base.h | 4 +-
common/fbs/CatalogObjects.fbs | 8 +-
common/fbs/IcebergObjects.fbs | 20 +++-
common/thrift/CatalogObjects.thrift | 11 +--
.../impala/catalog/IcebergContentFileStore.java | 104 +++++++++++++-------
.../apache/impala/catalog/IcebergDeleteTable.java | 2 +-
.../impala/catalog/IcebergFileDescriptor.java | 15 +++
.../impala/catalog/IcebergFileMetadataLoader.java | 68 +++++++------
.../org/apache/impala/planner/HdfsScanNode.java | 5 +
.../org/apache/impala/planner/IcebergScanNode.java | 16 +++-
.../apache/impala/planner/IcebergScanPlanner.java | 8 +-
.../java/org/apache/impala/util/IcebergUtil.java | 106 +++++++++++++++------
.../impala/catalog/FileMetadataLoaderTest.java | 45 +++++----
16 files changed, 301 insertions(+), 145 deletions(-)
diff --git a/be/src/exec/file-metadata-utils.cc b/be/src/exec/file-metadata-utils.cc
index 9dff7570e..45df6db4e 100644
--- a/be/src/exec/file-metadata-utils.cc
+++ b/be/src/exec/file-metadata-utils.cc
@@ -115,7 +115,7 @@ void FileMetadataUtils::AddFileLevelVirtualColumns(MemPool* mem_pool,
} else if (slot_desc->virtual_column_type() ==
TVirtualColumnType::ICEBERG_DATA_SEQUENCE_NUMBER) {
using namespace org::apache::impala::fb;
- const FbIcebergMetadata* ice_metadata =
+ const FbIcebergSplitMetadata* ice_metadata =
file_desc_->file_metadata->iceberg_metadata();
DCHECK(ice_metadata != nullptr);
@@ -130,7 +130,7 @@ void FileMetadataUtils::AddFileLevelVirtualColumns(MemPool* mem_pool,
} else if (slot_desc->virtual_column_type() ==
TVirtualColumnType::ICEBERG_FIRST_ROW_ID) {
using namespace org::apache::impala::fb;
- const FbIcebergMetadata* ice_metadata =
+ const FbIcebergSplitMetadata* ice_metadata =
file_desc_->file_metadata->iceberg_metadata();
DCHECK(ice_metadata != nullptr);
@@ -151,9 +151,9 @@ auto FileMetadataUtils::IcebergPartitionTransforms() const {
DCHECK(scan_node_->hdfs_table()->IsIcebergTable());
using namespace org::apache::impala::fb;
- const FbFileMetadata* file_metadata = file_desc_->file_metadata;
+ const FbSplitFileMetadata* file_metadata = file_desc_->file_metadata;
DCHECK(file_metadata != nullptr);
- const FbIcebergMetadata* ice_metadata = file_metadata->iceberg_metadata();
+ const FbIcebergSplitMetadata* ice_metadata = file_metadata->iceberg_metadata();
DCHECK(ice_metadata != nullptr);
return ice_metadata->partition_keys();
}
@@ -174,8 +174,8 @@ void FileMetadataUtils::AddIcebergColumns(MemPool* mem_pool, Tuple** template_tu
std::map<const SlotId, const SlotDescriptor*>* slot_descs_written) {
using namespace org::apache::impala::fb;
TextConverter text_converter = CreateTextConverter();
- const FbFileMetadata* file_metadata = file_desc_->file_metadata;
- const FbIcebergMetadata* ice_metadata = file_metadata->iceberg_metadata();
+ const FbSplitFileMetadata* file_metadata = file_desc_->file_metadata;
+ const FbIcebergSplitMetadata* ice_metadata = file_metadata->iceberg_metadata();
auto transforms = ice_metadata->partition_keys();
const TupleDescriptor* tuple_desc = scan_node_->tuple_desc();
@@ -238,7 +238,7 @@ void FileMetadataUtils::AddIcebergColumns(MemPool* mem_pool, Tuple** template_tu
}
void FileMetadataUtils::AddVirtualIcebergColumn(MemPool* mem_pool, Tuple* template_tuple,
- const org::apache::impala::fb::FbIcebergMetadata& ice_metadata,
+ const org::apache::impala::fb::FbIcebergSplitMetadata& ice_metadata,
const SlotDescriptor* slot_desc) {
DCHECK(slot_desc->IsVirtual());
if (slot_desc->virtual_column_type() == TVirtualColumnType::PARTITION_SPEC_ID) {
diff --git a/be/src/exec/file-metadata-utils.h b/be/src/exec/file-metadata-utils.h
index 41c5b235d..e93a8e5a3 100644
--- a/be/src/exec/file-metadata-utils.h
+++ b/be/src/exec/file-metadata-utils.h
@@ -25,7 +25,7 @@
#include <memory>
namespace org::apache::impala::fb {
- struct FbIcebergMetadata;
+ struct FbIcebergSplitMetadata;
}
namespace impala {
@@ -91,15 +91,15 @@ public:
private:
void AddFileLevelVirtualColumns(MemPool* mem_pool, Tuple* template_tuple);
- /// Writes the Iceberg columns from FbIcebergMetadata into template_tuple. Updates the
- /// slot_descs_written map with the SlotDescriptors that were written into the
+ /// Writes the Iceberg columns from FbIcebergSplitMetadata into template_tuple. Updates
+ /// the slot_descs_written map with the SlotDescriptors that were written into the
/// template_tuple.
void AddIcebergColumns(MemPool* mem_pool, Tuple** template_tuple,
std::map<const SlotId, const SlotDescriptor*>* slot_descs_written);
/// Writes Iceberg-related virtual column values to the template tuple.
void AddVirtualIcebergColumn(MemPool* mem_pool, Tuple* template_tuple,
- const org::apache::impala::fb::FbIcebergMetadata& ice_metadata,
+ const org::apache::impala::fb::FbIcebergSplitMetadata& ice_metadata,
const SlotDescriptor* slot_desc);
/// Returns a pointer to the container of partition transforms. Only valid to call
diff --git a/be/src/exec/hdfs-scan-node-base.cc b/be/src/exec/hdfs-scan-node-base.cc
index bb21561fa..c0599de0c 100644
--- a/be/src/exec/hdfs-scan-node-base.cc
+++ b/be/src/exec/hdfs-scan-node-base.cc
@@ -264,10 +264,11 @@ Status HdfsScanPlanNode::ProcessScanRangesAndInitSharedState(FragmentState* stat
for (const ScanRangeParamsPB& params : ranges->second.scan_ranges()) {
DCHECK(params.scan_range().has_hdfs_file_split());
const HdfsFileSplitPB& split = params.scan_range().hdfs_file_split();
- const org::apache::impala::fb::FbFileMetadata* file_metadata = nullptr;
+ const org::apache::impala::fb::FbSplitFileMetadata* file_metadata = nullptr;
if (params.scan_range().has_file_metadata()) {
- file_metadata = flatbuffers::GetRoot<org::apache::impala::fb::FbFileMetadata>(
- params.scan_range().file_metadata().c_str());
+ file_metadata =
+ flatbuffers::GetRoot<org::apache::impala::fb::FbSplitFileMetadata>(
+ params.scan_range().file_metadata().c_str());
}
HdfsPartitionDescriptor* partition_desc =
hdfs_table_->GetPartition(split.partition_id());
@@ -911,10 +912,11 @@ Status HdfsScanNodeBase::CreateAndOpenScannerHelper(HdfsPartitionDescriptor* par
DCHECK(context != nullptr);
DCHECK(scanner->get() == nullptr);
- const FbFileMetadata* file_metadata = context->GetStream(0)->file_desc()->file_metadata;
+ const FbSplitFileMetadata* file_metadata =
+ context->GetStream(0)->file_desc()->file_metadata;
if (file_metadata) {
// Iceberg tables can have different file format for each data file:
- const FbIcebergMetadata* ice_metadata = file_metadata->iceberg_metadata();
+ const FbIcebergSplitMetadata* ice_metadata = file_metadata->iceberg_metadata();
DCHECK(ice_metadata != nullptr);
switch (ice_metadata->file_format()) {
case FbIcebergDataFileFormat::FbIcebergDataFileFormat_PARQUET:
diff --git a/be/src/exec/hdfs-scan-node-base.h b/be/src/exec/hdfs-scan-node-base.h
index ed4c6948b..4b52af048 100644
--- a/be/src/exec/hdfs-scan-node-base.h
+++ b/be/src/exec/hdfs-scan-node-base.h
@@ -44,7 +44,7 @@
#include "util/unique-id-hash.h"
namespace org { namespace apache { namespace impala { namespace fb {
-struct FbFileMetadata;
+struct FbSplitFileMetadata;
}}}}
namespace impala {
@@ -95,7 +95,7 @@ struct HdfsFileDesc {
std::vector<io::ScanRange*> splits;
/// Extra file metadata, e.g. Iceberg-related file-level info.
- const ::org::apache::impala::fb::FbFileMetadata* file_metadata;
+ const ::org::apache::impala::fb::FbSplitFileMetadata* file_metadata;
/// Whether file is encrypted.
bool is_encrypted = false;
diff --git a/common/fbs/CatalogObjects.fbs b/common/fbs/CatalogObjects.fbs
index f0dc08824..e2a35a091 100644
--- a/common/fbs/CatalogObjects.fbs
+++ b/common/fbs/CatalogObjects.fbs
@@ -90,6 +90,12 @@ table FbFileDesc {
// Additional file-related metadata
table FbFileMetadata {
- // Iceberg-related metadata about the data file
+ // Iceberg-related metadata about the data file without partition information
iceberg_metadata : FbIcebergMetadata;
}
+
+// Additional file-related metadata for scan ranges
+table FbSplitFileMetadata {
+ // Iceberg-related metadata about the data file with partition information
+ iceberg_metadata : FbIcebergSplitMetadata;
+}
\ No newline at end of file
diff --git a/common/fbs/IcebergObjects.fbs b/common/fbs/IcebergObjects.fbs
index 0c067738a..e8e51e78c 100644
--- a/common/fbs/IcebergObjects.fbs
+++ b/common/fbs/IcebergObjects.fbs
@@ -42,17 +42,32 @@ table FbIcebergPartitionTransformValue {
source_id: int;
}
+table FbIcebergPartition {
+ spec_id : ushort;
+ partition_keys : [FbIcebergPartitionTransformValue];
+}
+
+// From Catalog to Coordinator
+// first_row_id cannot have -1 as default value here because it is not safe to mutate.
table FbIcebergMetadata {
file_format : FbIcebergDataFileFormat;
record_count : long;
data_sequence_number : long;
- spec_id : ushort;
- partition_keys : [FbIcebergPartitionTransformValue];
equality_field_ids : [int];
part_id : int = -1;
first_row_id : long;
}
+// From FE to BE. Included in ScanRanges.
+// For V1 and V2 tables, the first_row_id is -1. Setting it as default value saves space.
+table FbIcebergSplitMetadata {
+ file_format : FbIcebergDataFileFormat;
+ data_sequence_number : long;
+ spec_id : ushort;
+ partition_keys : [FbIcebergPartitionTransformValue];
+ first_row_id : long = -1;
+}
+
table FbIcebergColumnStats {
field_id: int;
total_compressed_byte_size: long;
@@ -66,6 +81,7 @@ table FbIcebergPartitionField {
field_value: [ubyte];
}
+// From Coordinator to Catalog. Used when committing Iceberg DML operations.
table FbIcebergDataFile {
path: string;
format: FbIcebergDataFileFormat = PARQUET;
diff --git a/common/thrift/CatalogObjects.thrift b/common/thrift/CatalogObjects.thrift
index 6471323c6..aba4695b9 100644
--- a/common/thrift/CatalogObjects.thrift
+++ b/common/thrift/CatalogObjects.thrift
@@ -357,7 +357,7 @@ struct THdfsFileDesc {
// TODO: Put this in a KRPC sidecar to avoid serialization cost.
1: required binary file_desc_data
- // Additional file metadata serialized into a FlatBuffer
+ // Additional file metadata serialized into a FlatBuffer (FbSplitFileMetadata)
// TODO: Put this in a KRPC sidecar to avoid serialization cost.
2: optional binary file_metadata
}
@@ -655,11 +655,6 @@ struct TIcebergPartitionSpec {
2: optional list<TIcebergPartitionField> partition_fields
}
-struct TIcebergPartition {
- 1: required i32 spec_id
- 2: required list<string> partition_values
-}
-
struct TIcebergPartitionStats {
1: required i64 num_files;
2: required i64 num_rows;
@@ -683,7 +678,9 @@ struct TIcebergContentFileStore {
6: optional bool has_orc
7: optional bool has_parquet
8: optional list<string> missing_files
- 9: optional list<TIcebergPartition> partitions
+ // Partition metadata serialized into a FlatBuffer
+ // (FbIcebergPartition defined in common/fbs/IcebergObjects.fbs).
+ 9: optional list<binary> partitions
10: optional map<THash128, Types.TIcebergDeletionVector> data_path_hash_to_dv
// Total number of files in this content file store. Only set in the first partial
// response (offset 0) so the coordinator knows upfront how many pages to expect.
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergContentFileStore.java b/fe/src/main/java/org/apache/impala/catalog/IcebergContentFileStore.java
index 063774b3a..6f84bdcc0 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergContentFileStore.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergContentFileStore.java
@@ -17,6 +17,7 @@
package org.apache.impala.catalog;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
@@ -34,7 +35,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.lang.ref.SoftReference;
import java.util.stream.Collectors;
-
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.ContentFile;
import org.apache.iceberg.DataFile;
@@ -50,7 +50,6 @@ import org.apache.impala.thrift.THash128;
import org.apache.impala.thrift.THdfsFileDesc;
import org.apache.impala.thrift.TIcebergContentFileStore;
import org.apache.impala.thrift.TIcebergDeletionVector;
-import org.apache.impala.thrift.TIcebergPartition;
import org.apache.impala.thrift.TNetworkAddress;
import org.apache.impala.util.Hash128;
import org.apache.impala.util.IcebergUtil;
@@ -219,7 +218,10 @@ public class IcebergContentFileStore {
private Set<String> missingFiles_ = new HashSet<>();
// Partitions with their corresponding ids that are used to refer to the partition info
// from the IcebergFileDescriptors.
- private Map<TIcebergPartition, Integer> partitions_;
+ private Map<ByteBuffer, Integer> partitionMap_;
+ // Current partitions in the order of their corresponding ids. Initialized and used to
+ // look up partitions only on the Frontend.
+ private List<ByteBuffer> partitionList_ = null;
// Flags to indicate file formats used in the table.
private boolean hasAvro_ = false;
@@ -236,8 +238,13 @@ public class IcebergContentFileStore {
private final ConcurrentMap<Hash128, EncodedFileDescriptor> oldFileDescMap_ =
new ConcurrentHashMap<>();
// Caches the partitions of file descriptors loaded during time-travel queries.
- private final ConcurrentMap<TIcebergPartition, Integer> oldPartitionMap_ =
+ private final ConcurrentMap<ByteBuffer, Integer> oldPartitionMap_ =
new ConcurrentHashMap<>();
+ // Old partitions in order of their corresponding ids offseted by the number of current
+ // partitions, e.g. first old partition in the list has the following id:
+ // [number_of_current_partitions + 0] where 0 is the index.
+ // Initialized and used to look up old partitions only on the Frontend.
+ private List<ByteBuffer> oldPartitionList_ = null;
// END: mutable members
/////////////////////////////////////////
@@ -246,13 +253,13 @@ public class IcebergContentFileStore {
public IcebergContentFileStore(
Table iceApiTable, List<IcebergFileDescriptor> fileDescriptors,
- GroupedContentFiles icebergFiles, Map<TIcebergPartition, Integer> partitions) {
+ GroupedContentFiles icebergFiles, Map<ByteBuffer, Integer> partitions) {
Preconditions.checkNotNull(iceApiTable);
Preconditions.checkNotNull(fileDescriptors);
Preconditions.checkNotNull(icebergFiles);
Preconditions.checkNotNull(partitions);
- partitions_ = partitions;
+ partitionMap_ = partitions;
Map<String, IcebergFileDescriptor> fileDescMap = new HashMap<>();
String apiTableLocation = iceApiTable.location();
@@ -291,7 +298,8 @@ public class IcebergContentFileStore {
ret.equalityDeleteFiles_ = equalityDeleteFiles_;
ret.dataFileToDV_ = dataFileToDV_;
ret.missingFiles_ = missingFiles_;
- ret.partitions_ = partitions_;
+ ret.partitionMap_ = partitionMap_;
+ ret.partitionList_ = partitionList_;
ret.hasAvro_ = hasAvro_;
ret.hasOrc_ = hasOrc_;
ret.hasParquet_ = hasParquet_;
@@ -312,7 +320,7 @@ public class IcebergContentFileStore {
public void mergeWithNewFiles(
GroupedContentFiles newIcebergFiles,
List<IcebergFileDescriptor> newFds,
- Map<TIcebergPartition, Integer> mergedPartitions,
+ Map<ByteBuffer, Integer> mergedPartitions,
String apiTableLocation) {
Preconditions.checkState(newIcebergFiles.dataFilesWithDeletes.isEmpty()
&& newIcebergFiles.positionDeleteFiles.isEmpty()
@@ -330,7 +338,7 @@ public class IcebergContentFileStore {
for (DataFile dataFile : newIcebergFiles.dataFilesWithoutDeletes) {
storeFile(dataFile, newFdMap, dataFilesWithoutDeletes_);
}
- partitions_ = mergedPartitions;
+ partitionMap_ = mergedPartitions;
}
private void storeFile(ContentFile<?> contentFile,
@@ -352,7 +360,8 @@ public class IcebergContentFileStore {
// This is only invoked during time travel, when we are querying a snapshot that has
// partitions which have been removed since.
- public void addOldPartition(TIcebergPartition partition, Integer id) {
+ public void addOldPartition(ByteBuffer partition, Integer id) {
+ Preconditions.checkState(partition.position() == 0);
oldPartitionMap_.put(partition, id);
}
@@ -373,7 +382,7 @@ public class IcebergContentFileStore {
return decode(oldFileDescMap_.get(pathHash));
}
- public Integer getOldPartition(TIcebergPartition partition) {
+ public Integer getOldPartition(ByteBuffer partition) {
return oldPartitionMap_.get(partition);
}
@@ -381,6 +390,40 @@ public class IcebergContentFileStore {
return oldPartitionMap_.size();
}
+ public int getNumPartitions() {
+ return partitionMap_.size();
+ }
+
+ public Map<ByteBuffer, Integer> getPartitionMap() {
+ return partitionMap_;
+ }
+
+ public List<ByteBuffer> getPartitionList() {
+ if (partitionList_ == null || partitionList_.size() < partitionMap_.size()) {
+ partitionList_ = convertPartitionMapToList(partitionMap_, 0);
+ }
+ return partitionList_;
+ }
+
+ public synchronized List<ByteBuffer> getOldPartitionList() {
+ //check size for freshness
+ if (oldPartitionList_ == null
+ || oldPartitionList_.size() != oldPartitionMap_.size()) {
+ // Old partition ids are offseted by the number of current partitions.
+ oldPartitionList_ = convertPartitionMapToList(oldPartitionMap_, getNumPartitions());
+ }
+ return oldPartitionList_;
+ }
+
+ public ByteBuffer getPartitionById(int id) {
+ // The old partitions are only loaded in the Coordinator in case of time travel.
+ // Their ids are greater than (offseted by) the ids of the current partitions.
+ if (id >= getNumPartitions()) {
+ return getOldPartitionList().get(id - getNumPartitions());
+ }
+ return getPartitionList().get(id);
+ }
+
public List<IcebergFileDescriptor> getDataFilesWithoutDeletes() {
return dataFilesWithoutDeletes_.getList();
}
@@ -442,18 +485,6 @@ public class IcebergContentFileStore {
equalityDeleteFiles_.getList());
}
- public Map<TIcebergPartition, Integer> getPartitionMap() {
- return partitions_;
- }
-
- public List<TIcebergPartition> getPartitionList() {
- return convertPartitionMapToList(partitions_);
- }
-
- public int getNumPartitions() {
- return partitions_.size();
- }
-
public boolean hasAvro() { return hasAvro_; }
public boolean hasOrc() { return hasOrc_; }
public boolean hasParquet() { return hasParquet_; }
@@ -533,7 +564,7 @@ public class IcebergContentFileStore {
ret.setHas_orc(hasOrc_);
ret.setHas_parquet(hasParquet_);
ret.setMissing_files(new ArrayList<>(missingFiles_));
- ret.setPartitions(convertPartitionMapToList(partitions_));
+ ret.setPartitions(convertPartitionMapToList(partitionMap_, 0));
Map<THash128, TIcebergDeletionVector> tdeletion_vectors = new HashMap<>();
for (Map.Entry<Hash128, TIcebergDeletionVector> entry : dataFileToDV_.entrySet()) {
tdeletion_vectors.put(entry.getKey().toThrift(), entry.getValue());
@@ -600,7 +631,7 @@ public class IcebergContentFileStore {
ret.setHas_orc(hasOrc_);
ret.setHas_parquet(hasParquet_);
ret.setMissing_files(new ArrayList<>(missingFiles_));
- ret.setPartitions(convertPartitionMapToList(partitions_));
+ ret.setPartitions(convertPartitionMapToList(partitionMap_, 0));
Map<THash128, TIcebergDeletionVector> tdeletion_vectors = new HashMap<>();
for (Map.Entry<Hash128, TIcebergDeletionVector> entry : dataFileToDV_.entrySet()) {
tdeletion_vectors.put(entry.getKey().toThrift(), entry.getValue());
@@ -646,29 +677,32 @@ public class IcebergContentFileStore {
ret.hasParquet_ = tFileStore.isSetHas_parquet() ? tFileStore.isHas_parquet() : false;
ret.missingFiles_ = tFileStore.isSetMissing_files() ?
new HashSet<>(tFileStore.getMissing_files()) : Collections.emptySet();
- ret.partitions_ = tFileStore.isSetPartitions() ?
+ ret.partitionList_ = tFileStore.isSetPartitions() ?
+ tFileStore.getPartitions() : new ArrayList<>();
+ ret.partitionMap_ = tFileStore.isSetPartitions() ?
convertPartitionListToMap(tFileStore.getPartitions()) : new HashMap<>();
return ret;
}
- static List<TIcebergPartition> convertPartitionMapToList(
- Map<TIcebergPartition, Integer> partitionMap) {
- List<TIcebergPartition> partitionList = new ArrayList<>(partitionMap.size());
+ @VisibleForTesting
+ static List<ByteBuffer> convertPartitionMapToList(
+ Map<ByteBuffer, Integer> partitionMap, int offset) {
+ List<ByteBuffer> partitionList = new ArrayList<>(partitionMap.size());
for (int i = 0; i < partitionMap.size(); i++) {
partitionList.add(null);
}
- for (Map.Entry<TIcebergPartition, Integer> partition : partitionMap.entrySet()) {
- partitionList.set(partition.getValue(), partition.getKey());
+ for (Map.Entry<ByteBuffer, Integer> partition : partitionMap.entrySet()) {
+ partitionList.set(partition.getValue() - offset, partition.getKey());
}
return partitionList;
}
- private static ImmutableMap<TIcebergPartition, Integer> convertPartitionListToMap(
- List<TIcebergPartition> partitionList) {
+ private static ImmutableMap<ByteBuffer, Integer> convertPartitionListToMap(
+ List<ByteBuffer> partitionList) {
Preconditions.checkState(partitionList != null);
- ImmutableMap.Builder<TIcebergPartition, Integer> builder = ImmutableMap.builder();
+ ImmutableMap.Builder<ByteBuffer, Integer> builder = ImmutableMap.builder();
for (int i = 0; i < partitionList.size(); ++i) {
- builder.put(partitionList.get(i), i);
+ builder.put(org.apache.thrift.TBaseHelper.copyBinary(partitionList.get(i)), i);
}
return builder.build();
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergDeleteTable.java b/fe/src/main/java/org/apache/impala/catalog/IcebergDeleteTable.java
index 38b5f4fd5..e7a6e76ae 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergDeleteTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergDeleteTable.java
@@ -105,7 +105,7 @@ public abstract class IcebergDeleteTable extends VirtualTable implements FeIcebe
@Override
public IcebergContentFileStore getContentFileStore() {
- throw new NotImplementedException("This should never be called.");
+ return baseTable_.getContentFileStore();
}
@Override
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergFileDescriptor.java b/fe/src/main/java/org/apache/impala/catalog/IcebergFileDescriptor.java
index 652a492d1..20796907b 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergFileDescriptor.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergFileDescriptor.java
@@ -63,6 +63,21 @@ public class IcebergFileDescriptor extends FileDescriptor {
return new IcebergFileDescriptor(fd.getFbFileDescriptor(), fileMetadata);
}
+ public IcebergFileDescriptor deepCopy() {
+ ByteBuffer newDescBuf = deepCopyByteBuffer(getFbFileDescriptor().getByteBuffer());
+ ByteBuffer newMetadataBuf = deepCopyByteBuffer(fbFileMetadata_.getByteBuffer());
+
+ return new IcebergFileDescriptor(
+ FbFileDesc.getRootAsFbFileDesc(newDescBuf),
+ FbFileMetadata.getRootAsFbFileMetadata(newMetadataBuf));
+ }
+
+ private static ByteBuffer deepCopyByteBuffer(ByteBuffer original) {
+ ByteBuffer newBuf = ByteBuffer.allocate(original.remaining()).put(original).flip();
+ original.rewind();
+ return newBuf;
+ }
+
@Override
public IcebergFileDescriptor cloneWithNewHostIndex(
List<TNetworkAddress> origIndex, ListMap<TNetworkAddress> dstIndex) {
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java b/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
index 036e94232..e12af8d83 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
@@ -27,6 +27,7 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -53,7 +54,6 @@ import org.apache.impala.common.FileSystemUtil;
import org.apache.impala.common.PrintUtils;
import org.apache.impala.common.Pair;
import org.apache.impala.fb.FbIcebergMetadata;
-import org.apache.impala.thrift.TIcebergPartition;
import org.apache.impala.thrift.TNetworkAddress;
import org.apache.impala.util.IcebergUtil;
import org.apache.impala.util.ListMap;
@@ -75,15 +75,16 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
private final Path tablePath_;
private final String tablePathStr_;
private final GroupedContentFiles icebergFiles_;
- private final List<TIcebergPartition> oldIcebergPartitions_;
+ // Partitions that were already loaded before the incremental refresh.
+ private final List<ByteBuffer> oldIcebergPartitions_;
private AtomicInteger nextPartitionId_ = new AtomicInteger(0);
// Map of the freshly loaded Iceberg partitions and their corresponding ids.
- private ConcurrentHashMap<TIcebergPartition, Integer> loadedIcebergPartitions_;
+ private ConcurrentHashMap<ByteBuffer, Integer> loadedIcebergPartitions_;
private final boolean requiresDataFilesInTableLocation_;
// When non-null, these partitions are pre-loaded into loadedIcebergPartitions_ at the
// start of loadInternal(). Used for incremental file loading to assign new partition
// IDs that do not conflict with already-existing ones.
- private final Map<TIcebergPartition, Integer> initialPartitions_;
+ private final Map<ByteBuffer, Integer> initialPartitions_;
/**
* 'existingPartitions', when non-null, is pre-loaded into the partition map so that
@@ -92,9 +93,9 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
*/
public IcebergFileMetadataLoader(org.apache.iceberg.Table iceTbl,
Iterable<IcebergFileDescriptor> oldFds, ListMap<TNetworkAddress> hostIndex,
- GroupedContentFiles icebergFiles, List<TIcebergPartition> partitions,
+ GroupedContentFiles icebergFiles, List<ByteBuffer> partitions,
boolean requiresDataFilesInTableLocation,
- Map<TIcebergPartition, Integer> existingPartitions) {
+ Map<ByteBuffer, Integer> existingPartitions) {
super(iceTbl.location(), true, oldFds, hostIndex, null, null,
HdfsFileFormat.ICEBERG);
iceTbl_ = iceTbl;
@@ -125,15 +126,11 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
.collect(Collectors.toList());
}
- public Map<TIcebergPartition, Integer> getIcebergPartitions() {
+ public Map<ByteBuffer, Integer> getIcebergPartitions() {
Preconditions.checkNotNull(loadedIcebergPartitions_);
return Collections.unmodifiableMap(loadedIcebergPartitions_);
}
- public List<TIcebergPartition> getIcebergPartitionList() {
- return IcebergContentFileStore.convertPartitionMapToList(getIcebergPartitions());
- }
-
private void loadInternal() throws CatalogException, IOException {
loadedFds_ = new ArrayList<>();
loadedIcebergPartitions_ = new ConcurrentHashMap<>();
@@ -201,26 +198,39 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
}
List<ContentFile<?>> newContentFiles = Lists.newArrayList();
for (ContentFile<?> contentFile : icebergFiles_.getAllContentFiles()) {
- IcebergFileDescriptor fd = getOldFd(contentFile);
- if (fd == null) {
+ IcebergFileDescriptor originalFd = getOldFd(contentFile);
+ if (originalFd == null) {
newContentFiles.add(contentFile);
} else {
- int oldPartId = fd.getFbFileMetadata().icebergMetadata().partId();
- TIcebergPartition partition = oldIcebergPartitions_.get(oldPartId);
+ FbIcebergMetadata originalIceMetadata =
+ originalFd.getFbFileMetadata().icebergMetadata();
+ // Look up the partition info in this old file descriptor from the partition list.
+ // Put the partition info in the new partitions map and get the new id.
+ int oldPartId = originalIceMetadata.partId();
+ ByteBuffer partition = oldIcebergPartitions_.get(oldPartId);
Integer newPartId = loadedIcebergPartitions_.computeIfAbsent(
partition, k -> nextPartitionId_.getAndIncrement());
- FbIcebergMetadata icebergMetadata = fd.getFbFileMetadata().icebergMetadata();
- // Look up the partition info in this old file descriptor from the partition list.
- // Put the partition info in the new partitions map and write the new partition id
- // to the file metadata of the fd.
- if (!icebergMetadata.mutatePartId(newPartId)) {
- throw new TableLoadingException(
- "Error modifying the partition id of the Iceberg file descriptor.");
- }
- if (icebergMetadata.firstRowId() == -1 && contentFile.firstRowId() != null) {
- if (!icebergMetadata.mutateFirstRowId(contentFile.firstRowId())) {
- throw new TableLoadingException(
- "Error modifying the first-row-id Iceberg file descriptor.");
+
+ IcebergFileDescriptor fd = originalFd;
+ boolean needToUpdatePartId = !newPartId.equals(oldPartId);
+ boolean needToUpdateFirstRowId =
+ originalIceMetadata.firstRowId() == -1 && contentFile.firstRowId() != null;
+ if (needToUpdatePartId || needToUpdateFirstRowId) {
+ // If we need to modify the metadata, first create a deep copy of the
+ // file descriptor for cache consistency.
+ fd = originalFd.deepCopy();
+ FbIcebergMetadata icebergMetadata = fd.getFbFileMetadata().icebergMetadata();
+ if (needToUpdatePartId) {
+ if (!icebergMetadata.mutatePartId(newPartId)) {
+ throw new TableLoadingException(
+ "Error modifying the partition id of the Iceberg file descriptor.");
+ }
+ }
+ if (needToUpdateFirstRowId) {
+ if (!icebergMetadata.mutateFirstRowId(contentFile.firstRowId())) {
+ throw new TableLoadingException(
+ "Error modifying the first-row-id Iceberg file descriptor.");
+ }
}
}
++loadStats_.skippedFiles;
@@ -265,8 +275,8 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
}
private int addPartitionInfo(ContentFile<?> contentFile) {
- TIcebergPartition partition =
- IcebergUtil.createIcebergPartitionInfo(iceTbl_, contentFile);
+ ByteBuffer partition =
+ IcebergUtil.createFbIcebergPartition(iceTbl_, contentFile);
return loadedIcebergPartitions_.computeIfAbsent(
partition, k -> nextPartitionId_.getAndIncrement());
}
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 c56042599..bb7668c46 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
@@ -18,6 +18,7 @@
package org.apache.impala.planner;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
@@ -55,6 +56,7 @@ import org.apache.impala.catalog.Column;
import org.apache.impala.catalog.ColumnStats;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
+import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.FileBlock;
import org.apache.impala.catalog.FileDescriptor;
@@ -76,6 +78,8 @@ import org.apache.impala.common.PrintUtils;
import org.apache.impala.common.RuntimeEnv;
import org.apache.impala.common.ThriftSerializationCtx;
import org.apache.impala.fb.FbFileBlock;
+import org.apache.impala.fb.FbFileMetadata;
+import org.apache.impala.fb.FbSplitFileMetadata;
import org.apache.impala.planner.RuntimeFilterGenerator.RuntimeFilter;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.thrift.TExplainLevel;
@@ -101,6 +105,7 @@ import org.apache.impala.thrift.TTableStats;
import org.apache.impala.util.AcidUtils;
import org.apache.impala.util.BitUtil;
import org.apache.impala.util.ExecutorMembershipSnapshot;
+import org.apache.impala.util.IcebergUtil;
import org.apache.impala.util.MathUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java b/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
index 9e2b18769..7d117d250 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
@@ -26,6 +26,7 @@ import java.util.BitSet;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
+import java.nio.ByteBuffer;
import org.apache.iceberg.Snapshot;
import org.apache.impala.analysis.Analyzer;
@@ -317,13 +318,17 @@ public class IcebergScanNode extends HdfsScanNode {
protected void decorateScanRange(FileDescriptor fileDesc, TScanRange scanRange) {
IcebergFileDescriptor iceFd = applyDeletionVector(fileDesc,
scanRange::setIceberg_deletion_vector);
- scanRange.setFile_metadata(iceFd.getFbFileMetadata().getByteBuffer());
+ scanRange.setFile_metadata(getIcebergSplitFileMetadata(iceFd));
}
@Override
protected void decorateSplitSpec(FileDescriptor fileDesc,
TFileSplitGeneratorSpec splitSpec) {
- applyDeletionVector(fileDesc, splitSpec::setIceberg_deletion_vector);
+ IcebergFileDescriptor iceFd = applyDeletionVector(fileDesc,
+ splitSpec::setIceberg_deletion_vector);
+ // Overwrite the raw FbFileMetadata (which only contains a part_id reference)
+ // with the denormalized FbSplitFileMetadata that includes the actual partition keys.
+ splitSpec.getFile_desc().setFile_metadata(getIcebergSplitFileMetadata(iceFd));
}
private IcebergFileDescriptor applyDeletionVector(FileDescriptor fileDesc,
@@ -345,4 +350,11 @@ public class IcebergScanNode extends HdfsScanNode {
}
return iceFd;
}
+
+ private ByteBuffer getIcebergSplitFileMetadata(IcebergFileDescriptor fileDesc) {
+ int partId = fileDesc.getFbFileMetadata().icebergMetadata().partId();
+ ByteBuffer partitionBuf =
+ ((FeIcebergTable) tbl_).getContentFileStore().getPartitionById(partId);
+ return IcebergUtil.transformFileMetadata(fileDesc, partitionBuf).getByteBuffer();
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java b/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
index 1c111c140..39921e3d9 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
@@ -23,6 +23,7 @@ import static org.apache.impala.util.IcebergUtil.getFilePathHash;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
@@ -95,7 +96,6 @@ import org.apache.impala.planner.JoinNode.DistributionMode;
import org.apache.impala.service.Frontend;
import org.apache.impala.thrift.TColumnStats;
import org.apache.impala.thrift.TIcebergDeletionVector;
-import org.apache.impala.thrift.TIcebergPartition;
import org.apache.impala.thrift.TIcebergPartitionTransformType;
import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.thrift.TVirtualColumnType;
@@ -954,9 +954,9 @@ public class IcebergScanPlanner {
Integer partitionId = 0;
- TIcebergPartition partition =
- IcebergUtil.createIcebergPartitionInfo(getIceTable().getIcebergApiTable(), cf);
- Map<TIcebergPartition, Integer> partitions = fileStore.getPartitionMap();
+ ByteBuffer partition =
+ IcebergUtil.createFbIcebergPartition(getIceTable().getIcebergApiTable(), cf);
+ Map<ByteBuffer, Integer> partitions = fileStore.getPartitionMap();
partitionId = partitions.get(partition);
// If we do not find the partition among the current partitions cached by catalog, try
// looking it up among the old descriptors' partitions.
diff --git a/fe/src/main/java/org/apache/impala/util/IcebergUtil.java b/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
index e54dcb6bb..bb222ea39 100644
--- a/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
+++ b/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
@@ -101,6 +101,7 @@ import org.apache.impala.catalog.Column;
import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.IcebergColumn;
+import org.apache.impala.catalog.IcebergFileDescriptor;
import org.apache.impala.catalog.IcebergTable;
import org.apache.impala.catalog.IcebergTableLoadingException;
import org.apache.impala.catalog.TableLoadingException;
@@ -115,11 +116,14 @@ import org.apache.impala.common.FileSystemUtil;
import org.apache.impala.common.ImpalaRuntimeException;
import org.apache.impala.common.Pair;
import org.apache.impala.fb.FbFileMetadata;
+import org.apache.impala.fb.FbSplitFileMetadata;
import org.apache.impala.fb.FbIcebergDataFile;
import org.apache.impala.fb.FbIcebergPartitionField;
import org.apache.impala.fb.FbIcebergDataFileFormat;
import org.apache.impala.fb.FbIcebergDeletionVector;
import org.apache.impala.fb.FbIcebergMetadata;
+import org.apache.impala.fb.FbIcebergSplitMetadata;
+import org.apache.impala.fb.FbIcebergPartition;
import org.apache.impala.fb.FbIcebergPartitionTransformValue;
import org.apache.impala.fb.FbIcebergTransformType;
import org.apache.impala.thrift.TCompressionCodec;
@@ -128,7 +132,6 @@ import org.apache.impala.thrift.THdfsFileFormat;
import org.apache.impala.thrift.TIcebergCatalog;
import org.apache.impala.thrift.TIcebergDeletionVector;
import org.apache.impala.thrift.TIcebergFileFormat;
-import org.apache.impala.thrift.TIcebergPartition;
import org.apache.impala.thrift.TIcebergPartitionField;
import org.apache.impala.thrift.TIcebergPartitionSpec;
import org.apache.impala.thrift.TIcebergPartitionTransformType;
@@ -1186,28 +1189,82 @@ public class IcebergUtil {
return ret;
}
- public static TIcebergPartition createIcebergPartitionInfo(
+ public static ByteBuffer createFbIcebergPartition(
Table iceApiTbl, ContentFile cf) {
- TIcebergPartition partInfo = new TIcebergPartition(cf.specId(), new ArrayList<>());
+ FlatBufferBuilder fbb = new FlatBufferBuilder();
PartitionSpec spec = iceApiTbl.specs().get(cf.specId());
- Preconditions.checkState(spec.fields().size() == cf.partition().size());
- List<String> partitionKeys = new ArrayList<>();
- for (int i = 0; i < spec.fields().size(); ++i) {
- Object partValue = cf.partition().get(i, Object.class);
- if (partValue != null) {
- if (partValue instanceof ByteBuffer) {
- String partValueString =
- StringUtils.fromByteBuffer((ByteBuffer) partValue, CHARSET_FOR_BINARY);
- partitionKeys.add(partValueString);
- } else {
- partitionKeys.add(partValue.toString());
- }
- } else {
- partitionKeys.add("NULL");
+ int partKeysOffset = -1;
+ if (spec != null && !spec.fields().isEmpty()) {
+ partKeysOffset = createPartitionKeys(fbb, spec, cf);
+ }
+ FbIcebergPartition.startFbIcebergPartition(fbb);
+ FbIcebergPartition.addSpecId(fbb, cf.specId());
+ if (partKeysOffset != -1) {
+ FbIcebergPartition.addPartitionKeys(fbb, partKeysOffset);
+ }
+ int iceOffset = FbIcebergPartition.endFbIcebergPartition(fbb);
+ fbb.finish(iceOffset);
+ ByteBuffer bb = fbb.dataBuffer().slice();
+ ByteBuffer compressedBb = ByteBuffer.allocate(bb.capacity());
+ compressedBb.put(bb);
+ return (ByteBuffer)compressedBb.flip();
+ }
+
+ /**
+ * Creates a FbSplitFileMetadata object that contains Iceberg file metadata in a
+ * denormalized way: it includes the partition metadata as well in the file descriptor.
+ */
+ public static FbSplitFileMetadata transformFileMetadata(IcebergFileDescriptor fileDesc,
+ ByteBuffer partitionBuf) {
+ FbIcebergPartition partition =
+ FbIcebergPartition.getRootAsFbIcebergPartition(partitionBuf.duplicate());
+ FbIcebergMetadata iceMetadata = fileDesc.getFbFileMetadata().icebergMetadata();
+ FlatBufferBuilder fbb = new FlatBufferBuilder();
+ int partitionKeysOffset = copyPartitionKeys(fbb, partition);
+
+ FbIcebergSplitMetadata.startFbIcebergSplitMetadata(fbb);
+ FbIcebergSplitMetadata.addFileFormat(fbb, iceMetadata.fileFormat());
+ FbIcebergSplitMetadata.addDataSequenceNumber(fbb, iceMetadata.dataSequenceNumber());
+ FbIcebergSplitMetadata.addFirstRowId(fbb, iceMetadata.firstRowId());
+ FbIcebergSplitMetadata.addSpecId(fbb, partition.specId());
+ if (partitionKeysOffset != -1) {
+ FbIcebergSplitMetadata.addPartitionKeys(fbb, partitionKeysOffset);
+ }
+ int iceOffset = FbIcebergSplitMetadata.endFbIcebergSplitMetadata(fbb);
+ fbb.finish(FbSplitFileMetadata.createFbSplitFileMetadata(fbb, iceOffset));
+ ByteBuffer bb = fbb.dataBuffer().slice();
+ ByteBuffer resultBb = ByteBuffer.allocate(bb.capacity());
+ resultBb.put(bb);
+ return FbSplitFileMetadata.getRootAsFbSplitFileMetadata((ByteBuffer)resultBb.flip());
+ }
+
+ private static int copyPartitionKeys(FlatBufferBuilder fbb,
+ FbIcebergPartition partition) {
+ if (partition.partitionKeysLength() == 0) {
+ return -1;
+ }
+ int[] partitionKeyOffsets = new int[partition.partitionKeysLength()];
+ for (int i = 0; i < partition.partitionKeysLength(); i++) {
+ FbIcebergPartitionTransformValue origKey = partition.partitionKeys(i);
+
+ int transformValueOffset = -1;
+ ByteBuffer buf = origKey.transformValueAsByteBuffer();
+ if (buf != null) {
+ byte[] bytes = new byte[buf.remaining()];
+ buf.get(bytes);
+ transformValueOffset = FbIcebergPartitionTransformValue
+ .createTransformValueVector(fbb, bytes);
}
+
+ partitionKeyOffsets[i] =
+ FbIcebergPartitionTransformValue.createFbIcebergPartitionTransformValue(
+ fbb,
+ origKey.transformType(),
+ origKey.transformParam(),
+ transformValueOffset,
+ origKey.sourceId());
}
- partInfo.setPartition_values(partitionKeys);
- return partInfo;
+ return FbIcebergSplitMetadata.createPartitionKeysVector(fbb, partitionKeyOffsets);
}
/**
@@ -1229,11 +1286,6 @@ public class IcebergUtil {
private static int createIcebergMetadata(Table iceApiTbl, FlatBufferBuilder fbb,
ContentFile cf, int partId) {
- int partKeysOffset = -1;
- PartitionSpec spec = iceApiTbl.specs().get(cf.specId());
- if (spec != null && !spec.fields().isEmpty()) {
- partKeysOffset = createPartitionKeys(fbb, spec, cf);
- }
int eqFieldIdsOffset = -1;
List<Integer> eqFieldIds = cf.equalityFieldIds();
if (eqFieldIds != null && !eqFieldIds.isEmpty()) {
@@ -1250,11 +1302,7 @@ public class IcebergUtil {
if (fileFormat != -1) {
FbIcebergMetadata.addFileFormat(fbb, fileFormat);
}
- FbIcebergMetadata.addSpecId(fbb, cf.specId());
FbIcebergMetadata.addRecordCount(fbb, cf.recordCount());
- if (partKeysOffset != -1) {
- FbIcebergMetadata.addPartitionKeys(fbb, partKeysOffset);
- }
if (cf.dataSequenceNumber() != null) {
FbIcebergMetadata.addDataSequenceNumber(fbb, cf.dataSequenceNumber());
} else {
@@ -1283,7 +1331,7 @@ public class IcebergUtil {
partitionKeyOffsets[i] =
createPartitionTransformValue(fbb, spec, cf, i);
}
- return FbIcebergMetadata.createPartitionKeysVector(fbb, partitionKeyOffsets);
+ return FbIcebergPartition.createPartitionKeysVector(fbb, partitionKeyOffsets);
}
private static int createPartitionTransformValue(
diff --git a/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java b/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
index c780c7a69..56043c962 100644
--- a/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
@@ -20,9 +20,11 @@ package org.apache.impala.catalog;
import static org.junit.Assert.*;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -35,7 +37,6 @@ import org.apache.impala.common.FileSystemUtil;
import org.apache.impala.compat.MetastoreShim;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.testutil.CatalogServiceTestCatalog;
-import org.apache.impala.thrift.TIcebergPartition;
import org.apache.impala.thrift.TNetworkAddress;
import org.apache.impala.util.IcebergUtil;
import org.apache.impala.util.ListMap;
@@ -193,7 +194,7 @@ public class FileMetadataLoaderTest {
CatalogServiceCatalog catalog = CatalogServiceTestCatalog.create();
IcebergFileMetadataLoader fml1 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
- /* oldFds = */ Collections.emptyList(), Collections.emptyList(),
+ /* oldFds = */ Collections.emptyList(), Collections.emptyMap(),
/* requiresDataFilesInTableLocation = */ true);
fml1.load();
assertEquals(20, fml1.getStats().loadedFiles);
@@ -208,7 +209,7 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml2 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
- /* oldFds = */ Collections.emptyList(), Collections.emptyList(),
+ /* oldFds = */ Collections.emptyList(), Collections.emptyMap(),
/* requiresDataFilesInTableLocation = */ true);
fml2.load();
assertEquals(20, fml2.getStats().loadedFiles);
@@ -233,7 +234,7 @@ public class FileMetadataLoaderTest {
CatalogServiceCatalog catalog = CatalogServiceTestCatalog.create();
IcebergFileMetadataLoader fml = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
- /* oldFds = */ Collections.emptyList(), Collections.emptyList(),
+ /* oldFds = */ Collections.emptyList(), Collections.emptyMap(),
/* requiresDataFilesInTableLocation = */ true);
fml.load();
List<IcebergFileDescriptor> fileDescs = fml.getLoadedIcebergFds();
@@ -251,18 +252,20 @@ public class FileMetadataLoaderTest {
CatalogServiceCatalog catalog = CatalogServiceTestCatalog.create();
IcebergFileMetadataLoader fml1 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
- /* oldFds = */ Collections.emptyList(), Collections.emptyList(),
+ /* oldFds = */ Collections.emptyList(), Collections.emptyMap(),
/* requiresDataFilesInTableLocation = */ true);
fml1.load();
+ assertEquals(3, fml1.getIcebergPartitions().size());
IcebergFileMetadataLoader fml1Refresh = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
- /* oldFds = */ fml1.getLoadedIcebergFds(), fml1.getIcebergPartitionList(),
+ /* oldFds = */ fml1.getLoadedIcebergFds(), fml1.getIcebergPartitions(),
/* requiresDataFilesInTableLocation = */ true);
fml1Refresh.load();
assertEquals(0, fml1Refresh.getStats().loadedFiles);
assertEquals(20, fml1Refresh.getStats().skippedFiles);
assertEquals(20, fml1Refresh.getLoadedFds().size());
+ assertEquals(3, fml1Refresh.getIcebergPartitions().size());
List<String> relPaths = new ArrayList<>(
Collections2.transform(fml1Refresh.getLoadedFds(),
@@ -273,18 +276,20 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml2 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
- /* oldFds = */ Collections.emptyList(), Collections.emptyList(),
+ /* oldFds = */ Collections.emptyList(), Collections.emptyMap(),
/* requiresDataFilesInTableLocation = */ true);
fml2.load();
+ assertEquals(1, fml2.getIcebergPartitions().size());
IcebergFileMetadataLoader fml2Refresh = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
- /* oldFds = */ fml2.getLoadedIcebergFds(), fml1.getIcebergPartitionList(),
+ /* oldFds = */ fml2.getLoadedIcebergFds(), fml2.getIcebergPartitions(),
/* requiresDataFilesInTableLocation = */ true);
fml2Refresh.load();
assertEquals(0, fml2Refresh.getStats().loadedFiles);
assertEquals(20, fml2Refresh.getStats().skippedFiles);
assertEquals(20, fml2Refresh.getLoadedFds().size());
+ assertEquals(1, fml2Refresh.getIcebergPartitions().size());
relPaths = new ArrayList<>(
Collections2.transform(fml2Refresh.getLoadedFds(),
@@ -299,35 +304,39 @@ public class FileMetadataLoaderTest {
CatalogServiceCatalog catalog = CatalogServiceTestCatalog.create();
IcebergFileMetadataLoader fml1 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
- /* oldFds = */ Collections.emptyList(), Collections.emptyList(),
+ /* oldFds = */ Collections.emptyList(), Collections.emptyMap(),
/* requiresDataFilesInTableLocation = */ true);
fml1.load();
+ assertEquals(3, fml1.getIcebergPartitions().size());
IcebergFileMetadataLoader fml1Refresh = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
/* oldFds = */ fml1.getLoadedIcebergFds().subList(0, 10),
- fml1.getIcebergPartitionList(),
+ fml1.getIcebergPartitions(),
/* requiresDataFilesInTableLocation = */ true);
fml1Refresh.load();
assertEquals(10, fml1Refresh.getStats().loadedFiles);
assertEquals(10, fml1Refresh.getStats().skippedFiles);
assertEquals(20, fml1Refresh.getLoadedFds().size());
+ assertEquals(3, fml1Refresh.getIcebergPartitions().size());
IcebergFileMetadataLoader fml2 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
- /* oldFds = */ Collections.emptyList(), Collections.emptyList(),
+ /* oldFds = */ Collections.emptyList(), Collections.emptyMap(),
/* requiresDataFilesInTableLocation = */ true);
fml2.load();
+ assertEquals(1, fml2.getIcebergPartitions().size());
IcebergFileMetadataLoader fml2Refresh = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
/* oldFds = */ fml2.getLoadedIcebergFds().subList(0, 10),
- fml1.getIcebergPartitionList(),
+ fml2.getIcebergPartitions(),
/* requiresDataFilesInTableLocation = */ true);
fml2Refresh.load();
assertEquals(10, fml2Refresh.getStats().loadedFiles);
assertEquals(10, fml2Refresh.getStats().skippedFiles);
assertEquals(20, fml2Refresh.getLoadedFds().size());
+ assertEquals(1, fml2Refresh.getIcebergPartitions().size());
}
@Test
@@ -336,14 +345,14 @@ public class FileMetadataLoaderTest {
BackendConfig.INSTANCE.setIcebergAllowDatafileInTableLocationOnly(false);
IcebergFileMetadataLoader fml1 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_multiple_storage_locations",
- /* oldFds = */ Collections.emptyList(), Collections.emptyList(),
+ /* oldFds = */ Collections.emptyList(), Collections.emptyMap(),
/* requiresDataFilesInTableLocation = */ false);
fml1.load();
IcebergFileMetadataLoader fml1Refresh1 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_multiple_storage_locations",
/* oldFds = */ fml1.getLoadedIcebergFds().subList(0, 1),
- fml1.getIcebergPartitionList(),
+ fml1.getIcebergPartitions(),
/* requiresDataFilesInTableLocation = */ false);
fml1Refresh1.load();
assertEquals(5, fml1Refresh1.getStats().loadedFiles);
@@ -353,7 +362,7 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml1Refresh5 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_multiple_storage_locations",
/* oldFds = */ fml1.getLoadedIcebergFds().subList(0, 5),
- fml1.getIcebergPartitionList(),
+ fml1.getIcebergPartitions(),
/* requiresDataFilesInTableLocation = */ false);
fml1Refresh5.load();
assertEquals(1, fml1Refresh5.getStats().loadedFiles);
@@ -363,7 +372,7 @@ public class FileMetadataLoaderTest {
private IcebergFileMetadataLoader getLoaderForIcebergTable(
CatalogServiceCatalog catalog, String dbName, String tblName,
- List<IcebergFileDescriptor> oldFds, List<TIcebergPartition> oldPartitions,
+ List<IcebergFileDescriptor> oldFds, Map<ByteBuffer, Integer> oldPartitionMap,
boolean requiresDataFilesInTableLocation)
throws CatalogException {
ListMap<TNetworkAddress> hostIndex = new ListMap<>();
@@ -372,8 +381,10 @@ public class FileMetadataLoaderTest {
Path location = new Path(iceT.getLocation());
GroupedContentFiles iceFiles = IcebergUtil.getIcebergFiles(iceT,
/*predicates=*/Collections.emptyList(), /*timeTravelSpec=*/null);
+ List<ByteBuffer> oldPartitionList =
+ IcebergContentFileStore.convertPartitionMapToList(oldPartitionMap, 0);
return new IcebergFileMetadataLoader(iceT.getIcebergApiTable(),
- oldFds, hostIndex, iceFiles, oldPartitions,
+ oldFds, hostIndex, iceFiles, oldPartitionList,
requiresDataFilesInTableLocation, null);
}