Merge pull request ClickHouse#78244 from ClickHouse/Add_system_iceberg_history_table

SmitaRKulkarni · web-flow · commit 8e294b90c245 · 2025-04-15T14:09:03.000Z
Add system iceberg history table
diff --git a/docs/en/operations/system-tables/iceberg_history.md b/docs/en/operations/system-tables/iceberg_history.md
@@ -0,0 +1,24 @@
+---
+description: 'System iceberg snapshot history'
+keywords: ['system iceberg_history']
+slug: /operations/system-tables/iceberg_history
+title: 'system.iceberg_history'
+---
+
+# system.iceberg_history
+
+Contains snapshot history of iceberg table.
+
+Columns:
+
+- `database` ([String](../../sql-reference/data-types/string.md)) — The name of the database the table is in.
+
+- `name` ([String](../../sql-reference/data-types/string.md)) — Table name.
+
+- `made_current_at` ([DateTime](../../sql-reference/data-types/uuid.md)) — Time when the snapshot was made current snapshot.
+
+- `snapshot_id` ([Int64](../../sql-reference/data-types/int-uint.md)) — Snapshot id.
+
+- `parent_id` ([Int64](../../sql-reference/data-types/int-uint.md)) - Snapshot id of the parent snapshot.
+
+- `is_current_ancestor` ([Bool](../../sql-reference/data-types/boolean.md)) - Flag that indicates whether this snapshot is an ancestor of the current snapshot.
diff --git a/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h b/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h
@@ -108,6 +108,8 @@ class DataLakeConfiguration : public BaseStorageConfiguration, public std::enabl
             && current_metadata->supportsSchemaEvolution();
     }
 
+    IDataLakeMetadata * getExternalMetadata() const override { return current_metadata.get(); }
+
     ColumnsDescription updateAndGetCurrentSchema(
         ObjectStoragePtr object_storage,
         ContextPtr context) override
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp
@@ -707,6 +707,85 @@ ManifestListPtr IcebergMetadata::getManifestList(const String & filename) const
     return manifest_list_ptr;
 }
 
