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 2026/04/24 04:32:36 UTC
(impala) 02/02: IMPALA-14588: Implement Iceberg deletion vector reading/writing
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
commit 7ceefb1ed84c2d3e69cdf5688d294adcb67ee28a
Author: Peter Rozsa <pr...@cloudera.com>
AuthorDate: Wed Mar 4 11:48:58 2026 +0100
IMPALA-14588: Implement Iceberg deletion vector reading/writing
This is the second part of a multi-part implementation adding support
for Iceberg deletion vectors stored in Puffin files. The first part
is added as IMPALA-14755, this patch covers IMPALA-14587 and
IMPALA-14588. This commit wires the Puffin reader/writer infrastructure
from part 1 into the query execution pipeline and catalog layer,
enabling DELETE on Iceberg V3 tables using deletion vectors.
Puffin file writing is partition-scoped: the delete sink creates one
Puffin file per output partition, and each blob inside that file is a
serialised RoaringBitmap64 covering exactly the deleted row positions
of one data file in that partition. When a data file already has a
deletion vector from a previous DELETE, the existing bitmap is OR-ed
with the new one before the merged blob is written, so each Puffin file
always holds the complete, up-to-date set of deleted positions for its
partition.
DELETE on a V3 table is rejected at analysis time if the table has any
existing V2 position- or equality-delete files. The table must first be
compacted with OPTIMIZE TABLE to remove those files before DELETE can be
used on them.
Testing:
- iceberg-v3-delete.test and iceberg-v3-delete-partition-sort.test
added.
- Manually validated that deletion vectors written by Spark can be read
from Impala and deletion vectors written by Impala can be read from
Spark.
Generated-by: Claude Sonnet 4.6
Change-Id: I5613c31a7aa46b94b7c70386c939c08cc68632cd
Reviewed-on: http://gerrit.cloudera.org:8080/24143
Reviewed-by: Zoltan Borok-Nagy <bo...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/blob-reader.h | 3 +-
be/src/exec/iceberg-delete-builder.cc | 50 +-
be/src/exec/iceberg-delete-builder.h | 33 +-
be/src/exec/iceberg-delete-sink-base.cc | 8 +-
be/src/exec/iceberg-delete-sink-base.h | 6 +-
be/src/exec/iceberg-delete-sink-config.cc | 4 +
be/src/exec/iceberg-delete-sink-config.h | 5 +
be/src/exec/output-partition.h | 3 +-
be/src/exec/puffin/puffin-writer.cc | 34 +-
be/src/exec/puffin/puffin-writer.h | 8 +
be/src/runtime/dml-exec-state.cc | 196 ++++-
be/src/runtime/dml-exec-state.h | 19 +
be/src/scheduling/scheduler.cc | 3 +
be/src/service/client-request-state.cc | 121 +--
be/src/service/client-request-state.h | 11 +-
be/src/util/roaring-bitmap.h | 4 +
common/fbs/IcebergObjects.fbs | 2 +
.../apache/impala/analysis/IcebergDeleteImpl.java | 17 +-
.../org/apache/impala/catalog/FeIcebergTable.java | 6 +-
.../impala/catalog/IcebergContentFileStore.java | 11 +-
.../apache/impala/catalog/IcebergDeleteTable.java | 10 +
.../catalog/iceberg/GroupedContentFiles.java | 16 +-
.../apache/impala/planner/DistributedPlanner.java | 6 +-
.../org/apache/impala/planner/ExchangeNode.java | 2 +-
.../org/apache/impala/planner/HdfsScanNode.java | 7 +-
.../impala/planner/IcebergBufferedDeleteSink.java | 20 +-
...gDeleteNode.java => IcebergDeleteJoinNode.java} | 15 +-
.../org/apache/impala/planner/IcebergScanNode.java | 40 +-
.../apache/impala/planner/IcebergScanPlanner.java | 60 +-
.../org/apache/impala/planner/JoinBuildSink.java | 4 +-
.../impala/service/IcebergCatalogOpExecutor.java | 127 ++-
.../java/org/apache/impala/util/IcebergUtil.java | 29 +-
.../QueryTest/iceberg-v3-delete-existing-dv.test | 116 +++
.../iceberg-v3-delete-partition-sort.test | 882 +++++++++++++++++++++
.../iceberg-v3-delete-v2-equality-upgrade.test | 102 +++
.../queries/QueryTest/iceberg-v3-delete.test | 423 ++++++++++
.../queries/QueryTest/iceberg-v3-negative.test | 10 -
tests/query_test/test_iceberg.py | 138 ++++
38 files changed, 2411 insertions(+), 140 deletions(-)
diff --git a/be/src/exec/blob-reader.h b/be/src/exec/blob-reader.h
index b2e131d29..5952b2cad 100644
--- a/be/src/exec/blob-reader.h
+++ b/be/src/exec/blob-reader.h
@@ -148,7 +148,8 @@ struct DeletionVectorDeserializer {
DCHECK(data != nullptr);
DCHECK(output != nullptr);
- RETURN_IF_ERROR(RoaringBitmap64::Deserialize(data, length, output));
+ RETURN_IF_ERROR(RoaringBitmap64::Deserialize(data + DELETION_VECTOR_BLOB_HEADER_SIZE,
+ length - DELETION_VECTOR_BLOB_HEADER_SIZE, output));
VLOG(2) << "Deserialized deletion vector with " << output->Cardinality()
<< " deleted positions (min: " << output->Min()
diff --git a/be/src/exec/iceberg-delete-builder.cc b/be/src/exec/iceberg-delete-builder.cc
index 7e074d0cf..8163dcc89 100644
--- a/be/src/exec/iceberg-delete-builder.cc
+++ b/be/src/exec/iceberg-delete-builder.cc
@@ -19,7 +19,6 @@
#include <filesystem>
-#include "exec/blob-reader.h"
#include "exec/exec-node.h"
#include "exec/join-op.h"
#include "runtime/exec-env.h"
@@ -181,9 +180,27 @@ Status IcebergDeleteBuilder::CalculateDataFiles() {
memcpy(ptr_copy, file_path_str.c_str(), file_path_str.length());
+ RoaringBitmap64 bitmap;
+ if (params.scan_range().has_iceberg_deletion_vector()) {
+ const auto* ice_del_vector =
+ flatbuffers::GetRoot<org::apache::impala::fb::FbIcebergDeletionVector>(
+ params.scan_range().iceberg_deletion_vector().c_str());
+ int64_t content_size = ice_del_vector->content_size_in_bytes();
+ if (content_size <= 0) {
+ return Status(strings::Substitute(
+ "Invalid deletion vector for file '$0': content_size_in_bytes is $1.",
+ file_path_str, content_size));
+ }
+ VLOG_FILE << "Found DV for file " << file_path_str << " => "
+ << ice_del_vector->path()->str();
+ SCOPED_TIMER(dv_load_timer_);
+ RETURN_IF_ERROR(dv_reader_.Load(reader_context_.get(), mem_tracker(), &obj_pool_,
+ ice_del_vector->path()->str(), ice_del_vector->content_offset(),
+ content_size, &bitmap));
+ }
deleted_rows_.emplace(std::piecewise_construct,
std::forward_as_tuple(ptr_copy, file_path_str.length()),
- std::forward_as_tuple());
+ std::forward_as_tuple(std::move(bitmap)));
}
}
@@ -194,10 +211,32 @@ Status IcebergDeleteBuilder::Prepare(
RuntimeState* state, MemTracker* parent_mem_tracker) {
RETURN_IF_ERROR(DataSink::Prepare(state, parent_mem_tracker));
- RETURN_IF_ERROR(CalculateDataFiles());
-
num_build_rows_ = ADD_COUNTER(profile(), "BuildRows", TUnit::UNIT);
+ dv_load_timer_ = ADD_TIMER(profile(), "DeletionVectorLoadTime");
+
+ reader_context_ = ExecEnv::GetInstance()->disk_io_mgr()->RegisterContext();
+
+ // Wire data cache counters into the request context so HdfsFileReader can update
+ // per-context cache statistics when reading deletion vector files.
+ data_cache_hit_count_ =
+ ADD_COUNTER(profile(), "DataCacheHitCount", TUnit::UNIT);
+ data_cache_partial_hit_count_ =
+ ADD_COUNTER(profile(), "DataCachePartialHitCount", TUnit::UNIT);
+ data_cache_miss_count_ =
+ ADD_COUNTER(profile(), "DataCacheMissCount", TUnit::UNIT);
+ data_cache_hit_bytes_ =
+ ADD_COUNTER(profile(), "DataCacheHitBytes", TUnit::BYTES);
+ data_cache_miss_bytes_ =
+ ADD_COUNTER(profile(), "DataCacheMissBytes", TUnit::BYTES);
+ reader_context_->set_data_cache_hit_counter(data_cache_hit_count_);
+ reader_context_->set_data_cache_partial_hit_counter(data_cache_partial_hit_count_);
+ reader_context_->set_data_cache_miss_counter(data_cache_miss_count_);
+ reader_context_->set_data_cache_hit_bytes_counter(data_cache_hit_bytes_);
+ reader_context_->set_data_cache_miss_bytes_counter(data_cache_miss_bytes_);
+
+ RETURN_IF_ERROR(CalculateDataFiles());
+
RETURN_IF_ERROR(DebugAction(state->query_options(), "ID_BUILDER_PREPARE"));
const auto& tuple_descs = build_row_desc_->tuple_descriptors();
@@ -241,6 +280,9 @@ Status IcebergDeleteBuilder::FinalizeBuild(RuntimeState* state) {
void IcebergDeleteBuilder::Close(RuntimeState* state) {
if (closed_) return;
+ if (reader_context_ != nullptr) {
+ ExecEnv::GetInstance()->disk_io_mgr()->UnregisterContext(reader_context_.get());
+ }
obj_pool_.Clear();
DataSink::Close(state);
closed_ = true;
diff --git a/be/src/exec/iceberg-delete-builder.h b/be/src/exec/iceberg-delete-builder.h
index ae25d3040..ecf096a20 100644
--- a/be/src/exec/iceberg-delete-builder.h
+++ b/be/src/exec/iceberg-delete-builder.h
@@ -22,6 +22,7 @@
#include "common/object-pool.h"
#include "common/status.h"
+#include "exec/blob-reader.h"
#include "exec/join-builder.h"
#include "util/roaring-bitmap.h"
@@ -31,6 +32,11 @@ class IcebergDeleteBuilder;
class RowDescriptor;
class RuntimeState;
+namespace io {
+class RequestContext;
+}
+
+
/// Iceberg Delete Builder Config class. This has a few extra methods to be used
/// directly by the IcebergDeletePlanNode.
class IcebergDeleteBuilderConfig : public JoinBuilderConfig {
@@ -93,9 +99,6 @@ class IcebergDeleteBuilder : public JoinBuilder {
int64_t max_row_buffer_size, RuntimeState* state);
~IcebergDeleteBuilder();
- // Collects the processed data files' file paths and fills 'deleted_rows_' with them.
- Status CalculateDataFiles();
-
/// Implementations of DataSink interface methods.
Status Prepare(RuntimeState* state, MemTracker* parent_mem_tracker) override;
Status Open(RuntimeState* state) override;
@@ -112,9 +115,12 @@ class IcebergDeleteBuilder : public JoinBuilder {
using DeleteRowHashTable =
std::unordered_map<impala::StringValue, RoaringBitmap64>;
- DeleteRowHashTable& deleted_rows() { return deleted_rows_; }
+ const DeleteRowHashTable& deleted_rows() const { return deleted_rows_; }
private:
+ // Collects the processed data files' file paths and fills 'deleted_rows_' with them.
+ Status CalculateDataFiles();
+
/// Reads the rows in build_batch and collects them into delete_hash_.
Status ProcessBuildBatch(RuntimeState* state, RowBatch* build_batch);
@@ -127,6 +133,8 @@ class IcebergDeleteBuilder : public JoinBuilder {
/// Helper method for FlushFinal() that does the actual work.
Status FinalizeBuild(RuntimeState* state);
+ RuntimeProfile::Counter* dv_load_timer_ = nullptr;
+
RuntimeState* const runtime_state_;
/// Pool for objects with same lifetime as builder.
@@ -140,7 +148,22 @@ class IcebergDeleteBuilder : public JoinBuilder {
int file_path_offset_;
int pos_offset_;
- // Stores {file_path: positions in roaring bitmap}
+ // Stores {file_path: deleted row positions as roaring bitmap}. Entries are
+ // pre-populated by CalculateDataFiles(): deletion vector bitmaps are loaded eagerly
+ // from Puffin files, while positional-delete bitmaps start empty and are filled
+ // incrementally during the build phase.
DeleteRowHashTable deleted_rows_;
+ std::unique_ptr<io::RequestContext> reader_context_;
+
+ /// Blob reader for loading deletion vector bitmaps from Puffin files.
+ DeletionVectorBlobReader dv_reader_;
+
+ /// Data cache counters, wired into reader_context_ so HdfsFileReader can update
+ /// per-context cache statistics when reading deletion vector files.
+ RuntimeProfile::Counter* data_cache_hit_count_ = nullptr;
+ RuntimeProfile::Counter* data_cache_partial_hit_count_ = nullptr;
+ RuntimeProfile::Counter* data_cache_miss_count_ = nullptr;
+ RuntimeProfile::Counter* data_cache_hit_bytes_ = nullptr;
+ RuntimeProfile::Counter* data_cache_miss_bytes_ = nullptr;
};
} // namespace impala
diff --git a/be/src/exec/iceberg-delete-sink-base.cc b/be/src/exec/iceberg-delete-sink-base.cc
index 7d3b07d20..ff542d679 100644
--- a/be/src/exec/iceberg-delete-sink-base.cc
+++ b/be/src/exec/iceberg-delete-sink-base.cc
@@ -32,9 +32,8 @@ namespace impala {
IcebergDeleteSinkBase::IcebergDeleteSinkBase(TDataSinkId sink_id,
const IcebergDeleteSinkConfig& sink_config, const std::string& name,
- RuntimeState* state) :
- TableSinkBase(sink_id, sink_config, name, state) {
-}
+ RuntimeState* state): TableSinkBase(sink_id, sink_config, name, state),
+ referenced_deletion_vectors_(&sink_config.referenced_deletion_vectors) {}
Status IcebergDeleteSinkBase::Prepare(RuntimeState* state,
MemTracker* parent_mem_tracker) {
@@ -133,6 +132,9 @@ Status IcebergDeleteSinkBase::ConstructPartitionInfo(
Status IcebergDeleteSinkBase::ConstructPartitionInfo(int32_t spec_id,
const std::string& partition_values_str, OutputPartition* output_partition) {
+
+ output_partition->referenced_deletion_vectors = referenced_deletion_vectors_;
+
if (partition_key_expr_evals_.empty()) {
DCHECK_EQ(spec_id, table_desc_->IcebergSpecId());
output_partition->iceberg_spec_id = spec_id;
diff --git a/be/src/exec/iceberg-delete-sink-base.h b/be/src/exec/iceberg-delete-sink-base.h
index f1e19481b..cf4058b0d 100644
--- a/be/src/exec/iceberg-delete-sink-base.h
+++ b/be/src/exec/iceberg-delete-sink-base.h
@@ -17,7 +17,9 @@
#pragma once
+#include <map>
#include "exec/table-sink-base.h"
+#include "gen-cpp/CatalogObjects_types.h"
namespace impala {
@@ -61,6 +63,8 @@ class IcebergDeleteSinkBase : public TableSinkBase {
std::string HumanReadablePartitionValue(
const TIcebergPartitionField& part_field, const std::string& value,
Status* transform_result);
+ private:
+ const std::map<THash128, TIcebergDeletionVector>* referenced_deletion_vectors_;
};
-}
\ No newline at end of file
+}
diff --git a/be/src/exec/iceberg-delete-sink-config.cc b/be/src/exec/iceberg-delete-sink-config.cc
index 6cf93e833..063d1cf66 100644
--- a/be/src/exec/iceberg-delete-sink-config.cc
+++ b/be/src/exec/iceberg-delete-sink-config.cc
@@ -36,6 +36,10 @@ Status IcebergDeleteSinkConfig::Init(
RETURN_IF_ERROR(DataSinkConfig::Init(tsink, input_row_desc, state));
DCHECK(tsink_->__isset.table_sink);
DCHECK(tsink_->table_sink.__isset.iceberg_delete_sink);
+ if (tsink_->table_sink.iceberg_delete_sink.__isset.deletion_vectors) {
+ referenced_deletion_vectors =
+ tsink_->table_sink.iceberg_delete_sink.deletion_vectors;
+ }
RETURN_IF_ERROR(
ScalarExpr::Create(tsink_->table_sink.iceberg_delete_sink.partition_key_exprs,
*input_row_desc_, state, &partition_key_exprs_));
diff --git a/be/src/exec/iceberg-delete-sink-config.h b/be/src/exec/iceberg-delete-sink-config.h
index c34111bb7..da8486e5a 100644
--- a/be/src/exec/iceberg-delete-sink-config.h
+++ b/be/src/exec/iceberg-delete-sink-config.h
@@ -17,7 +17,10 @@
#pragma once
+#include <map>
#include "exec/table-sink-base.h"
+#include "gen-cpp/CatalogObjects_types.h"
+#include "gen-cpp/Types_types.h"
namespace impala {
@@ -27,6 +30,8 @@ class IcebergDeleteSinkConfig : public TableSinkBaseConfig {
~IcebergDeleteSinkConfig() override {}
+ std::map<THash128, TIcebergDeletionVector> referenced_deletion_vectors;
+
protected:
Status Init(const TDataSink& tsink, const RowDescriptor* input_row_desc,
FragmentState* state) override;
diff --git a/be/src/exec/output-partition.h b/be/src/exec/output-partition.h
index 317c73825..2ee234ba6 100644
--- a/be/src/exec/output-partition.h
+++ b/be/src/exec/output-partition.h
@@ -125,7 +125,8 @@ struct OutputPartition {
/// Map from data file path hash to deletion vector metadata.
/// Used to track existing deletion vectors that need to be loaded and merged.
- std::map<THash128, TIcebergDeletionVector> referenced_deletion_vectors;
+ /// Points to the map owned by IcebergDeleteSinkConfig; not owned by this struct.
+ const std::map<THash128, TIcebergDeletionVector>* referenced_deletion_vectors = nullptr;
};
}
diff --git a/be/src/exec/puffin/puffin-writer.cc b/be/src/exec/puffin/puffin-writer.cc
index cf5518c32..db0a1065b 100644
--- a/be/src/exec/puffin/puffin-writer.cc
+++ b/be/src/exec/puffin/puffin-writer.cc
@@ -74,6 +74,25 @@ Status PuffinWriter::Init() {
obj_pool_ = state_->obj_pool();
DCHECK(obj_pool_ != nullptr);
+ // Wire data cache counters into the request context so HdfsFileReader can update
+ // per-context cache statistics when reading existing deletion vectors.
+ RuntimeProfile* profile = state_->runtime_profile();
+ data_cache_hit_count_ =
+ ADD_COUNTER(profile, "DataCacheHitCount", TUnit::UNIT);
+ data_cache_partial_hit_count_ =
+ ADD_COUNTER(profile, "DataCachePartialHitCount", TUnit::UNIT);
+ data_cache_miss_count_ =
+ ADD_COUNTER(profile, "DataCacheMissCount", TUnit::UNIT);
+ data_cache_hit_bytes_ =
+ ADD_COUNTER(profile, "DataCacheHitBytes", TUnit::BYTES);
+ data_cache_miss_bytes_ =
+ ADD_COUNTER(profile, "DataCacheMissBytes", TUnit::BYTES);
+ io_request_context_->set_data_cache_hit_counter(data_cache_hit_count_);
+ io_request_context_->set_data_cache_partial_hit_counter(data_cache_partial_hit_count_);
+ io_request_context_->set_data_cache_miss_counter(data_cache_miss_count_);
+ io_request_context_->set_data_cache_hit_bytes_counter(data_cache_hit_bytes_);
+ io_request_context_->set_data_cache_miss_bytes_counter(data_cache_miss_bytes_);
+
// Initialize deletion vector blob reader
dv_blob_reader_.reset(new DeletionVectorBlobReader());
@@ -173,9 +192,9 @@ Status PuffinWriter::Finalize() {
const std::string& data_file_path = entry.first;
// Check if this data file has an old DV that needs to be loaded and merged
- auto old_dv_it = output_->referenced_deletion_vectors.find(
+ auto old_dv_it = output_->referenced_deletion_vectors->find(
THash128FromFilePath(data_file_path));
- if (old_dv_it != output_->referenced_deletion_vectors.end()) {
+ if (old_dv_it != output_->referenced_deletion_vectors->end()) {
const TIcebergDeletionVector& old_dv_metadata = old_dv_it->second;
VLOG(2) << "Loading old DV for data file with new deletes: " << data_file_path;
@@ -184,7 +203,7 @@ Status PuffinWriter::Finalize() {
SCOPED_TIMER(load_dv_timer_);
RETURN_IF_ERROR(LoadExistingDeletionVector(
old_dv_metadata.path, data_file_path,
- old_dv_metadata.content_offset + DELETION_VECTOR_BLOB_HEADER_SIZE,
+ old_dv_metadata.content_offset,
old_dv_metadata.content_size_in_bytes));
// Track that this old DV is being replaced
@@ -215,6 +234,8 @@ Status PuffinWriter::Finalize() {
// Write the footer containing blob metadata
RETURN_IF_ERROR(WriteFooter());
+ output_->current_file_bytes = current_offset_;
+
blob_mem_pool_->Clear();
return Status::OK();
@@ -236,6 +257,7 @@ Status PuffinWriter::AddBlob(puffin::Blob& blob) {
Status PuffinWriter::AddDeletionVector(const std::string& data_file_path,
RoaringBitmap64& bitmap, int64_t snapshot_id, int64_t sequence_number) {
// Get the serialized size of the bitmap
+ bitmap.Optimize(); // Optimize the bitmap before serialization for better compression
size_t bitmap_serialized_size = bitmap.BitmapSizeInBytes();
if (bitmap_serialized_size == 0) {
return Status("Cannot add deletion vector: bitmap is empty");
@@ -253,6 +275,7 @@ Status PuffinWriter::AddDeletionVector(const std::string& data_file_path,
const size_t total_blob_size =
length_field_size + magic_size + bitmap_serialized_size + crc_size;
+ DCHECK_GT(total_blob_size, 0);
// Allocate buffer for the complete serialized blob
uint8_t* data = blob_mem_pool_->TryAllocate(total_blob_size);
if (UNLIKELY(data == nullptr)) {
@@ -315,10 +338,13 @@ Status PuffinWriter::WriteFooter() {
uint32_t magic_be = htonl(PUFFIN_MAGIC);
RETURN_IF_ERROR(Write(magic_be));
RETURN_IF_ERROR(Write(footer_json.data(), footer_json.size()));
+
uint32_t footer_size = static_cast<uint32_t>(footer_json.size());
RETURN_IF_ERROR(Write(footer_size));
- RETURN_IF_ERROR(Write(0));
+ RETURN_IF_ERROR(Write(0)); //flags (reserved for future use)
RETURN_IF_ERROR(Write(magic_be));
+ current_offset_ += 4 /*leading magic*/ + footer_size
+ + 4 /*footer size field*/ + 4 /*flags*/ + 4 /*trailing magic*/;
return Status::OK();
}
diff --git a/be/src/exec/puffin/puffin-writer.h b/be/src/exec/puffin/puffin-writer.h
index b58cc4437..fc7790d0e 100644
--- a/be/src/exec/puffin/puffin-writer.h
+++ b/be/src/exec/puffin/puffin-writer.h
@@ -149,6 +149,14 @@ class PuffinWriter : public HdfsTableWriter {
/// Timer for loading existing deletion vectors
RuntimeProfile::Counter* load_dv_timer_;
+ /// Data cache counters, wired into io_request_context_ so HdfsFileReader can
+ /// update per-context cache statistics when loading existing deletion vectors.
+ RuntimeProfile::Counter* data_cache_hit_count_ = nullptr;
+ RuntimeProfile::Counter* data_cache_partial_hit_count_ = nullptr;
+ RuntimeProfile::Counter* data_cache_miss_count_ = nullptr;
+ RuntimeProfile::Counter* data_cache_hit_bytes_ = nullptr;
+ RuntimeProfile::Counter* data_cache_miss_bytes_ = nullptr;
+
/// Magic bytes "PFA1" as per Puffin spec
static constexpr uint32_t PUFFIN_MAGIC = 0x50464131;
diff --git a/be/src/runtime/dml-exec-state.cc b/be/src/runtime/dml-exec-state.cc
index 408a760d5..94da22104 100644
--- a/be/src/runtime/dml-exec-state.cc
+++ b/be/src/runtime/dml-exec-state.cc
@@ -28,6 +28,9 @@
#include "common/logging.h"
#include "exec/data-sink.h"
#include "exec/output-partition.h"
+#define XXH_INLINE_ALL
+#include "thirdparty/xxhash/xxhash.h"
+#undef XXH_INLINE_ALL
#include "util/pretty-printer.h"
#include "util/container-util.h"
#include "util/hdfs-bulk-ops.h"
@@ -574,6 +577,81 @@ createIcebergColumnStats(
return stats_builder.Finish();
}
+// Helper function to create an Iceberg data file FlatBuffer with deletion vectors.
+// Used for puffin files where we need to create one entry per data file with DVs.
+// Parameters:
+// partition: The output partition (for partition info)
+// puffin_file_path: Path to the puffin file containing the deletion vector blobs
+// puffin_file_size: Total size of the puffin file in bytes
+// referenced_data_file: Path to the data file that this deletion vector references
+// old_dv: Old deletion vector to remove (can be nullptr)
+// new_dv: New deletion vector to add
+string createIcebergDataFileWithDeletionVectorString(
+ const OutputPartition& partition,
+ const string& puffin_file_path,
+ int64_t puffin_file_size,
+ const string& referenced_data_file,
+ const TIcebergDeletionVector* old_dv,
+ const TIcebergDeletionVector* new_dv) {
+ DCHECK(new_dv != nullptr);
+ using namespace org::apache::impala::fb;
+ flatbuffers::FlatBufferBuilder fbb;
+
+ // For puffin files with deletion vectors, we don't have column stats
+ vector<flatbuffers::Offset<FbIcebergColumnStats>> ice_col_stats_vec;
+
+ vector<flatbuffers::Offset<FbIcebergPartitionField>> raw_partition_fields;
+ for (const string& partition_name : partition.raw_partition_names) {
+ auto data = reinterpret_cast<const uint8_t*>(partition_name.data());
+ auto fb_vector = fbb.CreateVector(data, partition_name.size());
+ raw_partition_fields.push_back(CreateFbIcebergPartitionField(fbb, fb_vector));
+ }
+
+ THash128 hash = THash128FromFilePath(referenced_data_file);
+
+ // Create FlatBuffer deletion vector reference for old DV
+ flatbuffers::Offset<FbIcebergDeletionVector> old_dv_fb = 0;
+ if (old_dv != nullptr) {
+ old_dv_fb = CreateFbIcebergDeletionVector(fbb,
+ fbb.CreateString(old_dv->path),
+ old_dv->content_offset,
+ old_dv->content_size_in_bytes,
+ hash.high,
+ hash.low,
+ old_dv->__isset.record_count ? old_dv->record_count : 0);
+ }
+
+ // Create FlatBuffer deletion vector reference for new DV
+ flatbuffers::Offset<FbIcebergDeletionVector> new_dv_fb = CreateFbIcebergDeletionVector(
+ fbb,
+ fbb.CreateString(puffin_file_path),
+ new_dv->content_offset,
+ new_dv->content_size_in_bytes,
+ hash.high,
+ hash.low,
+ new_dv->__isset.record_count ? new_dv->record_count : 0,
+ puffin_file_size);
+
+ // For puffin files, use the deletion vector record count and the total
+ // Puffin file size.
+ int64_t num_rows = new_dv->__isset.record_count ? new_dv->record_count : 0;
+
+ flatbuffers::Offset<FbIcebergDataFile> data_file = CreateFbIcebergDataFile(fbb,
+ fbb.CreateString(puffin_file_path),
+ FbIcebergDataFileFormat::FbIcebergDataFileFormat_PUFFIN,
+ num_rows,
+ puffin_file_size,
+ partition.iceberg_spec_id,
+ fbb.CreateString(partition.partition_name),
+ fbb.CreateVector(raw_partition_fields),
+ fbb.CreateVector(ice_col_stats_vec),
+ old_dv_fb,
+ new_dv_fb);
+ fbb.Finish(data_file);
+ return string(reinterpret_cast<char*>(fbb.GetBufferPointer()), fbb.GetSize());
+}
+
+// Creates an Iceberg data file FlatBuffer for regular (non-puffin) data files.
string createIcebergDataFileString(
const OutputPartition& partition, const string& final_path, int64_t num_rows,
int64_t file_size, const IcebergFileStats& insert_stats) {
@@ -595,14 +673,15 @@ string createIcebergDataFileString(
flatbuffers::Offset<FbIcebergDataFile> data_file = CreateFbIcebergDataFile(fbb,
fbb.CreateString(final_path),
- // Currently we can only write Parquet to Iceberg
FbIcebergDataFileFormat::FbIcebergDataFileFormat_PARQUET,
num_rows,
file_size,
partition.iceberg_spec_id,
fbb.CreateString(partition.partition_name),
fbb.CreateVector(raw_partition_fields),
- fbb.CreateVector(ice_col_stats_vec));
+ fbb.CreateVector(ice_col_stats_vec),
+ 0, // No old_deletion_vector for regular files
+ 0); // No new_deletion_vector for regular files
fbb.Finish(data_file);
return string(reinterpret_cast<char*>(fbb.GetBufferPointer()), fbb.GetSize());
}
@@ -619,9 +698,73 @@ void DmlExecState::AddCreatedDeleteFile(const OutputPartition& partition,
AddFileAux(partition, /*is_iceberg=*/true, insert_stats, /*is_delete=*/true);
}
+// Helper function to add puffin file entries with deletion vectors.
+// Creates one FbIcebergDataFile entry per data file that has deletion vectors.
+// This is called for puffin files only.
+void DmlExecState::AddPuffinDeletionVectorEntries(const OutputPartition& partition) {
+ // Note: lock_ is already held by AddFileAux
+
+ const string& puffin_path = partition.current_file_final_name.empty()
+ ? partition.current_file_name
+ : partition.current_file_final_name;
+
+ // Build a unified map of all data files with deletion vectors (old and/or new)
+ // Map: data_file_path -> (old_dv_ptr, new_dv_ptr)
+ std::map<std::string,
+ std::pair<const TIcebergDeletionVector*, const TIcebergDeletionVector*>>
+ data_file_to_dvs;
+
+ // Collect old deletion vectors
+ for (const auto& entry : partition.puffin_result.old_deletion_vectors) {
+ data_file_to_dvs[entry.first].first = &entry.second;
+ }
+
+ // Collect new deletion vectors
+ for (const auto& entry : partition.puffin_result.new_deletion_vectors) {
+ data_file_to_dvs[entry.first].second = &entry.second;
+ }
+
+ // Create one FbIcebergDataFile entry per data file with deletion vectors
+ PartitionStatusMap::iterator partition_entry =
+ per_partition_status_.find(partition.partition_name);
+ DCHECK(partition_entry != per_partition_status_.end());
+
+ for (const auto& entry : data_file_to_dvs) {
+ const string& referenced_data_file = entry.first;
+ const TIcebergDeletionVector* old_dv = entry.second.first;
+ const TIcebergDeletionVector* new_dv = entry.second.second;
+
+ DmlFileStatusPb* file = partition_entry->second.add_created_delete_files();
+
+ file->set_final_path(puffin_path);
+ if (!partition.current_file_final_name.empty()) {
+ file->set_staging_path(partition.current_file_name);
+ }
+
+ // For deletion vectors, use the DV record count and the total puffin file size.
+ file->set_num_rows(new_dv && new_dv->__isset.record_count ? new_dv->record_count : 0);
+ file->set_size(partition.current_file_bytes);
+
+ file->set_iceberg_data_file_fb(
+ createIcebergDataFileWithDeletionVectorString(
+ partition, puffin_path, partition.current_file_bytes,
+ referenced_data_file, old_dv, new_dv));
+ }
+}
+
void DmlExecState::AddFileAux(const OutputPartition& partition, bool is_iceberg,
const IcebergFileStats& insert_stats, bool is_delete) {
lock_guard<mutex> l(lock_);
+
+ // Special handling for puffin files with deletion vectors
+ bool is_puffin = partition.writer && partition.writer->file_extension() == "puffin";
+ if (is_puffin && is_delete && is_iceberg) {
+ // For puffin files, create one entry per data file with deletion vectors
+ AddPuffinDeletionVectorEntries(partition);
+ return;
+ }
+
+ // Regular file handling (non-puffin or non-delete files)
PartitionStatusMap::iterator entry =
per_partition_status_.find(partition.partition_name);
DCHECK(entry != per_partition_status_.end());
@@ -674,6 +817,55 @@ vector<string> DmlExecState::CreateIcebergDeleteFilesVector() {
return ret;
}
+template<typename GetDV>
+vector<string> DmlExecState::CollectDeletionVectors(GetDV get_dv) {
+ using namespace org::apache::impala::fb;
+ vector<string> ret;
+
+ for (const PartitionStatusMap::value_type& partition : per_partition_status_) {
+ for (int i = 0; i < partition.second.created_delete_files_size(); ++i) {
+ const DmlFileStatusPb& file = partition.second.created_delete_files(i);
+ if (!file.has_iceberg_data_file_fb()) continue;
+
+ const string& fb_data = file.iceberg_data_file_fb();
+ const FbIcebergDataFile* data_file =
+ flatbuffers::GetRoot<FbIcebergDataFile>(fb_data.data());
+
+ const FbIcebergDeletionVector* dv = get_dv(data_file);
+ if (dv == nullptr) continue;
+
+ // Serialize the DV into its own FlatBuffer.
+ flatbuffers::FlatBufferBuilder builder(256);
+ auto path_offset = builder.CreateString(dv->path()->c_str());
+ auto dv_offset = CreateFbIcebergDeletionVector(builder,
+ path_offset,
+ dv->content_offset(),
+ dv->content_size_in_bytes(),
+ dv->referenced_data_file_hash_high(),
+ dv->referenced_data_file_hash_low(),
+ dv->record_count(),
+ data_file->file_size_in_bytes());
+ builder.Finish(dv_offset);
+
+ ret.push_back(string(reinterpret_cast<const char*>(builder.GetBufferPointer()),
+ builder.GetSize()));
+ }
+ }
+ return ret;
+}
+
+vector<string> DmlExecState::CreateIcebergAddedDeletionVectors() {
+ using namespace org::apache::impala::fb;
+ return CollectDeletionVectors(
+ [](const FbIcebergDataFile* f) { return f->new_deletion_vector(); });
+}
+
+vector<string> DmlExecState::CreateIcebergRemovedDeletionVectors() {
+ using namespace org::apache::impala::fb;
+ return CollectDeletionVectors(
+ [](const FbIcebergDataFile* f) { return f->old_deletion_vector(); });
+}
+
void DmlExecState::MergeDmlStats(const DmlStatsPB& src, DmlStatsPB* dst) {
dst->set_bytes_written(dst->bytes_written() + src.bytes_written());
if (src.has_kudu_stats()) {
diff --git a/be/src/runtime/dml-exec-state.h b/be/src/runtime/dml-exec-state.h
index e82a1bb09..3b778ea93 100644
--- a/be/src/runtime/dml-exec-state.h
+++ b/be/src/runtime/dml-exec-state.h
@@ -136,6 +136,14 @@ class DmlExecState {
// Encodes delete file list info in flatbuffer format expected by Iceberg API.
std::vector<std::string> CreateIcebergDeleteFilesVector();
+ // Creates list of deletion vectors to add to Iceberg metadata.
+ // Each entry is a FlatBuffer-encoded FbIcebergDeletionVector.
+ std::vector<std::string> CreateIcebergAddedDeletionVectors();
+
+ // Creates list of deletion vectors to remove from Iceberg metadata.
+ // Each entry is a FlatBuffer-encoded FbIcebergDeletionVector.
+ std::vector<std::string> CreateIcebergRemovedDeletionVectors();
+
// Returns vector of Iceberg data files referenced by position delete records by
// this DML statement.
const std::vector<std::string>& DataFilesReferencedByPositionDeletes() const {
@@ -157,6 +165,17 @@ class DmlExecState {
void AddFileAux(const OutputPartition& partition, bool is_iceberg,
const IcebergFileStats& insert_stats, bool is_delete);
+ /// Helper function to add puffin file entries with deletion vectors.
+ /// Creates one FbIcebergDataFile entry per data file that has deletion vectors.
+ /// Called by AddFileAux for puffin files only. Assumes lock_ is held.
+ void AddPuffinDeletionVectorEntries(const OutputPartition& partition);
+
+ /// Collects FlatBuffer-serialized FbIcebergDeletionVector entries from all created
+ /// delete files. 'get_dv' selects which deletion vector to extract from each
+ /// FbIcebergDataFile entry (e.g. new_deletion_vector or old_deletion_vector).
+ template<typename GetDV>
+ std::vector<std::string> CollectDeletionVectors(GetDV get_dv);
+
/// protects all fields below
std::mutex lock_;
diff --git a/be/src/scheduling/scheduler.cc b/be/src/scheduling/scheduler.cc
index 8f59dd8ce..690e65bd9 100644
--- a/be/src/scheduling/scheduler.cc
+++ b/be/src/scheduling/scheduler.cc
@@ -1728,6 +1728,9 @@ void TScanRangeToScanRangePB(const TScanRange& tscan_range, ScanRangePB* scan_ra
if (tscan_range.__isset.is_system_scan) {
scan_range_pb->set_is_system_scan(tscan_range.is_system_scan);
}
+ if (tscan_range.__isset.iceberg_deletion_vector) {
+ scan_range_pb->set_iceberg_deletion_vector(tscan_range.iceberg_deletion_vector);
+ }
}
void Scheduler::AssignmentCtx::RecordScanRangeAssignment(
diff --git a/be/src/service/client-request-state.cc b/be/src/service/client-request-state.cc
index d75ca6ee0..e39a8cad9 100644
--- a/be/src/service/client-request-state.cc
+++ b/be/src/service/client-request-state.cc
@@ -1804,64 +1804,82 @@ Status ClientRequestState::UpdateCatalog() {
return Status::OK();
}
+bool ClientRequestState::SetDeleteArtifacts(
+ DmlExecState* dml_exec_state, TIcebergOperationParam* cat_ice_op) {
+ auto added_dvs = dml_exec_state->CreateIcebergAddedDeletionVectors();
+ auto removed_dvs = dml_exec_state->CreateIcebergRemovedDeletionVectors();
+ cat_ice_op->__set_data_files_referenced_by_position_deletes(
+ dml_exec_state->DataFilesReferencedByPositionDeletes());
+ if (!added_dvs.empty() || !removed_dvs.empty()) {
+ // DV operation: set only the DV lists; delete files are not used.
+ cat_ice_op->iceberg_added_deletion_vectors_fb = std::move(added_dvs);
+ cat_ice_op->__isset.iceberg_added_deletion_vectors_fb = true;
+ cat_ice_op->iceberg_removed_deletion_vectors_fb = std::move(removed_dvs);
+ cat_ice_op->__isset.iceberg_removed_deletion_vectors_fb = true;
+ return true;
+ }
+ cat_ice_op->iceberg_delete_files_fb =
+ dml_exec_state->CreateIcebergDeleteFilesVector();
+ cat_ice_op->__isset.iceberg_delete_files_fb = true;
+
+ return !cat_ice_op->iceberg_delete_files_fb.empty();
+}
+
bool ClientRequestState::CreateIcebergCatalogOps(
const TFinalizeParams& finalize_params, TIcebergOperationParam* cat_ice_op) {
DCHECK(cat_ice_op != nullptr);
- const TIcebergDmlFinalizeParams& ice_finalize_params = finalize_params.iceberg_params;
- DmlExecState* dml_exec_state = GetCoordinator()->dml_exec_state();
- bool update_catalog = true;
- cat_ice_op->__set_operation(ice_finalize_params.operation);
- cat_ice_op->__set_initial_snapshot_id(
- ice_finalize_params.initial_snapshot_id);
- cat_ice_op->__set_spec_id(ice_finalize_params.spec_id);
- if (ice_finalize_params.operation == TIcebergOperation::INSERT) {
- cat_ice_op->__set_iceberg_data_files_fb(
- dml_exec_state->CreateIcebergDataFilesVector());
- cat_ice_op->__set_is_overwrite(finalize_params.is_overwrite);
- if (cat_ice_op->iceberg_data_files_fb.empty()) update_catalog = false;
- } else if (ice_finalize_params.operation == TIcebergOperation::DELETE) {
- cat_ice_op->__set_iceberg_delete_files_fb(
- dml_exec_state->CreateIcebergDeleteFilesVector());
- cat_ice_op->__set_data_files_referenced_by_position_deletes(
- dml_exec_state->DataFilesReferencedByPositionDeletes());
- if (cat_ice_op->iceberg_delete_files_fb.empty()) update_catalog = false;
- } else if (ice_finalize_params.operation == TIcebergOperation::UPDATE) {
- cat_ice_op->__set_iceberg_data_files_fb(
- dml_exec_state->CreateIcebergDataFilesVector());
- cat_ice_op->__set_iceberg_delete_files_fb(
- dml_exec_state->CreateIcebergDeleteFilesVector());
- cat_ice_op->__set_data_files_referenced_by_position_deletes(
- dml_exec_state->DataFilesReferencedByPositionDeletes());
- if (cat_ice_op->iceberg_delete_files_fb.empty()) {
- DCHECK(cat_ice_op->iceberg_data_files_fb.empty());
- update_catalog = false;
- }
- } else if (ice_finalize_params.operation == TIcebergOperation::OPTIMIZE) {
- DCHECK(ice_finalize_params.__isset.optimize_params);
- const TIcebergOptimizeParams& optimize_params = ice_finalize_params.optimize_params;
- if (optimize_params.mode == TIcebergOptimizationMode::NOOP) {
- update_catalog = false;
- } else {
- cat_ice_op->__set_iceberg_data_files_fb(
- dml_exec_state->CreateIcebergDataFilesVector());
- if (optimize_params.mode == TIcebergOptimizationMode::PARTIAL) {
- DCHECK(optimize_params.__isset.selected_data_files_without_deletes);
+ const TIcebergDmlFinalizeParams& ice_params = finalize_params.iceberg_params;
+ DmlExecState* dml = GetCoordinator()->dml_exec_state();
+
+ cat_ice_op->__set_operation(ice_params.operation);
+ cat_ice_op->__set_initial_snapshot_id(ice_params.initial_snapshot_id);
+ cat_ice_op->__set_spec_id(ice_params.spec_id);
+
+ bool has_data_files = false;
+ bool has_delete_artifacts = false;
+
+ switch (ice_params.operation) {
+ case TIcebergOperation::INSERT:
+ cat_ice_op->__set_iceberg_data_files_fb(dml->CreateIcebergDataFilesVector());
+ cat_ice_op->__set_is_overwrite(finalize_params.is_overwrite);
+ has_data_files = !cat_ice_op->iceberg_data_files_fb.empty();
+ break;
+
+ case TIcebergOperation::DELETE:
+ has_delete_artifacts = SetDeleteArtifacts(dml, cat_ice_op);
+ break;
+
+ case TIcebergOperation::UPDATE:
+ cat_ice_op->__set_iceberg_data_files_fb(dml->CreateIcebergDataFilesVector());
+ has_data_files = !cat_ice_op->iceberg_data_files_fb.empty();
+ has_delete_artifacts = SetDeleteArtifacts(dml, cat_ice_op);
+ // UPDATE rewrites rows as delete+insert pairs: delete artifacts must be present
+ // whenever data files are.
+ DCHECK(has_delete_artifacts || !has_data_files);
+ break;
+
+ case TIcebergOperation::MERGE:
+ cat_ice_op->__set_iceberg_data_files_fb(dml->CreateIcebergDataFilesVector());
+ has_data_files = !cat_ice_op->iceberg_data_files_fb.empty();
+ has_delete_artifacts = SetDeleteArtifacts(dml, cat_ice_op);
+ break;
+
+ case TIcebergOperation::OPTIMIZE: {
+ DCHECK(ice_params.__isset.optimize_params);
+ const TIcebergOptimizeParams& opt = ice_params.optimize_params;
+ if (opt.mode == TIcebergOptimizationMode::NOOP) break;
+ cat_ice_op->__set_iceberg_data_files_fb(dml->CreateIcebergDataFilesVector());
+ if (opt.mode == TIcebergOptimizationMode::PARTIAL) {
+ DCHECK(opt.__isset.selected_data_files_without_deletes);
cat_ice_op->__set_replaced_data_files_without_deletes(
- optimize_params.selected_data_files_without_deletes);
+ opt.selected_data_files_without_deletes);
}
- }
- } else if (ice_finalize_params.operation == TIcebergOperation::MERGE) {
- cat_ice_op->__set_iceberg_data_files_fb(
- dml_exec_state->CreateIcebergDataFilesVector());
- cat_ice_op->__set_iceberg_delete_files_fb(
- dml_exec_state->CreateIcebergDeleteFilesVector());
- cat_ice_op->__set_data_files_referenced_by_position_deletes(
- dml_exec_state->DataFilesReferencedByPositionDeletes());
- if (cat_ice_op->iceberg_delete_files_fb.empty()
- && cat_ice_op->iceberg_data_files_fb.empty()) {
- update_catalog = false;
+ has_data_files = true;
+ break;
}
}
+
+ bool update_catalog = has_data_files || has_delete_artifacts;
if (!update_catalog) query_events_->MarkEvent("No-op Iceberg DML statement");
return update_catalog;
}
@@ -2017,6 +2035,7 @@ Status ClientRequestState::UpdateTableAndColumnStats(
}
const TExecRequest& exec_req = exec_request();
+
std::optional<long> snapshot_id = getIcebergSnapshotId(exec_req);
Status status = catalog_op_executor_->ExecComputeStats(
GetCatalogServiceRequestHeader(),
diff --git a/be/src/service/client-request-state.h b/be/src/service/client-request-state.h
index 467e5aa55..5fee31d4b 100644
--- a/be/src/service/client-request-state.h
+++ b/be/src/service/client-request-state.h
@@ -920,14 +920,21 @@ class ClientRequestState {
bool CreateIcebergCatalogOps(const TFinalizeParams& finalize_params,
TIcebergOperationParam* cat_ice_op);
+ /// Sets either the deletion vector fields or the pos-delete fields on 'cat_ice_op'
+ /// (the two are mutually exclusive). Also populates
+ /// data_files_referenced_by_position_deletes. Returns true iff any delete artifacts
+ /// were produced (i.e. the delete side of the operation is not a no-op).
+ bool SetDeleteArtifacts(DmlExecState* dml_exec_state,
+ TIcebergOperationParam* cat_ice_op);
+
/// Copies results into request_result_set_
/// TODO: Have the FE return list<Data.TResultRow> so that this isn't necessary
void SetResultSet(const TDdlExecResponse* ddl_resp);
void SetResultSet(const std::vector<std::string>& results);
void SetResultSet(const std::vector<std::string>& col1,
const std::vector<std::string>& col2);
- void SetResultSet(const vector<string>& col1,
- const vector<string>& col2, const vector<string>& col3);
+ void SetResultSet(const std::vector<std::string>& col1,
+ const std::vector<std::string>& col2, const std::vector<std::string>& col3);
void SetResultSet(const std::vector<std::string>& col1,
const std::vector<std::string>& col2, const std::vector<std::string>& col3,
const std::vector<std::string>& col4);
diff --git a/be/src/util/roaring-bitmap.h b/be/src/util/roaring-bitmap.h
index 441913464..27cffc3e2 100644
--- a/be/src/util/roaring-bitmap.h
+++ b/be/src/util/roaring-bitmap.h
@@ -194,6 +194,10 @@ public:
return roaring64_bitmap_portable_size_in_bytes(rbitmap_);
}
+ bool Optimize() {
+ return roaring64_bitmap_run_optimize(rbitmap_);
+ }
+
/// Serializes this bitmap to 'buf'.
/// Caller must ensure that 'buf' has at least 'BitmapSizeInBytes()' bytes.
uint64_t Serialize(uint8_t* buf) {
diff --git a/common/fbs/IcebergObjects.fbs b/common/fbs/IcebergObjects.fbs
index db091e811..931fa624a 100644
--- a/common/fbs/IcebergObjects.fbs
+++ b/common/fbs/IcebergObjects.fbs
@@ -88,5 +88,7 @@ table FbIcebergDeletionVector {
referenced_data_file_hash_high: long;
referenced_data_file_hash_low: long;
record_count: long;
+ // Total size of the Puffin file containing this deletion vector, in bytes.
+ file_size_in_bytes: long;
}
diff --git a/fe/src/main/java/org/apache/impala/analysis/IcebergDeleteImpl.java b/fe/src/main/java/org/apache/impala/analysis/IcebergDeleteImpl.java
index 7eea62c0a..ccea56cad 100644
--- a/fe/src/main/java/org/apache/impala/analysis/IcebergDeleteImpl.java
+++ b/fe/src/main/java/org/apache/impala/analysis/IcebergDeleteImpl.java
@@ -21,6 +21,7 @@ import java.util.List;
import org.apache.iceberg.TableProperties;
import org.apache.impala.catalog.FeIcebergTable;
+import org.apache.impala.catalog.IcebergContentFileStore;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.common.Pair;
import org.apache.impala.planner.DataSink;
@@ -40,12 +41,26 @@ public class IcebergDeleteImpl extends IcebergModifyImpl {
@Override
public void analyze(Analyzer analyzer) throws AnalysisException {
super.analyze(analyzer);
- if (originalTargetTable_.getFormatVersion() > 2) {
+ if (originalTargetTable_.getFormatVersion() > 3) {
throw new AnalysisException(String.format(
"Impala does not support DELETE statements on Iceberg tables with format " +
"version %d", originalTargetTable_.getFormatVersion()));
}
+ if (originalTargetTable_.getFormatVersion() >= 3) {
+ IcebergContentFileStore fileStore = originalTargetTable_.getContentFileStore();
+ int positionDeletes = fileStore.getPositionDeleteFiles().size();
+ if (positionDeletes > 0) {
+ throw new AnalysisException(String.format(
+ "DELETE is not allowed on Iceberg format version 3 table '%s' as it "
+ + "has existing version 2 position delete file(s): "
+ + "%d position delete file(s). Run 'OPTIMIZE TABLE %s' to rewrite the "
+ + "files and remove deletion files before attempting DELETE.",
+ originalTargetTable_.getFullName(), positionDeletes,
+ originalTargetTable_.getFullName()));
+ }
+ }
+
// Make the virtual position delete table the new target table.
modifyStmt_.setTargetTable(icePosDelTable_);
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 03bd51267..3da814268 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
@@ -1173,14 +1173,14 @@ public interface FeIcebergTable extends FeFsTable {
TimeTravelSpec travelSpec) throws AnalysisException {
if (travelSpec == null) {
IcebergContentFileStore fileStore = table.getContentFileStore();
- return !fileStore.getPositionDeleteFiles().isEmpty()
- || !fileStore.getEqualityDeleteFiles().isEmpty();
+ return fileStore.hasDeletes();
} else {
try {
GroupedContentFiles groupedFiles =
IcebergUtil.getIcebergFiles(table, Lists.newArrayList(), travelSpec);
return !groupedFiles.positionDeleteFiles.isEmpty()
- || !groupedFiles.equalityDeleteFiles.isEmpty();
+ || !groupedFiles.equalityDeleteFiles.isEmpty()
+ || !groupedFiles.dataFileToDV.isEmpty();
} catch (TableLoadingException e) {
throw new AnalysisException("Failed to get record count of Iceberg V2 table: "
+ table.getFullName(), e);
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 e80605d5d..9083d729e 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergContentFileStore.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergContentFileStore.java
@@ -264,7 +264,10 @@ public class IcebergContentFileStore {
for (DeleteFile deleteFile : icebergFiles.equalityDeleteFiles) {
storeFile(deleteFile, fileDescMap, equalityDeleteFiles_);
}
- dataFileToDV_ = icebergFiles.dataFileToDV;
+ for (Map.Entry<Hash128, DeleteFile> entry : icebergFiles.dataFileToDV.entrySet()) {
+ dataFileToDV_.put(entry.getKey(),
+ IcebergUtil.createTIcebergDeletionVector(entry.getValue()));
+ }
}
// Creates a new instance with shared immutable members but
@@ -349,6 +352,12 @@ public class IcebergContentFileStore {
return equalityDeleteFiles_.getList();
}
+ public boolean hasDeletes() {
+ return !positionDeleteFiles_.fileDescList_.isEmpty()
+ || !equalityDeleteFiles_.fileDescList_.isEmpty()
+ || !dataFileToDV_.isEmpty();
+ }
+
public boolean hasMissingFile() {
return !missingFiles_.isEmpty();
}
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 08e8a2afa..38b5f4fd5 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergDeleteTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergDeleteTable.java
@@ -26,6 +26,7 @@ import org.apache.iceberg.Table;
import org.apache.impala.analysis.IcebergPartitionSpec;
import org.apache.impala.catalog.CatalogObject.ThriftObjectType;
import org.apache.impala.thrift.TCompressionCodec;
+import org.apache.impala.thrift.THdfsFileFormat;
import org.apache.impala.thrift.THdfsTable;
import org.apache.impala.thrift.TIcebergCatalog;
import org.apache.impala.thrift.TIcebergFileFormat;
@@ -86,6 +87,15 @@ public abstract class IcebergDeleteTable extends VirtualTable implements FeIcebe
TTableDescriptor desc =
baseTable_.toThriftDescriptor(tableId, referencedPartitions);
desc.setColumnDescriptors(getTColumnDescriptors());
+ // PUFFIN format is used for deletion vectors, which are only supported in Iceberg
+ // format version 3 and above. For earlier format versions, we leave the file format
+ // unchanged to avoid misrepresenting the underlying file format.
+ if (baseTable_.getFormatVersion() > 2) {
+ desc.getHdfsTable().partitions.forEach((aLong, tHdfsPartition) ->
+ tHdfsPartition.hdfs_storage_descriptor.fileFormat = THdfsFileFormat.PUFFIN);
+ desc.getHdfsTable().prototype_partition.hdfs_storage_descriptor.fileFormat =
+ THdfsFileFormat.PUFFIN;
+ }
if (desc.hdfsTable.isSetAvroSchema()) {
desc.hdfsTable.setAvroSchema(AvroSchemaConverter.convertColumns(getColumns(),
getFullName().replaceAll("-", "_")).toString());
diff --git a/fe/src/main/java/org/apache/impala/catalog/iceberg/GroupedContentFiles.java b/fe/src/main/java/org/apache/impala/catalog/iceberg/GroupedContentFiles.java
index 74a37f240..d81b957d7 100644
--- a/fe/src/main/java/org/apache/impala/catalog/iceberg/GroupedContentFiles.java
+++ b/fe/src/main/java/org/apache/impala/catalog/iceberg/GroupedContentFiles.java
@@ -32,8 +32,6 @@ import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.io.CloseableIterable;
-import org.apache.impala.thrift.TIcebergDeletionVector;
-
import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;
import org.apache.impala.util.Hash128;
@@ -54,7 +52,7 @@ public class GroupedContentFiles {
public List<DataFile> dataFilesWithDeletes = new ArrayList<>();
public Set<DeleteFile> positionDeleteFiles = new HashSet<>();
public Set<DeleteFile> equalityDeleteFiles = new HashSet<>();
- public Map<Hash128, TIcebergDeletionVector> dataFileToDV = new HashMap<>();
+ public Map<Hash128, DeleteFile> dataFileToDV = new HashMap<>();
public GroupedContentFiles() { }
@@ -68,10 +66,14 @@ public class GroupedContentFiles {
for (DeleteFile delFile : scanTask.deletes()) {
if (delFile.content() == FileContent.POSITION_DELETES) {
if (delFile.format() == FileFormat.PUFFIN) {
- Preconditions.checkState(scanTask.deletes().size() == 1);
- dataFileToDV.put(
- IcebergUtil.getFilePathHash(dataFile),
- IcebergUtil.createTIcebergDeletionVector(delFile));
+ // A data file can have at most one DV, but may also be covered by legacy
+ // position-delete files simultaneously (e.g. after a partial V3 migration
+ // by an external engine). Assert there is no second DV for this file.
+ Hash128 dataFileHash = IcebergUtil.getFilePathHash(dataFile);
+ DeleteFile existing = dataFileToDV.put(dataFileHash, delFile);
+ Preconditions.checkState(existing == null,
+ "More than one deletion vector found for data file: %s",
+ dataFile.location());
} else {
positionDeleteFiles.add(delFile);
}
diff --git a/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java b/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
index 1e97b5f17..b8b8e6956 100644
--- a/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
@@ -153,9 +153,9 @@ public class DistributedPlanner {
ctx_.getNextFragmentId(), root, DataPartition.UNPARTITIONED);
} else if (root instanceof CardinalityCheckNode) {
result = createCardinalityCheckNodeFragment((CardinalityCheckNode) root, childFragments);
- } else if (root instanceof IcebergDeleteNode) {
+ } else if (root instanceof IcebergDeleteJoinNode) {
Preconditions.checkState(childFragments.size() == 2);
- result = createIcebergDeleteFragment((IcebergDeleteNode) root,
+ result = createIcebergDeleteFragment((IcebergDeleteJoinNode) root,
childFragments.get(0), childFragments.get(1));
} else if (root instanceof IcebergMergeNode) {
childFragments.get(0).addPlanRoot(root);
@@ -677,7 +677,7 @@ public class DistributedPlanner {
* Similarly to a BROADCAST join, the left child of the join is in the same fragment
* with the join itself, while the right child is in a separate fragment.
*/
- private PlanFragment createIcebergDeleteFragment(IcebergDeleteNode node,
+ private PlanFragment createIcebergDeleteFragment(IcebergDeleteJoinNode node,
PlanFragment leftChildFragment, PlanFragment rightChildFragment)
throws ImpalaException {
Preconditions.checkState(node.getEqJoinConjuncts().size() == 2);
diff --git a/fe/src/main/java/org/apache/impala/planner/ExchangeNode.java b/fe/src/main/java/org/apache/impala/planner/ExchangeNode.java
index cb0729f44..0705c9ba2 100644
--- a/fe/src/main/java/org/apache/impala/planner/ExchangeNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/ExchangeNode.java
@@ -112,7 +112,7 @@ public class ExchangeNode extends PlanNode {
}
protected boolean isChildOfIcebergDeleteNode(PlanNode currNode) {
- if (currNode instanceof IcebergDeleteNode) {
+ if (currNode instanceof IcebergDeleteJoinNode) {
Preconditions.checkState(currNode.getChildCount() == 2);
if (currNode.getChild(1) == this) return true;
}
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 e0a7c443d..9822b1c7c 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
@@ -1574,10 +1574,7 @@ public class HdfsScanNode extends ScanNode {
hdfsFileSplit.setIs_encrypted(fileDesc.getIsEncrypted());
hdfsFileSplit.setIs_erasure_coded(fileDesc.getIsEc());
scanRange.setHdfs_file_split(hdfsFileSplit);
- if (fileDesc instanceof IcebergFileDescriptor) {
- scanRange.setFile_metadata(
- ((IcebergFileDescriptor)fileDesc).getFbFileMetadata().getByteBuffer());
- }
+ decorateScanRange(fileDesc, scanRange);
TScanRangeLocationList scanRangeLocations = new TScanRangeLocationList();
scanRangeLocations.scan_range = scanRange;
scanRangeLocations.locations = locations;
@@ -1601,6 +1598,8 @@ public class HdfsScanNode extends ScanNode {
return new Pair<Boolean, Long>(fileDescMissingDiskIds, fileMaxScanRangeBytes);
}
+ protected void decorateScanRange(FileDescriptor fileDesc, TScanRange scanRange) {}
+
/**
* Computes the average row size, input and output cardinalities, and estimates the
* number of nodes.
diff --git a/fe/src/main/java/org/apache/impala/planner/IcebergBufferedDeleteSink.java b/fe/src/main/java/org/apache/impala/planner/IcebergBufferedDeleteSink.java
index 660f84c33..0d3bc47c1 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergBufferedDeleteSink.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergBufferedDeleteSink.java
@@ -17,26 +17,35 @@
package org.apache.impala.planner;
+import com.google.common.base.Preconditions;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.stream.Collectors;
import org.apache.impala.analysis.DescriptorTable;
import org.apache.impala.analysis.Expr;
import org.apache.impala.catalog.FeIcebergTable;
+import org.apache.impala.catalog.IcebergDeleteTable;
import org.apache.impala.common.ByteUnits;
import org.apache.impala.thrift.TDataSink;
import org.apache.impala.thrift.TDataSinkType;
import org.apache.impala.thrift.TExplainLevel;
+import org.apache.impala.thrift.THash128;
import org.apache.impala.thrift.TIcebergDeleteSink;
+import org.apache.impala.thrift.TIcebergDeletionVector;
import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.thrift.TTableSink;
import org.apache.impala.thrift.TTableSinkType;
import java.util.List;
+import org.apache.impala.util.Hash128;
public class IcebergBufferedDeleteSink extends TableSink {
- final private int deleteTableId_;
+ private final int deleteTableId_;
// Exprs for computing the output partition(s).
protected final List<Expr> partitionKeyExprs_;
+ protected final Map<Hash128, TIcebergDeletionVector> referencedDVs_;
public IcebergBufferedDeleteSink(FeIcebergTable targetTable,
List<Expr> partitionKeyExprs, List<Expr> outputExprs) {
@@ -47,6 +56,9 @@ public class IcebergBufferedDeleteSink extends TableSink {
List<Expr> partitionKeyExprs, List<Expr> outputExprs,
int deleteTableId) {
super(targetTable, Op.DELETE, outputExprs);
+ Preconditions.checkState(targetTable instanceof IcebergDeleteTable);
+ FeIcebergTable originalTable = ((IcebergDeleteTable) targetTable).getBaseTable();
+ referencedDVs_ = originalTable.getContentFileStore().getDataFileToDV();
partitionKeyExprs_ = partitionKeyExprs;
deleteTableId_ = deleteTableId;
}
@@ -113,6 +125,12 @@ public class IcebergBufferedDeleteSink extends TableSink {
protected void toThriftImpl(TDataSink tsink) {
TIcebergDeleteSink icebergDeleteSink = new TIcebergDeleteSink();
icebergDeleteSink.setPartition_key_exprs(Expr.treesToThrift(partitionKeyExprs_));
+ Map<THash128, TIcebergDeletionVector> thriftDVs = referencedDVs_.entrySet().stream()
+ .collect(Collectors.toMap(
+ entry -> entry.getKey().toThrift(),
+ Entry::getValue
+ ));
+ icebergDeleteSink.setDeletion_vectors(thriftDVs);
TTableSink tTableSink = new TTableSink(DescriptorTable.TABLE_SINK_ID,
TTableSinkType.HDFS, sinkOp_.toThrift());
tTableSink.iceberg_delete_sink = icebergDeleteSink;
diff --git a/fe/src/main/java/org/apache/impala/planner/IcebergDeleteNode.java b/fe/src/main/java/org/apache/impala/planner/IcebergDeleteJoinNode.java
similarity index 96%
rename from fe/src/main/java/org/apache/impala/planner/IcebergDeleteNode.java
rename to fe/src/main/java/org/apache/impala/planner/IcebergDeleteJoinNode.java
index 198630b3e..91ecaffad 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergDeleteNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergDeleteJoinNode.java
@@ -44,12 +44,18 @@ import com.google.common.base.MoreObjects;
import com.google.common.base.Preconditions;
import org.apache.impala.util.ExprUtil;
-public class IcebergDeleteNode extends JoinNode {
- private final static Logger LOG = LoggerFactory.getLogger(IcebergDeleteNode.class);
- public IcebergDeleteNode(
- PlanNode outer, PlanNode inner, List<BinaryPredicate> eqJoinConjuncts) {
+public class IcebergDeleteJoinNode extends JoinNode {
+ private final static Logger LOG = LoggerFactory.getLogger(IcebergDeleteJoinNode.class);
+
+ // Record count from deletion vectors associated with the scanned data files.
+ private final long dvDeletesRecordCount_;
+
+ public IcebergDeleteJoinNode(
+ PlanNode outer, PlanNode inner, List<BinaryPredicate> eqJoinConjuncts,
+ long dvDeletesRecordCount) {
super(outer, inner, true, DistributionMode.NONE, JoinOperator.ICEBERG_DELETE_JOIN,
eqJoinConjuncts, Collections.emptyList(), "ICEBERG DELETE");
+ dvDeletesRecordCount_ = dvDeletesRecordCount;
Preconditions.checkNotNull(eqJoinConjuncts);
Preconditions.checkState(joinOp_ == JoinOperator.ICEBERG_DELETE_JOIN);
Preconditions.checkState(conjuncts_.isEmpty());
@@ -101,6 +107,7 @@ public class IcebergDeleteNode extends JoinNode {
TableSampleClause leftSampleParams = leftScanChild.getSampleParams();
long leftCardWithSelectivity = leftScanChild.cardinality_;
long rightCard = getChild(1).cardinality_;
+ rightCard = MathUtil.addCardinalities(rightCard, dvDeletesRecordCount_);
// Both sides should have non-negative cardinalities.
Preconditions.checkState(leftCardWithSelectivity >= 0);
Preconditions.checkState(rightCard >= 0);
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 d6a67a0bd..5ea0d4602 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
@@ -17,6 +17,8 @@
package org.apache.impala.planner;
+import static org.apache.impala.util.IcebergUtil.getFilePathHash;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -39,8 +41,14 @@ import org.apache.impala.catalog.Type;
import org.apache.impala.common.ImpalaRuntimeException;
import org.apache.impala.common.ThriftSerializationCtx;
import org.apache.impala.fb.FbIcebergDataFileFormat;
+import org.apache.impala.fb.FbIcebergDeletionVector;
import org.apache.impala.thrift.TExplainLevel;
+import org.apache.impala.thrift.THash128;
+import org.apache.impala.thrift.TIcebergDeletionVector;
import org.apache.impala.thrift.TPlanNode;
+import org.apache.impala.thrift.TScanRange;
+import org.apache.impala.util.Hash128;
+import org.apache.impala.util.IcebergUtil;
import org.apache.impala.util.MathUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,6 +66,8 @@ public class IcebergScanNode extends HdfsScanNode {
// See IMPALA-12765 for details.
private List<IcebergFileDescriptor> fileDescs_;
+ private final Map<Hash128, TIcebergDeletionVector> dataFileToDV_;
+
// Number of actual partitions in the table, inferred from file metadata.
// We treat Iceberg tables as unpartitioned HDFS tables, so instead of storing the file
// descriptors per partition in the partition map in FeFsTable, we keep track of
@@ -90,16 +100,18 @@ public class IcebergScanNode extends HdfsScanNode {
MultiAggregateInfo aggInfo, List<IcebergFileDescriptor> fileDescs,
int numPartitions,
List<Expr> nonIdentityConjuncts, List<Expr> skippedConjuncts, long snapshotId,
- boolean isPartitionKeyScan) {
+ boolean isPartitionKeyScan, Map<Hash128, TIcebergDeletionVector> dataFileToDV) {
this(id, tblRef, conjuncts, aggInfo, fileDescs, numPartitions, nonIdentityConjuncts,
- skippedConjuncts, null, snapshotId, isPartitionKeyScan);
+ skippedConjuncts, null, snapshotId, isPartitionKeyScan, dataFileToDV);
}
public IcebergScanNode(PlanNodeId id, TableRef tblRef, List<Expr> conjuncts,
MultiAggregateInfo aggInfo, List<IcebergFileDescriptor> fileDescs,
int numPartitions,
- List<Expr> nonIdentityConjuncts, List<Expr> skippedConjuncts, PlanNodeId deleteId,
- long snapshotId, boolean isPartitionKeyScan) {
+ List<Expr> nonIdentityConjuncts, List<Expr> skippedConjuncts,
+ PlanNodeId deleteId,
+ long snapshotId, boolean isPartitionKeyScan,
+ Map<Hash128, TIcebergDeletionVector> dataFileToDV) {
super(id, tblRef.getDesc(), conjuncts,
getIcebergPartition(((FeIcebergTable)tblRef.getTable()).getFeFsTable()), tblRef,
aggInfo, null, isPartitionKeyScan);
@@ -122,6 +134,7 @@ public class IcebergScanNode extends HdfsScanNode {
snapshotId_ = snapshotId;
this.skippedConjuncts_ = skippedConjuncts;
this.deleteFileScanNodeId = deleteId;
+ this.dataFileToDV_ = dataFileToDV;
}
/**
@@ -296,4 +309,23 @@ public class IcebergScanNode extends HdfsScanNode {
if (hasOrc) fileFormats_.add(HdfsFileFormat.ORC);
if (hasAvro) fileFormats_.add(HdfsFileFormat.AVRO);
}
+ @Override
+ protected void decorateScanRange(FileDescriptor fileDesc, TScanRange scanRange) {
+ Preconditions.checkState(fileDesc instanceof IcebergFileDescriptor);
+ scanRange.setFile_metadata(
+ ((IcebergFileDescriptor)fileDesc).getFbFileMetadata().getByteBuffer());
+ if (dataFileToDV_ != null) {
+ String absPath = fileDesc.getAbsolutePath(tbl_.getLocation());
+ Hash128 pathHash = getFilePathHash(absPath);
+ TIcebergDeletionVector delVec = dataFileToDV_.get(pathHash);
+ if (delVec != null) {
+ long recordCount = delVec.isSetRecord_count() ? delVec.getRecord_count() : 0;
+ FbIcebergDeletionVector fbIceDelVec = IcebergUtil.createFbIcebergDeletionVector(
+ delVec.getPath(), delVec.getContent_offset(),
+ delVec.getContent_size_in_bytes(), pathHash.getHigh(),
+ pathHash.getLow(), recordCount);
+ scanRange.setIceberg_deletion_vector(fbIceDelVec.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 aa6aad43e..a543ce537 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
@@ -18,6 +18,8 @@
package org.apache.impala.planner;
import com.google.common.base.Joiner;
+import static org.apache.impala.util.IcebergUtil.getFilePathHash;
+
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import java.io.IOException;
@@ -149,7 +151,13 @@ public class IcebergScanPlanner {
// Statistics about the data and delete files. Useful for memory estimates of the
// ANTI JOIN
+ // Record count of legacy position-delete files only (not deletion vectors).
private long positionDeletesRecordCount_ = 0;
+ // Record count contributed solely by deletion vectors, tracked separately so it can
+ // be passed to IcebergDeleteJoinNode for cardinality estimation. The delete scan
+ // node's fileDescs_ only contains legacy position-delete files, so its cardinality_
+ // misses DV-deleted rows; the join node adds dvDeletesRecordCount_ to compensate.
+ private long dvDeletesRecordCount_ = 0;
private long dataFilesWithDeletesSumPaths_ = 0;
private long dataFilesWithDeletesMaxPath_ = 0;
// Stores how many delete records are involved broken down by equality field ID lists.
@@ -268,18 +276,15 @@ public class IcebergScanPlanner {
aggInfo_, dataFilesWithoutDeletes_,
getIceTable().getContentFileStore().getNumPartitions(),
nonIdentityConjuncts_,
- getSkippedConjuncts(), snapshotId_, isPartitionKeyScan);
+ getSkippedConjuncts(), snapshotId_, isPartitionKeyScan, dataFileToDV_);
ret.init(analyzer_);
return ret;
}
- if (!dataFileToDV_.isEmpty()) {
- throw new ImpalaRuntimeException(
- "Iceberg tables with Deletion Vectors are not supported yet: " +
- getIceTable().getFullName());
- }
PlanNode joinNode = null;
- if (!positionDeleteFiles_.isEmpty()) joinNode = createPositionJoinNode();
+ if (!positionDeleteFiles_.isEmpty() || !dataFileToDV_.isEmpty()) {
+ joinNode = createPositionJoinNode();
+ }
if (!equalityIdsToDeleteFiles_.isEmpty()) joinNode = createEqualityJoinNode(joinNode);
Preconditions.checkNotNull(joinNode);
@@ -298,7 +303,8 @@ public class IcebergScanPlanner {
IcebergScanNode dataScanNode = new IcebergScanNode(
ctx_.getNextNodeId(), tblRef_, conjuncts_, aggInfo_, dataFilesWithoutDeletes_,
getIceTable().getContentFileStore().getNumPartitions(),
- nonIdentityConjuncts_, getSkippedConjuncts(), snapshotId_, isPartitionKeyScan);
+ nonIdentityConjuncts_, getSkippedConjuncts(), snapshotId_,
+ isPartitionKeyScan, dataFileToDV_);
dataScanNode.init(analyzer_);
List<Expr> outputExprs = tblRef_.getDesc().getSlots().stream().map(
SlotRef::new).collect(Collectors.toList());
@@ -360,7 +366,8 @@ public class IcebergScanPlanner {
}
private PlanNode createPositionJoinNode() throws ImpalaException {
- Preconditions.checkState(positionDeletesRecordCount_ != 0);
+ Preconditions.checkState(
+ positionDeletesRecordCount_ != 0 || !dataFileToDV_.isEmpty());
Preconditions.checkState(dataFilesWithDeletesSumPaths_ != 0);
Preconditions.checkState(dataFilesWithDeletesMaxPath_ != 0);
// The followings just create separate scan nodes for data files and position delete
@@ -381,7 +388,7 @@ public class IcebergScanPlanner {
dataScanNodeId, tblRef_, conjuncts_, aggInfo_, dataFilesWithDeletes_,
getIceTable().getContentFileStore().getNumPartitions(),
nonIdentityConjuncts_, getSkippedConjuncts(), deleteScanNodeId, snapshotId_,
- false /*isPartitionKeyScan*/);
+ false /*isPartitionKeyScan*/, dataFileToDV_);
dataScanNode.init(analyzer_);
IcebergScanNode deleteScanNode = new IcebergScanNode(
deleteScanNodeId,
@@ -393,7 +400,7 @@ public class IcebergScanPlanner {
Collections.emptyList(), /*nonIdentityConjuncts*/
Collections.emptyList(), /*skippedConjuncts*/
snapshotId_,
- false /*isPartitionKeyScan*/);
+ false /*isPartitionKeyScan*/, dataFileToDV_);
deleteScanNode.init(analyzer_);
// Now let's create the JOIN node
@@ -402,13 +409,15 @@ public class IcebergScanPlanner {
TQueryOptions queryOpts = analyzer_.getQueryCtx().client_request.query_options;
JoinNode joinNode = null;
- if (queryOpts.disable_optimized_iceberg_v2_read) {
+ if (queryOpts.disable_optimized_iceberg_v2_read
+ && getIceTable().getFormatVersion() == 2) {
joinNode = new HashJoinNode(dataScanNode, deleteScanNode,
/*straight_join=*/true, DistributionMode.NONE, JoinOperator.LEFT_ANTI_JOIN,
positionJoinConjuncts, /*otherJoinConjuncts=*/Collections.emptyList());
} else {
joinNode =
- new IcebergDeleteNode(dataScanNode, deleteScanNode, positionJoinConjuncts);
+ new IcebergDeleteJoinNode(dataScanNode, deleteScanNode, positionJoinConjuncts,
+ dvDeletesRecordCount_);
}
joinNode.setId(ctx_.getNextNodeId());
joinNode.init(analyzer_);
@@ -596,7 +605,7 @@ public class IcebergScanPlanner {
dataScanNodeId, tblRef_, conjuncts_, aggInfo_, dataFilesWithDeletes_,
getIceTable().getContentFileStore().getNumPartitions(),
nonIdentityConjuncts_, getSkippedConjuncts(), snapshotId_,
- false /*isPartitionKeyScan*/);
+ false /*isPartitionKeyScan*/, dataFileToDV_);
addAllSlotsForEqualityDeletes(tblRef_);
dataScanNode.init(analyzer_);
@@ -637,8 +646,7 @@ public class IcebergScanPlanner {
getIceTable().getContentFileStore().getNumPartitions(),
Collections.emptyList(), /*nonIdentityConjuncts*/
Collections.emptyList(), /*skippedConjuncts*/
- snapshotId_,
- false /*isPartitionKeyScan*/);
+ snapshotId_, false /*isPartitionKeyScan*/, dataFileToDV_);
deleteScanNode.init(analyzer_);
Pair<List<BinaryPredicate>, List<Expr>> equalityJoinConjuncts =
@@ -726,10 +734,15 @@ public class IcebergScanPlanner {
addEqualityDeletesAndIds(delFileDesc.first);
} else {
if (delFile.format() == FileFormat.PUFFIN) {
- Preconditions.checkState(fileScanTask.deletes().size() == 1);
- dataFileToDV_.put(
- IcebergUtil.getFilePathHash(dataFile.location()),
+ // A data file can have at most one DV, but may also be covered by legacy
+ // position-delete files simultaneously (e.g. after a partial V3 migration
+ // by an external engine). Assert there is no second DV for this file.
+ Hash128 dataFileHash = IcebergUtil.getFilePathHash(dataFile.location());
+ TIcebergDeletionVector dv = dataFileToDV_.put(dataFileHash,
IcebergUtil.createTIcebergDeletionVector(delFile));
+ Preconditions.checkState(dv == null,
+ "More than one deletion vector found for data file: %s",
+ dataFile.location());
} else {
positionDeleteFiles_.add(delFileDesc.first);
}
@@ -829,6 +842,15 @@ public class IcebergScanPlanner {
equalityIdsToDeleteFiles_.entrySet()) {
updateEqualityDeleteFilesStatistics(entry.getKey(), entry.getValue());
}
+ updateDeletionVectorStatistics();
+ }
+
+ private void updateDeletionVectorStatistics() {
+ for (TIcebergDeletionVector dv : dataFileToDV_.values()) {
+ if (dv.isSetRecord_count()) {
+ dvDeletesRecordCount_ += dv.getRecord_count();
+ }
+ }
}
private void updateDataFilesWithDeletesStatistics(IcebergFileDescriptor fd) {
diff --git a/fe/src/main/java/org/apache/impala/planner/JoinBuildSink.java b/fe/src/main/java/org/apache/impala/planner/JoinBuildSink.java
index 227ae4919..f5b63c6b9 100644
--- a/fe/src/main/java/org/apache/impala/planner/JoinBuildSink.java
+++ b/fe/src/main/java/org/apache/impala/planner/JoinBuildSink.java
@@ -77,7 +77,7 @@ public class JoinBuildSink extends DataSink implements SpillableOperator {
tBuildSink.setDest_node_id(joinNode_.getId().asInt());
tBuildSink.setJoin_op(joinNode_.getJoinOp().toThrift());
if (joinNode_ instanceof HashJoinNode ||
- joinNode_ instanceof IcebergDeleteNode) {
+ joinNode_ instanceof IcebergDeleteJoinNode) {
tBuildSink.setEq_join_conjuncts(
joinNode_.getThriftEquiJoinConjuncts(new ThriftSerializationCtx()));
tBuildSink.setHash_seed(joinNode_.getFragment().getHashSeed());
@@ -97,7 +97,7 @@ public class JoinBuildSink extends DataSink implements SpillableOperator {
} else if (joinNode_ instanceof NestedLoopJoinNode) {
return TDataSinkType.NESTED_LOOP_JOIN_BUILDER;
} else {
- Preconditions.checkState(joinNode_ instanceof IcebergDeleteNode);
+ Preconditions.checkState(joinNode_ instanceof IcebergDeleteJoinNode);
return TDataSinkType.ICEBERG_DELETE_BUILDER;
}
}
diff --git a/fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java
index 3904dc5c0..59e2ce3fb 100644
--- a/fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java
@@ -18,10 +18,13 @@
package org.apache.impala.service;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Collections;
+import java.util.Set;
import java.util.concurrent.ExecutorService;
import org.apache.iceberg.AppendFiles;
@@ -31,6 +34,7 @@ import org.apache.iceberg.DataFiles;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.DeleteFiles;
import org.apache.iceberg.ExpireSnapshots;
+import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileMetadata;
import org.apache.iceberg.ManageSnapshots;
import org.apache.iceberg.Metrics;
@@ -52,6 +56,7 @@ import org.apache.impala.analysis.IcebergPartitionSpec;
import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeIcebergTable;
+import org.apache.impala.catalog.IcebergFileDescriptor;
import org.apache.impala.catalog.IcebergTable;
import org.apache.impala.catalog.TableLoadingException;
import org.apache.impala.catalog.TableNotFoundException;
@@ -64,15 +69,19 @@ import org.apache.impala.catalog.iceberg.ImpalaRepairIcebergTable;
import org.apache.impala.common.ImpalaRuntimeException;
import org.apache.impala.fb.FbIcebergColumnStats;
import org.apache.impala.fb.FbIcebergDataFile;
+import org.apache.impala.fb.FbIcebergDataFileFormat;
+import org.apache.impala.fb.FbIcebergDeletionVector;
import org.apache.impala.thrift.TAlterTableDropPartitionParams;
import org.apache.impala.thrift.TAlterTableExecuteExpireSnapshotsParams;
import org.apache.impala.thrift.TAlterTableExecuteRemoveOrphanFilesParams;
import org.apache.impala.thrift.TAlterTableExecuteRollbackParams;
import org.apache.impala.thrift.TColumn;
+import org.apache.impala.thrift.THash128;
import org.apache.impala.thrift.TIcebergCatalog;
import org.apache.impala.thrift.TIcebergOperationParam;
import org.apache.impala.thrift.TIcebergPartitionSpec;
import org.apache.impala.thrift.TRollbackType;
+import org.apache.impala.util.Hash128;
import org.apache.impala.util.IcebergSchemaConverter;
import org.apache.impala.util.IcebergUtil;
import org.apache.thrift.TException;
@@ -407,11 +416,15 @@ public class IcebergCatalogOpExecutor {
private static void deleteRows(FeIcebergTable feIcebergTable, Transaction txn,
TIcebergOperationParam icebergOp) throws ImpalaRuntimeException {
List<ByteBuffer> deleteFilesFb = icebergOp.getIceberg_delete_files_fb();
+ List<ByteBuffer> addedDeletionVectorsFb =
+ icebergOp.getIceberg_added_deletion_vectors_fb();
+ List<ByteBuffer> removedDeletionVectorsFb =
+ icebergOp.getIceberg_removed_deletion_vectors_fb();
RowDelta rowDelta = txn.newRowDelta();
- for (ByteBuffer buf : deleteFilesFb) {
- DeleteFile deleteFile = createDeleteFile(feIcebergTable, buf);
- rowDelta.addDeletes(deleteFile);
- }
+
+ applyDeletes(feIcebergTable, removedDeletionVectorsFb, rowDelta,
+ addedDeletionVectorsFb, deleteFilesFb);
+
// Validate that there are no conflicting data files, because if data files are
// added in the meantime, they potentially contain records that should have been
// affected by this DELETE operation.
@@ -423,6 +436,84 @@ public class IcebergCatalogOpExecutor {
rowDelta.commit();
}
+ private static void applyDeletes(FeIcebergTable feIcebergTable,
+ List<ByteBuffer> removedDeletionVectorsFb, RowDelta rowDelta,
+ List<ByteBuffer> addedDeletionVectorsFb, List<ByteBuffer> deleteFilesFb)
+ throws ImpalaRuntimeException {
+
+ if ((removedDeletionVectorsFb != null && !removedDeletionVectorsFb.isEmpty())
+ || (addedDeletionVectorsFb != null && !addedDeletionVectorsFb.isEmpty())) {
+ // DV operation: deleteFilesFb must not be set (the two are mutually exclusive).
+ Preconditions.checkState(deleteFilesFb == null || deleteFilesFb.isEmpty(),
+ "iceberg_delete_files_fb must be empty when deletion vectors are present");
+ applyDeletionVectors(feIcebergTable, removedDeletionVectorsFb, rowDelta,
+ addedDeletionVectorsFb);
+ } else {
+ for (ByteBuffer buf : deleteFilesFb) {
+ rowDelta.addDeletes(createDeleteFile(feIcebergTable, buf));
+ }
+ }
+ }
+
+ private static void applyDeletionVectors(FeIcebergTable feIcebergTable,
+ List<ByteBuffer> removedDeletionVectorsFb, RowDelta rowDelta,
+ List<ByteBuffer> addedDeletionVectorsFb) throws ImpalaRuntimeException {
+ if (removedDeletionVectorsFb != null) {
+ for (ByteBuffer buf : removedDeletionVectorsFb) {
+ FbIcebergDeletionVector dv =
+ FbIcebergDeletionVector.getRootAsFbIcebergDeletionVector(buf);
+ DeleteFile deleteFile = createDeletionVector(feIcebergTable, dv);
+ rowDelta.removeDeletes(deleteFile);
+ }
+ }
+
+ if (addedDeletionVectorsFb != null) {
+ for (ByteBuffer buf : addedDeletionVectorsFb) {
+ FbIcebergDeletionVector dv =
+ FbIcebergDeletionVector.getRootAsFbIcebergDeletionVector(buf);
+ DeleteFile deleteFile = createDeletionVector(feIcebergTable, dv);
+ rowDelta.addDeletes(deleteFile);
+ }
+ }
+ }
+
+ /**
+ * Creates a DeleteFile from a FlatBuffer-encoded deletion vector.
+ * Used for adding/removing individual deletion vectors from Iceberg metadata.
+ */
+ private static DeleteFile createDeletionVector(
+ FeIcebergTable feIcebergTable, FbIcebergDeletionVector dv)
+ throws ImpalaRuntimeException {
+
+ // Get the referenced data file path hash
+ THash128 referencedDataFileHash = new THash128(
+ dv.referencedDataFileHashHigh(), dv.referencedDataFileHashLow());
+ IcebergFileDescriptor dataFileDescriptor = feIcebergTable.getContentFileStore()
+ .getDataFileDescriptor(Hash128.fromThrift(referencedDataFileHash));
+ if (dataFileDescriptor == null) {
+ throw new ImpalaRuntimeException(
+ "Could not find data file for deletion vector with hash: " +
+ referencedDataFileHash);
+ }
+ String referencedDataFile = dataFileDescriptor
+ .getAbsolutePath(feIcebergTable.getLocation());
+
+ PartitionSpec partSpec = feIcebergTable.getIcebergApiTable().spec();
+
+ // Build the delete file metadata
+ FileMetadata.Builder builder = FileMetadata.deleteFileBuilder(partSpec)
+ .ofPositionDeletes()
+ .withPath(dv.path())
+ .withFormat(FileFormat.PUFFIN)
+ .withContentOffset(dv.contentOffset())
+ .withContentSizeInBytes(dv.contentSizeInBytes())
+ .withReferencedDataFile(referencedDataFile)
+ .withRecordCount(dv.recordCount())
+ .withFileSizeInBytes(dv.fileSizeInBytes());
+
+ return builder.build();
+ }
+
private static void updateRows(FeIcebergTable feIcebergTable, Transaction txn,
TIcebergOperationParam icebergOp) throws ImpalaRuntimeException {
List<ByteBuffer> deleteFilesFb = icebergOp.getIceberg_delete_files_fb();
@@ -491,6 +582,10 @@ public class IcebergCatalogOpExecutor {
IcebergUtil.PartitionData partitionData = IcebergUtil.partitionDataFromDataFile(
partSpec.partitionType(), impPartSpec, deleteFile);
if (partitionData != null) builder.withPartition(partitionData);
+ if (deleteFile.format() == FbIcebergDataFileFormat.PUFFIN) {
+ throw new ImpalaRuntimeException(
+ "Unexpected PUFFIN format in non-DV delete file path: " + deleteFile.path());
+ }
return builder.build();
}
@@ -574,6 +669,9 @@ public class IcebergCatalogOpExecutor {
for (DeleteFile deleteFile : contentFiles.positionDeleteFiles) {
rewrite.deleteFile(deleteFile);
}
+ for (DeleteFile dvDeleteFile : contentFiles.dataFileToDV.values()) {
+ rewrite.deleteFile(dvDeleteFile);
+ }
for (DeleteFile deleteFile : contentFiles.equalityDeleteFiles) {
rewrite.deleteFile(deleteFile);
}
@@ -624,6 +722,27 @@ public class IcebergCatalogOpExecutor {
if (icebergOp.isSetIceberg_delete_files_fb()) {
deleteIcebergFiles(icebergOp.getIceberg_delete_files_fb());
}
+ // For DV operations iceberg_delete_files_fb is not set; clean up puffin files
+ // referenced by the added deletion vectors instead.
+ if (icebergOp.isSetIceberg_added_deletion_vectors_fb()) {
+ deleteDvPuffinFiles(icebergOp.getIceberg_added_deletion_vectors_fb());
+ }
+ }
+
+ /**
+ * Deletes the puffin files referenced by the given deletion vector FlatBuffers.
+ * Multiple deletion vectors may share a single puffin file, so paths are deduplicated.
+ */
+ private static void deleteDvPuffinFiles(Iterable<ByteBuffer> dvFbs) {
+ Set<String> paths = new HashSet<>();
+ for (ByteBuffer buf : dvFbs) {
+ FbIcebergDeletionVector dv =
+ FbIcebergDeletionVector.getRootAsFbIcebergDeletionVector(buf);
+ if (dv.path() != null) paths.add(dv.path());
+ }
+ for (String path : paths) {
+ deleteIfExists(new org.apache.hadoop.fs.Path(path));
+ }
}
private static void deleteIcebergFiles(Iterable<ByteBuffer> icebergFiles) {
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 ba9ba4c2b..cc42f7fc3 100644
--- a/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
+++ b/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
@@ -112,6 +112,7 @@ import org.apache.impala.common.Pair;
import org.apache.impala.fb.FbFileMetadata;
import org.apache.impala.fb.FbIcebergDataFile;
import org.apache.impala.fb.FbIcebergDataFileFormat;
+import org.apache.impala.fb.FbIcebergDeletionVector;
import org.apache.impala.fb.FbIcebergMetadata;
import org.apache.impala.fb.FbIcebergPartitionTransformValue;
import org.apache.impala.fb.FbIcebergTransformType;
@@ -759,6 +760,8 @@ public class IcebergUtil {
switch (fbFileFormat){
case org.apache.impala.fb.FbIcebergDataFileFormat.PARQUET:
return org.apache.iceberg.FileFormat.PARQUET;
+ case FbIcebergDataFileFormat.PUFFIN:
+ return FileFormat.PUFFIN;
default:
throw new ImpalaRuntimeException(String.format("Unexpected file format: %s",
org.apache.impala.fb.FbIcebergDataFileFormat.name(fbFileFormat)));
@@ -1146,7 +1149,7 @@ public class IcebergUtil {
*/
public static FbFileMetadata createIcebergMetadata(
Table iceApiTbl, ContentFile cf, int partId) {
- FlatBufferBuilder fbb = new FlatBufferBuilder(1);
+ FlatBufferBuilder fbb = new FlatBufferBuilder();
int iceOffset = createIcebergMetadata(iceApiTbl, fbb, cf, partId);
fbb.finish(FbFileMetadata.createFbFileMetadata(fbb, iceOffset));
ByteBuffer bb = fbb.dataBuffer().slice();
@@ -1173,6 +1176,8 @@ public class IcebergUtil {
if (cf.format() == FileFormat.PARQUET) fileFormat = FbIcebergDataFileFormat.PARQUET;
else if (cf.format() == FileFormat.ORC) fileFormat = FbIcebergDataFileFormat.ORC;
else if (cf.format() == FileFormat.AVRO) fileFormat = FbIcebergDataFileFormat.AVRO;
+ else if (cf.format() == FileFormat.PUFFIN) fileFormat =
+ FbIcebergDataFileFormat.PUFFIN;
if (fileFormat != -1) {
FbIcebergMetadata.addFileFormat(fbb, fileFormat);
}
@@ -1352,6 +1357,24 @@ public class IcebergUtil {
return props;
}
+ public static FbIcebergDeletionVector createFbIcebergDeletionVector(
+ String path, long contentOffset, long contentSizeInBytes,
+ long referencedDataFileHashHigh, long referencedDataFileHashLow,
+ long recordCount) {
+ FlatBufferBuilder fbb = new FlatBufferBuilder();
+ int pathOffset = fbb.createString(path);
+
+ fbb.finish(
+ FbIcebergDeletionVector.createFbIcebergDeletionVector(fbb, pathOffset,
+ contentOffset, contentSizeInBytes, referencedDataFileHashHigh,
+ referencedDataFileHashLow, recordCount, 0));
+ ByteBuffer bb = fbb.dataBuffer().slice();
+ ByteBuffer compressedBb = ByteBuffer.allocate(bb.capacity());
+ compressedBb.put(bb);
+ return FbIcebergDeletionVector.getRootAsFbIcebergDeletionVector(
+ compressedBb.flip());
+ }
+
/**
* Checks the table before insert for unsupported types and file formats.
*/
@@ -1474,9 +1497,11 @@ public class IcebergUtil {
DeleteFile delFile) {
Preconditions.checkState(delFile.content() == FileContent.POSITION_DELETES);
Preconditions.checkState(delFile.format() == FileFormat.PUFFIN);
- return new TIcebergDeletionVector(
+ TIcebergDeletionVector dv = new TIcebergDeletionVector(
delFile.location(),
delFile.contentOffset(), delFile.contentSizeInBytes());
+ dv.setRecord_count(delFile.recordCount());
+ return dv;
}
/**
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete-existing-dv.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete-existing-dv.test
new file mode 100644
index 000000000..ec8bb5fd3
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete-existing-dv.test
@@ -0,0 +1,116 @@
+====
+---- QUERY
+# Verify the file structure of the Spark-written table: 5 data files and the
+# single Puffin DV file that Spark created (it covers 2 data files, marking rows
+# 2 and 4 as deleted). The Puffin file references two data files, so it appears
+# as 2 entries in the virtual 'files' table.
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.iceberg_v3_deletion_vectors.`files`
+ORDER BY content, file_format;
+---- RESULTS
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Verify initial visible rows: Spark DVs already deleted rows 2 and 4.
+SELECT * FROM iceberg_v3_deletion_vectors ORDER BY i;
+---- RESULTS
+1
+3
+5
+---- TYPES
+INT
+====
+---- QUERY
+# Delete row i=1 from the Spark-written DV table. Row 1 lives in a data file that
+# has no prior DV, so Impala creates a new Puffin DV entry for it.
+DELETE FROM iceberg_v3_deletion_vectors WHERE i = 1;
+---- RUNTIME_PROFILE
+NumModifiedRows: 0
+NumDeletedRows: 1
+====
+---- QUERY
+SELECT * FROM iceberg_v3_deletion_vectors ORDER BY i;
+---- RESULTS
+3
+5
+---- TYPES
+INT
+====
+---- QUERY
+# After Impala's first delete: the original Spark Puffin covers 2 data files (2
+# entries) and the new Impala Puffin covers 1 data file (1 entry) = 3 PUFFIN rows.
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.iceberg_v3_deletion_vectors.`files`
+ORDER BY content, file_format;
+---- RESULTS
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+1,'PUFFIN',true,true
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Delete row i=3. Impala creates another new Puffin entry for row 3's data file.
+DELETE FROM iceberg_v3_deletion_vectors WHERE i = 3;
+---- RUNTIME_PROFILE
+NumModifiedRows: 0
+NumDeletedRows: 1
+====
+---- QUERY
+SELECT * FROM iceberg_v3_deletion_vectors ORDER BY i;
+---- RESULTS
+5
+---- TYPES
+INT
+====
+---- QUERY
+# After two Impala deletes: 2 (Spark) + 2 (Impala) = 4 PUFFIN rows total.
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.iceberg_v3_deletion_vectors.`files`
+ORDER BY content, file_format;
+---- RESULTS
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+1,'PUFFIN',true,true
+1,'PUFFIN',true,true
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Compact the table: OPTIMIZE TABLE rewrites all 5 data files (merging out the 4
+# DV-deleted rows) into a single clean Parquet file with no deletion vectors.
+OPTIMIZE TABLE iceberg_v3_deletion_vectors;
+SELECT * FROM iceberg_v3_deletion_vectors ORDER BY i;
+---- RESULTS
+5
+---- TYPES
+INT
+====
+---- QUERY
+# After compaction: exactly 1 data file, no Puffin DV entries.
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.iceberg_v3_deletion_vectors.`files`
+ORDER BY content, file_format;
+---- RESULTS
+0,'PARQUET',false,false
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete-partition-sort.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete-partition-sort.test
new file mode 100644
index 000000000..15ab6f2d1
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete-partition-sort.test
@@ -0,0 +1,882 @@
+====
+---- QUERY
+# Identity partition - deletion vector created in the correct partition directory
+CREATE TABLE ice_v3_dv_id_part (id int, cat string, val int)
+PARTITIONED BY SPEC (cat)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_id_part VALUES
+ (1, 'A', 10), (2, 'A', 20), (3, 'B', 30), (4, 'B', 40), (5, 'C', 50);
+
+DELETE FROM ice_v3_dv_id_part WHERE cat = 'A' AND id = 1;
+SELECT * FROM ice_v3_dv_id_part ORDER BY id;
+---- RESULTS
+2,'A',20
+3,'B',30
+4,'B',40
+5,'C',50
+---- TYPES
+INT,STRING,INT
+====
+---- QUERY
+# After the first DELETE (only partition 'A' touched), we have 3 data files (one per
+# identity partition A/B/C) and 1 DV file for the affected partition.
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.ice_v3_dv_id_part.`files`
+ORDER BY content, file_format;
+---- RESULTS
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Second delete in the same partition - DV merge within a single partition
+DELETE FROM ice_v3_dv_id_part WHERE cat = 'A' AND id = 2;
+SELECT * FROM ice_v3_dv_id_part ORDER BY id;
+---- RESULTS
+3,'B',30
+4,'B',40
+5,'C',50
+---- TYPES
+INT,STRING,INT
+====
+---- QUERY
+# Cross-partition delete - deletion vectors created per partition
+DELETE FROM ice_v3_dv_id_part WHERE id IN (3, 5);
+SELECT * FROM ice_v3_dv_id_part ORDER BY id;
+---- RESULTS
+4,'B',40
+---- TYPES
+INT,STRING,INT
+====
+---- QUERY
+# Delete remaining partition to empty the table, verify count is zero
+DELETE FROM ice_v3_dv_id_part WHERE cat = 'B';
+SELECT count(*) FROM ice_v3_dv_id_part;
+---- RESULTS
+0
+---- TYPES
+BIGINT
+====
+---- QUERY
+# Truncate partition transform - rows sharing a truncated key go to same partition
+CREATE TABLE ice_v3_dv_trunc_part (id int, name string, val int)
+PARTITIONED BY SPEC (truncate(2, name))
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_trunc_part VALUES
+ (1, 'apple', 10), (2, 'apricot', 20), (3, 'banana', 30), (4, 'berry', 40), (5, 'cherry', 50);
+
+# 'apple' and 'apricot' share truncated key 'ap'; deleting by LIKE hits the same partition
+DELETE FROM ice_v3_dv_trunc_part WHERE name LIKE 'ap%';
+SELECT * FROM ice_v3_dv_trunc_part ORDER BY id;
+---- RESULTS
+3,'banana',30
+4,'berry',40
+5,'cherry',50
+---- TYPES
+INT,STRING,INT
+====
+---- QUERY
+# Second delete in the truncated partition - DV merge on truncated partition
+DELETE FROM ice_v3_dv_trunc_part WHERE name LIKE 'ba%';
+SELECT * FROM ice_v3_dv_trunc_part ORDER BY id;
+---- RESULTS
+4,'berry',40
+5,'cherry',50
+---- TYPES
+INT,STRING,INT
+====
+---- QUERY
+# Bucket partition - deletes distribute across buckets
+CREATE TABLE ice_v3_dv_bucket_part (id int, hash_col int, string_data string)
+PARTITIONED BY SPEC (bucket(5, hash_col))
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_bucket_part VALUES
+ (1, 10, 'a'), (2, 20, 'b'), (3, 30, 'c'), (4, 40, 'd'), (5, 50, 'e');
+
+DELETE FROM ice_v3_dv_bucket_part WHERE hash_col IN (20, 40);
+SELECT * FROM ice_v3_dv_bucket_part ORDER BY id;
+---- RESULTS
+1,10,'a'
+3,30,'c'
+5,50,'e'
+---- TYPES
+INT,INT,STRING
+====
+---- QUERY
+# Second delete - cumulative DV merge across buckets
+DELETE FROM ice_v3_dv_bucket_part WHERE id = 3;
+SELECT * FROM ice_v3_dv_bucket_part ORDER BY id;
+---- RESULTS
+1,10,'a'
+5,50,'e'
+---- TYPES
+INT,INT,STRING
+====
+---- QUERY
+# Multi-column partition spec - deletion vectors per (col1, col2) partition cell
+CREATE TABLE ice_v3_dv_multi_part (id int, region string, yr int, val int)
+PARTITIONED BY SPEC (region, bucket(3, yr))
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_multi_part VALUES
+ (1, 'US', 2022, 100), (2, 'US', 2023, 200),
+ (3, 'EU', 2022, 300), (4, 'EU', 2023, 400),
+ (5, 'AP', 2022, 500);
+
+# Partition-pruning delete: only touches EU partitions
+DELETE FROM ice_v3_dv_multi_part WHERE region = 'EU';
+SELECT * FROM ice_v3_dv_multi_part ORDER BY id;
+---- RESULTS
+1,'US',2022,100
+2,'US',2023,200
+5,'AP',2022,500
+---- TYPES
+INT,STRING,INT,INT
+====
+---- QUERY
+# Cross-partition delete on non-partition column
+DELETE FROM ice_v3_dv_multi_part WHERE yr = 2022;
+SELECT * FROM ice_v3_dv_multi_part ORDER BY id;
+---- RESULTS
+2,'US',2023,200
+---- TYPES
+INT,STRING,INT,INT
+====
+---- QUERY
+# Temporal partition (DAY transform) - delete within and across day partitions
+CREATE TABLE ice_v3_dv_day_part (id int, ts timestamp, payload string)
+PARTITIONED BY SPEC (day(ts))
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_day_part VALUES
+ (1, '2024-01-01 08:00:00', 'a'),
+ (2, '2024-01-01 20:00:00', 'b'),
+ (3, '2024-01-02 10:00:00', 'c'),
+ (4, '2024-01-03 12:00:00', 'd');
+
+DELETE FROM ice_v3_dv_day_part WHERE id IN (1, 3);
+SELECT * FROM ice_v3_dv_day_part ORDER BY id;
+---- RESULTS
+2,2024-01-01 20:00:00,'b'
+4,2024-01-03 12:00:00,'d'
+---- TYPES
+INT,TIMESTAMP,STRING
+====
+---- QUERY
+# YEAR partition transform - multiple rows per year, DV merge across years
+CREATE TABLE ice_v3_dv_year_part (id int, event_date date, amount decimal(10,2))
+PARTITIONED BY SPEC (year(event_date))
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_year_part VALUES
+ (1, '2022-03-15', 100.00),
+ (2, '2022-09-20', 200.00),
+ (3, '2023-01-10', 300.00),
+ (4, '2023-11-05', 400.00),
+ (5, '2024-06-01', 500.00);
+
+DELETE FROM ice_v3_dv_year_part WHERE year(event_date) = 2022;
+SELECT * FROM ice_v3_dv_year_part ORDER BY id;
+---- RESULTS
+3,2023-01-10,300.00
+4,2023-11-05,400.00
+5,2024-06-01,500.00
+---- TYPES
+INT,DATE,DECIMAL
+====
+---- QUERY
+# Second delete - merges DV with prior vector in 2023 partition
+DELETE FROM ice_v3_dv_year_part WHERE id = 4;
+SELECT count(*) FROM ice_v3_dv_year_part;
+---- RESULTS
+2
+---- TYPES
+BIGINT
+====
+---- QUERY
+# NULL partition values with deletion vectors
+CREATE TABLE ice_v3_dv_null_part (id int, part_key string, val int)
+PARTITIONED BY SPEC (part_key)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_null_part VALUES
+ (1, 'X', 10), (2, NULL, 20), (3, 'X', 30), (4, NULL, 40);
+
+# Delete from NULL partition
+DELETE FROM ice_v3_dv_null_part WHERE part_key IS NULL;
+SELECT * FROM ice_v3_dv_null_part ORDER BY id;
+---- RESULTS
+1,'X',10
+3,'X',30
+---- TYPES
+INT,STRING,INT
+====
+---- QUERY
+# NULL-partition delete: 2 data files (partition 'X' and NULL partition) + 1 DV for the NULL partition
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.ice_v3_dv_null_part.`files`
+ORDER BY content, file_format;
+---- RESULTS
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Aggregate correctness after partition-specific deletes
+CREATE TABLE ice_v3_dv_agg_part (region string, category string, sales int)
+PARTITIONED BY SPEC (region)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_agg_part VALUES
+ ('US', 'A', 100), ('US', 'A', 200), ('US', 'B', 150),
+ ('EU', 'A', 300), ('EU', 'B', 50), ('EU', 'B', 75);
+
+DELETE FROM ice_v3_dv_agg_part WHERE region = 'US' AND sales < 150;
+
+SELECT region, category, sum(sales), count(*)
+FROM ice_v3_dv_agg_part
+GROUP BY region, category
+ORDER BY region, category;
+---- RESULTS
+'EU','A',300,1
+'EU','B',125,2
+'US','A',200,1
+'US','B',150,1
+---- TYPES
+STRING,STRING,BIGINT,BIGINT
+====
+---- QUERY
+# Large-scale partitioned delete - stress RoaringBitmap per-partition
+CREATE TABLE ice_v3_dv_large_part (id bigint, part_key int)
+PARTITIONED BY SPEC (bucket(10, part_key))
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_large_part
+SELECT id, id % 100
+FROM functional.alltypes
+ORDER BY id LIMIT 5000;
+
+# Delete roughly half the rows spread across all 10 buckets
+DELETE FROM ice_v3_dv_large_part WHERE id % 2 = 0;
+SELECT count(*) FROM ice_v3_dv_large_part;
+---- RESULTS
+2500
+---- TYPES
+BIGINT
+====
+---- QUERY
+# Second large delete - DV merge per partition bucket
+DELETE FROM ice_v3_dv_large_part WHERE id % 3 = 0;
+SELECT count(*) FROM ice_v3_dv_large_part;
+---- RESULTS
+1667
+---- TYPES
+BIGINT
+====
+---- QUERY
+# Basic SORT BY (linear sort) - deletion vectors on a sorted table
+CREATE TABLE ice_v3_dv_sorted (id int, score int, label string)
+SORT BY (score)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_sorted VALUES
+ (1, 85, 'good'), (2, 90, 'great'), (3, 75, 'ok'), (4, 95, 'excellent'), (5, 60, 'poor');
+
+DELETE FROM ice_v3_dv_sorted WHERE score < 80;
+SELECT * FROM ice_v3_dv_sorted ORDER BY id;
+---- RESULTS
+1,85,'good'
+2,90,'great'
+4,95,'excellent'
+---- TYPES
+INT,INT,STRING
+====
+---- QUERY
+# Unpartitioned sorted table: 1 data file + 1 DV file after first delete
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.ice_v3_dv_sorted.`files`
+ORDER BY content;
+---- RESULTS
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Second delete - DV merge on sorted file
+DELETE FROM ice_v3_dv_sorted WHERE id = 2;
+SELECT * FROM ice_v3_dv_sorted ORDER BY id;
+---- RESULTS
+1,85,'good'
+4,95,'excellent'
+---- TYPES
+INT,INT,STRING
+====
+---- QUERY
+# Multi-column SORT BY - deletion vector covers rows sorted by (cat, score)
+CREATE TABLE ice_v3_dv_multi_sort (id int, cat string, score int, ts timestamp)
+SORT BY (cat, score)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_multi_sort VALUES
+ (1, 'A', 85, '2024-01-01 10:00:00'),
+ (2, 'A', 90, '2024-01-01 11:00:00'),
+ (3, 'A', 70, '2024-01-01 12:00:00'),
+ (4, 'B', 80, '2024-01-02 09:00:00'),
+ (5, 'B', 95, '2024-01-02 10:00:00');
+
+# Delete rows from the middle of the sort order
+DELETE FROM ice_v3_dv_multi_sort WHERE cat = 'A' AND score < 90;
+SELECT * FROM ice_v3_dv_multi_sort ORDER BY id;
+---- RESULTS
+2,'A',90,2024-01-01 11:00:00
+4,'B',80,2024-01-02 09:00:00
+5,'B',95,2024-01-02 10:00:00
+---- TYPES
+INT,STRING,INT,TIMESTAMP
+====
+---- QUERY
+# Second delete across both sort-key groups
+DELETE FROM ice_v3_dv_multi_sort WHERE score > 85;
+SELECT * FROM ice_v3_dv_multi_sort ORDER BY id;
+---- RESULTS
+4,'B',80,2024-01-02 09:00:00
+---- TYPES
+INT,STRING,INT,TIMESTAMP
+====
+---- QUERY
+# SORT BY ZORDER - deletion vectors on a Z-order-sorted table
+CREATE TABLE ice_v3_dv_zorder (id int, x int, y int, string_data string)
+SORT BY ZORDER (x, y)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_zorder VALUES
+ (1, 10, 20, 'a'), (2, 15, 25, 'b'), (3, 10, 30, 'c'),
+ (4, 20, 20, 'd'), (5, 20, 30, 'e');
+
+DELETE FROM ice_v3_dv_zorder WHERE x <= 10;
+SELECT * FROM ice_v3_dv_zorder ORDER BY id;
+---- RESULTS
+2,15,25,'b'
+4,20,20,'d'
+5,20,30,'e'
+---- TYPES
+INT,INT,INT,STRING
+====
+---- QUERY
+# Z-order sorted unpartitioned table: 1 data file + 1 DV file after first delete
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.ice_v3_dv_zorder.`files`
+ORDER BY content;
+---- RESULTS
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Second ZORDER delete - DV merge with previous vector
+DELETE FROM ice_v3_dv_zorder WHERE y > 25;
+SELECT * FROM ice_v3_dv_zorder ORDER BY id;
+---- RESULTS
+2,15,25,'b'
+4,20,20,'d'
+---- TYPES
+INT,INT,INT,STRING
+====
+---- QUERY
+# SORT BY ZORDER with more dimensions and cumulative deletes
+CREATE TABLE ice_v3_dv_zorder3d (id int, a int, b int, c int)
+SORT BY ZORDER (a, b, c)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_zorder3d VALUES
+ (1, 1, 1, 1), (2, 1, 2, 3), (3, 2, 1, 4),
+ (4, 2, 3, 1), (5, 3, 2, 2), (6, 3, 3, 3);
+
+DELETE FROM ice_v3_dv_zorder3d WHERE a = 1;
+DELETE FROM ice_v3_dv_zorder3d WHERE b = 3;
+SELECT * FROM ice_v3_dv_zorder3d ORDER BY id;
+---- RESULTS
+3,2,1,4
+5,3,2,2
+---- TYPES
+INT,INT,INT,INT
+====
+---- QUERY
+# SORT BY on a timestamp column - verify correctness after deletes
+CREATE TABLE ice_v3_dv_ts_sort (id int, created timestamp, status string)
+SORT BY (created)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_ts_sort VALUES
+ (1, '2024-01-01 08:00:00', 'active'),
+ (2, '2024-01-02 09:00:00', 'inactive'),
+ (3, '2024-01-03 10:00:00', 'active'),
+ (4, '2024-01-04 11:00:00', 'inactive');
+
+DELETE FROM ice_v3_dv_ts_sort WHERE status = 'inactive';
+SELECT * FROM ice_v3_dv_ts_sort ORDER BY id;
+---- RESULTS
+1,2024-01-01 08:00:00,'active'
+3,2024-01-03 10:00:00,'active'
+---- TYPES
+INT,TIMESTAMP,STRING
+====
+---- QUERY
+# Sequential inserts and deletes on sorted table - multiple data files with DVs
+CREATE TABLE ice_v3_dv_sort_seq (id int, batch int, val int)
+SORT BY (val)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_sort_seq VALUES (1, 1, 100), (2, 1, 200), (3, 1, 300);
+DELETE FROM ice_v3_dv_sort_seq WHERE val = 200;
+
+INSERT INTO ice_v3_dv_sort_seq VALUES (4, 2, 150), (5, 2, 250), (6, 2, 350);
+DELETE FROM ice_v3_dv_sort_seq WHERE val > 300;
+
+SELECT * FROM ice_v3_dv_sort_seq ORDER BY id;
+---- RESULTS
+1,1,100
+3,1,300
+4,2,150
+5,2,250
+---- TYPES
+INT,INT,INT
+====
+---- QUERY
+# Identity partition + single-column sort - per-partition DVs on sorted data
+CREATE TABLE ice_v3_dv_part_sort (id int, region string, score int, label string)
+PARTITIONED BY SPEC (region)
+SORT BY (score)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_part_sort VALUES
+ (1, 'US', 85, 'a'), (2, 'US', 90, 'b'), (3, 'US', 70, 'c'),
+ (4, 'EU', 80, 'd'), (5, 'EU', 95, 'e'), (6, 'EU', 60, 'f');
+
+# Delete within a single partition, from a non-boundary position in sort order
+DELETE FROM ice_v3_dv_part_sort WHERE region = 'US' AND score < 85;
+SELECT * FROM ice_v3_dv_part_sort ORDER BY id;
+---- RESULTS
+1,'US',85,'a'
+2,'US',90,'b'
+4,'EU',80,'d'
+5,'EU',95,'e'
+6,'EU',60,'f'
+---- TYPES
+INT,STRING,INT,STRING
+====
+---- QUERY
+# Identity-partitioned sorted table: 2 data files (US and EU) + 1 DV for the US partition
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.ice_v3_dv_part_sort.`files`
+ORDER BY content, file_format;
+---- RESULTS
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Cross-partition delete - DVs created in both partition directories
+DELETE FROM ice_v3_dv_part_sort WHERE score > 88;
+SELECT * FROM ice_v3_dv_part_sort ORDER BY id;
+---- RESULTS
+1,'US',85,'a'
+4,'EU',80,'d'
+6,'EU',60,'f'
+---- TYPES
+INT,STRING,INT,STRING
+====
+---- QUERY
+# Aggregate correctness after partition+sort deletes
+SELECT region, avg(score), count(*)
+FROM ice_v3_dv_part_sort
+GROUP BY region
+ORDER BY region;
+---- RESULTS
+'EU',70.0,2
+'US',85.0,1
+---- TYPES
+STRING,DOUBLE,BIGINT
+====
+---- QUERY
+# Bucket partition + multi-column sort - DV interacts with both dimensions
+CREATE TABLE ice_v3_dv_bucket_msort (id int, user_id int, category string, amount double)
+PARTITIONED BY SPEC (bucket(4, user_id))
+SORT BY (category, amount)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_bucket_msort VALUES
+ (1, 100, 'A', 1.5), (2, 100, 'B', 2.5), (3, 200, 'A', 3.5),
+ (4, 200, 'B', 4.5), (5, 300, 'A', 5.5), (6, 300, 'B', 6.5);
+
+DELETE FROM ice_v3_dv_bucket_msort WHERE category = 'A' AND amount < 4.0;
+SELECT * FROM ice_v3_dv_bucket_msort ORDER BY id;
+---- RESULTS
+2,100,'B',2.5
+4,200,'B',4.5
+5,300,'A',5.5
+6,300,'B',6.5
+---- TYPES
+INT,INT,STRING,DOUBLE
+====
+---- QUERY
+# Second delete - DV merge for buckets already written in previous delete
+DELETE FROM ice_v3_dv_bucket_msort WHERE user_id = 200;
+SELECT * FROM ice_v3_dv_bucket_msort ORDER BY id;
+---- RESULTS
+2,100,'B',2.5
+5,300,'A',5.5
+6,300,'B',6.5
+---- TYPES
+INT,INT,STRING,DOUBLE
+====
+---- QUERY
+# Truncate partition + SORT BY - truncated keys route rows; sort order is
+# per-file within each truncated partition
+CREATE TABLE ice_v3_dv_trunc_sort (id int, name string, val int)
+PARTITIONED BY SPEC (truncate(3, name))
+SORT BY (val)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_trunc_sort VALUES
+ (1, 'alpha', 30), (2, 'almond', 10), (3, 'beta', 50),
+ (4, 'belong', 20), (5, 'gamma', 40);
+
+# 'alpha' and 'almond' share truncated key 'alm'/'alp'... truncate(3,'alpha')='alp', truncate(3,'almond')='alm'
+# delete within two different truncated partitions
+DELETE FROM ice_v3_dv_trunc_sort WHERE val IN (10, 50);
+SELECT * FROM ice_v3_dv_trunc_sort ORDER BY id;
+---- RESULTS
+1,'alpha',30
+4,'belong',20
+5,'gamma',40
+---- TYPES
+INT,STRING,INT
+====
+---- QUERY
+# Year partition + SORT BY - time-series pattern
+CREATE TABLE ice_v3_dv_year_sort (id int, event_date date, metric string, value int)
+PARTITIONED BY SPEC (year(event_date))
+SORT BY (event_date, metric)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_year_sort VALUES
+ (1, '2022-03-01', 'cpu', 80),
+ (2, '2022-07-15', 'mem', 60),
+ (3, '2023-02-10', 'cpu', 70),
+ (4, '2023-08-20', 'mem', 90),
+ (5, '2024-01-05', 'cpu', 50);
+
+DELETE FROM ice_v3_dv_year_sort WHERE year(event_date) = 2022 AND metric = 'cpu';
+SELECT * FROM ice_v3_dv_year_sort ORDER BY id;
+---- RESULTS
+2,2022-07-15,'mem',60
+3,2023-02-10,'cpu',70
+4,2023-08-20,'mem',90
+5,2024-01-05,'cpu',50
+---- TYPES
+INT,DATE,STRING,INT
+====
+---- QUERY
+# Cross-year delete - multiple partition DVs created in one statement
+DELETE FROM ice_v3_dv_year_sort WHERE metric = 'mem';
+SELECT * FROM ice_v3_dv_year_sort ORDER BY id;
+---- RESULTS
+3,2023-02-10,'cpu',70
+5,2024-01-05,'cpu',50
+---- TYPES
+INT,DATE,STRING,INT
+====
+---- QUERY
+# Multi-column partition + ZORDER sort - most complex structural combination
+CREATE TABLE ice_v3_dv_mc_part_zorder (
+ id int, country string, yr int, x int, y int, val double)
+PARTITIONED BY SPEC (country, bucket(3, yr))
+SORT BY ZORDER (x, y)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_mc_part_zorder VALUES
+ (1, 'US', 2023, 10, 10, 1.1),
+ (2, 'US', 2023, 20, 20, 2.2),
+ (3, 'US', 2024, 10, 30, 3.3),
+ (4, 'EU', 2023, 15, 15, 4.4),
+ (5, 'EU', 2024, 25, 25, 5.5);
+
+DELETE FROM ice_v3_dv_mc_part_zorder WHERE country = 'US' AND x <= 10;
+SELECT * FROM ice_v3_dv_mc_part_zorder ORDER BY id;
+---- RESULTS
+2,'US',2023,20,20,2.2
+4,'EU',2023,15,15,4.4
+5,'EU',2024,25,25,5.5
+---- TYPES
+INT,STRING,INT,INT,INT,DOUBLE
+====
+---- QUERY
+# Cross-partition ZORDER delete
+DELETE FROM ice_v3_dv_mc_part_zorder WHERE y > 20;
+SELECT * FROM ice_v3_dv_mc_part_zorder ORDER BY id;
+---- RESULTS
+2,'US',2023,20,20,2.2
+4,'EU',2023,15,15,4.4
+---- TYPES
+INT,STRING,INT,INT,INT,DOUBLE
+====
+---- QUERY
+# Cumulative DV merges - partition+sort table with many sequential deletes
+CREATE TABLE ice_v3_dv_cumulative (id int, grp string, sort_key int, string_data string)
+PARTITIONED BY SPEC (grp)
+SORT BY (sort_key)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_cumulative VALUES
+ (1, 'X', 10, 'a'), (2, 'X', 20, 'b'), (3, 'X', 30, 'c'),
+ (4, 'Y', 10, 'd'), (5, 'Y', 20, 'e'), (6, 'Y', 30, 'f');
+
+# Delete 1
+DELETE FROM ice_v3_dv_cumulative WHERE sort_key = 10;
+SELECT count(*) FROM ice_v3_dv_cumulative;
+---- RESULTS
+4
+---- TYPES
+BIGINT
+====
+---- QUERY
+# First cumulative delete hits both partitions (X and Y): 2 data files + 2 DV files
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.ice_v3_dv_cumulative.`files`
+ORDER BY content, file_format;
+---- RESULTS
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Delete 2 - merges with existing DV per partition
+DELETE FROM ice_v3_dv_cumulative WHERE sort_key = 30;
+SELECT count(*) FROM ice_v3_dv_cumulative;
+---- RESULTS
+2
+---- TYPES
+BIGINT
+====
+---- QUERY
+# Delete 3 - removes the remaining row in one partition entirely
+DELETE FROM ice_v3_dv_cumulative WHERE grp = 'X';
+SELECT * FROM ice_v3_dv_cumulative ORDER BY id;
+---- RESULTS
+5,'Y',20,'e'
+---- TYPES
+INT,STRING,INT,STRING
+====
+---- QUERY
+# Sequential inserts between deletes on partitioned+sorted table
+CREATE TABLE ice_v3_dv_ps_seq (id int, part_key string, sort_key int)
+PARTITIONED BY SPEC (part_key)
+SORT BY (sort_key)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_ps_seq VALUES (1, 'A', 100), (2, 'A', 200), (3, 'B', 100);
+DELETE FROM ice_v3_dv_ps_seq WHERE sort_key = 200;
+
+INSERT INTO ice_v3_dv_ps_seq VALUES (4, 'A', 300), (5, 'B', 200);
+DELETE FROM ice_v3_dv_ps_seq WHERE sort_key = 100;
+
+SELECT * FROM ice_v3_dv_ps_seq ORDER BY id;
+---- RESULTS
+4,'A',300
+5,'B',200
+---- TYPES
+INT,STRING,INT
+====
+---- QUERY
+# DELETE with subquery on partitioned+sorted table
+CREATE TABLE ice_v3_dv_ps_sub (id int, part_key string, value int)
+PARTITIONED BY SPEC (part_key)
+SORT BY (value)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_ps_sub VALUES
+ (1, 'A', 10), (2, 'A', 50), (3, 'B', 20), (4, 'B', 80);
+
+DELETE FROM ice_v3_dv_ps_sub
+WHERE id IN (SELECT id FROM ice_v3_dv_ps_sub WHERE value > 40);
+SELECT * FROM ice_v3_dv_ps_sub ORDER BY id;
+---- RESULTS
+1,'A',10
+3,'B',20
+---- TYPES
+INT,STRING,INT
+====
+---- QUERY
+# FILE__POSITION delete on partitioned+sorted table
+CREATE TABLE ice_v3_dv_ps_filepos (id int, part_key string, sort_key int)
+PARTITIONED BY SPEC (part_key)
+SORT BY (sort_key)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_ps_filepos VALUES (10, 'A', 300), (20, 'A', 100), (30, 'B', 200);
+
+# FILE__POSITION=0 refers to the first record in a file; on sorted data that is the
+# row with the lowest sort_key written by the INSERT above ('A',100)
+DELETE FROM ice_v3_dv_ps_filepos WHERE part_key = 'A' AND FILE__POSITION = 0;
+SELECT * FROM ice_v3_dv_ps_filepos ORDER BY id;
+---- RESULTS: VERIFY_IS_SUBSET
+10,'A',300
+30,'B',200
+---- TYPES
+INT,STRING,INT
+====
+---- QUERY
+# Large-scale delete on partitioned+sorted table - RoaringBitmap at scale
+CREATE TABLE ice_v3_dv_ps_large (id bigint, bucket_key int, sort_key int)
+PARTITIONED BY SPEC (bucket(8, bucket_key))
+SORT BY (sort_key)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_ps_large
+SELECT id, id % 80, id % 500
+FROM functional.alltypes
+ORDER BY id LIMIT 8000;
+
+# Delete about half the rows across all 8 buckets
+DELETE FROM ice_v3_dv_ps_large WHERE id % 2 = 0;
+SELECT count(*) FROM ice_v3_dv_ps_large;
+---- RESULTS
+3650
+---- TYPES
+BIGINT
+====
+---- QUERY
+# Second large batch delete - DV merge per bucket partition
+DELETE FROM ice_v3_dv_ps_large WHERE sort_key < 100;
+SELECT count(*) FROM ice_v3_dv_ps_large;
+---- RESULTS
+2900
+---- TYPES
+BIGINT
+====
+---- QUERY
+# Partition evolution from unpartitioned to partitioned+sorted
+CREATE TABLE ice_v3_dv_evolve (id int, cat string, val int)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+# Phase 1: unpartitioned, no sort
+INSERT INTO ice_v3_dv_evolve VALUES (1, 'A', 10), (2, 'B', 20), (3, 'A', 30);
+DELETE FROM ice_v3_dv_evolve WHERE id = 2;
+SELECT * FROM ice_v3_dv_evolve ORDER BY id;
+---- RESULTS
+1,'A',10
+3,'A',30
+---- TYPES
+INT,STRING,INT
+====
+---- QUERY
+ALTER TABLE ice_v3_dv_evolve SET PARTITION SPEC (cat);
+# Phase 2: new inserts go into identity partition; old unpartitioned data still readable
+INSERT INTO ice_v3_dv_evolve VALUES (4, 'A', 40), (5, 'B', 50);
+DELETE FROM ice_v3_dv_evolve WHERE val = 40;
+SELECT * FROM ice_v3_dv_evolve ORDER BY id;
+---- RESULTS
+1,'A',10
+3,'A',30
+5,'B',50
+---- TYPES
+INT,STRING,INT
+====
+---- QUERY
+# JOIN between two partitioned+sorted v3 tables - DVs applied on both sides
+CREATE TABLE ice_v3_dv_ps_join1 (id int, grp string, score int)
+PARTITIONED BY SPEC (grp)
+SORT BY (score)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+CREATE TABLE ice_v3_dv_ps_join2 (id int, grp string, detail string)
+PARTITIONED BY SPEC (grp)
+SORT BY (id)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_ps_join1 VALUES
+ (1, 'X', 80), (2, 'X', 90), (3, 'Y', 70), (4, 'Y', 95);
+INSERT INTO ice_v3_dv_ps_join2 VALUES
+ (1, 'X', 'detail_1'), (2, 'X', 'detail_2'), (3, 'Y', 'detail_3'), (4, 'Y', 'detail_4');
+
+DELETE FROM ice_v3_dv_ps_join1 WHERE score < 80;
+DELETE FROM ice_v3_dv_ps_join2 WHERE id IN (2, 4);
+
+SELECT j1.id, j1.grp, j1.score, j2.detail
+FROM ice_v3_dv_ps_join1 j1
+JOIN ice_v3_dv_ps_join2 j2 ON j1.id = j2.id
+ORDER BY j1.id;
+---- RESULTS
+1,'X',80,'detail_1'
+---- TYPES
+INT,STRING,INT,STRING
+====
+---- QUERY
+# Window function correctness after partition+sort deletes
+CREATE TABLE ice_v3_dv_ps_window (id int, grp string, sort_key int, val int)
+PARTITIONED BY SPEC (grp)
+SORT BY (sort_key)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_ps_window VALUES
+ (1, 'A', 1, 10), (2, 'A', 2, 20), (3, 'A', 3, 30),
+ (4, 'B', 1, 40), (5, 'B', 2, 50);
+
+DELETE FROM ice_v3_dv_ps_window WHERE id IN (1, 4);
+
+SELECT id, grp, val,
+ row_number() OVER (PARTITION BY grp ORDER BY sort_key) AS rn
+FROM ice_v3_dv_ps_window
+ORDER BY id;
+---- RESULTS
+2,'A',20,1
+3,'A',30,2
+5,'B',50,1
+---- TYPES
+INT,STRING,INT,BIGINT
+====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete-v2-equality-upgrade.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete-v2-equality-upgrade.test
new file mode 100644
index 000000000..9030a7318
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete-v2-equality-upgrade.test
@@ -0,0 +1,102 @@
+====
+---- QUERY
+# The iceberg_v2_delete_equality table was written by Flink with equality deletes.
+# Verify the initial visible state: equality deletes produced the final view of 2 rows.
+SELECT * FROM iceberg_v2_delete_equality ORDER BY id;
+---- RESULTS
+1,'test_1_base'
+2,'test_2_updated'
+---- TYPES
+BIGINT,STRING
+====
+---- QUERY
+# Verify the file structure: 3 data files (content=0) and 3 equality delete files
+# (content=2) written by Flink. No position or DV delete files yet.
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.iceberg_v2_delete_equality.`files`
+ORDER BY content, file_format;
+---- RESULTS
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+2,'PARQUET',false,false
+2,'PARQUET',false,false
+2,'PARQUET',false,false
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Upgrade the table from v2 to v3.
+ALTER TABLE iceberg_v2_delete_equality SET TBLPROPERTIES ('format-version'='3');
+====
+---- QUERY
+# After upgrade, the existing equality deletes are still applied correctly.
+SELECT * FROM iceberg_v2_delete_equality ORDER BY id;
+---- RESULTS
+1,'test_1_base'
+2,'test_2_updated'
+---- TYPES
+BIGINT,STRING
+====
+---- QUERY
+# Delete one of the remaining rows using Impala. On a v3 table Impala writes a
+# deletion vector (Puffin file) for this delete.
+DELETE FROM iceberg_v2_delete_equality WHERE id = 1;
+---- RUNTIME_PROFILE
+NumModifiedRows: 0
+NumDeletedRows: 1
+====
+---- QUERY
+SELECT * FROM iceberg_v2_delete_equality ORDER BY id;
+---- RESULTS
+2,'test_2_updated'
+---- TYPES
+BIGINT,STRING
+====
+---- QUERY
+# The file structure now contains the original equality delete files (content=2) from
+# Flink and a new Puffin DV file (content=1) written by Impala.
+# The Puffin DV's record_count equals the data file's record_count, confirming every
+# record in that data file is covered by the delete.
+SELECT content, file_format, record_count, referenced_data_file IS NOT NULL,
+ content_offset IS NOT NULL
+FROM $DATABASE.iceberg_v2_delete_equality.`files`
+ORDER BY content, file_format;
+---- RESULTS
+0,'PARQUET',1,false,false
+0,'PARQUET',1,false,false
+0,'PARQUET',1,false,false
+1,'PUFFIN',1,true,true
+2,'PARQUET',1,false,false
+2,'PARQUET',1,false,false
+2,'PARQUET',1,false,false
+---- TYPES
+INT,STRING,BIGINT,BOOLEAN,BOOLEAN
+====
+---- QUERY
+OPTIMIZE TABLE iceberg_v2_delete_equality;
+SELECT * FROM iceberg_v2_delete_equality ORDER BY id;
+---- RESULTS
+2,'test_2_updated'
+---- TYPES
+BIGINT,STRING
+====
+---- QUERY
+# After OPTIMIZE: 2 equality delete files (content=2) remain due to the
+# bug IMPALA-14908. The DV (content=1) is correctly removed. TODO:
+SELECT COUNT(*) FROM $DATABASE.iceberg_v2_delete_equality.`files`
+WHERE content = 2;
+---- RESULTS
+2
+---- TYPES
+BIGINT
+====
+---- QUERY
+# No position delete file remains in the table.
+SELECT COUNT(*) FROM $DATABASE.iceberg_v2_delete_equality.`files`
+WHERE content = 1;
+---- RESULTS
+0
+---- TYPES
+BIGINT
+====
\ No newline at end of file
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete.test
new file mode 100644
index 000000000..12c292c20
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete.test
@@ -0,0 +1,423 @@
+====
+---- QUERY
+SELECT * FROM iceberg_v3_deletion_vectors;
+---- RESULTS
+1
+3
+5
+---- TYPES
+INT
+====
+---- QUERY
+# Basic DELETE creates deletion vector (Puffin file)
+CREATE TABLE ice_v3_dv (i int, s string)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv VALUES (1, 'one'), (2, 'two'), (3, 'three'), (4, 'four'), (5, 'five');
+DELETE FROM ice_v3_dv WHERE i = 3;
+---- RUNTIME_PROFILE
+NumModifiedRows: 0
+NumDeletedRows: 1
+====
+---- QUERY
+SELECT * FROM ice_v3_dv ORDER BY i;
+---- RESULTS
+1,'one'
+2,'two'
+4,'four'
+5,'five'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+# Count after deletion
+SELECT count(*) FROM ice_v3_dv;
+---- RESULTS
+4
+---- TYPES
+BIGINT
+====
+---- QUERY
+# Verify that the DV file is a PUFFIN position-delete entry with referenced_data_file
+# and content_offset populated; the data file has neither of those DV-specific fields.
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.ice_v3_dv.`files`
+ORDER BY content;
+---- RESULTS
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Multiple deletes accumulate in deletion vector (merge test)
+DELETE FROM ice_v3_dv WHERE i IN (1, 4);
+---- RUNTIME_PROFILE
+NumModifiedRows: 0
+NumDeletedRows: 2
+====
+---- QUERY
+SELECT * FROM ice_v3_dv ORDER BY i;
+---- RESULTS
+2,'two'
+5,'five'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+# Third delete - cumulative DV merge
+DELETE FROM ice_v3_dv WHERE i = 5;
+---- RUNTIME_PROFILE
+NumModifiedRows: 0
+NumDeletedRows: 1
+====
+---- QUERY
+SELECT * FROM ice_v3_dv ORDER BY i;
+---- RESULTS
+2,'two'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+# WHERE clause with multiple conditions
+CREATE TABLE ice_v3_dv_where (id int, cat string, val int)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_where VALUES
+ (1, 'A', 10), (2, 'A', 20), (3, 'B', 30), (4, 'B', 40), (5, 'C', 50);
+
+DELETE FROM ice_v3_dv_where WHERE cat = 'B' AND val > 35;
+---- RUNTIME_PROFILE
+NumModifiedRows: 0
+NumDeletedRows: 1
+====
+---- QUERY
+SELECT * FROM ice_v3_dv_where ORDER BY id;
+---- RESULTS
+1,'A',10
+2,'A',20
+3,'B',30
+5,'C',50
+---- TYPES
+INT,STRING,INT
+====
+---- QUERY
+# DELETE with subquery
+CREATE TABLE ice_v3_dv_sub (id int, value int)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_sub VALUES (1, 100), (2, 200), (3, 150), (4, 250);
+DELETE FROM ice_v3_dv_sub WHERE id IN (SELECT id FROM ice_v3_dv_sub WHERE value > 180);
+SELECT * FROM ice_v3_dv_sub ORDER BY id;
+---- RESULTS
+1,100
+3,150
+---- TYPES
+INT,INT
+====
+---- QUERY
+# JOIN with deletion vectors
+CREATE TABLE ice_v3_dv_join1 (id int, name string)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+CREATE TABLE ice_v3_dv_join2 (id int, detail string)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_join1 VALUES (1, 'a'), (2, 'b'), (3, 'c');
+INSERT INTO ice_v3_dv_join2 VALUES (2, 'd2'), (3, 'd3'), (4, 'd4');
+
+DELETE FROM ice_v3_dv_join1 WHERE id = 2;
+DELETE FROM ice_v3_dv_join2 WHERE id = 4;
+
+SELECT j1.id, j1.name, j2.detail
+FROM ice_v3_dv_join1 j1
+JOIN ice_v3_dv_join2 j2 ON j1.id = j2.id
+ORDER BY j1.id;
+---- RESULTS
+3,'c','d3'
+---- TYPES
+INT,STRING,STRING
+====
+---- QUERY
+# Verify that ice_v3_dv_join1 has one data file and one DV file after the delete.
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.ice_v3_dv_join1.`files`
+ORDER BY content;
+---- RESULTS
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Verify that ice_v3_dv_join2 also got its own DV file.
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.ice_v3_dv_join2.`files`
+ORDER BY content;
+---- RESULTS
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Aggregates with deletion vectors
+CREATE TABLE ice_v3_dv_agg (cat string, val int)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_agg VALUES
+ ('A', 10), ('A', 20), ('B', 30), ('B', 40), ('C', 50);
+
+DELETE FROM ice_v3_dv_agg WHERE val = 20;
+
+SELECT cat, sum(val), count(*) FROM ice_v3_dv_agg GROUP BY cat ORDER BY cat;
+---- RESULTS
+'A',10,1
+'B',70,2
+'C',50,1
+---- TYPES
+STRING,BIGINT,BIGINT
+====
+---- QUERY
+# MIN/MAX with deletion vectors
+SELECT min(val), max(val) FROM ice_v3_dv_agg;
+---- RESULTS
+10,50
+---- TYPES
+INT,INT
+====
+---- QUERY
+# FILE__POSITION delete
+CREATE TABLE ice_v3_dv_pos (id int, string_data string)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_pos VALUES (10, 'a'), (20, 'b'), (30, 'c');
+DELETE FROM ice_v3_dv_pos WHERE FILE__POSITION = 0;
+SELECT * FROM ice_v3_dv_pos ORDER BY id;
+---- RESULTS
+20,'b'
+30,'c'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+# Verify that the FILE__POSITION delete also produces a valid DV Puffin file.
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.ice_v3_dv_pos.`files`
+ORDER BY content;
+---- RESULTS
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Merge with overlapping deletes (large scale)
+CREATE TABLE ice_v3_dv_merge (id int)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_merge SELECT id FROM functional.alltypes ORDER BY id LIMIT 1000;
+DELETE FROM ice_v3_dv_merge WHERE id < 300;
+---- RUNTIME_PROFILE
+NumModifiedRows: 0
+NumDeletedRows: 300
+====
+---- QUERY
+SELECT count(*) FROM ice_v3_dv_merge;
+---- RESULTS
+700
+---- TYPES
+BIGINT
+====
+---- QUERY
+# Second delete with overlap - tests DV merge
+DELETE FROM ice_v3_dv_merge WHERE id >= 200 AND id < 500;
+SELECT count(*) FROM ice_v3_dv_merge;
+---- RESULTS
+500
+---- TYPES
+BIGINT
+====
+---- QUERY
+# NULL handling with deletion vectors
+CREATE TABLE ice_v3_dv_null (id int, nullable string)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_null VALUES (1, 'a'), (2, NULL), (3, 'c'), (4, NULL);
+DELETE FROM ice_v3_dv_null WHERE nullable IS NULL;
+SELECT * FROM ice_v3_dv_null ORDER BY id;
+---- RESULTS
+1,'a'
+3,'c'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+# Verify DV and data file entries in the files metadata table after NULL-value delete
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.ice_v3_dv_null.`files`
+ORDER BY content;
+---- RESULTS
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# DELETE with LIKE pattern
+CREATE TABLE ice_v3_dv_like (id int, text string)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_like VALUES (1, 'apple'), (2, 'banana'), (3, 'apricot'), (4, 'cherry');
+DELETE FROM ice_v3_dv_like WHERE text LIKE 'a%';
+SELECT * FROM ice_v3_dv_like ORDER BY id;
+---- RESULTS
+2,'banana'
+4,'cherry'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+# DELETE no matching rows (no-op)
+CREATE TABLE ice_v3_dv_noop (id int)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_noop VALUES (1), (2), (3);
+DELETE FROM ice_v3_dv_noop WHERE id = 999;
+SELECT count(*) FROM ice_v3_dv_noop;
+---- RESULTS
+3
+---- TYPES
+BIGINT
+====
+---- QUERY
+# No-op delete should not create any DV/Puffin files; only the data file remains
+SELECT count(*) FROM $DATABASE.ice_v3_dv_noop.`files` WHERE content = 1;
+---- RESULTS
+0
+---- TYPES
+BIGINT
+====
+---- QUERY
+# DELETE from empty table
+CREATE TABLE ice_v3_dv_empty (id int)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+DELETE FROM ice_v3_dv_empty WHERE id = 1;
+SELECT count(*) FROM ice_v3_dv_empty;
+---- RESULTS
+0
+---- TYPES
+BIGINT
+====
+---- QUERY
+# Empty table delete should produce no files at all
+SELECT count(*) FROM $DATABASE.ice_v3_dv_empty.`files`;
+---- RESULTS
+0
+---- TYPES
+BIGINT
+====
+---- QUERY
+# ORDER BY and LIMIT with deletion vectors
+CREATE TABLE ice_v3_dv_order (id int, score int)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_order VALUES (1, 85), (2, 90), (3, 75), (4, 95), (5, 80);
+DELETE FROM ice_v3_dv_order WHERE score < 85;
+SELECT * FROM ice_v3_dv_order ORDER BY score DESC LIMIT 2;
+---- RESULTS
+4,95
+2,90
+---- TYPES
+INT,INT
+====
+---- QUERY
+# DISTINCT with deletion vectors
+CREATE TABLE ice_v3_dv_distinct (cat string, val int)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_distinct VALUES ('A', 1), ('A', 2), ('B', 3), ('B', 4);
+DELETE FROM ice_v3_dv_distinct WHERE val IN (2, 4);
+SELECT DISTINCT cat FROM ice_v3_dv_distinct ORDER BY cat;
+---- RESULTS
+'A'
+'B'
+---- TYPES
+STRING
+====
+---- QUERY
+# Sequential deletes with inserts between
+CREATE TABLE ice_v3_dv_seq (id int, batch int)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_seq VALUES (1, 1), (2, 1);
+DELETE FROM ice_v3_dv_seq WHERE id = 2;
+
+INSERT INTO ice_v3_dv_seq VALUES (3, 2), (4, 2);
+DELETE FROM ice_v3_dv_seq WHERE id = 4;
+
+SELECT * FROM ice_v3_dv_seq ORDER BY id;
+---- RESULTS
+1,1
+3,2
+---- TYPES
+INT,INT
+====
+---- QUERY
+# Sequential inserts+deletes produce one DV per data file batch
+SELECT content, file_format, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.ice_v3_dv_seq.`files`
+ORDER BY content, file_format;
+---- RESULTS
+0,'PARQUET',false,false
+0,'PARQUET',false,false
+1,'PUFFIN',true,true
+1,'PUFFIN',true,true
+---- TYPES
+INT,STRING,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# Large delete batch - tests RoaringBitmap serialization
+CREATE TABLE ice_v3_dv_large (id bigint)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='3');
+
+INSERT INTO ice_v3_dv_large
+SELECT id FROM functional.alltypes ORDER BY id LIMIT 5000;
+
+DELETE FROM ice_v3_dv_large WHERE id % 3 = 0;
+SELECT count(*) FROM ice_v3_dv_large;
+---- RESULTS
+3333
+---- TYPES
+BIGINT
+====
+---- QUERY
+# v2 table with position delete upgraded to v3, DELETE must be blocked.
+CREATE TABLE ice_v3_upgraded_from_v2 (id int, val string)
+STORED BY ICEBERG
+TBLPROPERTIES ('format-version'='2');
+
+INSERT INTO ice_v3_upgraded_from_v2 VALUES (1, 'a'), (2, 'b'), (3, 'c');
+DELETE FROM ice_v3_upgraded_from_v2 WHERE id = 2;
+ALTER TABLE ice_v3_upgraded_from_v2 SET TBLPROPERTIES ('format-version'='3');
+DELETE FROM ice_v3_upgraded_from_v2 WHERE id = 1;
+---- CATCH
+DELETE is not allowed on Iceberg format version 3 table
+====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-negative.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-negative.test
index fcd4a8950..59e9157ec 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-negative.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-negative.test
@@ -9,11 +9,6 @@ INSERT INTO ice_v3 VALUES (2, 'iceberg');
INSERT INTO ice_v3 VALUES (3, 'spark');
====
---- QUERY
-DELETE FROM ice_v3 WHERE i = 1;
----- CATCH
-Impala does not support DELETE statements on Iceberg tables with format version 3
-====
----- QUERY
UPDATE ice_v3 SET s = 'updated' WHERE i = 1;
---- CATCH
Impala does not support UPDATE statements on Iceberg tables with format version 3
@@ -26,11 +21,6 @@ WHEN MATCHED THEN UPDATE SET target.s = source.s;
Impala does not support MERGE statements on Iceberg tables with format version 3
====
---- QUERY
-SELECT * FROM iceberg_v3_deletion_vectors;
----- CATCH
-Iceberg tables with Deletion Vectors are not supported yet
-====
----- QUERY
# The first snapshot has no deletion vectors.
SELECT *
FROM iceberg_v3_deletion_vectors FOR SYSTEM_VERSION AS OF 2700858159721908397;
diff --git a/tests/query_test/test_iceberg.py b/tests/query_test/test_iceberg.py
index 3eb4d31da..1c27a3266 100644
--- a/tests/query_test/test_iceberg.py
+++ b/tests/query_test/test_iceberg.py
@@ -23,7 +23,9 @@ import logging
import os
import random
import re
+import struct
from subprocess import check_call, check_output
+import tempfile
import time
from avro.datafile import DataFileReader
@@ -1780,6 +1782,7 @@ class TestIcebergV2Table(IcebergTestSuite):
if IS_HDFS and self.should_run_for_hive(vector):
self._delete_partitioned_hive_tests(unique_database)
+
def _delete_partitioned_hive_tests(self, db):
hive_output = self.run_stmt_in_hive("SELECT * FROM {}.{} ORDER BY i".format(
db, "id_part"))
@@ -2340,6 +2343,31 @@ class TestIcebergV3Table(IcebergTestSuite):
self.load_table(unique_database, "iceberg_v3_row_lineage_orc", format="orc")
self.run_test_case('QueryTest/iceberg-v3-row-lineage', vector, unique_database)
+ def test_v3_delete(self, vector, unique_database):
+ """Test Iceberg V3 deletion vectors (Puffin files)."""
+ self.load_table(unique_database, "iceberg_v3_deletion_vectors")
+ self.run_test_case('QueryTest/iceberg-v3-delete', vector, unique_database)
+ self.run_test_case('QueryTest/iceberg-v3-delete-partition-sort',
+ vector, unique_database)
+
+ def test_v3_delete_on_existing_dv_table(self, vector, unique_database):
+ """Test DELETE on a table that already has deletion vectors written by Spark.
+ Verifies that Impala correctly reads and extends pre-existing Puffin DV files
+ produced by an external engine."""
+ self.load_table(unique_database, "iceberg_v3_deletion_vectors")
+ self.run_test_case('QueryTest/iceberg-v3-delete-existing-dv', vector, unique_database)
+
+ def test_v3_delete_on_v2_equality_delete_table(self, vector, unique_database):
+ """Test that a v2 table with Flink-written equality delete files can be upgraded to
+ v3 and that both the existing equality deletes and new Impala DV-based deletes are
+ applied correctly after the upgrade."""
+ create_iceberg_table_from_directory(self.client, unique_database,
+ "iceberg_v2_delete_equality", "parquet",
+ table_location="${IMPALA_HOME}/testdata/data/iceberg_test/hadoop_catalog/ice",
+ warehouse_prefix=os.getenv("FILESYSTEM_PREFIX"))
+ self.run_test_case('QueryTest/iceberg-v3-delete-v2-equality-upgrade',
+ vector, unique_database)
+
@SkipIf.not_dfs
def test_v3_row_lineage_file_schema(self, unique_database):
"""Test that plain INSERTs only write user columns, not hidden metadata columns."""
@@ -2381,6 +2409,116 @@ class TestIcebergV3Table(IcebergTestSuite):
self.load_table(unique_database, "test_complex_default")
self.run_test_case('QueryTest/iceberg-v3-default-values', vector, unique_database)
+ def _assert_puffin_file_layout(self, local_file, reported_size,
+ content_offset, content_size):
+ """Validate the binary layout of a downloaded Puffin file.
+
+ Checks:
+ - Actual on-disk size matches manifest-reported file_size_in_bytes.
+ - DV blob extent (content_offset, content_size) fits within the file.
+ - PFA1 magic bytes at both ends.
+ - Blob bytes are readable at content_offset.
+ - Full structural size reconstruction from footer size field matches actual size.
+ """
+ actual_size = os.path.getsize(local_file)
+
+ assert actual_size == reported_size, (
+ "Puffin file size mismatch: actual=%d, manifest-reported=%d"
+ % (actual_size, reported_size))
+
+ assert content_offset >= 0, \
+ "content_offset must be non-negative, got %d" % content_offset
+ assert content_size > 0, \
+ "content_size_in_bytes must be positive, got %d" % content_size
+ assert content_offset + content_size <= actual_size, (
+ "DV blob extent [%d, %d) exceeds file size %d"
+ % (content_offset, content_offset + content_size, actual_size))
+
+ PUFFIN_MAGIC = b'\x50\x46\x41\x31'
+ with open(local_file, 'rb') as f:
+ header = f.read(4)
+ assert header == PUFFIN_MAGIC, \
+ "Unexpected Puffin header magic: %r" % header
+ f.seek(-4, 2)
+ trailer = f.read(4)
+ assert trailer == PUFFIN_MAGIC, \
+ "Unexpected Puffin trailer magic: %r" % trailer
+
+ # Read the footer size field (uint32 LE) at bytes [-12:-8], then reconstruct
+ # the expected total file size from first principles:
+ #
+ # Puffin file layout:
+ # [File magic: 4]
+ # [Blobs: total_blobs_size bytes]
+ # [Footer leading magic: 4]
+ # [Footer JSON: footer_json_len bytes]
+ # [Footer size field (LE uint32): 4] <- at bytes [-12:-8]
+ # [Flags (reserved): 4] <- at bytes [-8:-4]
+ # [Trailing magic: 4] <- at bytes [-4:]
+ f.seek(-12, 2)
+ footer_json_len = struct.unpack('<I', f.read(4))[0]
+
+ # The footer leading magic sits immediately before the footer JSON.
+ # Everything between the file magic and the footer leading magic is blob data.
+ total_blobs_size = actual_size - 4 - 4 - footer_json_len - 4 - 4 - 4
+
+ expected_size = (
+ 4 # file magic
+ + total_blobs_size # all blob data
+ + 4 # footer leading magic
+ + footer_json_len # footer JSON
+ + 4 # footer size field
+ + 4 # flags
+ + 4 # trailing magic
+ )
+ assert actual_size == expected_size, (
+ "Puffin file size does not match structural calculation: "
+ "actual=%d, expected=%d "
+ "(content_offset=%d, content_size=%d, footer_json_len=%d)"
+ % (actual_size, expected_size,
+ content_offset, content_size, footer_json_len))
+
+ @SkipIf.not_dfs
+ def test_v3_delete_puffin_file_size(self, unique_database):
+ """Test that the Puffin file size recorded in the manifest matches the actual binary
+ size of the file on the filesystem, and that the DV blob extent (content_offset +
+ content_size_in_bytes) fits within that file. This guards against off-by-N bugs in
+ current_offset_ tracking inside PuffinWriter (e.g. the leading footer magic not being
+ accounted for)."""
+ table = "ice_puffin_size"
+ fqn = "%s.%s" % (unique_database, table)
+
+ self.client.execute("""
+ CREATE TABLE {fqn} (id INT, val STRING)
+ STORED BY ICEBERG
+ TBLPROPERTIES ('format-version'='3')
+ """.format(fqn=fqn))
+ self.client.execute(
+ "INSERT INTO {fqn} VALUES (1, 'a'), (2, 'b'), (3, 'c')".format(fqn=fqn))
+ self.client.execute(
+ "DELETE FROM {fqn} WHERE id = 2".format(fqn=fqn))
+
+ # Retrieve the Puffin file path, manifest-reported size, and DV blob extent.
+ result = self.client.execute("""
+ SELECT file_path, file_size_in_bytes, content_offset, content_size_in_bytes
+ FROM {fqn}.`files`
+ WHERE content = 1 AND file_format = 'PUFFIN'
+ """.format(fqn=fqn))
+ assert len(result.data) == 1, \
+ "Expected exactly one Puffin DV file, got: %s" % result.data
+ row = result.data[0].split('\t')
+ hdfs_path = row[0]
+ reported_size = int(row[1])
+ content_offset = int(row[2])
+ content_size = int(row[3])
+
+ # Download the Puffin file to a temporary directory and run layout checks.
+ with tempfile.TemporaryDirectory() as tmp_dir:
+ local_file = os.path.join(tmp_dir, "dv.puffin")
+ check_call(['hadoop', 'fs', '-copyToLocal', hdfs_path, local_file])
+ self._assert_puffin_file_layout(local_file, reported_size,
+ content_offset, content_size)
+
# Tests to exercise the DIRECTED distribution mode for V2 Iceberg tables. Note, that most
# of the test coverage is in TestIcebergV2Table.test_read_position_deletes but since it