Merge branch 'antalya-25.8' into backports/antalya-25.8/90490

Author: zvonand

ci/docker/integration/runner/requirements.txt

@@ -15,8 +15,8 @@ azure-core==1.30.1
 azure-storage-blob==12.19.0
 bcrypt==4.1.3
 beautifulsoup4==4.12.3
-boto3[all]==1.37.7
-botocore==1.37.7
+boto3==1.39.11
+botocore==1.39.11
 bs4==0.0.2
 cassandra-driver==3.29.0
 certifi==2025.4.26
@@ -97,7 +97,8 @@ redis==5.0.1
 requests-kerberos==0.14.0
 requests==2.32.4
 rich==13.9.4
-s3transfer==0.11.4
+s3fs==2024.12.0
+s3transfer==0.13.0
 setuptools==78.1.1
 simplejson==3.19.2
 sortedcontainers==2.4.0

src/Interpreters/DDLWorker.cpp

@@ -1140,7 +1140,6 @@ bool DDLWorker::initializeMainThread()
             auto zookeeper = getAndSetZooKeeper();
             zookeeper->createAncestors(fs::path(queue_dir) / "");
             initializeReplication();
-            markReplicasActive(true);
             initialized = true;
             return true;
         }
@@ -1213,6 +1212,14 @@ void DDLWorker::runMainThread()
             }
 
             cleanup_event->set();
+            try
+            {
+                markReplicasActive(reinitialized);
+            }
+            catch (...)
+            {
+                tryLogCurrentException(log, "An error occurred when markReplicasActive: ");
+            }
             scheduleTasks(reinitialized);
             subsequent_errors_count = 0;
 
@@ -1291,20 +1298,23 @@ void DDLWorker::createReplicaDirs(const ZooKeeperPtr & zookeeper, const NameSet
         zookeeper->createAncestors(fs::path(replicas_dir) / host_id / "");
 }
 
-void DDLWorker::markReplicasActive(bool /*reinitialized*/)
+void DDLWorker::markReplicasActive(bool reinitialized)
 {
     auto zookeeper = getZooKeeper();
 
-    // Reset all active_node_holders
-    for (auto & it : active_node_holders)
+    if (reinitialized)
     {
-        auto & active_node_holder = it.second.second;
-        if (active_node_holder)
-            active_node_holder->setAlreadyRemoved();
-        active_node_holder.reset();
-    }
+        // Reset all active_node_holders
+        for (auto & it : active_node_holders)
+        {
+            auto & active_node_holder = it.second.second;
+            if (active_node_holder)
+                active_node_holder->setAlreadyRemoved();
+            active_node_holder.reset();
+        }
 
-    active_node_holders.clear();
+        active_node_holders.clear();
+    }
 
     for (auto it = active_node_holders.begin(); it != active_node_holders.end();)
     {
@@ -1385,12 +1395,7 @@ void DDLWorker::markReplicasActive(bool /*reinitialized*/)
         {
             zookeeper->deleteEphemeralNodeIfContentMatches(active_path, active_id);
         }
-        Coordination::Requests ops;
-        ops.emplace_back(zkutil::makeCreateRequest(active_path, active_id, zkutil::CreateMode::Ephemeral));
-        /// To bump node mtime
-        ops.emplace_back(zkutil::makeSetRequest(fs::path(replicas_dir) / host_id, "", -1));
-        zookeeper->multi(ops);
-
+        zookeeper->create(active_path, active_id, zkutil::CreateMode::Ephemeral);
         auto active_node_holder_zookeeper = zookeeper;
         auto active_node_holder = zkutil::EphemeralNodeHolder::existing(active_path, *active_node_holder_zookeeper);
         active_node_holders[host_id] = {active_node_holder_zookeeper, active_node_holder};

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

@@ -990,12 +990,16 @@ std::optional<size_t> IcebergMetadata::totalBytes(ContextPtr local_context) cons
 std::optional<String> IcebergMetadata::partitionKey(ContextPtr) const
 {
     SharedLockGuard lock(mutex);
+    if (!relevant_snapshot)
+        return {};
     return relevant_snapshot->partition_key;
 }
 
 std::optional<String> IcebergMetadata::sortingKey(ContextPtr) const
 {
     SharedLockGuard lock(mutex);
+    if (!relevant_snapshot)
+        return {};
     return relevant_snapshot->sorting_key;
 }
 

tests/integration/compose/docker_compose_iceberg_nessie_catalog.yml

@@ -0,0 +1,62 @@
+services:
+  nessie:
+    image: ghcr.io/projectnessie/nessie:latest
+    depends_on:
+      minio:
+        condition: service_started
+    ports:
+      - "19120:19120"
+    environment:
+      - nessie.version.store.type=IN_MEMORY
+      - nessie.catalog.default-warehouse=warehouse
+      - nessie.catalog.warehouses.warehouse.location=s3://warehouse-rest/
+      - nessie.catalog.service.s3.default-options.endpoint=http://minio:9000/
+      - nessie.catalog.service.s3.default-options.external-endpoint=http://127.0.0.1:9002/
+      - nessie.catalog.service.s3.default-options.access-key=urn:nessie-secret:quarkus:nessie.catalog.secrets.access-key
+      - nessie.catalog.service.s3.default-options.path-style-access=true
+      - nessie.catalog.service.s3.default-options.auth-type=STATIC
+      - nessie.catalog.secrets.access-key.name=minio
+      - nessie.catalog.secrets.access-key.secret=ClickHouse_Minio_P@ssw0rd
+      - nessie.catalog.service.s3.default-options.region=us-east-1
+      - nessie.server.authentication.enabled=false
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:19120/api/v2/config"]
+      interval: 30s
+      timeout: 10s
+      retries: 5
+      start_period: 60s
+
+  # TODO: can we simply use with_minio=True instead?
+  minio:
+    image: minio/minio:RELEASE.2024-07-31T05-46-26Z
+    environment:
+      - MINIO_ROOT_USER=minio
+      - MINIO_ROOT_PASSWORD=ClickHouse_Minio_P@ssw0rd
+      - MINIO_DOMAIN=minio
+    networks:
+      default:
+        aliases:
+          - warehouse-rest.minio
+    ports:
+      - "9001:9001"
+      - "9002:9000"
+    command: ["server", "/data", "--console-address", ":9001"]
+
+  # TODO: move this code to cluster.py
+  mc:
+    depends_on:
+      - minio
+    # Stick to version with "mc config"
+    image: minio/mc:RELEASE.2025-04-16T18-13-26Z
+    environment:
+      - AWS_ACCESS_KEY_ID=minio
+      - AWS_SECRET_ACCESS_KEY=ClickHouse_Minio_P@ssw0rd
+      - AWS_REGION=us-east-1
+    entrypoint: >
+      /bin/sh -c "
+      until (/usr/bin/mc config host add minio http://minio:9000 minio ClickHouse_Minio_P@ssw0rd) do echo '...waiting...' && sleep 1; done;
+      /usr/bin/mc rm -r --force minio/warehouse-rest;
+      /usr/bin/mc mb minio/warehouse-rest --ignore-existing;
+      /usr/bin/mc policy set public minio/warehouse-rest;
+      tail -f /dev/null
+      "

tests/integration/test_database_iceberg_nessie_catalog/__init__.py

@@ -0,0 +1 @@
+#!/usr/bin/env python3

tests/integration/test_database_iceberg_nessie_catalog/configs/backups.xml

@@ -0,0 +1,5 @@
+<clickhouse>
+    <backups>
+        <allowed_path>/backups</allowed_path>
+    </backups>
+</clickhouse>