+IcebergMetadata::IcebergHistory IcebergMetadata::getHistory() const
+{
+    auto configuration_ptr = configuration.lock();
+
+    const auto [metadata_version, metadata_file_path] = getLatestOrExplicitMetadataFileAndVersion(object_storage, *configuration_ptr, getContext(), log.get());
+
+    chassert(metadata_version == last_metadata_version);
+
+    auto metadata_object = readJSON(metadata_file_path, object_storage, getContext(), log);
+
+    chassert(format_version == metadata_object->getValue<int>(FORMAT_VERSION_FIELD));
+
+    /// History
+    std::vector<Iceberg::IcebergHistoryRecord> iceberg_history;
+
+    auto snapshots = metadata_object->get("snapshots").extract<Poco::JSON::Array::Ptr>();
+    auto snapshot_logs = metadata_object->get("snapshot-log").extract<Poco::JSON::Array::Ptr>();
+
+    std::vector<Int64> ancestors;
+    std::map<Int64, Int64> parents_list;
+    for (size_t i = 0; i < snapshots->size(); ++i)
+    {
+        const auto snapshot = snapshots->getObject(static_cast<UInt32>(i));
+        auto snapshot_id = snapshot->getValue<Int64>("snapshot-id");
+
+        if (snapshot->has("parent-snapshot-id") && !snapshot->isNull("parent-snapshot-id"))
+            parents_list[snapshot_id] = snapshot->getValue<Int64>("parent-snapshot-id");
+        else
+            parents_list[snapshot_id] = 0;
+    }
+
+    auto current_snapshot_id = metadata_object->getValue<Int64>("current-snapshot-id");
+
+    /// Add current snapshot-id to ancestors list
+    ancestors.push_back(current_snapshot_id);
+    while (parents_list[current_snapshot_id] != 0)
+    {
+        ancestors.push_back(parents_list[current_snapshot_id]);
+        current_snapshot_id = parents_list[current_snapshot_id];
+    }
+
+    for (size_t i = 0; i < snapshots->size(); ++i)
+    {
+        IcebergHistoryRecord history_record;
+
+        const auto snapshot = snapshots->getObject(static_cast<UInt32>(i));
+        history_record.snapshot_id = snapshot->getValue<Int64>("snapshot-id");
+
+        if (snapshot->has("parent-snapshot-id") && !snapshot->isNull("parent-snapshot-id"))
+            history_record.parent_id = snapshot->getValue<Int64>("parent-snapshot-id");
+        else
+            history_record.parent_id = 0;
+
+        for (size_t j = 0; j < snapshot_logs->size(); ++j)
+        {
+            const auto snapshot_log = snapshot_logs->getObject(static_cast<UInt32>(j));
+            if (snapshot_log->getValue<Int64>("snapshot-id") == history_record.snapshot_id)
+            {
+                auto value = snapshot_log->getValue<std::string>("timestamp-ms");
+                ReadBufferFromString in(value);
+                DateTime64 time = 0;
+                readDateTime64Text(time, 6, in);
+
+                history_record.made_current_at = time;
+                break;
+            }
+        }
+
+        if (std::find(ancestors.begin(), ancestors.end(), history_record.snapshot_id) != ancestors.end())
+            history_record.is_current_ancestor = true;
+        else
+            history_record.is_current_ancestor = false;
+
+        iceberg_history.push_back(history_record);
+    }
+
+    return iceberg_history;
+}
+
 ManifestFilePtr IcebergMetadata::getManifestFile(const String & filename, Int64 inherited_sequence_number) const
 {
     auto configuration_ptr = configuration.lock();
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h
@@ -29,7 +29,7 @@ class IcebergMetadata : public IDataLakeMetadata, private WithContext
 public:
     using ConfigurationObserverPtr = StorageObjectStorage::ConfigurationObserverPtr;
     using ConfigurationPtr = StorageObjectStorage::ConfigurationPtr;
-
+    using IcebergHistory = std::vector<Iceberg::IcebergHistoryRecord>;
 
     static constexpr auto name = "Iceberg";
 
@@ -82,6 +82,8 @@ class IcebergMetadata : public IDataLakeMetadata, private WithContext
 
     bool update(const ContextPtr & local_context) override;
 
+    IcebergHistory getHistory() const;
+
     std::optional<size_t> totalRows() const override;
     std::optional<size_t> totalBytes() const override;
 
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Snapshot.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/Snapshot.h
@@ -4,6 +4,7 @@
 #if USE_AVRO
 
 #include <Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h>
+#include <DataTypes/DataTypeDateTime64.h>
 
 namespace Iceberg
 {
@@ -18,6 +19,15 @@ struct IcebergSnapshot
     std::optional<size_t> total_rows;
     std::optional<size_t> total_bytes;
 };
+
+struct IcebergHistoryRecord
+{
+    Int64 snapshot_id;
+    DB::DateTime64 made_current_at;
+    Int64 parent_id;
+    bool is_current_ancestor;
+};
+
 }
 
 #endif
diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.cpp b/src/Storages/ObjectStorage/StorageObjectStorage.cpp
@@ -197,6 +197,11 @@ bool StorageObjectStorage::hasExternalDynamicMetadata() const
     return configuration->hasExternalDynamicMetadata();
 }
 
