diff --git a/parquet/configs/users.d/users.xml b/parquet/configs/users.d/users.xml
index a0cbec01b..1cd56d40b 100644
--- a/parquet/configs/users.d/users.xml
+++ b/parquet/configs/users.d/users.xml
@@ -88,6 +88,11 @@
             <!-- Allow access management -->
             <access_management>1</access_management>
 
+            <!-- Allow access to named collections -->
+            <named_collection_control>1</named_collection_control>
+            <show_named_collections>1</show_named_collections>
+            <show_named_collections_secrets>1</show_named_collections_secrets>
+
             <!-- Example of row level security policy. -->
             <!-- <databases>
                 <test>
diff --git a/parquet/configs/users.d/users_disabled_caching.xml b/parquet/configs/users.d/users_disabled_caching.xml
index c7d0ecae6..1668b2e27 100644
--- a/parquet/configs/users.d/users_disabled_caching.xml
+++ b/parquet/configs/users.d/users_disabled_caching.xml
@@ -87,6 +87,11 @@
             <!-- Allow access management -->
             <access_management>1</access_management>
 
+            <!-- Allow access to named collections -->
+            <named_collection_control>1</named_collection_control>
+            <show_named_collections>1</show_named_collections>
+            <show_named_collections_secrets>1</show_named_collections_secrets>
+
             <!-- Example of row level security policy. -->
             <!-- <databases>
                 <test>
diff --git a/parquet/tests/s3.py b/parquet/tests/s3.py
index c08134bd8..610604e19 100644
--- a/parquet/tests/s3.py
+++ b/parquet/tests/s3.py
@@ -269,7 +269,7 @@ def insert_into_function(self):
 
     with When("I insert test data into `s3` table function in Parquet format"):
         node.query(
-            f"INSERT INTO FUNCTION s3('{self.context.uri}{file_name}.Parquet', '{self.context.access_key_id}', '{self.context.secret_access_key}', 'Parquet', '{func_def}', '{compression_type.lower()}') VALUES {','.join(total_values)}",
+            f"INSERT INTO FUNCTION s3(s3_credentials, url='{self.context.uri}{file_name}.Parquet', format='Parquet', structure='{func_def}', compression_method='{compression_type.lower()}') VALUES {','.join(total_values)}",
             settings=[("allow_suspicious_low_cardinality_types", 1)],
         )
 
@@ -328,7 +328,7 @@ def select_from_function_manual_cast_types(self):
                     executor=executor,
                 )(
                     sql=f"SELECT {column.name}, toTypeName({column.name}) FROM \
-                    s3('{self.context.uri}{table_name}.Parquet', '{self.context.access_key_id}', '{self.context.secret_access_key}', 'Parquet', '{table_def}')",
+                    s3(s3_credentials, url='{self.context.uri}{table_name}.Parquet', format='Parquet', structure='{table_def}')",
                 )
             join()
 
@@ -373,7 +373,7 @@ def select_from_function_auto_cast_types(self):
                     executor=executor,
                 )(
                     sql=f"SELECT {column.name}, toTypeName({column.name}) FROM \
-                    s3('{self.context.uri}{table_name}.Parquet', '{self.context.access_key_id}', '{self.context.secret_access_key}', 'Parquet')",
+                    s3(s3_credentials, url='{self.context.uri}{table_name}.Parquet', format='Parquet')",
                 )
             join()
 
@@ -442,6 +442,13 @@ def outline(self, compression_type):
     self.context.compression_type = compression_type
     self.context.node = self.context.cluster.node("clickhouse1")
 
+    with Given("I add S3 credentials configuration"):
+        named_s3_credentials(
+            access_key_id=self.context.access_key_id,
+            secret_access_key=self.context.secret_access_key,
+            restart=True,
+        )
+
     Suite(run=engine)
     Suite(run=function)
 
diff --git a/s3/configs/clickhouse/users.xml b/s3/configs/clickhouse/users.xml
index 86b2cd9e1..5b79d2417 100644
--- a/s3/configs/clickhouse/users.xml
+++ b/s3/configs/clickhouse/users.xml
@@ -87,6 +87,11 @@
             <!-- Allow access management -->
             <access_management>1</access_management>
 
+            <!-- Allow access to named collections -->
+            <named_collection_control>1</named_collection_control>
+            <show_named_collections>1</show_named_collections>
+            <show_named_collections_secrets>1</show_named_collections_secrets>
+
             <!-- Example of row level security policy. -->
             <!-- <databases>
                 <test>
