Merge pull request #615 from Altinity/project-antalya-24.12.2-distributed-table-engine

Distributed request to tables with Object Storage Engines

Author: ianton-ru

tests/integration/test_s3_cluster/test.py

@@ -630,3 +630,72 @@ def test_cluster_default_expression(started_cluster):
     )
 
     assert result == expected_result
+
+
+def test_distributed_s3_table_engine(started_cluster):
+    node = started_cluster.instances["s0_0_0"]
+
+    resp_def = node.query(
+        """
+        SELECT * from s3Cluster(
+            'cluster_simple',
+            'http://minio1:9001/root/data/{clickhouse,database}/*', 'minio', 'minio123', 'CSV',
+            'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') ORDER BY (name, value, polygon)
+        """
+    )
+
+    node.query("DROP TABLE IF EXISTS single_node");
+    node.query(
+        """
+        CREATE TABLE single_node
+            (name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64))))
+            ENGINE=S3('http://minio1:9001/root/data/{clickhouse,database}/*', 'minio', 'minio123', 'CSV')
+        """
+    )
+    query_id_engine_single_node = str(uuid.uuid4())
+    resp_engine_single_node = node.query(
+        """
+        SELECT * FROM single_node ORDER BY (name, value, polygon)
+        """,
+        query_id = query_id_engine_single_node
+    )
+    assert resp_def == resp_engine_single_node
+
+    node.query("DROP TABLE IF EXISTS distributed");
+    node.query(
+        """
+        CREATE TABLE distributed
+            (name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64))))
+            ENGINE=S3('http://minio1:9001/root/data/{clickhouse,database}/*', 'minio', 'minio123', 'CSV')
+            SETTINGS object_storage_cluster='cluster_simple'
+        """
+    )
+    query_id_engine_distributed = str(uuid.uuid4())
+    resp_engine_distributed = node.query(
+        """
+        SELECT * FROM distributed ORDER BY (name, value, polygon)
+        """,
+        query_id = query_id_engine_distributed
+    )
+    assert resp_def == resp_engine_distributed
+
+    node.query("SYSTEM FLUSH LOGS ON CLUSTER 'cluster_simple'")
+
+    hosts_engine_single_node = node.query(
+        f"""
+        SELECT uniq(hostname)
+            FROM clusterAllReplicas('cluster_simple', system.query_log)
+            WHERE type='QueryFinish' AND initial_query_id='{query_id_engine_single_node}'
+        """
+    )
+    assert int(hosts_engine_single_node) == 1
+    hosts_engine_distributed = node.query(
+        f"""
+        SELECT uniq(hostname)
+            FROM clusterAllReplicas('cluster_simple', system.query_log)
+            WHERE type='QueryFinish' AND initial_query_id='{query_id_engine_distributed}'
+        """
+    )
+    assert int(hosts_engine_distributed) == 3
+
+

tests/integration/test_storage_iceberg/test.py

@@ -202,6 +202,7 @@ def get_creation_expression(
     allow_dynamic_metadata_for_data_lakes=False,
     use_version_hint=False,
     run_on_cluster=False,
+    object_storage_cluster=False,
     explicit_metadata_path="",
     **kwargs,
 ):
@@ -215,6 +216,9 @@ def get_creation_expression(
     if use_version_hint:
         settings_array.append("iceberg_use_version_hint = true")
 
+    if object_storage_cluster:
+        settings_array.append(f"object_storage_cluster = '{object_storage_cluster}'")
+
     if settings_array:
         settings_expression = " SETTINGS " + ",".join(settings_array)
     else:
@@ -313,10 +317,18 @@ def create_iceberg_table(
     table_name,
     cluster,
     format="Parquet",
+    object_storage_cluster=False,
     **kwargs,
 ):
     node.query(
-        get_creation_expression(storage_type, table_name, cluster, format, **kwargs)
+        get_creation_expression(
+            storage_type,
+            table_name,
+            cluster,
+            format,
+            object_storage_cluster=object_storage_cluster,
+            **kwargs,
+        )
     )
 
 
@@ -657,61 +669,81 @@ def add_df(mode):
         .split()
     )
 
