Unmute ydb/tests/olap/scenario/test_insert.py.TestInsert.test[read_data_during_bulk_upsert] #13113 (#17467)

Author: XJIE6

.github/config/muted_ya.txt

@@ -118,7 +118,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

@@ -95,7 +95,7 @@ def test_multi(self, ctx: TestContext):
         assert exit_codes == [0] * num_threads, exit_codes
 
     def test(self, ctx: TestContext):
-        self.def_inserts_count = 200
+        self.def_inserts_count = 50
         exit_codes = [None]
         self._test_suffix(ctx, get_external_param("table_suffix", ""), exit_codes, 0)
 

ydb/tests/olap/scenario/test_insert.py

@@ -8,7 +8,11 @@
 from ydb.tests.olap.common.thread_helper import TestThread, TestThreads
 from ydb import PrimitiveType
 from typing import List, Dict, Any
-from ydb.tests.olap.lib.utils import get_external_param, external_param_is_true
+from ydb.tests.olap.lib.utils import get_external_param
+
+import random
+import logging
+logger = logging.getLogger(__name__)
 
 
 class TestInsert(BaseTestSet):
@@ -26,16 +30,25 @@ class TestInsert(BaseTestSet):
     )
 
     def _loop_upsert(self, ctx: TestContext, data: list, table: str):
+        min_time = 10.0 / len(data)
+        max_time = min_time * 10
+
         sth = ScenarioTestHelper(ctx)
         table_name = "log" + table
         for batch in data:
             sth.bulk_upsert_data(table_name, self.schema_log, batch)
+            logger.info("Upsert")
+            time.sleep(random.uniform(min_time, max_time))
+
+    def _loop_insert(self, ctx: TestContext, rows_count: int, table: str):
+        min_time = 1.0 / rows_count
+        max_time = min_time * 10
 
-    def _loop_insert(self, ctx: TestContext, rows_count: int, table: str, ignore_read_errors: bool):
         sth = ScenarioTestHelper(ctx)
         log: str = sth.get_full_path("log" + table)
         cnt: str = sth.get_full_path("cnt" + table)
         for i in range(rows_count):
+            logger.info("Insert")
             for c in range(10):
                 try:
                     result = sth.execute_query(
@@ -50,12 +63,9 @@ def _loop_insert(self, ctx: TestContext, rows_count: int, table: str, ignore_rea
 
                     break
                 except Exception:
-                    if ignore_read_errors:
-                        pass
-                    else:
-                        if c >= 9:
-                            raise
-                time.sleep(1)
+                    if c >= 9:
+                        raise
+                time.sleep(random.uniform(min_time, max_time))
 
     def scenario_read_data_during_bulk_upsert(self, ctx: TestContext):
         sth = ScenarioTestHelper(ctx)
@@ -65,7 +75,6 @@ def scenario_read_data_during_bulk_upsert(self, ctx: TestContext):
         rows_count = int(get_external_param("rows_count", "1000"))
         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):
             sth.execute_scheme_query(
                 CreateTable(cnt_table_name + str(table)).with_schema(self.schema_cnt)
@@ -86,7 +95,7 @@ def scenario_read_data_during_bulk_upsert(self, ctx: TestContext):
         for table in range(tables_count):
             thread1.append(TestThread(target=self._loop_upsert, args=[ctx, data, str(table)]))
         for table in range(tables_count):
-            thread2.append(TestThread(target=self._loop_insert, args=[ctx, inserts_count, str(table), ignore_read_errors]))
+            thread2.append(TestThread(target=self._loop_insert, args=[ctx, inserts_count, str(table)]))
 
         thread1.start_all()
         thread2.start_all()