Merge pull request ClickHouse#88537 from ClickHouse/backport/25.8/87903

Backport ClickHouse#87903 to 25.8: Add more information for Iceberg SELECT queries profiling

Author: clickhouse-gh[bot]

src/Common/ProfileEvents.cpp

@@ -87,6 +87,10 @@
     M(IcebergMetadataFilesCacheHits, "Number of times iceberg metadata files have been found in the cache.", ValueType::Number) \
     M(IcebergMetadataFilesCacheMisses, "Number of times iceberg metadata files have not been found in the iceberg metadata cache and had to be read from (remote) disk.", ValueType::Number) \
     M(IcebergMetadataFilesCacheWeightLost, "Approximate number of bytes evicted from the iceberg metadata cache.", ValueType::Number) \
+    M(IcebergMetadataReadWaitTimeMicroseconds, "Total time data readers spend waiting for iceberg metadata files to be read and parsed, summed across all reader threads.", ValueType::Microseconds) \
+    M(IcebergIteratorInitializationMicroseconds, "Total time spent on synchronous initialization of iceberg data iterators.", ValueType::Microseconds) \
+    M(IcebergMetadataUpdateMicroseconds, "Total time spent on synchronous initialization of iceberg data iterators.", ValueType::Microseconds) \
+    M(IcebergMetadataReturnedObjectInfos, "Total number of returned object infos from iceberg iterator.", ValueType::Number) \
     M(VectorSimilarityIndexCacheHits, "Number of times an index granule has been found in the vector index cache.", ValueType::Number) \
     M(VectorSimilarityIndexCacheMisses, "Number of times an index granule has not been found in the vector index cache and had to be read from disk.", ValueType::Number) \
     M(VectorSimilarityIndexCacheWeightLost, "Approximate number of bytes evicted from the vector index cache.", ValueType::Number) \

src/Interpreters/IcebergMetadataLog.cpp

@@ -44,6 +44,12 @@ namespace
 
 const DataTypePtr rowType = makeNullable(std::make_shared<DataTypeUInt64>());
 
+auto iceberg_pruning_status_datatype = std::make_shared<DataTypeEnum8>(DataTypeEnum8::Values{
+    {"NotPruned", static_cast<Int8>(Iceberg::PruningReturnStatus::NOT_PRUNED)},
+    {"PartitionPruned", static_cast<Int8>(Iceberg::PruningReturnStatus::PARTITION_PRUNED)},
+    {"MinMaxIndexPruned", static_cast<Int8>(Iceberg::PruningReturnStatus::MIN_MAX_INDEX_PRUNED)}});
+
+const DataTypePtr iceberg_pruning_status_datatype_nullable = makeNullable(iceberg_pruning_status_datatype);
 }
 
 ColumnsDescription IcebergMetadataLogElement::getColumnsDescription()
@@ -64,7 +70,8 @@ ColumnsDescription IcebergMetadataLogElement::getColumnsDescription()
         {"table_path", std::make_shared<DataTypeString>(), "Table path."},
         {"file_path", std::make_shared<DataTypeString>(), "File path."},
         {"content", std::make_shared<DataTypeString>(), "Content in a JSON format (json file content, avro metadata or avro entry)."},
-        {"row_in_file", rowType, "Row in file."}};
+        {"row_in_file", rowType, "Row in file."},
+        {"pruning_status", iceberg_pruning_status_datatype_nullable, "Status of partition pruning or min-max index pruning for the file."}};
 }
 
 void IcebergMetadataLogElement::appendToBlock(MutableColumns & columns) const
@@ -79,6 +86,7 @@ void IcebergMetadataLogElement::appendToBlock(MutableColumns & columns) const
     columns[column_index++]->insert(file_path);
     columns[column_index++]->insert(metadata_content);
     columns[column_index++]->insert(row_in_file ? *row_in_file : rowType->getDefault());
+    columns[column_index++]->insert(pruning_status ? *pruning_status : iceberg_pruning_status_datatype_nullable->getDefault());
 }
 
 void insertRowToLogTable(
@@ -87,7 +95,8 @@ void insertRowToLogTable(
     IcebergMetadataLogLevel row_log_level,
     const String & table_path,
     const String & file_path,
-    std::optional<UInt64> row_in_file)
+    std::optional<UInt64> row_in_file,
+    std::optional<Iceberg::PruningReturnStatus> pruning_status)
 {
     IcebergMetadataLogLevel set_log_level = local_context->getSettingsRef()[Setting::iceberg_metadata_log_level].value;
     if (set_log_level < row_log_level)
@@ -104,6 +113,7 @@ void insertRowToLogTable(
             .table_path = table_path,
             .file_path = file_path,
             .metadata_content = row,
-            .row_in_file = row_in_file});
+            .row_in_file = row_in_file,
+            .pruning_status = pruning_status});
 }
 }

