diff --git a/src/Common/CacheBase.h b/src/Common/CacheBase.h
index 29d06f6382bc8..6baf4996cfe5e 100644
--- a/src/Common/CacheBase.h
+++ b/src/Common/CacheBase.h
@@ -269,7 +269,9 @@ class CacheBase
         return cache_policy->dump();
     }
 
-    void clear()
+    /// Virtual so a derived cache that keeps state outside `cache_policy` is not skipped
+    /// when the call goes through a `CacheBase` pointer.
+    virtual void clear()
     {
         std::lock_guard lock(mutex);
         insert_tokens.clear();
@@ -333,13 +335,15 @@ class CacheBase
         return cache_policy->maxCount();
     }
 
-    void setMaxCount(size_t max_count)
+    /// Virtual for the same reason as `clear`.
+    virtual void setMaxCount(size_t max_count)
     {
         std::lock_guard lock(mutex);
         cache_policy->setMaxCount(max_count);
     }
 
-    void setMaxSizeInBytes(size_t max_size_in_bytes)
+    /// Virtual for the same reason as `clear`.
+    virtual void setMaxSizeInBytes(size_t max_size_in_bytes)
     {
         std::lock_guard lock(mutex);
         cache_policy->setMaxSizeInBytes(max_size_in_bytes);
diff --git a/src/IO/tests/gtest_readbuffer_s3.cpp b/src/IO/tests/gtest_readbuffer_s3.cpp
index f9e99df39bb9a..855042407c8e3 100644
--- a/src/IO/tests/gtest_readbuffer_s3.cpp
+++ b/src/IO/tests/gtest_readbuffer_s3.cpp
@@ -35,6 +35,8 @@ namespace DB::Setting
     extern const SettingsBool filesystem_cache_prefer_bigger_buffer_size;
     extern const SettingsBool read_from_filesystem_cache_if_exists_otherwise_bypass_cache;
     extern const SettingsUInt64 remote_read_min_bytes_for_seek;
+    extern const SettingsBool s3_validate_etag_on_read;
+    extern const SettingsBool table_engine_read_through_distributed_cache;
 }
 
 namespace DB::FileCacheSetting
@@ -141,6 +143,7 @@ struct ClientFake : DB::S3::Client
     using ListObjectsV2Fn = std::function<Aws::S3::Model::ListObjectsV2Outcome(const Aws::S3::Model::ListObjectsV2Request &)>;
     std::optional<ListObjectsV2Fn> listObjectsV2Impl;
     mutable std::optional<std::string> last_start_after;
+    mutable std::string last_if_match;
     struct ListRequest
     {
         std::string start_after;
@@ -163,8 +166,9 @@ struct ClientFake : DB::S3::Client
         };
     }
 
-    Aws::S3::Model::GetObjectOutcome GetObject([[maybe_unused]] const Aws::S3::Model::GetObjectRequest & request) const override
+    Aws::S3::Model::GetObjectOutcome GetObject(const Aws::S3::Model::GetObjectRequest & request) const override
     {
+        last_if_match = request.GetIfMatch();
         chassert(getObjectImpl);
         return (*getObjectImpl)(request);
     }
@@ -438,4 +442,65 @@ TEST_F(ReadBufferFromS3Test, HavingZeroBytes)
 
     DB::FileCacheFactory::instance().clear();
 }
+
+TEST_F(ReadBufferFromS3Test, CreateReadBufferCanPinListedEtagWhenValidationIsOff)
+{
+    /// The Puffin deletion-vector cache stores bytes under the etag from its HEAD. That read must
+    /// send `If-Match` even when `s3_validate_etag_on_read` is off, or a newer object can be cached
+    /// under the listed etag.
+    DB::ServerUUID::setRandomForUnitTests();
+    auto query_context = DB::Context::createCopy(getContext().context);
+    query_context->makeQueryContext();
+    query_context->setCurrentQueryId("puffin_etag_pin");
+    const auto & settings = query_context->getSettingsRef();
+    const_cast<DB::Settings &>(settings)[DB::Setting::s3_validate_etag_on_read] = false;
+    const_cast<DB::Settings &>(settings)[DB::Setting::filesystem_cache_name] = "";
+    const_cast<DB::Settings &>(settings)[DB::Setting::table_engine_read_through_distributed_cache] = false;
+
+    auto make_storage = []()
+    {
+        auto client = std::make_unique<ClientFake>();
+        auto * fake = client.get();
+        DB::S3::URI uri;
+        uri.bucket = "test_bucket";
+        DB::S3Capabilities cap;
+        DB::ObjectStorageKeyGeneratorPtr gen;
+        auto object_storage = std::make_shared<DB::S3ObjectStorage>(
+            std::move(client), std::make_unique<DB::S3Settings>(), std::move(uri), cap, gen, "s3");
+        return std::pair{object_storage, fake};
+    };
+
+    DB::ReadSettings read_settings;
+    read_settings.remote_fs_settings.method = DB::RemoteFSReadMethod::read;
+    read_settings.remote_fs_settings.prefetch = false;
+    read_settings.enable_filesystem_cache = false;
+    read_settings.use_page_cache_for_object_storage = false;
+
+    const auto read_with_pin = [&](bool pin_to_metadata_etag)
+    {
+        auto [object_storage, fake] = make_storage();
+        DB::ObjectMetadata object_metadata;
+        object_metadata.size_bytes = 4;
+        object_metadata.etag = "\"etag-at-listing\"";
+        DB::RelativePathWithMetadata relative_path("test_key", object_metadata);
+
+        auto session = std::make_shared<CountedSession>();
+        const auto stream_buf = std::make_shared<StringHTTPBasicStreamBuf>("abcd");
+        fake->setGetObjectSuccess(session, stream_buf.get());
+
+        auto buf = DB::createReadBuffer(
+            relative_path,
+            object_storage,
+            query_context,
+            getLogger("test"),
+            read_settings,
+            /*allow_page_cache=*/ false,
+            pin_to_metadata_etag);
+        EXPECT_TRUE(buf->next());
+        return fake->last_if_match;
+    };
+
+    EXPECT_EQ(read_with_pin(false), "");
+    EXPECT_EQ(read_with_pin(true), "\"etag-at-listing\"");
+}
 #endif
