Update on reads, tests

ianton-ru · ianton-ru · commit 31448574d1d2 · 2025-11-03T20:00:50.000+01:00
diff --git a/src/Databases/DataLake/DatabaseDataLake.cpp b/src/Databases/DataLake/DatabaseDataLake.cpp
@@ -434,17 +434,18 @@ StoragePtr DatabaseDataLake::tryGetTableImpl(const String & name, ContextPtr con
         configuration,
         configuration->createObjectStorage(context_copy, /* is_readonly */ false),
         StorageID(getDatabaseName(), name),
-        /* columns */columns,
-        /* constraints */ConstraintsDescription{},
-        /* partition_by */nullptr,
+        /* columns */ columns,
+        /* constraints */ ConstraintsDescription{},
+        /* partition_by */ nullptr,
         context_copy,
-        /* comment */"",
+        /* comment */ "",
         getFormatSettings(context_copy),
         LoadingStrictnessLevel::CREATE,
         getCatalog(),
-        /* if_not_exists*/true,
-        /* is_datalake_query*/true,
-        /* lazy_init */true);
+        /* if_not_exists */ true,
+        /* is_datalake_query */ true,
+        /* is_table_function */ true,
+        /* lazy_init */ true);
 }
 
 void DatabaseDataLake::dropTable( /// NOLINT
diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp
@@ -288,6 +288,8 @@ void IStorageCluster::read(
         return;
     }
 
+    updateConfigurationIfNeeded(context);
+
     storage_snapshot->check(column_names);
 
     const auto & settings = context->getSettingsRef();
diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h
@@ -97,6 +97,8 @@ class IStorageCluster : public IStorage
         throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Method writeFallBackToPure is not supported by storage {}", getName());
     }
 
+    virtual void updateConfigurationIfNeeded(ContextPtr /* context */) {}
+
 private:
     static ClusterPtr getClusterImpl(ContextPtr context, const String & cluster_name_, size_t max_hosts = 0);
 
diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp
@@ -91,6 +91,7 @@ StorageObjectStorageCluster::StorageObjectStorageCluster(
     std::shared_ptr<DataLake::ICatalog> catalog,
     bool if_not_exists,
     bool is_datalake_query,
+    bool is_table_function,
     bool lazy_init)
     : IStorageCluster(
         cluster_name_, table_id_, getLogger(fmt::format("{}({})", configuration_->getEngineName(), table_id_.table_name)))
@@ -145,6 +146,10 @@ StorageObjectStorageCluster::StorageObjectStorageCluster(
         tryLogCurrentException(log_);
     }
 
+    // For tables need to update configuration on each read
+    // because data can be changed after previous update
+    update_configuration_on_read_write = !is_table_function;
+
     ColumnsDescription columns{columns_in_table_or_function_definition};
     std::string sample_path;
     if (need_resolve_columns_or_format)
@@ -743,6 +748,18 @@ IDataLakeMetadata * StorageObjectStorageCluster::getExternalMetadata(ContextPtr
     return configuration->getExternalMetadata();
 }
 
+void StorageObjectStorageCluster::updateConfigurationIfNeeded(ContextPtr context)
+{
+    if (update_configuration_on_read_write)
+    {
+        configuration->update(
+            object_storage,
+            context,
+            /* if_not_updated_before */false,
+            /* check_consistent_with_previous_metadata */false);
+    }
+}
+
 void StorageObjectStorageCluster::checkAlterIsPossible(const AlterCommands & commands, ContextPtr context) const
 {
     if (getClusterName(context).empty())
diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h
@@ -25,7 +25,8 @@ class StorageObjectStorageCluster : public IStorageCluster
         std::shared_ptr<DataLake::ICatalog> catalog,
         bool if_not_exists,
         bool is_datalake_query,
-        bool lazy_init = false);
+        bool is_table_function,
+        bool lazy_init);
 
     std::string getName() const override;
 
@@ -154,6 +155,8 @@ class StorageObjectStorageCluster : public IStorageCluster
         ContextPtr context,
         bool async_insert) override;
 
+    void updateConfigurationIfNeeded(ContextPtr context) override;
+
     /*
     In case the table was created with `object_storage_cluster` setting,
     modify the AST query object so that it uses the table function implementation
@@ -176,6 +179,7 @@ class StorageObjectStorageCluster : public IStorageCluster
 
     /// non-clustered storage to fall back on pure realisation if needed
     std::shared_ptr<StorageObjectStorage> pure_storage;
+    bool update_configuration_on_read_write;
 };
 
 }
diff --git a/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp b/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp
@@ -98,7 +98,9 @@ createStorageObjectStorage(const StorageFactory::Arguments & args, StorageObject
         args.mode,
         configuration->getCatalog(context, args.query.attach),
         args.query.if_not_exists,
-        /* is_datalake_query*/ false);
+        /* is_datalake_query */ false,
+        /* is_table_function */ false,
+        /* lazy_init */ false);
 }
 
 #endif
diff --git a/src/TableFunctions/TableFunctionObjectStorage.cpp b/src/TableFunctions/TableFunctionObjectStorage.cpp
@@ -204,9 +204,11 @@ StoragePtr TableFunctionObjectStorage<Definition, Configuration, is_data_lake>::
             /* comment */ String{},
             /* format_settings */ std::nullopt, /// No format_settings
             /* mode */ LoadingStrictnessLevel::CREATE,
-            configuration->getCatalog(context, /*attach*/ false),
+            configuration->getCatalog(context, /* attach */ false),
             /* if_not_exists */ false,
-            /* is_datalake_query*/ false);
+            /* is_datalake_query */ false,
+            /* is_table_function */ true,
+            /* lazy_init */ false);
 
         storage->startup();
         return storage;
