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:34 UTC
(impala) branch master updated (6c3324ee7 -> 7ceefb1ed)
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 6c3324ee7 IMPALA-14790: Avoid bug in psutil 7.2.2
new 09e8b963b IMPALA-14774: Increase HMS event sync wait time to avoid timeout
new 7ceefb1ed IMPALA-14588: Implement Iceberg deletion vector reading/writing
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
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/metadata/test_event_processing.py | 2 +-
tests/metadata/test_event_processing_base.py | 2 +-
tests/query_test/test_iceberg.py | 138 ++++
40 files changed, 2413 insertions(+), 142 deletions(-)
rename fe/src/main/java/org/apache/impala/planner/{IcebergDeleteNode.java => IcebergDeleteJoinNode.java} (96%)
create mode 100644 testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete-existing-dv.test
create mode 100644 testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete-partition-sort.test
create mode 100644 testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete-v2-equality-upgrade.test
create mode 100644 testdata/workloads/functional-query/queries/QueryTest/iceberg-v3-delete.test
(impala) 01/02: IMPALA-14774: Increase HMS event sync wait time to avoid timeout
Posted by st...@apache.org.
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 09e8b963b9543ae42fb8c598fd674e8d47522df6
Author: Venu Reddy <k....@gmail.com>
AuthorDate: Tue Apr 21 16:11:47 2026 +0530
IMPALA-14774: Increase HMS event sync wait time to avoid timeout
metadata/test_ddl.py::TestDdlStatements::test_truncate_table and
metadata/test_event_processing.py::TestEventSyncWaiting::test_txn
running concurrently on the same cluster has caused the issue.
TRUNCATE_TABLE with 24 partitions held the table write lock
for about 13 seconds. Event processing thread waited to acquire
the lock for test_truncate_table test.
Another thread waited for the HMS event sync to finish
(sync_hms_events_wait_time_s=10sec) till the latest event ID
for test_txn test. Since the truncate took over 10 seconds,
HMS event sync timed out and failed test.
This issue doesn't happen after IMPALA-14535 because HMS event
sync is optimised to wait for table level granularity instead
of waiting for global latest event id.
This change reduces the chance of such failure and is easy to
backport, even though the issue didn't occur since IMPALA-14535.
Testing:
- Executed existing tests.
Change-Id: Ib89c6f1928a1190942ec3c8001e0d8671d896857
Reviewed-on: http://gerrit.cloudera.org:8080/24216
Reviewed-by: Csaba Ringhofer <cs...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
tests/metadata/test_event_processing.py | 2 +-
tests/metadata/test_event_processing_base.py | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/tests/metadata/test_event_processing.py b/tests/metadata/test_event_processing.py
index 32c2607f7..892b6b2dd 100644
--- a/tests/metadata/test_event_processing.py
+++ b/tests/metadata/test_event_processing.py
@@ -32,7 +32,7 @@ from tests.common.test_vector import HS2
from tests.metadata.test_event_processing_base import TestEventProcessingBase
from tests.util.event_processor_utils import EventProcessorUtils
-PROCESSING_TIMEOUT_S = 10
+PROCESSING_TIMEOUT_S = 30
LOG = logging.getLogger(__name__)
@SkipIfFS.hive
diff --git a/tests/metadata/test_event_processing_base.py b/tests/metadata/test_event_processing_base.py
index d2199e418..f46543250 100644
--- a/tests/metadata/test_event_processing_base.py
+++ b/tests/metadata/test_event_processing_base.py
@@ -20,7 +20,7 @@ from tests.common.impala_test_suite import ImpalaTestSuite
from tests.util.event_processor_utils import EventProcessorUtils
EVENT_SYNC_QUERY_OPTIONS = {
- "sync_hms_events_wait_time_s": 10,
+ "sync_hms_events_wait_time_s": 30,
"sync_hms_events_strict_mode": True
}
(impala) 02/02: IMPALA-14588: Implement Iceberg deletion vector reading/writing
Posted by st...@apache.org.
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