diff --git a/src/Interpreters/ClusterFunctionReadTask.cpp b/src/Interpreters/ClusterFunctionReadTask.cpp
index 05a7519bfbe7b..bdd62a0ebc57f 100644
--- a/src/Interpreters/ClusterFunctionReadTask.cpp
+++ b/src/Interpreters/ClusterFunctionReadTask.cpp
@@ -27,6 +27,18 @@ namespace Setting
     extern const SettingsBool cluster_function_process_archive_on_multiple_nodes;
 }
 
+namespace
+{
+
+/// An empty DAG is the on-wire stand-in for a missing `schema_transform`. A transform that only
+/// adds columns has outputs and no inputs; that is still schema evolution and must be kept.
+bool schemaTransformCarriesColumns(const ActionsDAG & dag)
+{
+    return !dag.getInputs().empty() || !dag.getOutputs().empty();
+}
+
+}
+
 ClusterFunctionReadTaskResponse::ClusterFunctionReadTaskResponse(ObjectInfoPtr object, const ContextPtr & context)
 {
     if (!object)
@@ -96,10 +108,11 @@ void ClusterFunctionReadTaskResponse::serialize(WriteBuffer & out, size_t worker
         = std::min(static_cast<UInt64>(worker_protocol_version), static_cast<UInt64>(DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION));
 
     /// Fail closed: protocol < 2 omits `schema_transform`, so workers would skip data-lake schema
-    /// evolution and return wrong columns / values.
+    /// evolution and return wrong columns / values. An output-only DAG (added columns, no inputs)
+    /// is still evolution; only a fully empty DAG is the missing-transform sentinel.
     if (protocol_version < DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_DATA_LAKE_METADATA
         && data_lake_metadata.schema_transform
-        && !data_lake_metadata.schema_transform->getInputs().empty())
+        && schemaTransformCarriesColumns(*data_lake_metadata.schema_transform))
     {
         throw Exception(
             ErrorCodes::UNKNOWN_PROTOCOL,
@@ -256,7 +269,7 @@ void ClusterFunctionReadTaskResponse::deserialize(ReadBuffer & in, UInt64 antaly
         /// Trusted intra-cluster metadata: decode types without the input complexity limit.
         auto transform = std::make_shared<ActionsDAG>(ActionsDAG::deserialize(in, registry, Context::getGlobalContextInstance(), 0));
 
-        if (!path.empty() && !transform->getInputs().empty())
+        if (!path.empty() && schemaTransformCarriesColumns(*transform))
         {
             data_lake_metadata.schema_transform = std::move(transform);
         }
diff --git a/src/Interpreters/tests/gtest_cluster_function_read_task.cpp b/src/Interpreters/tests/gtest_cluster_function_read_task.cpp
index 63b0c0dfb8d1f..621602f38feaa 100644
--- a/src/Interpreters/tests/gtest_cluster_function_read_task.cpp
+++ b/src/Interpreters/tests/gtest_cluster_function_read_task.cpp
@@ -49,6 +49,79 @@ TEST(ClusterFunctionReadTaskResponse, RejectsSchemaTransformOnProtocolBeforeData
     }
 }
 
+namespace
+{
+
+std::shared_ptr<ActionsDAG> makeOutputOnlySchemaTransform()
+{
+    auto dag = std::make_shared<ActionsDAG>();
+    auto type = std::make_shared<DataTypeUInt64>();
+    auto constant = type->createColumnConstWithDefaultValue(0);
+    const auto & added = dag->addColumn(std::move(constant), type, "added");
+    dag->getOutputs().push_back(&added);
+    return dag;
+}
+
+}
+
+TEST(ClusterFunctionReadTaskResponse, RejectsOutputOnlySchemaTransformOnProtocolBeforeDataLakeMetadata)
+{
+    ClusterFunctionReadTaskResponse response;
+    response.path = "/path/file.parquet";
+    response.data_lake_metadata.schema_transform = makeOutputOnlySchemaTransform();
+
+    String serialized;
+    WriteBufferFromString out(serialized);
+    try
+    {
+        response.serialize(out, DBMS_CLUSTER_INITIAL_PROCESSING_PROTOCOL_VERSION, DBMS_ANTALYA_PROTOCOL_VERSION);
+        FAIL() << "Expected exception";
+    }
+    catch (const Exception & e)
+    {
+        EXPECT_EQ(e.code(), ErrorCodes::UNKNOWN_PROTOCOL);
+        EXPECT_NE(e.message().find("schema_transform"), std::string::npos);
+    }
+}
+
+TEST(ClusterFunctionReadTaskResponse, RoundTripsOutputOnlySchemaTransformOnSupportedProtocol)
+{
+    ClusterFunctionReadTaskResponse response;
+    response.path = "/path/file.parquet";
+    response.data_lake_metadata.schema_transform = makeOutputOnlySchemaTransform();
+
+    String serialized;
+    WriteBufferFromString out(serialized);
+    response.serialize(out, DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_DATA_LAKE_METADATA, DBMS_ANTALYA_PROTOCOL_VERSION);
+    out.finalize();
+
+    ReadBufferFromString in(serialized);
+    ClusterFunctionReadTaskResponse deserialized;
+    deserialized.deserialize(in, DBMS_ANTALYA_PROTOCOL_VERSION);
+
+    ASSERT_TRUE(deserialized.data_lake_metadata.schema_transform);
+    EXPECT_TRUE(deserialized.data_lake_metadata.schema_transform->getInputs().empty());
+    ASSERT_EQ(deserialized.data_lake_metadata.schema_transform->getOutputs().size(), 1u);
+    EXPECT_EQ(deserialized.data_lake_metadata.schema_transform->getOutputs().front()->result_name, "added");
+}
+
+TEST(ClusterFunctionReadTaskResponse, DropsEmptySchemaTransformSentinel)
+{
+    ClusterFunctionReadTaskResponse response;
+    response.path = "/path/file.parquet";
+
+    String serialized;
+    WriteBufferFromString out(serialized);
+    response.serialize(out, DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_DATA_LAKE_METADATA, DBMS_ANTALYA_PROTOCOL_VERSION);
+    out.finalize();
+
+    ReadBufferFromString in(serialized);
+    ClusterFunctionReadTaskResponse deserialized;
+    deserialized.deserialize(in, DBMS_ANTALYA_PROTOCOL_VERSION);
+
+    EXPECT_FALSE(deserialized.data_lake_metadata.schema_transform);
+}
+
 TEST(ClusterFunctionReadTaskResponse, AllowsEmptySchemaTransformOnProtocolBeforeDataLakeMetadata)
 {
     ClusterFunctionReadTaskResponse response;
diff --git a/src/Processors/Formats/Impl/PuffinBlockInputFormat.cpp b/src/Processors/Formats/Impl/PuffinBlockInputFormat.cpp
index db9559027330b..e55a0fbec1ed8 100644
--- a/src/Processors/Formats/Impl/PuffinBlockInputFormat.cpp
+++ b/src/Processors/Formats/Impl/PuffinBlockInputFormat.cpp
@@ -700,6 +700,15 @@ roaring::Roaring readRoaringPortableSafe(const char * data, size_t size, Int32 k
 
 std::vector<UInt64> deserializeRoaringPositionBitmap(std::string_view bytes, std::optional<UInt64> expected_cardinality)
 {
+    /// A missing declared cardinality must not skip the materialization ceiling. Callers that pass
+    /// one still fail here before `positions.reserve`.
+    if (expected_cardinality.has_value() && *expected_cardinality > PUFFIN_DV_MAX_MATERIALIZED_POSITIONS)
+        throw Exception(
+            ErrorCodes::BAD_ARGUMENTS,
+            "Deletion vector cardinality {} exceeds materialization limit {}",
+            *expected_cardinality,
+            PUFFIN_DV_MAX_MATERIALIZED_POSITIONS);
+
     if (bytes.size() < sizeof(Int64))
         throw Exception(ErrorCodes::BAD_ARGUMENTS, "Deletion vector bitmap is too small");
 
@@ -742,31 +751,36 @@ std::vector<UInt64> deserializeRoaringPositionBitmap(std::string_view bytes, std
         if (bitmap_size > remaining)
             throw Exception(ErrorCodes::BAD_ARGUMENTS, "Deletion vector roaring bitmap at key {} exceeds blob size", key);
 
-        if (expected_cardinality.has_value())
-        {
-            const UInt64 bitmap_cardinality = bitmap.cardinality();
-            UInt64 new_running_cardinality = 0;
-            if (common::addOverflow(running_cardinality, bitmap_cardinality, new_running_cardinality))
-                throw Exception(
-                    ErrorCodes::BAD_ARGUMENTS,
-                    "Deletion vector cardinality exceeds declared cardinality {}",
-                    *expected_cardinality);
+        /// Cap on the bitmap's own cardinality, before expanding it into `positions`. This runs
+        /// even when the caller did not declare a cardinality.
+        const UInt64 bitmap_cardinality = bitmap.cardinality();
+        UInt64 new_running_cardinality = 0;
+        if (common::addOverflow(running_cardinality, bitmap_cardinality, new_running_cardinality)
+            || new_running_cardinality > PUFFIN_DV_MAX_MATERIALIZED_POSITIONS)
+            throw Exception(
+                ErrorCodes::BAD_ARGUMENTS,
+                "Deletion vector cardinality exceeds materialization limit {}",
+                PUFFIN_DV_MAX_MATERIALIZED_POSITIONS);
 
-            if (new_running_cardinality > *expected_cardinality)
-                throw Exception(
-                    ErrorCodes::BAD_ARGUMENTS,
-                    "Deletion vector cardinality {} exceeds declared cardinality {}",
-                    new_running_cardinality,
-                    *expected_cardinality);
+        if (expected_cardinality.has_value() && new_running_cardinality > *expected_cardinality)
+            throw Exception(
+                ErrorCodes::BAD_ARGUMENTS,
+                "Deletion vector cardinality {} exceeds declared cardinality {}",
+                new_running_cardinality,
+                *expected_cardinality);
 
-            running_cardinality = new_running_cardinality;
-        }
+        running_cardinality = new_running_cardinality;
 
         for (UInt32 sub_position : bitmap)
         {
             const UInt64 position = positionFromKeyAndSubPosition(static_cast<UInt32>(key), sub_position);
             if (position > static_cast<UInt64>(DELETION_VECTOR_MAX_POSITION))
                 throw Exception(ErrorCodes::BAD_ARGUMENTS, "Deletion vector position {} is out of supported range", position);
+            if (positions.size() >= PUFFIN_DV_MAX_MATERIALIZED_POSITIONS)
+                throw Exception(
+                    ErrorCodes::BAD_ARGUMENTS,
+                    "Deletion vector cardinality exceeds materialization limit {}",
+                    PUFFIN_DV_MAX_MATERIALIZED_POSITIONS);
             positions.push_back(position);
         }
 
diff --git a/src/Processors/Formats/Impl/PuffinBlockInputFormat.h b/src/Processors/Formats/Impl/PuffinBlockInputFormat.h
index 251e96ef60f52..cdbf07e485640 100644
--- a/src/Processors/Formats/Impl/PuffinBlockInputFormat.h
+++ b/src/Processors/Formats/Impl/PuffinBlockInputFormat.h
@@ -37,6 +37,8 @@ std::vector<PuffinBlob> readPuffinFooterFromSeekable(SeekableReadBuffer & seekab
 
 /// Shared deletion-vector-v1 payload helpers (also used by `PuffinDeletionVectorReader`).
 std::string_view extractDeletionVectorPayload(std::string_view blob);
+/// `expected_cardinality` may be omitted. Materialization is still capped at
+/// `PUFFIN_DV_MAX_MATERIALIZED_POSITIONS` either way.
 std::vector<UInt64> deserializeRoaringPositionBitmap(
     std::string_view bytes, std::optional<UInt64> expected_cardinality = std::nullopt);
 void deserializeDeletionVectorV1(std::string_view blob, UInt64 expected_cardinality, ColumnUInt64 & positions);
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDeletionVector.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDeletionVector.cpp
index a03becacda081..065e7cb6aa96f 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDeletionVector.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDeletionVector.cpp
@@ -20,6 +20,7 @@
 
 #include <atomic>
 #include <optional>
+#include <string_view>
 
 namespace DB
 {
@@ -83,6 +84,37 @@ namespace
 
 using FooterBlobsPtr = PuffinFilesCache::FooterBlobsPtr;
 
+/// Marker for the local/HDFS/Web generation check. S3 and Azure report the same race with their own
+/// error codes, from the `If-Match` on the listed etag. The mismatch is not retried: a later object
+/// must not be applied under the listed offset, length, and cardinality.
+constexpr std::string_view PUFFIN_LISTED_ETAG_MISMATCH = "changed while loading a cached Puffin deletion vector";
+
+bool storagePinsReadToListedEtag(const IObjectStorage & object_storage)
+{
+    const auto type = object_storage.getType();
+    return type == ObjectStorageType::S3 || type == ObjectStorageType::Azure;
+}
+
+/// S3 and Azure enforce the listed etag on the GET itself. Other backends ignore `StoredObject::etag`,
+/// so compare the etag again after the bytes have been read and refuse to cache a different generation.
+void throwIfListedEtagDrifted(const IObjectStorage & object_storage, const String & puffin_path, const ObjectMetadata & listed)
+{
+    if (!listed.isEtagUsableAsCacheKey() || storagePinsReadToListedEtag(object_storage))
+        return;
+
+    const auto current = object_storage.getObjectMetadata(puffin_path, /*with_tags=*/ false);
+    if (current.etag == listed.etag)
+        return;
+
+    throw Exception(
+        ErrorCodes::BAD_ARGUMENTS,
+        "Puffin object '{}' {} (listed etag {}, current etag {})",
+        puffin_path,
+        PUFFIN_LISTED_ETAG_MISMATCH,
+        listed.etag,
+        current.etag);
+}
+
 void logUndersizedPuffinFilesCacheOnce(LoggerPtr log, size_t max_size_in_bytes, UInt64 minimum_entry_weight)
 {
     static std::atomic_flag logged = ATOMIC_FLAG_INIT;
@@ -103,17 +135,21 @@ FooterBlobsPtr readFooterBlobs(
     const std::optional<ObjectMetadata> & puffin_metadata,
     ContextPtr context,
     LoggerPtr log,
-    bool disable_filesystem_cache)
+    bool disable_filesystem_cache,
+    bool pin_listed_etag)
 {
     /// Seeding the metadata matters: `createReadBuffer` issues its own `getObjectMetadata`
     /// (an S3 `HEAD`) only when it is empty, so a caller that already has it pays one request
     /// for the whole deletion-vector load instead of one per buffer it opens.
+    /// `pin_listed_etag` sends that same etag as `If-Match` on the GET, including when
+    /// `s3_validate_etag_on_read` / `azure_validate_etag_on_read` is off.
     RelativePathWithMetadata puffin_object{puffin_path, puffin_metadata};
     auto read_settings = context->getReadSettings();
     if (disable_filesystem_cache)
         read_settings.enable_filesystem_cache = false;
 
-    auto read_buffer = createReadBuffer(puffin_object, object_storage, context, log, read_settings);
+    auto read_buffer = createReadBuffer(
+        puffin_object, object_storage, context, log, read_settings, /*allow_page_cache=*/ true, pin_listed_etag);
 
     auto * seekable = dynamic_cast<SeekableReadBuffer *>(read_buffer.get());
     if (!seekable)
@@ -123,7 +159,10 @@ FooterBlobsPtr readFooterBlobs(
     if (!file_size)
         throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot determine Puffin file size for '{}'", puffin_path);
 
-    return std::make_shared<const std::vector<PuffinBlob>>(readPuffinFooterBlobsFromSeekable(*seekable, *file_size));
+    auto footer = std::make_shared<const std::vector<PuffinBlob>>(readPuffinFooterBlobsFromSeekable(*seekable, *file_size));
+    if (pin_listed_etag && puffin_metadata)
+        throwIfListedEtagDrifted(*object_storage, puffin_path, *puffin_metadata);
+    return footer;
 }
 
 DataLakeObjectMetadata::ExcludedRowsPtr loadDeletionVectorUncached(
@@ -140,14 +179,16 @@ DataLakeObjectMetadata::ExcludedRowsPtr loadDeletionVectorUncached(
     bool disable_filesystem_cache,
     FooterBlobsPtr preloaded_footer,
     PuffinFilesCache * footer_cache = nullptr,
-    const std::optional<PuffinFooterCacheKey> & footer_key = {})
+    const std::optional<PuffinFooterCacheKey> & footer_key = {},
+    bool pin_listed_etag = false)
 {
     RelativePathWithMetadata puffin_object{puffin_path, puffin_metadata};
     auto read_settings = context->getReadSettings();
     if (disable_filesystem_cache)
         read_settings.enable_filesystem_cache = false;
 
-    auto read_buffer = createReadBuffer(puffin_object, object_storage, context, log, read_settings);
+    auto read_buffer = createReadBuffer(
+        puffin_object, object_storage, context, log, read_settings, /*allow_page_cache=*/ true, pin_listed_etag);
 
     auto * seekable = dynamic_cast<SeekableReadBuffer *>(read_buffer.get());
     if (!seekable)
@@ -170,7 +211,13 @@ DataLakeObjectMetadata::ExcludedRowsPtr loadDeletionVectorUncached(
                 footer_owner = footer_cache->getOrSetFooter(*footer_key, [&]()
                 {
                     return readFooterBlobs(
-                        object_storage, puffin_path, puffin_metadata, context, log, /*disable_filesystem_cache=*/ true);
+                        object_storage,
+                        puffin_path,
+                        puffin_metadata,
+                        context,
+                        log,
+                        /*disable_filesystem_cache=*/ true,
+                        pin_listed_etag);
                 });
             }
             else
@@ -193,6 +240,9 @@ DataLakeObjectMetadata::ExcludedRowsPtr loadDeletionVectorUncached(
 
     validateDeletionVectorPositionsAgainstDataFile(deleted_positions, expected_cardinality, data_file_record_count);
 
+    if (pin_listed_etag && puffin_metadata)
+        throwIfListedEtagDrifted(*object_storage, puffin_path, *puffin_metadata);
+
     if (deleted_positions.empty())
         return nullptr;
 
@@ -336,11 +386,11 @@ DataLakeObjectMetadata::ExcludedRowsPtr loadDeletionVector(
     if (!puffin_object.metadata)
         puffin_object.metadata = object_storage->getObjectMetadata(puffin_object.getPath(), /*with_tags=*/ false);
 
-    if (puffin_object.metadata->etag.empty())
+    if (!puffin_object.metadata->isEtagUsableAsCacheKey())
     {
         LOG_TRACE(
             log,
-            "Not using Puffin files cache for '{}', because etag is empty",
+            "Not using Puffin files cache for '{}', because etag is empty or not a strong content identifier",
             puffin_path);
         return loadDeletionVectorUncached(
             object_storage,
@@ -368,7 +418,7 @@ DataLakeObjectMetadata::ExcludedRowsPtr loadDeletionVector(
         expected_cardinality_u64,
         static_cast<UInt64>(data_file_record_count));
 
-    /// Empty etag is the only reason tryCreate* returns nullopt; that case is handled above.
+    /// Empty etag is the only reason tryCreate* returns nullopt; a weak etag never reaches here.
     if (!footer_key || !cache_key)
     {
         throw Exception(
@@ -403,6 +453,9 @@ DataLakeObjectMetadata::ExcludedRowsPtr loadDeletionVector(
     /// Footer is keyed by file identity only, so N DV slices in one coalesced Puffin share one parse.
     /// Resolve the footer only on a Puffin deletion-vector cache miss (nested memo lookup).
     /// Delta `.bin` objects have no Puffin footer; detection inside the uncached loader skips memo.
+    /// The load is pinned to the etag in the cache key. If the object is overwritten in between, the
+    /// loader throws and `getOrSetDeletionVector` does not insert. The query fails instead of reading
+    /// the new generation under the listed offset, length, and cardinality.
     return cache->getOrSetDeletionVector(*cache_key, [&]()
     {
         return loadDeletionVectorUncached(
@@ -419,7 +472,8 @@ DataLakeObjectMetadata::ExcludedRowsPtr loadDeletionVector(
             true,
             nullptr,
             cache.get(),
-            footer_key);
+            footer_key,
+            /*pin_listed_etag=*/ true);
     });
 }
 
diff --git a/src/Storages/ObjectStorage/DataLakes/PuffinDeletionVectorReader.h b/src/Storages/ObjectStorage/DataLakes/PuffinDeletionVectorReader.h
index 4412015160d1e..7b65aa7ce0aed 100644
--- a/src/Storages/ObjectStorage/DataLakes/PuffinDeletionVectorReader.h
+++ b/src/Storages/ObjectStorage/DataLakes/PuffinDeletionVectorReader.h
@@ -82,6 +82,8 @@ void checkDeletionVectorBlobReadLimits(Int64 length, std::optional<UInt64> expec
 void validateDeletionVectorEnvelope(const UInt8 * header, Int64 length);
 
 /// Deserialize a deletion-vector-v1 blob (magic + CRC wrapper + roaring bitmap payload).
+/// `expected_cardinality` may be omitted. Materialization is still capped at
+/// `PUFFIN_DV_MAX_MATERIALIZED_POSITIONS`.
 std::vector<UInt64> deserializeDeletionVectorV1Blob(std::string_view blob_bytes, std::optional<UInt64> expected_cardinality = std::nullopt);
 
 /// Read a deletion-vector-v1 blob from a Puffin file at the given offset and length.
diff --git a/src/Storages/ObjectStorage/DataLakes/PuffinFilesCache.h b/src/Storages/ObjectStorage/DataLakes/PuffinFilesCache.h
index 8196380ae50dd..67a4d61929123 100644
--- a/src/Storages/ObjectStorage/DataLakes/PuffinFilesCache.h
+++ b/src/Storages/ObjectStorage/DataLakes/PuffinFilesCache.h
@@ -135,10 +135,10 @@ class PuffinFilesCache : public CacheBase<PuffinFilesCacheKey, PuffinFilesCacheC
         const String & etag);
 
     /// Clears deletion-vector entries and footer memo.
-    void clear();
+    void clear() override;
 
-    void setMaxSizeInBytes(size_t max_size_in_bytes);
-    void setMaxCount(size_t max_count);
+    void setMaxSizeInBytes(size_t max_size_in_bytes) override;
+    void setMaxCount(size_t max_count) override;
 
     /// Test/observability helpers for the footer memo.
     size_t footerMemoEntries() const;
diff --git a/src/Storages/ObjectStorage/DataLakes/tests/gtest_puffin_deletion_vector_cardinality.cpp b/src/Storages/ObjectStorage/DataLakes/tests/gtest_puffin_deletion_vector_cardinality.cpp
index 2253f4ac548d2..c851511088e6f 100644
--- a/src/Storages/ObjectStorage/DataLakes/tests/gtest_puffin_deletion_vector_cardinality.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/tests/gtest_puffin_deletion_vector_cardinality.cpp
@@ -3,7 +3,10 @@
 #include <Common/Exception.h>
 #include <IO/ReadBufferFromMemory.h>
 #include <Storages/ObjectStorage/DataLakes/PuffinDeletionVectorReader.h>
+#include <roaring/roaring.hh>
+#include <zlib.h>
 
+#include <string>
 #include <string_view>
 
 using namespace DB;
@@ -28,6 +31,45 @@ constexpr UInt8 two_position_dv_blob[] = {
 
 constexpr Int64 large_declared_length = 64 * 1024 * 1024;
 
+void appendLittleEndian(std::string & out, UInt64 value, size_t width)
+{
+    for (size_t i = 0; i < width; ++i)
+        out.push_back(static_cast<char>((value >> (8 * i)) & 0xFF));
+}
+
+void appendBigEndian32(std::string & out, UInt32 value)
+{
+    out.push_back(static_cast<char>((value >> 24) & 0xFF));
+    out.push_back(static_cast<char>((value >> 16) & 0xFF));
+    out.push_back(static_cast<char>((value >> 8) & 0xFF));
+    out.push_back(static_cast<char>(value & 0xFF));
+}
+
+/// deletion-vector-v1 envelope around one portable roaring bitmap at key 0.
+std::string makeDeletionVectorBlob(const roaring::Roaring & bitmap)
+{
+    const size_t roaring_size = bitmap.getSizeInBytes(/*portable=*/true);
+    std::string roaring_bytes(roaring_size, '\0');
+    const size_t written = bitmap.write(roaring_bytes.data(), /*portable=*/true);
+    EXPECT_EQ(written, roaring_size);
+
+    std::string payload;
+    appendLittleEndian(payload, /*bitmap_count=*/1, sizeof(UInt64));
+    appendLittleEndian(payload, /*key=*/0, sizeof(UInt32));
+    payload.append(roaring_bytes);
+
+    std::string crc_input;
+    crc_input.append(reinterpret_cast<const char *>(DELETION_VECTOR_MAGIC), sizeof(DELETION_VECTOR_MAGIC));
+    crc_input.append(payload);
+    const auto crc = static_cast<UInt32>(crc32_z(0L, reinterpret_cast<const unsigned char *>(crc_input.data()), crc_input.size()));
+
+    std::string blob;
+    appendBigEndian32(blob, static_cast<UInt32>(crc_input.size()));
+    blob.append(crc_input);
+    appendBigEndian32(blob, crc);
+    return blob;
+}
+
 }
 
 TEST(PuffinDeletionVectorCardinality, RejectsCardinalityAboveMaterializationLimitBeforeParse)
@@ -94,3 +136,34 @@ TEST(PuffinDeletionVectorCardinality, AcceptsMatchingCardinality)
     EXPECT_EQ(positions[0], 2u);
     EXPECT_EQ(positions[1], 5u);
 }
+
+TEST(PuffinDeletionVectorCardinality, AcceptsOmittedCardinalityBelowMaterializationLimit)
+{
+    const std::string_view blob(
+        reinterpret_cast<const char *>(two_position_dv_blob), sizeof(two_position_dv_blob));
+
+    const auto positions = deserializeDeletionVectorV1Blob(blob);
+    ASSERT_EQ(positions.size(), 2u);
+    EXPECT_EQ(positions[0], 2u);
+    EXPECT_EQ(positions[1], 5u);
+}
+
+TEST(PuffinDeletionVectorCardinality, RejectsOmittedCardinalityAboveMaterializationLimit)
+{
+    /// A run-length bitmap stays small on disk and expands only when positions are materialized.
+    /// The ceiling must reject it before that expansion, even with no declared cardinality.
+    roaring::Roaring bitmap;
+    bitmap.addRange(0, PUFFIN_DV_MAX_MATERIALIZED_POSITIONS + 1);
+    const std::string blob = makeDeletionVectorBlob(bitmap);
+
+    try
+    {
+        deserializeDeletionVectorV1Blob(blob);
+        FAIL() << "Expected exception";
+    }
+    catch (const Exception & e)
+    {
+        EXPECT_EQ(e.code(), ErrorCodes::BAD_ARGUMENTS);
+        EXPECT_NE(e.message().find("exceeds materialization limit"), std::string::npos);
+    }
+}
diff --git a/src/Storages/ObjectStorage/DataLakes/tests/gtest_puffin_footer_cache.cpp b/src/Storages/ObjectStorage/DataLakes/tests/gtest_puffin_footer_cache.cpp
index cd7ea18fd5536..e2842442a1371 100644
--- a/src/Storages/ObjectStorage/DataLakes/tests/gtest_puffin_footer_cache.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/tests/gtest_puffin_footer_cache.cpp
@@ -293,3 +293,36 @@ TEST(PuffinFooterMemo, SharesBudgetAndMetricsWithDeletionVectors)
     EXPECT_EQ(CurrentMetrics::get(CurrentMetrics::PuffinFilesCacheBytes), bytes_before);
     EXPECT_EQ(CurrentMetrics::get(CurrentMetrics::PuffinFilesCacheFiles), files_before);
 }
+
+TEST(PuffinFooterMemo, BasePointerUpdatesFooterMemo)
+{
+    /// `clear` / `setMaxSizeInBytes` / `setMaxCount` must reach the footer memo even when the
+    /// static type is `CacheBase`. A non-virtual base method would leave the memo resident.
+    PuffinFilesCache cache("SLRU", 1'000'000, 100, 0.5);
+    PuffinFilesCache::Base & base = cache;
+
+    const auto key_a = PuffinFilesCache::tryCreateFooterKey("Local:////test", "a.puffin", "etag-a");
+    const auto key_b = PuffinFilesCache::tryCreateFooterKey("Local:////test", "b.puffin", "etag-b");
+    ASSERT_TRUE(key_a.has_value());
+    ASSERT_TRUE(key_b.has_value());
+
+    ASSERT_NE(cache.getOrSetFooter(*key_a, loadFixtureFooter), nullptr);
+    EXPECT_EQ(cache.footerMemoEntries(), 1u);
+
+    base.clear();
+    EXPECT_EQ(cache.footerMemoEntries(), 0u);
+    EXPECT_EQ(cache.footerMemoBytes(), 0u);
+
+    ASSERT_NE(cache.getOrSetFooter(*key_a, loadFixtureFooter), nullptr);
+    base.setMaxSizeInBytes(0);
+    EXPECT_EQ(cache.footerMemoEntries(), 0u);
+    EXPECT_EQ(cache.footerMemoBytes(), 0u);
+
+    base.setMaxSizeInBytes(1'000'000);
+    ASSERT_NE(cache.getOrSetFooter(*key_a, loadFixtureFooter), nullptr);
+    ASSERT_NE(cache.getOrSetFooter(*key_b, loadFixtureFooter), nullptr);
+    EXPECT_EQ(cache.footerMemoEntries(), 2u);
+
+    base.setMaxCount(1);
+    EXPECT_EQ(cache.footerMemoEntries(), 1u);
+}
diff --git a/src/Storages/ObjectStorage/IObjectIterator.h b/src/Storages/ObjectStorage/IObjectIterator.h
index ee017e44c2d99..6eadf0870999e 100644
--- a/src/Storages/ObjectStorage/IObjectIterator.h
+++ b/src/Storages/ObjectStorage/IObjectIterator.h
@@ -67,8 +67,8 @@ struct ObjectInfo
     /// Sorted absolute row indexes within the file, see FormatFilterInfo::rows_to_read.
     std::shared_ptr<const PaddedPODArray<UInt64>> rows_to_read;
 
-    /// Polymorphic copy: preserves the dynamic type (e.g. `IcebergDataObjectInfo` with equality /
-    /// position deletes, resolved storage, and metadata path) where a plain copy would slice it.
+    /// Polymorphic copy: preserves the dynamic type (`IcebergDataObjectInfo`, `ObjectInfoInArchive`,
+    /// `ObjectStorageQueueObjectInfo`) where a plain copy would slice it.
     virtual std::shared_ptr<ObjectInfo> clone() const { return std::make_shared<ObjectInfo>(*this); }
 
     String getIdentifier(bool include_file_bucket_info = true) const;
diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp
index e73a1383c6007..afb0dd69448bc 100644
--- a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp
+++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp
@@ -1927,7 +1927,8 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBuffer(
     const ContextPtr & context_,
     const LoggerPtr & log,
     const std::optional<ReadSettings> & read_settings,
-    bool allow_page_cache)
+    bool allow_page_cache,
+    bool pin_to_metadata_etag)
 {
     const auto & settings = context_->getSettingsRef();
     const auto & effective_read_settings = read_settings.has_value() ? read_settings.value() : context_->getReadSettings();
@@ -2056,7 +2057,9 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBuffer(
     /// Pin the read to the object generation seen here (etag from the LIST/HEAD): a GET with a
     /// different ETag means an in-place overwrite, reported as S3_OBJECT_CHANGED_DURING_READ or
     /// AZURE_OBJECT_CHANGED_DURING_READ instead of torn cross-generation data.
-    if (validate_etag_on_read && object_info.metadata.has_value())
+    /// `pin_to_metadata_etag` is independent of the backend setting: a caller that will cache the
+    /// bytes under this etag cannot opt out of the pin.
+    if (object_info.metadata.has_value() && (validate_etag_on_read || pin_to_metadata_etag))
         stored_object.etag = object_info.metadata->etag;
     pipeline.setSource(object_storage, StoredObjects{stored_object}, modified_read_settings);
 
diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.h b/src/Storages/ObjectStorage/StorageObjectStorageSource.h
index 196beb0e6bbf8..27d8a4406820d 100644
--- a/src/Storages/ObjectStorage/StorageObjectStorageSource.h
+++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.h
@@ -392,6 +392,10 @@ class StorageObjectStorageSource::ArchiveIterator : public IObjectIterator, priv
 
         size_t fileSizeInArchive() const override { return file_info.uncompressed_size; }
 
+        /// Bucket splitting copies through `clone`. The base implementation slices to `ObjectInfo`
+        /// and drops the archive path, reader, and file info.
+        std::shared_ptr<ObjectInfo> clone() const override { return std::make_shared<ObjectInfoInArchive>(*this); }
+
         const ObjectInfoPtr archive_object;
         const std::string path_in_archive;
         const std::shared_ptr<IArchiveReader> archive_reader;
diff --git a/src/Storages/ObjectStorage/Utils.h b/src/Storages/ObjectStorage/Utils.h
index 096b401f9852e..794c9743ccfeb 100644
--- a/src/Storages/ObjectStorage/Utils.h
+++ b/src/Storages/ObjectStorage/Utils.h
@@ -31,13 +31,17 @@ void validateSupportedColumns(
 /// An empty column name has no identifier to render it with, so it cannot survive analysis.
 void validateLakeSchemaColumnNames(const NamesAndTypesList & schema, std::string_view lake_name);
 
+/// `pin_to_metadata_etag` pins the read to `object_info.metadata->etag` even when the backend's
+/// `validate_etag_on_read` setting is off. The Puffin deletion-vector cache uses it so bytes stored
+/// under a listed etag cannot come from a later in-place overwrite.
 std::unique_ptr<ReadBufferFromFileBase> createReadBuffer(
     RelativePathWithMetadata & object_info,
     const ObjectStoragePtr & object_storage,
     const ContextPtr & context_,
     const LoggerPtr & log,
     const std::optional<ReadSettings> & read_settings = std::nullopt,
-    bool allow_page_cache = true);
+    bool allow_page_cache = true,
+    bool pin_to_metadata_etag = false);
 
 ASTs::iterator getFirstKeyValueArgument(ASTs & args);
 std::unordered_map<std::string, Field> parseKeyValueArguments(const ASTs & function_args, ContextPtr context);
diff --git a/src/Storages/ObjectStorage/tests/gtest_object_info_clone.cpp b/src/Storages/ObjectStorage/tests/gtest_object_info_clone.cpp
new file mode 100644
index 0000000000000..57742737b3436
--- /dev/null
+++ b/src/Storages/ObjectStorage/tests/gtest_object_info_clone.cpp
@@ -0,0 +1,52 @@
+#include <gtest/gtest.h>
+
+#include <IO/Archives/IArchiveReader.h>
+#include <Storages/ObjectStorage/StorageObjectStorageSource.h>
+#include <Storages/ObjectStorageQueue/ObjectStorageQueueSource.h>
+
+using namespace DB;
+
+TEST(ObjectInfoClone, ArchiveEntryKeepsArchiveIdentity)
+{
+    auto archive = std::make_shared<ObjectInfo>("bucket/file.zip");
+    IArchiveReader::FileInfo file_info;
+    file_info.uncompressed_size = 42;
+    file_info.compressed_size = 20;
+
+    auto original = std::make_shared<StorageObjectStorageSource::ArchiveIterator::ObjectInfoInArchive>(
+        archive, "inner.parquet", /*archive_reader=*/ nullptr, std::move(file_info));
+
+    ObjectInfo sliced_value = *original;
+    auto sliced = std::make_shared<ObjectInfo>(sliced_value);
+    EXPECT_FALSE(sliced->isArchive());
+    EXPECT_FALSE(std::dynamic_pointer_cast<StorageObjectStorageSource::ArchiveIterator::ObjectInfoInArchive>(sliced));
+
+    auto cloned = original->clone();
+    auto archive_cloned = std::dynamic_pointer_cast<StorageObjectStorageSource::ArchiveIterator::ObjectInfoInArchive>(cloned);
+    ASSERT_TRUE(archive_cloned);
+    EXPECT_TRUE(archive_cloned->isArchive());
+    EXPECT_EQ(archive_cloned->getPathToArchive(), "bucket/file.zip");
+    EXPECT_EQ(archive_cloned->getFileName(), "inner.parquet");
+    EXPECT_EQ(archive_cloned->fileSizeInArchive(), 42u);
+    EXPECT_EQ(archive_cloned->archive_object.get(), archive.get());
+    EXPECT_EQ(archive_cloned->file_info.compressed_size, 20u);
+}
+
+TEST(ObjectInfoClone, QueueEntryKeepsFileMetadataType)
+{
+    ObjectInfo listed("queue/file.parquet");
+    auto original = std::make_shared<ObjectStorageQueueSource::ObjectStorageQueueObjectInfo>(listed, /*file_metadata=*/ nullptr);
+    original->data_lake_metadata.emplace();
+
+    ObjectInfo sliced_value = *original;
+    auto sliced = std::make_shared<ObjectInfo>(sliced_value);
+    EXPECT_FALSE(std::dynamic_pointer_cast<ObjectStorageQueueSource::ObjectStorageQueueObjectInfo>(sliced));
+
+    auto cloned = original->clone();
+    auto queue_cloned = std::dynamic_pointer_cast<ObjectStorageQueueSource::ObjectStorageQueueObjectInfo>(cloned);
+    ASSERT_TRUE(queue_cloned);
+    EXPECT_EQ(queue_cloned->getPath(), "queue/file.parquet");
+    EXPECT_FALSE(queue_cloned->file_metadata);
+    EXPECT_TRUE(queue_cloned->data_lake_metadata.has_value());
+    EXPECT_NE(queue_cloned.get(), original.get());
+}
diff --git a/src/Storages/ObjectStorageQueue/ObjectStorageQueueSource.h b/src/Storages/ObjectStorageQueue/ObjectStorageQueueSource.h
index 6767b30c9b537..2f64534b89cfe 100644
--- a/src/Storages/ObjectStorageQueue/ObjectStorageQueueSource.h
+++ b/src/Storages/ObjectStorageQueue/ObjectStorageQueueSource.h
@@ -38,6 +38,10 @@ class ObjectStorageQueueSource final : public ISource, WithContext
             const ObjectInfo & object_info,
             FileMetadataPtr file_metadata_);
 
+        /// Bucket splitting copies through `clone`. The base implementation slices to `ObjectInfo`
+        /// and drops `file_metadata`.
+        std::shared_ptr<ObjectInfo> clone() const override { return std::make_shared<ObjectStorageQueueObjectInfo>(*this); }
+
         FileMetadataPtr file_metadata;
     };