Fixes for Iceberg Puffin Delete Vectors

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);

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

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);
         }

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;

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);
         }
 

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);