diff --git a/src/TableFunctions/TableFunctionObjectStorageCluster.cpp b/src/TableFunctions/TableFunctionObjectStorageCluster.cpp
@@ -76,7 +76,8 @@ StoragePtr TableFunctionObjectStorageCluster<Definition, Configuration, is_data_
             /* mode */ LoadingStrictnessLevel::CREATE,
             /* catalog*/ nullptr,
             /* if_not_exists */ false,
-            /* is_datalake_query*/ false,
+            /* is_datalake_query */ false,
+            /* is_table_function */ true,
             /* lazy_init */ true);
     }
 
diff --git a/tests/integration/helpers/iceberg_utils.py b/tests/integration/helpers/iceberg_utils.py
@@ -288,22 +288,26 @@ def get_creation_expression(
                 )
 
     elif storage_type == "local":
-        assert not run_on_cluster
-
-        if table_function:
+        if run_on_cluster:
+            assert table_function
             return f"""
-                iceberg{engine_part}({storage_arg}, path = '/iceberg_data/default/{table_name}/', format={format})
+                iceberg{engine_part}Cluster('cluster_single_node', {storage_arg}, path = '/iceberg_data/default/{table_name}/', format={format})
             """
         else:
-            return (
-                f"""
-                DROP TABLE IF EXISTS {table_name};
-                CREATE TABLE {if_not_exists_prefix} {table_name} {schema}
-                ENGINE=Iceberg{engine_part}({storage_arg}, path = '/iceberg_data/default/{table_name}/', format={format})
-                {partition_by}
-                {settings_expression}
+            if table_function:
+                return f"""
+                    iceberg{engine_part}({storage_arg}, path = '/iceberg_data/default/{table_name}/', format={format})
                 """
-            )
+            else:
+                return (
+                    f"""
+                    DROP TABLE IF EXISTS {table_name};
+                    CREATE TABLE {if_not_exists_prefix} {table_name} {schema}
+                    ENGINE=Iceberg{engine_part}({storage_arg}, path = '/iceberg_data/default/{table_name}/', format={format})
+                    {partition_by}
+                    {settings_expression}
+                    """
+                )
 
     else:
         raise Exception(f"Unknown iceberg storage type: {storage_type}")
diff --git a/tests/integration/test_storage_iceberg/configs/config.d/cluster.xml b/tests/integration/test_storage_iceberg/configs/config.d/cluster.xml
@@ -16,5 +16,13 @@
                 </replica>
             </shard>
         </cluster_simple>
+        <cluster_single_node>
+            <shard>
+                <replica>
+                    <host>node1</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </cluster_single_node>
     </remote_servers>
 </clickhouse>
diff --git a/tests/integration/test_storage_iceberg/test.py b/tests/integration/test_storage_iceberg/test.py
@@ -1214,7 +1214,7 @@ def test_filesystem_cache(started_cluster, storage_type):
 
 @pytest.mark.parametrize(
     "storage_type, run_on_cluster",
-    [("s3", False), ("s3", True), ("azure", False), ("azure", True), ("local", False)],
+    [("s3", False), ("s3", True), ("azure", False), ("azure", True), ("local", False), ("local", True)],
 )
 def test_partition_pruning(started_cluster, storage_type, run_on_cluster):
     instance = started_cluster.instances["node1"]
@@ -2053,8 +2053,6 @@ def test_explicit_metadata_file(started_cluster, storage_type):
 @pytest.mark.parametrize("storage_type", ["s3", "azure", "local"])
 @pytest.mark.parametrize("run_on_cluster", [False, True])
 def test_minmax_pruning_with_null(started_cluster, storage_type, run_on_cluster):
-    if run_on_cluster and storage_type == "local":
-        pytest.skip("Local storage is not supported on cluster")
     instance = started_cluster.instances["node1"]
     spark = started_cluster.spark_session
     TABLE_NAME = "test_minmax_pruning_with_null" + storage_type + "_" + get_uuid_str()

Original file line number	Diff line number	Diff line change
`@@ -288,6 +288,8 @@ void IStorageCluster::read(`
`288`	`288`	`return;`
`289`	`289`	`}`
`290`	`290`
	`291`	`+ updateConfigurationIfNeeded(context);`
	`292`	`+`
`291`	`293`	`storage_snapshot->check(column_names);`
`292`	`294`
`293`	`295`	`const auto & settings = context->getSettingsRef();`
Original file line number	Diff line number	Diff line change
`@@ -97,6 +97,8 @@ class IStorageCluster : public IStorage`
`97`	`97`	`throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Method writeFallBackToPure is not supported by storage {}", getName());`
`98`	`98`	`}`
`99`	`99`
	`100`	`+ virtual void updateConfigurationIfNeeded(ContextPtr /* context */) {}`
	`101`	`+`
`100`	`102`	`private:`
`101`	`103`	`static ClusterPtr getClusterImpl(ContextPtr context, const String & cluster_name_, size_t max_hosts = 0);`
`102`	`104`
Original file line number	Diff line number	Diff line change
`@@ -76,7 +76,8 @@ StoragePtr TableFunctionObjectStorageCluster<Definition, Configuration, is_data_`
`76`	`76`	`/* mode */ LoadingStrictnessLevel::CREATE,`
`77`	`77`	`/* catalog*/ nullptr,`
`78`	`78`	`/* if_not_exists */ false,`
`79`		`- /* is_datalake_query*/ false,`
	`79`	`+ /* is_datalake_query */ false,`
	`80`	`+ /* is_table_function */ true,`
`80`	`81`	`/* lazy_init */ true);`
`81`	`82`	`}`
`82`	`83`