+    create_iceberg_table(storage_type, instance, TABLE_NAME, started_cluster, object_storage_cluster='cluster_simple')
+    query_id_cluster_table_engine = str(uuid.uuid4())
+    select_cluster_table_engine = (
+        instance.query(
+            f"""
+            SELECT * FROM {TABLE_NAME}
+            """,
+            query_id=query_id_cluster_table_engine,
+        )
+        .strip()
+        .split()
+    )
+
+    # write 3 times
+    assert int(instance.query(f"SELECT count() FROM {table_function_expr_cluster}")) == 100 * 3
+
+    select_remote_cluster = (
+        instance.query(f"SELECT * FROM remote('node2',{table_function_expr_cluster})")
+        .strip()
+        .split()
+    )
+
+    instance.query(f"DROP TABLE IF EXISTS `{TABLE_NAME}` SYNC")
+    create_iceberg_table(storage_type, instance, TABLE_NAME, started_cluster)
+    query_id_pure_table_engine = str(uuid.uuid4())
+    select_pure_table_engine = (
+        instance.query(
+            f"""
+            SELECT * FROM {TABLE_NAME}
+            """,
+            query_id=query_id_pure_table_engine,
+        )
+        .strip()
+        .split()
+    )
+    query_id_pure_table_engine_cluster = str(uuid.uuid4())
+    select_pure_table_engine_cluster = (
+        instance.query(
+            f"""
+            SELECT * FROM {TABLE_NAME}
+            SETTINGS object_storage_cluster='cluster_simple'
+            """,
+            query_id=query_id_pure_table_engine_cluster,
+        )
+        .strip()
+        .split()
+    )
+
     # Simple size check
     assert len(select_regular) == 600
     assert len(select_cluster) == 600
     assert len(select_cluster_alt_syntax) == 600
+    assert len(select_cluster_table_engine) == 600
+    assert len(select_remote_cluster) == 600
+    assert len(select_pure_table_engine) == 600
+    assert len(select_pure_table_engine_cluster) == 600
 
     # Actual check
     assert select_cluster == select_regular
     assert select_cluster_alt_syntax == select_regular
+    assert select_cluster_table_engine == select_regular
+    assert select_remote_cluster == select_regular
+    assert select_pure_table_engine == select_regular
+    assert select_pure_table_engine_cluster == select_regular
 
     # Check query_log
     for replica in started_cluster.instances.values():
         replica.query("SYSTEM FLUSH LOGS")
 
-    for node_name, replica in started_cluster.instances.items():
-        cluster_secondary_queries = (
-            replica.query(
-                f"""
-                SELECT query, type, is_initial_query, read_rows, read_bytes FROM system.query_log
-                WHERE
-                    type = 'QueryStart'
-                    AND NOT is_initial_query
-                    AND initial_query_id='{query_id_cluster}'
-            """
-            )
-            .strip()
-            .split("\n")
-        )
-
-        logging.info(
-            f"[{node_name}] cluster_secondary_queries: {cluster_secondary_queries}"
-        )
-        assert len(cluster_secondary_queries) == 1
-
-    for node_name, replica in started_cluster.instances.items():
-        cluster_secondary_queries = (
-            replica.query(
-                f"""
-                SELECT query, type, is_initial_query, read_rows, read_bytes FROM system.query_log
-                WHERE
-                    type = 'QueryStart'
-                    AND NOT is_initial_query
-                    AND initial_query_id='{query_id_cluster_alt_syntax}'
-            """
-            )
-            .strip()
-            .split("\n")
-        )
-
-        logging.info(
-            f"[{node_name}] cluster_secondary_queries: {cluster_secondary_queries}"
-        )
-        assert len(cluster_secondary_queries) == 1
+    count_secondary_subqueries(started_cluster, query_id_cluster, 1, "table function")
+    count_secondary_subqueries(started_cluster, query_id_cluster_alt_syntax, 1, "table function alt syntax")
+    count_secondary_subqueries(started_cluster, query_id_cluster_table_engine, 1, "cluster table engine")
+    count_secondary_subqueries(started_cluster, query_id_pure_table_engine, 0, "table engine")
+    count_secondary_subqueries(started_cluster, query_id_pure_table_engine_cluster, 1, "table engine with cluster setting")
 
-    # write 3 times
-    assert int(instance.query(f"SELECT count() FROM {table_function_expr_cluster}")) == 100 * 3
 
 @pytest.mark.parametrize("format_version", ["1", "2"])
 @pytest.mark.parametrize("storage_type", ["s3", "azure", "local"])