Add test

tests/integration/test_replicated_merge_tree_s3/configs/config.d/storage_conf.xml

@@ -7,6 +7,12 @@
                 <access_key_id>minio</access_key_id>
                 <secret_access_key>minio123</secret_access_key>
             </s3>
+            <s3_cache>
+                <type>cache</type>
+                <disk>s3</disk>
+                <path>./s3_cache/</path>
+                <max_size>1000000</max_size>
+            </s3_cache>
         </disks>
         <policies>
             <s3>
@@ -16,6 +22,13 @@
                     </main>
                 </volumes>
             </s3>
+            <s3_cache>
+                <volumes>
+                    <main>
+                        <disk>s3_cache</disk>
+                    </main>
+                </volumes>
+            </s3_cache>
         </policies>
     </storage_configuration>
 

tests/integration/test_replicated_merge_tree_s3/test.py

@@ -5,6 +5,8 @@ import string
 import pytest
 from helpers.cluster import ClickHouseCluster
 
+TABLE_NAME = "s3_test"
+
 
 @pytest.fixture(scope="module")
 def cluster():
@@ -58,8 +60,13 @@ def generate_values(date_str, count, sign=1):
 
 
 def create_table(cluster, additional_settings=None):
-    create_table_statement = """
-        CREATE TABLE s3_test ON CLUSTER cluster(
+    settings = {
+        "storage_policy": "s3",
+    }
+    settings.update(additional_settings)
+
+    create_table_statement = f"""
+        CREATE TABLE {TABLE_NAME} ON CLUSTER cluster(
             dt Date,
             id Int64,
             data String,
@@ -67,20 +74,42 @@ def create_table(cluster, additional_settings=None):
         ) ENGINE=ReplicatedMergeTree()
         PARTITION BY dt
         ORDER BY (dt, id)
-        SETTINGS storage_policy='s3'
+        SETTINGS {",".join((k+"="+repr(v) for k, v in settings.items()))}
         """
-    if additional_settings:
-        create_table_statement += ","
-        create_table_statement += additional_settings
 
     list(cluster.instances.values())[0].query(create_table_statement)
 
 
+def insert(cluster, node_idxs, verify=True):
+    all_values = ""
+    for node_idx in node_idxs:
+        node = cluster.instances["node" + str(node_idx)]
+        values = generate_values("2020-01-0" + str(node_idx), 4096)
+        node.query(
+            f"INSERT INTO {TABLE_NAME} VALUES {values}",
+            settings={"insert_quorum": 3},
+        )
+        if node_idx != 1:
+            all_values += ","
+        all_values += values
+
+    if verify:
+        for node_idx in node_idxs:
+            node = cluster.instances["node" + str(node_idx)]
+            assert (
+                node.query(
+                    f"SELECT * FROM {TABLE_NAME} order by dt, id FORMAT Values",
+                    settings={"select_sequential_consistency": 1},
+                )
+                == all_values
+            )
+
+
 @pytest.fixture(autouse=True)
 def drop_table(cluster):
     yield
     for node in list(cluster.instances.values()):
-        node.query("DROP TABLE IF EXISTS s3_test")
+        node.query(f"DROP TABLE IF EXISTS {TABLE_NAME}")
 
     minio = cluster.minio_client
     # Remove extra objects to prevent tests cascade failing
@@ -95,32 +124,37 @@ def drop_table(cluster):
 def test_insert_select_replicated(cluster, min_rows_for_wide_part, files_per_part):
     create_table(
         cluster,
-        additional_settings="min_rows_for_wide_part={}".format(min_rows_for_wide_part),
+        additional_settings=f"min_rows_for_wide_part={min_rows_for_wide_part}",
     )
 
-    all_values = ""
-    for node_idx in range(1, 4):
-        node = cluster.instances["node" + str(node_idx)]
-        values = generate_values("2020-01-0" + str(node_idx), 4096)
-        node.query(
-            "INSERT INTO s3_test VALUES {}".format(values),
-            settings={"insert_quorum": 3},
-        )
-        if node_idx != 1:
-            all_values += ","
-        all_values += values
-
-    for node_idx in range(1, 4):
-        node = cluster.instances["node" + str(node_idx)]
-        assert (
-            node.query(
-                "SELECT * FROM s3_test order by dt, id FORMAT Values",
-                settings={"select_sequential_consistency": 1},
-            )
-            == all_values
-        )
+    insert(node_idxs=[1, 2, 3, 4], verify=True)
 
     minio = cluster.minio_client
     assert len(list(minio.list_objects(cluster.minio_bucket, "data/"))) == 3 * (
         FILES_OVERHEAD + files_per_part * 3
     )
+
+
+def test_drop_cache_on_cluster(cluster):
+    create_table(
+        cluster,
+        additional_settings={"storage_policy":"s3_cache"},
+    )
+
+    insert(cluster, node_idxs=[1, 2, 3], verify=True)
+
+    node1 = cluster.instances["node1"]
+    node2 = cluster.instances["node2"]
+    node3 = cluster.instances["node2"]
+
+    node1.query(f"select * from clusterAllReplicas(cluster, default, {TABLE_NAME}) format Null")
+
+    assert int(node1.query("select count() from system.filesystem_cache")) > 0
+    assert int(node2.query("select count() from system.filesystem_cache")) > 0
+    assert int(node3.query("select count() from system.filesystem_cache")) > 0
+
+    node1.query("system drop filesystem cache on cluster cluster")
+
+    assert int(node1.query("select count() from system.filesystem_cache")) == 0
+    assert int(node2.query("select count() from system.filesystem_cache")) == 0
+    assert int(node3.query("select count() from system.filesystem_cache")) == 0