diff --git a/s3/tests/common.py b/s3/tests/common.py
index f0c68a25a..010d57e1d 100644
--- a/s3/tests/common.py
+++ b/s3/tests/common.py
@@ -209,6 +209,25 @@ def s3_storage(
     return add_config(config, restart=restart, nodes=nodes, timeout=timeout)
 
 
+@TestStep(Given)
+def named_s3_credentials(
+    self, access_key_id, secret_access_key, restart=False, nodes=None, timeout=60
+):
+    """Add S3 connection configuration as a named collection."""
+    config = create_xml_config_content(
+        entries={
+            "named_collections": {
+                "s3_credentials": {
+                    "access_key_id": access_key_id,
+                    "secret_access_key": secret_access_key,
+                }
+            }
+        },
+        config_file="s3_credentials.xml",
+    )
+    return add_config(config, restart=restart, nodes=nodes, timeout=timeout)
+
+
 @contextmanager
 def s3_endpoints(
     endpoints,
@@ -1847,16 +1866,14 @@ def insert_to_s3_function(
 ):
     """Write a table to a file in s3. File will be overwritten from an empty table during cleanup."""
 
-    access_key_id = self.context.access_key_id
-    secret_access_key = self.context.secret_access_key
     uri = uri or self.context.uri
     node = current().context.node
 
     try:
-        query = f"INSERT INTO FUNCTION s3('{uri}{filename}', '{access_key_id}','{secret_access_key}', 'CSVWithNames', '{columns}'"
+        query = f"INSERT INTO FUNCTION s3(s3_credentials, url='{uri}{filename}', format='CSVWithNames', structure='{columns}'"
 
         if compression:
-            query += f", '{compression}'"
+            query += f", compression_method='{compression}'"
 
         query += f") SELECT * FROM {table_name}"
 
@@ -1868,7 +1885,7 @@ def insert_to_s3_function(
         yield
 
     finally:
-        query = f"INSERT INTO FUNCTION s3('{uri}{filename}', '{access_key_id}','{secret_access_key}', 'CSV', '{columns}'"
+        query = f"INSERT INTO FUNCTION s3(s3_credentials, url='{uri}{filename}', format='CSV', structure='{columns}'"
         query += f") SELECT * FROM null('{columns}')"
 
         node.query(query)
@@ -1887,18 +1904,16 @@ def insert_from_s3_function(
     no_checks=False,
 ):
     """Import data from a file in s3 to a table."""
-    access_key_id = self.context.access_key_id
-    secret_access_key = self.context.secret_access_key
     uri = uri or self.context.uri
     node = current().context.node
 
     if cluster_name is None:
-        query = f"INSERT INTO {table_name} SELECT * FROM s3('{uri}{filename}', '{access_key_id}','{secret_access_key}', 'CSVWithNames', '{columns}'"
+        query = f"INSERT INTO {table_name} SELECT * FROM s3(s3_credentials, url='{uri}{filename}', format='CSVWithNames', structure='{columns}'"
     else:
-        query = f"INSERT INTO {table_name} SELECT * FROM s3Cluster('{cluster_name}', '{uri}{filename}', '{access_key_id}','{secret_access_key}', 'CSVWithNames', '{columns}'"
+        query = f"INSERT INTO {table_name} SELECT * FROM s3Cluster('{cluster_name}', s3_credentials, url='{uri}{filename}', format='CSVWithNames', structure='{columns}'"
 
     if compression:
-        query += f", '{compression}'"
+        query += f", compression_method='{compression}'"
 
     query += ")"
 
diff --git a/s3/tests/table_function.py b/s3/tests/table_function.py
index 55beae992..2df02373e 100644
--- a/s3/tests/table_function.py
+++ b/s3/tests/table_function.py
@@ -470,16 +470,13 @@ def credentials_s3Cluster(self):
 @Requirements(RQ_SRS_015_S3_Settings_PartitionBy("1.0"))
 def partition(self):
     """Check that ClickHouse can export partitioned data."""
-    access_key_id = self.context.access_key_id
-    secret_access_key = self.context.secret_access_key
     uri = self.context.uri
     node = current().context.node
 
     with When("I export the data to S3 using the table function"):
         sql = (
-            f"INSERT INTO FUNCTION s3('{uri}_partition_export_"
-            + "{_partition_id}.csv'"
-            + f", '{access_key_id}','{secret_access_key}', 'CSV', 'a String') PARTITION BY a VALUES ('x'),('y'),('z')"
+            f"INSERT INTO FUNCTION s3(s3_credentials, url='{uri}_partition_export_"
+            + "{_partition_id}.csv', format='CSV', structure='a String') PARTITION BY a VALUES ('x'),('y'),('z')"
         )
         node.query(sql)
 
@@ -487,7 +484,7 @@ def partition(self):
         with Then(f"I check the data in the {partition_id} partition"):
             output = node.query(
                 f"""SELECT * FROM
-                s3('{uri}_partition_export_{partition_id}.csv', '{access_key_id}','{secret_access_key}', 'CSV', 'a String') FORMAT TabSeparated"""
+                s3(s3_credentials, url='{uri}_partition_export_{partition_id}.csv', format='CSV', structure='a String') FORMAT TabSeparated"""
             ).output
             assert output == partition_id, error()
 
@@ -496,8 +493,6 @@ def partition(self):
 @Requirements(RQ_SRS_015_S3_Settings_PartitionBy("1.0"))
 def partition_s3Cluster(self):
     """Check that ClickHouse can export partitioned data using s3Cluster table function."""
-    access_key_id = self.context.access_key_id
-    secret_access_key = self.context.secret_access_key
     uri = self.context.uri
     node = current().context.node
 
@@ -505,9 +500,8 @@ def partition_s3Cluster(self):
         with Combination(f"cluster_name = {cluster_name}"):
             with When("I export the data to S3 using the table function"):
                 sql = (
-                    f"INSERT INTO FUNCTION s3('{uri}_partition_export_"
-                    + "{_partition_id}.csv'"
-                    + f", '{access_key_id}','{secret_access_key}', 'CSV', 'a String') PARTITION BY a VALUES ('x'),('y'),('z')"
+                    f"INSERT INTO FUNCTION s3(s3_credentials, url='{uri}_partition_export_"
+                    + "{_partition_id}.csv', format='CSV', structure='a String') PARTITION BY a VALUES ('x'),('y'),('z')"
                 )
                 node.query(sql)
 
@@ -519,7 +513,7 @@ def partition_s3Cluster(self):
                                 self.context.cluster.node("clickhouse1")
                                 .query(
                                     f"""SELECT * FROM
-                                s3Cluster('{cluster_name}', '{uri}_partition_export_{partition_id}.csv', '{access_key_id}','{secret_access_key}', 'CSV', 'a String') FORMAT TabSeparated"""
+                                s3Cluster('{cluster_name}', s3_credentials, url='{uri}_partition_export_{partition_id}.csv', format='CSV', structure='a String') FORMAT TabSeparated"""
                                 )
                                 .output
                             )
@@ -690,8 +684,6 @@ def remote_host_filter(self):
     """
     table1_name = "table_" + getuid()
     table2_name = "table_" + getuid()
-    access_key_id = self.context.access_key_id
-    secret_access_key = self.context.secret_access_key
     uri = self.context.uri
     node = current().context.node
     expected = "427"
@@ -717,7 +709,7 @@ def remote_host_filter(self):
             node.query(
                 f"""
                 INSERT INTO FUNCTION
-                s3('{uri}', '{access_key_id}','{secret_access_key}', 'CSVWithNames', 'd UInt64')
+                s3(s3_credentials, url='{uri}', format='CSVWithNames', structure='d UInt64')
                 SELECT * FROM {table1_name}""",
                 message=f'DB::Exception: URL "{uri}" is not allowed',
             )
@@ -727,8 +719,6 @@ def remote_host_filter(self):
 @Requirements(RQ_SRS_015_S3_TableFunction_MeasureFileSize("1.0"))
 def measure_file_size(self):
 
-    access_key_id = self.context.access_key_id
-    secret_access_key = self.context.secret_access_key
     table1_name = "table_" + getuid()
     uri = self.context.uri + "measure-file-size/"
     bucket_path = self.context.bucket_path + "/measure-file-size"
@@ -756,7 +746,7 @@ def measure_file_size(self):
 
     with Then("I compare the size that clickhouse reports"):
         r = node.query(
-            f"SELECT sum(_size) FROM s3('{uri}**', '{access_key_id}','{secret_access_key}', 'One') FORMAT TSV"
+            f"SELECT sum(_size) FROM s3(s3_credentials, url='{uri}**', format='One') FORMAT TSV"
         )
 
         for retry in retries(timeout=30, delay=5):
@@ -769,8 +759,6 @@ def measure_file_size(self):
 @Requirements(RQ_SRS_015_S3_TableFunction_MeasureFileSize("1.0"))
 def measure_file_size_s3Cluster(self):
     """Check that ClickHouse can measure file size using s3Cluster table function."""
-    access_key_id = self.context.access_key_id
-    secret_access_key = self.context.secret_access_key
     table1_name = "table_" + getuid()
     uri = self.context.uri + "measure-file-size/"
     bucket_path = self.context.bucket_path + "/measure-file-size"
@@ -804,7 +792,7 @@ def measure_file_size_s3Cluster(self):
                 for attempt in retries(timeout=10, delay=1):
                     with attempt:
                         r = node.query(
-                            f"SELECT sum(_size) FROM s3Cluster('{cluster_name}', '{uri}**', '{access_key_id}','{secret_access_key}', 'One') FORMAT TSV"
+                            f"SELECT sum(_size) FROM s3Cluster('{cluster_name}', s3_credentials, url='{uri}**', format='One') FORMAT TSV"
                         )
                         size_clickhouse = int(r.output.strip())
                         debug(
@@ -820,6 +808,13 @@ def measure_file_size_s3Cluster(self):
 def outline(self):
     """Test S3 and S3 compatible storage through storage disks."""
 
+    with Given("I add S3 credentials configuration"):
+        named_s3_credentials(
+            access_key_id=self.context.access_key_id,
+            secret_access_key=self.context.secret_access_key,
+            restart=True,
+        )
+
     for scenario in loads(current_module(), Scenario):
         with allow_s3_truncate(self.context.node):
             Scenario(run=scenario, flags=TE)
@@ -867,7 +862,7 @@ def ssec_encryption_check(self):
                         node.query(
                             f"""
                                 INSERT INTO FUNCTION
-                                s3('{self.context.uri}encrypted.csv', '{self.context.access_key_id}','{self.context.secret_access_key}', 'CSV', 'd UInt64')
+                                s3(s3_credentials, url='{self.context.uri}encrypted.csv', format='CSV', structure='d UInt64')
                                 SELECT * FROM {name}"""
                         )
 
diff --git a/s3/tests/table_function_invalid.py b/s3/tests/table_function_invalid.py
index 98414d21d..a8c43dabf 100644
--- a/s3/tests/table_function_invalid.py
+++ b/s3/tests/table_function_invalid.py
@@ -12,22 +12,18 @@ def insert_to_s3_function_invalid(
     columns="d UInt64",
     compression=None,
     file_format="CSVWithNames",
-    access_key_id=None,
-    secret_access_key=None,
     message=None,
     exitcode=None,
     timeout=60,
 ):
     """Write a table to a file in s3. File will be overwritten from an empty table during cleanup."""
 
-    access_key_id = access_key_id or self.context.access_key_id
-    secret_access_key = secret_access_key or self.context.secret_access_key
     node = current().context.node
 
-    query = f"INSERT INTO FUNCTION s3('{path}', '{access_key_id}','{secret_access_key}', '{file_format}', '{columns}'"
+    query = f"INSERT INTO FUNCTION s3(s3_credentials, url='{path}', format='{file_format}', structure='{columns}'"
 
     if compression:
-        query += f", '{compression}'"
+        query += f", compression_method='{compression}'"
 
     query += f") SELECT * FROM {table_name}"
 
@@ -48,15 +44,13 @@ def insert_from_s3_function_invalid(
     timeout=60,
 ):
     """Import data from a file in s3 to a table and catch fail."""
-    access_key_id = self.context.access_key_id
-    secret_access_key = self.context.secret_access_key
     uri = uri or self.context.uri
     node = current().context.node
 
-    query = f"INSERT INTO {table_name} SELECT * FROM s3('{uri}{filename}', '{access_key_id}','{secret_access_key}', 'CSVWithNames', '{columns}'"
+    query = f"INSERT INTO {table_name} SELECT * FROM s3(s3_credentials, url='{uri}{filename}', format='CSVWithNames', structure='{columns}'"
 
     if compression:
-        query += f", '{compression}'"
+        query += f", compression_method='{compression}'"
 
     query += ")"
 
@@ -373,14 +367,8 @@ def invalid_credentials(self):
         """I export the data to S3 using the table function with invalid
                credentials, expecting failure"""
     ):
-        insert_to_s3_function_invalid(
-            table_name=name_table1,
-            path=f"{uri}invalid.csv",
-            access_key_id=access_key_id,
-            secret_access_key=secret_access_key,
-            message=expected,
-            exitcode=243,
-        )
+        query = f"INSERT INTO FUNCTION s3('{uri}invalid.csv', '{access_key_id}', '{secret_access_key}', 'CSV', 'd UInt64') SELECT * FROM {name_table1}"
+        node.query(query, message=expected, exitcode=243)
 
 
 @TestOutline(Feature)
@@ -390,6 +378,13 @@ def outline(self, uri):
 
     self.context.uri = uri
 
+    with Given("I add S3 credentials configuration"):
+        named_s3_credentials(
+            access_key_id=self.context.access_key_id,
+            secret_access_key=self.context.secret_access_key,
+            restart=True,
+        )
+
     for scenario in loads(current_module(), Scenario):
         with allow_s3_truncate(self.context.node):
             scenario()
diff --git a/s3/tests/table_function_performance.py b/s3/tests/table_function_performance.py
index e35f2e7ed..52085b22a 100644
--- a/s3/tests/table_function_performance.py
+++ b/s3/tests/table_function_performance.py
@@ -17,8 +17,6 @@ def s3_create_many_files(self):
     num_files_per_folder = 5
 
     node = current().context.node
-    access_key_id = self.context.access_key_id
-    secret_access_key = self.context.secret_access_key
     table_name = "table_" + getuid()
     my_random = random.Random("many_files")
 
@@ -28,7 +26,7 @@ def s3_create_many_files(self):
     with Then("I check if the files already exist"):
         with By(f"checking if the last folder exists"):
             r = node.query(
-                f"""SELECT _path FROM s3('{self.context.many_files_uri}id={folder_ids[-1]}/*', '{access_key_id}','{secret_access_key}', 'One') FORMAT TabSeparated"""
+                f"""SELECT _path FROM s3(s3_credentials, url='{self.context.many_files_uri}id={folder_ids[-1]}/*', format='One') FORMAT TabSeparated"""
             )
 
     if r.output:
@@ -41,7 +39,7 @@ def insert_files(self, folder_id, iteration):
             with attempt:
                 node.query(
                     f"""INSERT INTO TABLE FUNCTION 
-                    s3('{self.context.many_files_uri}id={folder_id}/file_{{_partition_id}}.csv','{access_key_id}','{secret_access_key}','CSV','d UInt64') 
+                    s3(s3_credentials, url='{self.context.many_files_uri}id={folder_id}/file_{{_partition_id}}.csv', format='CSV', structure='d UInt64') 
                     PARTITION BY (d % {num_files_per_folder}) SELECT * FROM {table_name} 
                     -- {iteration}/{num_folders}"""
                 )
@@ -87,14 +85,12 @@ def wildcard(self, wildcard, expected_time, expect_result):
     """Check the performance of using wildcards in s3 paths."""
 
     node = current().context.node
-    access_key_id = self.context.access_key_id
-    secret_access_key = self.context.secret_access_key
 
     for i in range(1, 3):
         with Then(f"""I query S3 using the wildcard '{wildcard}'"""):
             t_start = time.time()
             r = node.query(
-                f"""SELECT median(d) FROM s3('{self.context.many_files_uri}id={wildcard}/*', '{access_key_id}','{secret_access_key}', 'CSV', 'd UInt64') FORMAT TabSeparated""",
+                f"""SELECT median(d) FROM s3(s3_credentials, url='{self.context.many_files_uri}id={wildcard}/*', format='CSV', structure='d UInt64') FORMAT TabSeparated""",
                 timeout=expected_time,
             )
             t_elapsed = time.time() - t_start
@@ -114,6 +110,13 @@ def outline(self, uri):
     self.context.uri = uri
     self.context.many_files_uri = self.context.uri + "many_files_benchmark/"
 
+    with Given("I add S3 credentials configuration"):
+        named_s3_credentials(
+            access_key_id=self.context.access_key_id,
+            secret_access_key=self.context.secret_access_key,
+            restart=True,
+        )
+
     with allow_s3_truncate(self.context.node):
         Scenario(run=s3_create_many_files)