[ymq] make merge test more light

1) тесты падают из-за получения ошибки %%Query result size limit exceeded. (157287674 > 50331648)%%
Максимальный размер ответа ограничен в 48МБ  на запрос https://a.yandex-team.ru/arcadia/ydb/core/engine/mkql_engine_flat.h?rev=rXXXXXX#L23
поэтому мы теперь получаем изначально количество строк в таблице и распределяем между ними 45 мегабайт информации https://a.yandex-team.ru/review/3375983/files/2#file-ydb/tests/functional/sqs/merge_split_common_table/test.py:R87
PS мы создаем отдельную колонку в таблице и пихаем туда множество данных для того чтобы стриггерить merge партиций по размеру
2) периодически тест таймаутиться, если мы не дожидаемся мержа до 1 партиции, т.к. после если и какая то партиция рассплититься, то происходит ситуация, когда не происходит балансирования партиций по нодам и уезда лидеров на другие ноды не происходит.
https://a.yandex-team.ru/review/3375983/files/2#file-ydb/tests/functional/sqs/merge_split_common_table/test.py:R136
3) Логи тестов (при их большом объеме) автоматически обрезаются %%..[truncated]..%%, поэтому в долгих тестах непонятно что происходило минутами,  поэтому в этот тесте отключил логирование всех каунтеров - они не приносят полезной информации
https://a.yandex-team.ru/review/3375983/files/2#file-ydb/tests/functional/sqs/merge_split_common_table/test.py:R33

2 files changed, 17 insertions, 6 deletions

diff --git a/ydb/tests/functional/sqs/merge_split_common_table/test.py b/ydb/tests/functional/sqs/merge_split_common_table/test.py
index 6bfa1e28bdd..6d2652e34e0 100644
--- a/ydb/tests/functional/sqs/merge_split_common_table/test.py
+++ b/ydb/tests/functional/sqs/merge_split_common_table/test.py
@@ -30,7 +30,7 @@ class TestSqsSplitMergeTables(KikimrSqsTestBase):
         nodes = len(self.cluster.nodes)

         leaders = []

         for node_index in range(nodes):

-            counters = self._get_counters(node_index, 'utils', counters_format='json')

+            counters = self._get_counters(node_index, 'utils', counters_format='json', dump_to_log=False)

             labels = {

                 'activity': 'SQS_QUEUE_LEADER_ACTOR',

                 'sensor': 'ActorsAliveByActivity'

@@ -69,12 +69,22 @@ class TestSqsSplitMergeTables(KikimrSqsTestBase):
             ),

         ))

 

+    def get_lines_count(self, table_path):

+        settings = ydb.RetrySettings()

+        settings.max_retries = 1

+        session = ydb.retry_operation_sync(lambda: self._driver.table_client.session().create(), retry_settings=settings)

+        res = session.transaction().execute(f'select count(*) as rows_cnt FROM `{table_path}`', commit_tx=True)

+        return res[0].rows[0]["rows_cnt"]

+

     def force_split(self, table_path):

         logging.info(f'force split {table_path}...')

         settings = ydb.RetrySettings()

         settings.max_retries = 1

         session = ydb.retry_operation_sync(lambda: self._driver.table_client.session().create(), retry_settings=settings)

-        session.transaction().execute(f'update `{table_path}` SET {self.__column_to_force_split}="{random_string(5*1024*1024)}"', commit_tx=True)

+

+        lines = self.get_lines_count(table_path)

+        row_size = int(45 * 1024 * 1024 / lines)

+        session.transaction().execute(f'update `{table_path}` SET {self.__column_to_force_split}="{random_string(row_size)}"', commit_tx=True)

 

     def get_nodes_with_leaders(self):

         leaders_per_node = self.get_leaders_per_nodes()

@@ -122,8 +132,8 @@ class TestSqsSplitMergeTables(KikimrSqsTestBase):
             leaders_per_node = self.get_leaders_per_nodes()

             nodes_with_leaders = len(list(filter(bool, leaders_per_node)))

             cur_partitions = self.get_partitions(balancing_table_path)

-            logging.debug(f'wait merge... partitions={cur_partitions}, nodes_with_leaders={nodes_with_leaders} all_leaders={sum(leaders_per_node)} : {leaders_per_node}')

-            if cur_partitions < partitions and nodes_with_leaders <= cur_partitions and sum(leaders_per_node) == queues_count:

+            logging.info(f'wait merge... partitions={cur_partitions}, nodes_with_leaders={nodes_with_leaders} all_leaders={sum(leaders_per_node)} : {leaders_per_node}')

+            if cur_partitions == 1 and nodes_with_leaders == 1 and sum(leaders_per_node) == queues_count:

                 break

             time.sleep(5)

 

diff --git a/ydb/tests/library/sqs/test_base.py b/ydb/tests/library/sqs/test_base.py
index ee76346d30b..dc01a0cdf0a 100644
--- a/ydb/tests/library/sqs/test_base.py
+++ b/ydb/tests/library/sqs/test_base.py
@@ -567,7 +567,7 @@ class KikimrSqsTestBase(object):
     def _get_ymq_counters(self, cloud, folder, node_index=0, counters_format='json'):
         return self._get_counters(node_index, "ymq_public", counters_format, cloud=cloud, folder=folder)
 
-    def _get_counters(self, node_index, component, counters_format, cloud=None, folder=None):
+    def _get_counters(self, node_index, component, counters_format, cloud=None, folder=None, dump_to_log=True):
         mon_port = self._get_mon_port(node_index)
 
         if counters_format == 'json':
@@ -593,7 +593,8 @@ class KikimrSqsTestBase(object):
             ret = reply.text
 
         assert_that(ret, not_none())
-        logging.debug('Got counters from node {}:\n{}'.format(node_index, json.dumps(ret, sort_keys=True, indent=2)))
+        if dump_to_log:
+            logging.debug('Got counters from node {}:\n{}'.format(node_index, json.dumps(ret, sort_keys=True, indent=2)))
         return ret
 
     def _get_counter(self, counters, labels):