+IDataLakeMetadata * StorageObjectStorage::getExternalMetadata() const
+{
+    return configuration->getExternalMetadata();
+}
+
 void StorageObjectStorage::updateExternalDynamicMetadata(ContextPtr context_ptr)
 {
     StorageInMemoryMetadata metadata;
diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.h b/src/Storages/ObjectStorage/StorageObjectStorage.h
@@ -9,6 +9,7 @@
 #include <Common/threadPoolCallbackRunner.h>
 #include <Interpreters/ActionsDAG.h>
 #include <Storages/ColumnsDescription.h>
+#include <Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h>
 
 #include <memory>
 namespace DB
@@ -140,6 +141,8 @@ class StorageObjectStorage : public IStorage
 
     void updateExternalDynamicMetadata(ContextPtr) override;
 
+    IDataLakeMetadata * getExternalMetadata() const;
+
     std::optional<UInt64> totalRows(ContextPtr query_context) const override;
     std::optional<UInt64> totalBytes(ContextPtr query_context) const override;
 protected:
@@ -229,6 +232,8 @@ class StorageObjectStorage::Configuration
 
     virtual bool hasExternalDynamicMetadata() { return false; }
 
+    virtual IDataLakeMetadata * getExternalMetadata() const { return nullptr; }
+
     virtual std::shared_ptr<NamesAndTypesList> getInitialSchemaByPath(const String &) const { return {}; }
 
     virtual std::shared_ptr<const ActionsDAG> getSchemaTransformer(const String &) const { return {}; }
diff --git a/src/Storages/System/StorageSystemIcebergHistory.cpp b/src/Storages/System/StorageSystemIcebergHistory.cpp
@@ -0,0 +1,104 @@
+#include <Storages/System/StorageSystemIcebergHistory.h>
+#include <mutex>
+#include <DataTypes/DataTypesNumber.h>
+#include <DataTypes/DataTypeString.h>
+#include <DataTypes/DataTypeMap.h>
+#include <DataTypes/DataTypeDateTime.h>
+#include <DataTypes/DataTypeDate.h>
+#include <DataTypes/DataTypeUUID.h>
+#include <DataTypes/DataTypeNullable.h>
+#include <DataTypes/DataTypeDateTime64.h>
+#include <Interpreters/InterpreterSelectQuery.h>
+#include <Processors/LimitTransform.h>
+#include <Processors/Port.h>
+#include <Processors/QueryPlan/QueryPlan.h>
+#include <Processors/QueryPlan/ReadFromSystemNumbersStep.h>
+#include <Storages/SelectQueryInfo.h>
+#include <Storages/ObjectStorage/StorageObjectStorage.h>
+#include <Access/ContextAccess.h>
+#include <Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h>
+#include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h>
+#include <Interpreters/DatabaseCatalog.h>
+#include <Core/Settings.h>
+
+static constexpr auto TIME_SCALE = 6;
+
+namespace DB
+{
+
+namespace Setting
+{
+    extern const SettingsSeconds lock_acquire_timeout;
+}
+
+ColumnsDescription StorageSystemIcebergHistory::getColumnsDescription()
+{
+    return ColumnsDescription
+    {
+        {"database_name",std::make_shared<DataTypeString>(),"Database name"},
+        {"table_name",std::make_shared<DataTypeString>(),"Table name."},
+        {"made_current_at",std::make_shared<DataTypeNullable>(std::make_shared<DataTypeDateTime64>(TIME_SCALE)),"date & time when this snapshot was made current snapshot"},
+        {"snapshot_id",std::make_shared<DataTypeUInt64>(),"snapshot id which is used to identify a snapshot."},
+        {"parent_id",std::make_shared<DataTypeUInt64>(),"parent id of this snapshot."},
+        {"is_current_ancestor",std::make_shared<DataTypeUInt8>(),"Flag that indicates if this snapshot is an ancestor of the current snapshot."}
+    };
+}
+
+void StorageSystemIcebergHistory::fillData([[maybe_unused]] MutableColumns & res_columns, [[maybe_unused]] ContextPtr context, const ActionsDAG::Node *, std::vector<UInt8>) const
+{
+#if USE_AVRO
+    const auto access = context->getAccess();
+
+    auto add_history_record = [&](const DatabaseTablesIteratorPtr & it, StorageObjectStorage * object_storage)
+    {
+        if (!access->isGranted(AccessType::SHOW_TABLES, it->databaseName(), it->name()))
+        {
+            return;
+        }
+
+        auto * current_metadata = object_storage->getExternalMetadata();
+
+        if (current_metadata && dynamic_cast<IcebergMetadata *>(current_metadata))
+        {
+            auto * iceberg_metadata = dynamic_cast<IcebergMetadata *>(current_metadata);
+            IcebergMetadata::IcebergHistory iceberg_history_items = iceberg_metadata->getHistory();
+
+            for (auto & iceberg_history_item : iceberg_history_items)
+            {
+                size_t column_index = 0;
+                res_columns[column_index++]->insert(it->databaseName());
+                res_columns[column_index++]->insert(it->name());
+                res_columns[column_index++]->insert(iceberg_history_item.made_current_at);
+                res_columns[column_index++]->insert(iceberg_history_item.snapshot_id);
+                res_columns[column_index++]->insert(iceberg_history_item.parent_id);
+                res_columns[column_index++]->insert(iceberg_history_item.is_current_ancestor);
+            }
+        }
+    };
+
+    const bool show_tables_granted = access->isGranted(AccessType::SHOW_TABLES);
+
+    if (show_tables_granted)
+    {
+        auto databases = DatabaseCatalog::instance().getDatabases();
+        for (const auto &db: databases)
+        {
+            for (auto iterator = db.second->getLightweightTablesIterator(context); iterator->isValid(); iterator->next())
+            {
+                StoragePtr storage = iterator->table();
+
+                TableLockHolder lock = storage->tryLockForShare(context->getCurrentQueryId(), context->getSettingsRef()[Setting::lock_acquire_timeout]);
+                if (!lock)
+                    // Table was dropped while acquiring the lock, skipping table
+                    continue;
+
+                if (auto *object_storage_table = dynamic_cast<StorageObjectStorage *>(storage.get()))
+                {
+                    add_history_record(iterator, object_storage_table);
+                }
+            }
+        }
+    }
+#endif
+}
+}
diff --git a/src/Storages/System/StorageSystemIcebergHistory.h b/src/Storages/System/StorageSystemIcebergHistory.h
@@ -0,0 +1,37 @@
+#pragma once
+
+#include <optional>
+#include <Parsers/ASTIdentifier.h>
+#include <Storages/System/IStorageSystemOneBlock.h>
+
+
+namespace DB
+{
+
+class Context;
+
+/** Implements a table engine for the system "iceberg_history".
+ *
+ * db_name String
+ * table_name String
+ * made_current_at DateTime64,
+ * snapshot_id UInt64,
+ * parent_id UInt64,
+ * is_current_ancestor Bool
+ *
+ */
+
+class StorageSystemIcebergHistory final : public IStorageSystemOneBlock
+{
+public:
+    std::string getName() const override { return "SystemIcebergHistory"; }
+
+    static ColumnsDescription getColumnsDescription();
+
+protected:
+    using IStorageSystemOneBlock::IStorageSystemOneBlock;
+
+    void fillData([[maybe_unused]] MutableColumns & res_columns, [[maybe_unused]] ContextPtr context, const ActionsDAG::Node *, std::vector<UInt8>) const override;
+};
+
+}
diff --git a/src/Storages/System/attachSystemTables.cpp b/src/Storages/System/attachSystemTables.cpp
@@ -100,6 +100,7 @@
 #include <Storages/System/StorageSystemViewRefreshes.h>
 #include <Storages/System/StorageSystemDNSCache.h>
 #include <Storages/System/StorageSystemLatencyBuckets.h>
+#include <Storages/System/StorageSystemIcebergHistory.h>
 
 #include <Poco/Util/LayeredConfiguration.h>
 
@@ -234,6 +235,7 @@ void attachSystemTablesServer(ContextPtr context, IDatabase & system_database, b
     attach<StorageSystemViewRefreshes>(context, system_database, "view_refreshes", "Lists all Refreshable Materialized Views of current server.");
     attach<StorageSystemWorkloads>(context, system_database, "workloads", "Contains a list of all currently existing workloads.");
     attach<StorageSystemResources>(context, system_database, "resources", "Contains a list of all currently existing resources.");
+    attach<StorageSystemIcebergHistory>(context, system_database, "iceberg_history", "Contains a list of iceberg history.");
 
     if (has_zookeeper)
     {
diff --git a/tests/integration/test_storage_iceberg/test.py b/tests/integration/test_storage_iceberg/test.py

Original file line number	Diff line number	Diff line change
`@@ -108,6 +108,8 @@ class DataLakeConfiguration : public BaseStorageConfiguration, public std::enabl`
`108`	`108`	`&& current_metadata->supportsSchemaEvolution();`
`109`	`109`	`}`
`110`	`110`
	`111`	`+ IDataLakeMetadata * getExternalMetadata() const override { return current_metadata.get(); }`
	`112`	`+`
`111`	`113`	`ColumnsDescription updateAndGetCurrentSchema(`
`112`	`114`	`ObjectStoragePtr object_storage,`
`113`	`115`	`ContextPtr context) override`
Original file line number	Diff line number	Diff line change
`@@ -197,6 +197,11 @@ bool StorageObjectStorage::hasExternalDynamicMetadata() const`
`197`	`197`	`return configuration->hasExternalDynamicMetadata();`
`198`	`198`	`}`
`199`	`199`
	`200`	`+IDataLakeMetadata * StorageObjectStorage::getExternalMetadata() const`
	`201`	`+{`
	`202`	`+ return configuration->getExternalMetadata();`
	`203`	`+}`
	`204`	`+`
`200`	`205`	`void StorageObjectStorage::updateExternalDynamicMetadata(ContextPtr context_ptr)`
`201`	`206`	`{`
`202`	`207`	`StorageInMemoryMetadata metadata;`