Add functional tests for spilling (ydb-platform#15358)

Author: lll-phill-lll

ydb/tests/functional/tpc/large/test_tpcds.py

@@ -5,9 +5,13 @@
 class TestTpcdsS1(tpcds.TestTpcds1, FunctionalTestBase):
     iterations: int = 1
 
+    memory_controller_config = {
+        'hard_limit_bytes': 107374182400,
+    }
+
     @classmethod
     def setup_class(cls) -> None:
-        cls.setup_cluster()
+        cls.setup_cluster(memory_controller_config=cls.memory_controller_config)
         cls.run_cli(['workload', 'tpcds', '-p', 'olap_yatests/tpcds/s1', 'init', '--store=column'])
         cls.run_cli(['workload', 'tpcds', '-p', 'olap_yatests/tpcds/s1', 'import', 'generator', '--scale=1'])
         tpcds.TestTpcds1.setup_class()

ydb/tests/functional/tpc/large/test_tpch_spilling.py

@@ -0,0 +1,37 @@
+import ydb.tests.olap.load.lib.tpch as tpch
+from ydb.tests.functional.tpc.lib.conftest import FunctionalTestBase
+
+
+class TestTpchSpillingS10(tpch.TestTpch10, FunctionalTestBase):
+    iterations: int = 1
+    query_settings = {i : tpch.TestTpch10.QuerySettings(query_prefix='pragma ydb.UseGraceJoinCoreForMap = "true";') for i in range(1, 23)}
+    # temporary exclude q5: https://github.com/ydb-platform/ydb/issues/15359
+    skip_tests: list = [5]
+
+    table_service_config = {
+        'enable_spilling_nodes': 'All',
+        'spilling_service_config': {
+            'local_file_config': {
+                'enable': True,
+                'max_total_size': 536870912000,
+                'max_file_size': 107374182400,
+            }
+        },
+        'resource_manager': {
+            'verbose_memory_limit_exception': True
+        },
+    }
+
+    memory_controller_config = {
+        'activities_limit_percent': 60,
+        'query_execution_limit_percent': 50,
+        'hard_limit_bytes': 6 * 1073741824,
+    }
+
+    @classmethod
+    def setup_class(cls) -> None:
+        cls.setup_cluster(table_service_config=cls.table_service_config, memory_controller_config=cls.memory_controller_config)
+        cls.run_cli(['workload', 'tpch', '-p', 'olap_yatests/tpch/s10', 'init', '--store=column'])
+        cls.run_cli(['workload', 'tpch', '-p', 'olap_yatests/tpch/s10', 'import', 'generator', '--scale=10'])
+
+        tpch.TestTpch10.setup_class()

ydb/tests/functional/tpc/large/ya.make

@@ -1,8 +1,8 @@
 PY3TEST()
-ENV(YDB_HARD_MEMORY_LIMIT_BYTES="107374182400")
 
 TEST_SRCS(
     test_tpcds.py
+    test_tpch_spilling.py
 )
 
 SIZE(LARGE)

ydb/tests/functional/tpc/lib/conftest.py

@@ -9,12 +9,19 @@ class FunctionalTestBase:
     cluster = None
 
     @classmethod
-    def setup_cluster(cls) -> None:
-        cls.cluster = KiKiMR(configurator=KikimrConfigGenerator(
+    def setup_cluster(cls, table_service_config: dict = {}, memory_controller_config: dict = {}) -> None:
+        config_generator = KikimrConfigGenerator(
             domain_name='local',
             extra_feature_flags=["enable_resource_pools"],
             use_in_memory_pdisks=True,
-        ))
+        )
+        if table_service_config:
+            config_generator.yaml_config["table_service_config"] = table_service_config
+
+        if memory_controller_config:
+            config_generator.yaml_config["memory_controller_config"] = memory_controller_config
+
+        cls.cluster = KiKiMR(configurator=config_generator)
         cls.cluster.start()
         node = cls.cluster.nodes[1]
         YdbCluster.reset(

ydb/tests/olap/load/lib/tpch.py

@@ -11,6 +11,7 @@ class TpchSuiteBase(LoadSuiteBase):
     workload_type: WorkloadType = WorkloadType.TPC_H
     iterations: int = 3
     tables_size: dict[str, int] = {}
+    skip_tests: list = []
 
     @classmethod
     def _get_tables_size(cls) -> dict[str, int]:
@@ -42,6 +43,8 @@ def do_setup_class(cls):
 
     @pytest.mark.parametrize('query_num', [i for i in range(1, 23)])
     def test_tpch(self, query_num: int):
+        if query_num in self.skip_tests:
+            return
         self.run_workload_test(self._get_path(), query_num)