src/Interpreters/IcebergMetadataLog.h

@@ -2,6 +2,7 @@
 
 #include <Interpreters/SystemLog.h>
 #include <Storages/ColumnsDescription.h>
+#include <Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h>
 
 namespace DB
 {
@@ -15,6 +16,7 @@ struct IcebergMetadataLogElement
     String file_path;
     String metadata_content;
     std::optional<UInt64> row_in_file;
+    std::optional<Iceberg::PruningReturnStatus> pruning_status;
 
     static std::string name() { return "IcebergMetadataLog"; }
 
@@ -29,7 +31,8 @@ void insertRowToLogTable(
     IcebergMetadataLogLevel row_log_level,
     const String & table_path,
     const String & file_path,
-    std::optional<UInt64> row_in_file);
+    std::optional<UInt64> row_in_file,
+    std::optional<Iceberg::PruningReturnStatus> pruning_status);
 
 class IcebergMetadataLog : public SystemLog<IcebergMetadataLogElement>
 {

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp

@@ -1,4 +1,3 @@
-
 #include "config.h"
 #if USE_AVRO
 
@@ -45,13 +44,21 @@
 
 #include <Storages/ObjectStorage/DataLakes/Iceberg/StatelessMetadataFileGetter.h>
 
+#include <Common/ProfileEvents.h>
 #include <Common/SharedLockGuard.h>
 #include <Common/logger_useful.h>
 
+#include <Interpreters/IcebergMetadataLog.h>
+#include <base/wide_integer_to_string.h>
+#include <Common/ElapsedTimeProfileEventIncrement.h>
+
+
 namespace ProfileEvents
 {
 extern const Event IcebergPartitionPrunedFiles;
 extern const Event IcebergMinMaxIndexPrunedFiles;
+extern const Event IcebergMetadataReadWaitTimeMicroseconds;
+extern const Event IcebergMetadataReturnedObjectInfos;
 };
 
 
@@ -158,6 +165,14 @@ std::optional<ManifestFileEntry> SingleThreadIcebergKeysIterator::next()
                     local_context);
             }
             auto pruning_status = current_pruner ? current_pruner->canBePruned(manifest_file_entry) : PruningReturnStatus::NOT_PRUNED;
