Scenario test for issue: #11186 (#12138)

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

@@ -395,6 +395,26 @@ def execute_scan_query(
         allure.attach(json.dumps(rows), 'result', allure.attachment_type.JSON)
         return ret
 
+    @allure.step('Execute query')
+    def execute_query(
+        self, yql: str, expected_status: ydb.StatusCode | Set[ydb.StatusCode] = ydb.StatusCode.SUCCESS
+    ):
+        """Run a query on the tested database.
+
+        Args:
+            yql: Query text.
+            expected_status: Expected status or set of database response statuses. If the response status is not in the expected set, an exception is thrown.
+
+        Example:
+            tablename = 'testTable'
+            sth = ScenarioTestHelper(ctx)
+            sth.execute_query(f'INSERT INTO `{sth.get_full_path("tablename") }` (key, c) values(1, 100)')
+        """
+
+        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), expected_status)
+
     def drop_if_exist(self, names: List[str], operation) -> None:
         """Erase entities in the tested database, if it exists.
 

ydb/tests/olap/scenario/test_insert.py

@@ -0,0 +1,87 @@
+from conftest import BaseTestSet
+from ydb.tests.olap.scenario.helpers import (
+    ScenarioTestHelper,
+    TestContext,
+    CreateTable,
+)
+
+from ydb import PrimitiveType
+from typing import List, Dict, Any
+from ydb.tests.olap.lib.utils import get_external_param
+import threading
+
+
+class TestInsert(BaseTestSet):
+    schema_cnt = (
+        ScenarioTestHelper.Schema()
+        .with_column(name="key", type=PrimitiveType.Int32, not_null=True)
+        .with_column(name="c", type=PrimitiveType.Int64)
+        .with_key_columns("key")
+    )
+
+    schema_log = (
+        ScenarioTestHelper.Schema()
+        .with_column(name="key", type=PrimitiveType.Int32, not_null=True)
+        .with_key_columns("key")
+    )
+
+    def _loop_upsert(self, ctx: TestContext, data: list):
+        sth = ScenarioTestHelper(ctx)
+        for batch in data:
+            sth.bulk_upsert_data("log", self.schema_log, batch)
+
+    def _loop_insert(self, ctx: TestContext, rows_count: int):
+        sth = ScenarioTestHelper(ctx)
+        log: str = sth.get_full_path("log")
+        cnt: str = sth.get_full_path("cnt")
+        for i in range(rows_count):
+            sth.execute_query(
+                f'$cnt = SELECT CAST(COUNT(*) AS INT64) from `{log}`; INSERT INTO `{cnt}` (key, c) values({i}, $cnt)'
+            )
+
+    def scenario_read_data_during_bulk_upsert(self, ctx: TestContext):
+        sth = ScenarioTestHelper(ctx)
+        cnt_table_name: str = "cnt"
+        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"))
+        sth.execute_scheme_query(
+            CreateTable(cnt_table_name).with_schema(self.schema_cnt)
+        )
+        sth.execute_scheme_query(
+            CreateTable(log_table_name).with_schema(self.schema_log)
+        )
+        data: List = []
+        for i in range(batches_count):
+            batch: List[Dict[str, Any]] = []
+            for j in range(rows_count):
+                batch.append({"key": j + rows_count * i})
+            data.append(batch)
+
+        thread1 = threading.Thread(target=self._loop_upsert, args=[ctx, data])
+        thread2 = threading.Thread(target=self._loop_insert, args=[ctx, inserts_count])
+
+        thread1.start()
+        thread2.start()
+
+        thread2.join()
+        thread1.join()
+
+        rows: int = sth.get_table_rows_count(cnt_table_name)
+        assert rows == inserts_count
+        scan_result = sth.execute_scan_query(
+            f"SELECT key, c FROM `{sth.get_full_path(cnt_table_name)}` ORDER BY key"
+        )
+        for i in range(rows):
+            if scan_result.result_set.rows[i]["key"] != i:
+                assert False, f"{i} ?= {scan_result.result_set.rows[i]['key']}"
+
+        rows: int = sth.get_table_rows_count(log_table_name)
+        assert rows == rows_count * batches_count
+        scan_result = sth.execute_scan_query(
+            f"SELECT key FROM `{sth.get_full_path(log_table_name)}` ORDER BY key"
+        )
+        for i in range(rows):
+            if scan_result.result_set.rows[i]["key"] != i:
+                assert False, f"{i} ?= {scan_result.result_set.rows[i]['key']}"

ydb/tests/olap/scenario/ya.make

@@ -10,6 +10,7 @@ PY3TEST()
         test_simple.py
         test_scheme_load.py
         test_alter_tiering.py
+        test_insert.py
     )
 
     PEERDIR(