Improve secondary index stress-test (#30457)

Author: nikvas0

ydb/tests/stress/common/common.py

@@ -19,12 +19,13 @@ def __init__(self, endpoint, database, use_query_service=False, sessions=100):
     def wait_connection(self, timeout=5):
         self.driver.wait(timeout, fail_fast=True)
 
-    def query(self, statement, is_ddl, parameters=None, retry_settings=None):
+    def query(self, statement, is_ddl, parameters=None, retry_settings=None, log_error=True):
         if self.use_query_service:
             try:
                 return self.session_pool.execute_with_retries(query=statement, parameters=parameters, retry_settings=retry_settings)
             except Exception as e:
-                logger.error(f"Error: {e} while executing query: {statement}")
+                if log_error:
+                    logger.error(f"Error: {e} while executing query: {statement}")
                 raise e
         else:
             if is_ddl:

ydb/tests/stress/oltp_workload/__main__.py

@@ -1,9 +1,12 @@
 # -*- coding: utf-8 -*-
 import argparse
+import logging
+import os
 
 from ydb.tests.stress.oltp_workload.workload import WorkloadRunner
 from ydb.tests.stress.common.common import YdbClient
 
+logging.basicConfig(level=os.getenv('LOG_LEVEL', 'INFO'))
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(

ydb/tests/stress/oltp_workload/workload/type/secondary_index.py

@@ -23,7 +23,7 @@ class WorkloadConfig:
     cover_max_columns: int = 2
     allowed_secondary_indexes_count: List[int] = field(default_factory=lambda: [1, 2, 3, 8, 16])
     unique_indexes_allowed: bool = True
-    unique_index_probability: float = 0.1
+    unique_index_probability: float = 0.05
     null_probability: float = 0.05
 
     # Column types
@@ -476,6 +476,8 @@ def run_for_table(self, table_name: str) -> None:
         self.verify_table(table_name, table_info)
         self.drop_table(table_name)
 
+        print(f"Workload for table {table_name} completed.")
+
     def _get_batch(self, pk_size: int, pk_only: bool, table_info: TableInfo = None) -> List[List[Any]]:
         """Generate a batch of rows for operations"""
         batch_rows = []
@@ -550,9 +552,9 @@ def _execute_query_with_retry(self, query: str, is_ddl: bool, parameters: Dict[s
             try:
                 if parameters:
                     logger.debug(f"Query: {query} Params: {parameters}")
-                    return self.client.query(query, is_ddl, parameters)
+                    return self.client.query(query, is_ddl, parameters, log_error=False)
                 else:
-                    return self.client.query(query, is_ddl)
+                    return self.client.query(query, is_ddl, log_error=False)
             except (ydb.issues.Aborted, ydb.issues.Unavailable, ydb.issues.Undetermined) as e:
                 last_exception = e
                 if attempt < self.config.max_retries - 1:
@@ -820,9 +822,10 @@ def _verify_index_uniqueness(
 
     def _loop(self):
         """Main loop for the workload"""
+        iteration = 0
         while not self.is_stop_requested():
             threads = []
-            for table_name in [f'test{i}' for i in range(self.config.tables_inflight)]:
+            for table_name in [f'test_{iteration}_{i}' for i in range(self.config.tables_inflight)]:
                 threads.append(
                     threading.Thread(target=lambda t=table_name: self.run_for_table(t), name=f'{table_name}')
                 )
@@ -833,6 +836,10 @@ def _loop(self):
             for thread in threads:
                 thread.join()
 
+            print(f"Iteration {iteration} finished.")
+            print(self.stats.get_stats())
+            iteration += 1
+
     def get_workload_thread_funcs(self):
         """Get the thread functions for the workload"""
         return [self._loop]