+            insertRowToLogTable(
+                local_context,
+                "",
+                DB::IcebergMetadataLogLevel::ManifestFileEntry,
+                configuration.lock()->getRawPath().path,
+                current_manifest_file_content->getPathToManifestFile(),
+                manifest_file_entry.row_number,
+                pruning_status);
             switch (pruning_status)
             {
                 case PruningReturnStatus::NOT_PRUNED:
@@ -321,6 +336,7 @@ IcebergIterator::IcebergIterator(
 
 ObjectInfoPtr IcebergIterator::next(size_t)
 {
+    ProfileEventTimeIncrement<Microseconds> watch(ProfileEvents::IcebergMetadataReadWaitTimeMicroseconds);
     Iceberg::ManifestFileEntry manifest_file_entry;
     if (blocking_queue.pop(manifest_file_entry))
     {
@@ -333,6 +349,7 @@ ObjectInfoPtr IcebergIterator::next(size_t)
         {
             object_info->addEqualityDeleteObject(equality_delete);
         }
+        ProfileEvents::increment(ProfileEvents::IcebergMetadataReturnedObjectInfos);
         return object_info;
     }
     {

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp

@@ -68,7 +68,9 @@
 
 namespace ProfileEvents
 {
-    extern const Event IcebergTrivialCountOptimizationApplied;
+extern const Event IcebergIteratorInitializationMicroseconds;
+extern const Event IcebergMetadataUpdateMicroseconds;
+extern const Event IcebergTrivialCountOptimizationApplied;
 }
 
 namespace DB
@@ -249,6 +251,7 @@ bool IcebergMetadata::update(const ContextPtr & local_context)
         DB::IcebergMetadataLogLevel::Metadata,
         configuration_ptr->getRawPath().path,
         metadata_file_path,
+        std::nullopt,
         std::nullopt);
 
     if (previous_snapshot_id != relevant_snapshot_id)
@@ -587,6 +590,7 @@ DataLakeMetadataPtr IcebergMetadata::create(
         DB::IcebergMetadataLogLevel::Metadata,
         configuration_ptr->getRawPath().path,
         metadata_file_path,
+        std::nullopt,
         std::nullopt);
     return std::make_unique<IcebergMetadata>(object_storage, configuration_ptr, local_context, metadata_version, format_version, object, cache_ptr, compression_method);
 }
@@ -782,6 +786,8 @@ ObjectIterator IcebergMetadata::iterate(
 {
     SharedLockGuard lock(mutex);
 
+    ProfileEventTimeIncrement<Microseconds> watch(ProfileEvents::IcebergIteratorInitializationMicroseconds);
+
     auto table_snapshot
         = std::make_shared<IcebergTableStateSnapshot>(last_metadata_version, relevant_snapshot_schema_id, relevant_snapshot_id);
     return std::make_shared<IcebergIterator>(

src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.cpp

@@ -160,6 +160,7 @@ ManifestFileContent::ManifestFileContent(
         DB::IcebergMetadataLogLevel::ManifestFileMetadata,
         common_path,
         path_to_manifest_file,
+        std::nullopt,
         std::nullopt);
 
     for (const auto & column_name : {f_status, f_data_file})
@@ -235,7 +236,8 @@ ManifestFileContent::ManifestFileContent(
             DB::IcebergMetadataLogLevel::ManifestFileEntry,
             common_path,
             path_to_manifest_file,
-            i);
+            i,
+            std::nullopt);
         FileContentType content_type = FileContentType::DATA;
         if (format_version_ > 1)
             content_type = FileContentType(manifest_file_deserializer.getValueFromRowByName(i, c_data_file_content, TypeIndex::Int32).safeGet<UInt64>());
@@ -418,6 +420,7 @@ ManifestFileContent::ManifestFileContent(
                 this->data_files_without_deleted.emplace_back(
                     file_path_key,
                     file_path,
+                    i,
                     status,
                     added_sequence_number,
                     snapshot_id,
@@ -444,6 +447,7 @@ ManifestFileContent::ManifestFileContent(
                 this->position_deletes_files_without_deleted.emplace_back(
                     file_path_key,
                     file_path,
+                    i,
                     status,
                     added_sequence_number,
                     snapshot_id,
@@ -472,6 +476,7 @@ ManifestFileContent::ManifestFileContent(
                 this->equality_deletes_files.emplace_back(
                     file_path_key,
                     file_path,
+                    i,
                     status,
                     added_sequence_number,
                     snapshot_id,

src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h

@@ -61,6 +61,7 @@ struct ManifestFileEntry
     String file_path_key;
     // It's a processed file path to be used by Object Storage
     String file_path;
+    Int64 row_number;
 
     ManifestEntryStatus status;
     Int64 added_sequence_number;

src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h

@@ -1,7 +1,6 @@
 #pragma once
 #include "config.h"
 
-#if USE_AVRO
 
 #include <Core/NamesAndTypes.h>
 #include <Parsers/IAST_fwd.h>
@@ -21,6 +20,13 @@ enum class PruningReturnStatus
     MIN_MAX_INDEX_PRUNED
 };
 
+}
+
+#if USE_AVRO
+
+namespace DB::Iceberg
+{
+
 struct ManifestFileEntry;
 class ManifestFileContent;
 

src/Storages/ObjectStorage/DataLakes/Iceberg/StatelessMetadataFileGetter.cpp

@@ -153,6 +153,7 @@ ManifestFileCacheKeys getManifestList(
             DB::IcebergMetadataLogLevel::ManifestListMetadata,
             configuration_ptr->getRawPath().path,
             filename,
+            std::nullopt,
             std::nullopt);
 
         for (size_t i = 0; i < manifest_list_deserializer.rows(); ++i)
@@ -187,7 +188,8 @@ ManifestFileCacheKeys getManifestList(
                 DB::IcebergMetadataLogLevel::ManifestListEntry,
                 configuration_ptr->getRawPath().path,
                 filename,
-                i);
+                i,
+                std::nullopt);
         }
         /// We only return the list of {file name, seq number} for cache.
         /// Because ManifestList holds a list of ManifestFilePtr which consume much memory space.
@@ -248,7 +250,6 @@ std::pair<Poco::JSON::Object::Ptr, Int32> parseTableSchemaV1Method(const Poco::J
     auto current_schema_id = schema->getValue<int>(f_schema_id);
     return {schema, current_schema_id};
 }
-
 }
 }