Fix multitest (#14910)

Author: aavdonkin

.github/config/muted_ya.txt

@@ -119,7 +119,6 @@ ydb/tests/olap/oom overlapping_portions.py.TestOverlappingPortions.test
 ydb/tests/olap/scenario sole chunk chunk
 ydb/tests/olap/scenario test_alter_compression.py.TestAlterCompression.test[alter_compression]
 ydb/tests/olap/scenario test_alter_tiering.py.TestAlterTiering.test[many_tables]
-ydb/tests/olap/scenario test_insert.py.TestInsert.test[read_data_during_bulk_upsert]
 ydb/tests/olap/scenario test_read_update_write_load.py.TestReadUpdateWriteLoad.test[read_update_write_load]
 ydb/tests/olap/ttl_tiering [data_migration_when_alter_ttl.py] chunk chunk
 ydb/tests/olap/ttl_tiering [ttl_delete_s3.py] chunk chunk

ydb/tests/olap/scenario/conftest.py

@@ -2,10 +2,13 @@
 import inspect
 import pytest
 import time
+import sys
+import copy
+from threading import Thread
+
 from ydb.tests.olap.lib.results_processor import ResultsProcessor
 from ydb.tests.olap.scenario.helpers.scenario_tests_helper import TestContext, ScenarioTestHelper
 from ydb.tests.olap.lib.ydb_cluster import YdbCluster
-from ydb.tests.olap.lib.utils import external_param_is_true
 from ydb.tests.olap.lib.utils import get_external_param
 from ydb.tests.olap.lib.allure_utils import allure_test_description
 from ydb.tests.library.harness.kikimr_runner import KiKiMR
@@ -71,19 +74,38 @@ def setup_class(cls):
         ydb_database = get_external_param('ydb-db', "").lstrip('/')
         cls._ydb_instance = YdbClusterInstance(ydb_endpoint, ydb_database, cls._get_cluster_config())
         YdbCluster.reset(cls._ydb_instance.endpoint(), cls._ydb_instance.database(), cls._ydb_instance.mon_port(), cls._ydb_instance.dyn_nodes_count())
-        if not external_param_is_true('reuse-tables'):
-            ScenarioTestHelper(None).remove_path(cls.get_suite_name())
 
     @classmethod
     def teardown_class(cls):
-        if not external_param_is_true('keep-tables'):
-            ScenarioTestHelper(None).remove_path(cls.get_suite_name())
         cls._ydb_instance.stop()
 
+    def test_multi(self, ctx: TestContext):
+        if self.__class__.__name__ != 'TestInsert':
+            return
+        self.def_inserts_count = 50
+        num_threads = int(get_external_param("num_threads", "10"))
+        threads = []
+        exit_codes = [None] * num_threads
+        for p in range(num_threads):
+            threads.append(Thread(target=self._test_suffix, args=(copy.deepcopy(ctx), str(p), exit_codes, p)))
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+        assert exit_codes == [0] * num_threads, exit_codes
+
     def test(self, ctx: TestContext):
-        test_path = ctx.test + get_external_param("table_suffix", "")
-        ScenarioTestHelper(None).remove_path(test_path, ctx.suite)
+        self.def_inserts_count = 200
+        exit_codes = [None]
+        self._test_suffix(ctx, get_external_param("table_suffix", ""), exit_codes, 0)
+
+    def _test_suffix(self, ctx: TestContext, table_suffix: str, exit_codes, num: int):
         start_time = time.time()
+        ctx.test += table_suffix
+        test_path = ctx.test
+        print('test_suffix, num {}, table path {} start_time {}'.format(num, test_path, start_time), file=sys.stderr)
+        ScenarioTestHelper(None).remove_path(test_path, ctx.suite)
+        print('Path {} removed'.format(test_path), file=sys.stderr)
         try:
             ctx.executable(self, ctx)
             ResultsProcessor.upload_results(
@@ -95,9 +117,11 @@ def test(self, ctx: TestContext):
                 is_successful=True,
             )
         except pytest.skip.Exception:
+            print('Caught skip exception, num {}'.format(num), file=sys.stderr)
             allure_test_description(ctx.suite, ctx.test, start_time=start_time, end_time=time.time())
             raise
-        except BaseException:
+        except BaseException as e:
+            print('Caught base exception, num {} message {}'.format(num, str(e)), file=sys.stderr)
             ResultsProcessor.upload_results(
                 kind='Scenario',
                 suite=ctx.suite,
@@ -109,11 +133,22 @@ def test(self, ctx: TestContext):
             allure_test_description(ctx.suite, ctx.test, start_time=start_time, end_time=time.time())
             raise
         allure_test_description(ctx.suite, ctx.test, start_time=start_time, end_time=time.time())
-        ScenarioTestHelper(None).remove_path(test_path, ctx.suite)
+        ScenarioTestHelper(None).remove_path(ctx.test, ctx.suite)
+        print('Path {} removed'.format(ctx.test), file=sys.stderr)
+        exit_codes[num] = 0
 
     @classmethod
     def _get_cluster_config(cls):
-        return KikimrConfigGenerator(extra_feature_flags=["enable_column_store"])
+        return KikimrConfigGenerator(
+            extra_feature_flags={
+                "enable_column_store": True,
+                "enable_external_data_sources": True,
+                "enable_tiering_in_column_shard": True,
+            },
+            query_service_config=dict(
+                available_external_data_sources=["ObjectStorage"]
+            )
+        )
 
 
 def pytest_generate_tests(metafunc):

ydb/tests/olap/scenario/helpers/scenario_tests_helper.py

@@ -10,7 +10,6 @@
 from abc import abstractmethod, ABC
 from typing import Set, List, Dict, Any, Callable, Optional
 from time import sleep
-from ydb.tests.olap.lib.utils import get_external_param
 
 
 class TestContext:
@@ -319,7 +318,7 @@ def _add_not_empty(p: str, dir: str):
         result = os.path.join('/', YdbCluster.ydb_database, YdbCluster.tables_path)
         if self.test_context is not None:
             result = _add_not_empty(result, self.test_context.suite)
-            result = _add_not_empty(result, self.test_context.test) + get_external_param("table_suffix", "")
+            result = _add_not_empty(result, self.test_context.test)
         result = _add_not_empty(result, path)
         return result
 
@@ -358,6 +357,7 @@ def _run_with_expected_status(
             sleep(3)
         if fail_on_error:
             pytest.fail(f'Retries exceeded with unexpected status: must be in {repr(expected_status)}, but get {repr(error or status)}')
+        return 1
 
     def _bulk_upsert_impl(
         self, tablename: str, data_generator: ScenarioTestHelper.IDataGenerator, expected_status: ydb.StatusCode | Set[ydb.StatusCode]
@@ -485,7 +485,7 @@ def execute_query(
 
         allure.attach(yql, 'request', allure.attachment_type.TEXT)
         with ydb.QuerySessionPool(YdbCluster.get_ydb_driver()) as pool:
-            self._run_with_expected_status(lambda: pool.execute_with_retries(yql, None, ydb.RetrySettings(max_retries=retries)), expected_status, fail_on_error=fail_on_error)
+            return self._run_with_expected_status(lambda: pool.execute_with_retries(yql, None, ydb.RetrySettings(max_retries=retries)), expected_status, fail_on_error=fail_on_error)
 
     def drop_if_exist(self, names: List[str], operation) -> None:
         """Erase entities in the tested database, if it exists.

ydb/tests/olap/scenario/multitest.sh

@@ -113,6 +113,9 @@ def _get_cluster_config(cls):
                 'TX_TIERING_BLOBS_TIER': LogLevels.TRACE,
                 'TX_COLUMNSHARD_ACTUALIZATION': LogLevels.TRACE,
             },
+            query_service_config=dict(
+                available_external_data_sources=["ObjectStorage"]
+            ),
         )
 
     def _setup_tiering_test(self, ctx):

ydb/tests/olap/scenario/test.mk

@@ -38,15 +38,23 @@ def _loop_insert(self, ctx: TestContext, rows_count: int, table: str, ignore_rea
         for i in range(rows_count):
             for c in range(10):
                 try:
-                    sth.execute_query(
+                    result = sth.execute_query(
                         yql=f'$cnt = SELECT CAST(COUNT(*) AS INT64) from `{log}`; INSERT INTO `{cnt}` (key, c) values({i}, $cnt)', retries=20, fail_on_error=False
                     )
+                    if result == 1:
+                        if c >= 9:
+                            raise Exception('Insert failed table {}'.format(table))
+                        else:
+                            time.sleep(1)
+                            continue
+
                     break
                 except Exception:
                     if ignore_read_errors:
                         pass
                     else:
-                        raise
+                        if c >= 9:
+                            raise
                 time.sleep(1)
 
     def scenario_read_data_during_bulk_upsert(self, ctx: TestContext):
@@ -55,7 +63,7 @@ def scenario_read_data_during_bulk_upsert(self, ctx: TestContext):
         log_table_name: str = "log"
         batches_count = int(get_external_param("batches_count", "10"))
         rows_count = int(get_external_param("rows_count", "1000"))
-        inserts_count = int(get_external_param("inserts_count", "200"))
+        inserts_count = int(get_external_param("inserts_count", str(self.def_inserts_count)))
         tables_count = int(get_external_param("tables_count", "1"))
         ignore_read_errors = external_param_is_true("ignore_read_errors")
         for table in range(tables_count):