YQL-17284: Add hybtrid run for sql tests (#1602)

Author: mxkovalev

ydb/library/yql/tests/common/test_framework/yql_utils.py

@@ -446,7 +446,7 @@ def get_tables(suite, cfg, DATA_PATH, def_attr=None):
 
 
 def get_supported_providers(cfg):
-    providers = 'yt', 'kikimr', 'dq'
+    providers = 'yt', 'kikimr', 'dq', 'hybrid'
     for item in cfg:
         if item[0] == 'providers':
             providers = [i.strip() for i in ''.join(item[1:]).split(',')]
@@ -901,6 +901,21 @@ def pytest_get_current_part(path):
     return (current, 1 + maxpart)
 
 
+def normalize_result(res, sort):
+    res = cyson.loads(res) if res else cyson.loads("[]")
+    res = replace_vals(res)
+    for r in res:
+        for data in r['Write']:
+            if sort and 'Data' in data:
+                data['Data'] = sorted(data['Data'])
+            if 'Ref' in data:
+                data['Ref'] = []
+                data['Truncated'] = True
+            if 'Data' in data and len(data['Data']) == 0:
+                del data['Data']
+    return res
+
+
 class LoggingDowngrade(object):
 
     def __init__(self, loggers, level=logging.CRITICAL):

ydb/library/yql/tests/sql/dq_file.py

@@ -2,17 +2,12 @@
 import os
 import pytest
 import re
-import json
-import yql_utils
-import cyson
 
 import yatest.common
-from yql_utils import execute_sql, get_tables, get_files, get_http_files, replace_vals, get_supported_providers, \
-    KSV_ATTR, yql_binary_path, is_xfail, is_skip_forceblocks, get_param, normalize_source_code_path, dump_table_yson, \
-    get_gateway_cfg_suffix, do_custom_query_check
-from yqlrun import YQLRun
+from yql_utils import get_supported_providers, yql_binary_path, is_xfail, is_skip_forceblocks, get_param, \
+    normalize_source_code_path, dump_table_yson, get_gateway_cfg_suffix, do_custom_query_check, normalize_result
 
-from utils import get_config, get_parameters_json, DATA_PATH
+from utils import get_config, DATA_PATH
 from file_common import run_file, run_file_no_cache
 
 ASTDIFF_PATH = yql_binary_path('ydb/library/yql/tools/astdiff/astdiff')
@@ -45,20 +40,6 @@ def run_test(suite, case, cfg, tmpdir, what, yql_http_file_server):
 
     if what == 'Results' or force_blocks:
         if not xfail:
-            def normalize_res(res, sort):
-                res = cyson.loads(res) if res else cyson.loads("[]")
-                res = replace_vals(res)
-                for r in res:
-                    for data in r['Write']:
-                        if sort and 'Data' in data:
-                            data['Data'] = sorted(data['Data'])
-                        if 'Ref' in data:
-                            data['Ref'] = []
-                            data['Truncated'] = True
-                        if 'Data' in data and len(data['Data']) == 0:
-                            del data['Data']
-                return res
-
             program_sql = os.path.join(DATA_PATH, suite, '%s.sql' % case)
             with codecs.open(program_sql, encoding='utf-8') as program_file_descr:
                 sql_query = program_file_descr.read()
@@ -68,7 +49,7 @@ def normalize_res(res, sort):
 
             sort = not 'order' in sql_query.lower()
 
-            dq_res_yson = normalize_res(res.results, sort)
+            dq_res_yson = normalize_result(res.results, sort)
 
             if 'ytfile can not' in sql_query or 'yt' not in get_supported_providers(config):
                 if force_blocks:
@@ -94,7 +75,7 @@ def normalize_res(res, sort):
 
                 if do_custom_query_check(yqlrun_res, sql_query):
                     return None
-                yqlrun_res_yson = normalize_res(yqlrun_res.results, sort)
+                yqlrun_res_yson = normalize_result(yqlrun_res.results, sort)
 
                 # Compare results
                 assert dq_res_yson == yqlrun_res_yson, 'RESULTS_DIFFER\n' \

ydb/library/yql/tests/sql/file_common.py

@@ -16,7 +16,7 @@
 from utils import get_config, get_parameters_json, DATA_PATH
 
 
-def get_gateways_config(http_files, yql_http_file_server, force_blocks=False):
+def get_gateways_config(http_files, yql_http_file_server, force_blocks=False, is_hybrid=False):
     config = None
 
     if http_files or force_blocks:
@@ -29,6 +29,13 @@ def get_gateways_config(http_files, yql_http_file_server, force_blocks=False):
             config_message.SqlCore.TranslationFlags.extend(['EmitAggApply'])
             flags = config_message.YqlCore.Flags.add()
             flags.Name = 'UseBlocks'
+        if is_hybrid:
+            activate_hybrid = config_message.Yt.DefaultSettings.add()
+            activate_hybrid.Name = "HybridDqExecution"
+            activate_hybrid.Value = "1"
+            deactivate_dq = config_message.Dq.DefaultSettings.add()
+            deactivate_dq.Name = "AnalyzeQuery"
+            deactivate_dq.Value = "0"
         config = text_format.MessageToString(config_message)
 
     return config
@@ -54,13 +61,14 @@ def run_file_no_cache(provider, suite, case, cfg, config, yql_http_file_server,
     http_files_urls = yql_http_file_server.register_files({}, http_files)
 
     program_sql = os.path.join(DATA_PATH, suite, '%s.sql' % case)
+    is_hybrid = provider == 'hybrid'
 
     with codecs.open(program_sql, encoding='utf-8') as program_file_descr:
         sql_query = program_file_descr.read()
         if get_param('TARGET_PLATFORM'):
             if "Yson::" in sql_query:
                 pytest.skip('yson udf is not supported on non-default target platform')
-        if provider + 'file can not' in sql_query:
+        if (provider + 'file can not' in sql_query) or (is_hybrid and ('ytfile can not' in sql_query)):
             pytest.skip(provider + ' can not execute this')
 
     pragmas.append(sql_query)
@@ -81,7 +89,7 @@ def run_file_no_cache(provider, suite, case, cfg, config, yql_http_file_server,
         prov=provider,
         keep_temp=not re.search(r"yt\.ReleaseTempData", sql_query),
         binary=yqlrun_binary,
-        gateway_config=get_gateways_config(http_files, yql_http_file_server, force_blocks=force_blocks),
+        gateway_config=get_gateways_config(http_files, yql_http_file_server, force_blocks=force_blocks, is_hybrid=is_hybrid),
         extra_args=extra_args,
         udfs_dir=yql_binary_path('ydb/library/yql/tests/common/test_framework/udfs_deps')
     )

ydb/library/yql/tests/sql/hybrid_file.make

@@ -0,0 +1,48 @@
+PY2TEST()
+
+TEST_SRCS(
+    test.py
+)
+
+IF (SANITIZER_TYPE OR WITH_VALGRIND)
+    TIMEOUT(1800)
+    SIZE(LARGE)
+    TAG(ya:fat sb:ttl=2)
+ELSE()
+    TIMEOUT(600)
+    SIZE(MEDIUM)
+    TAG(sb:ttl=2)
+ENDIF()
+
+FORK_TESTS()
+FORK_SUBTESTS()
+SPLIT_FACTOR(10)
+
+DEPENDS(
+    ydb/library/yql/tools/astdiff
+    ydb/library/yql/tools/dqrun
+    ydb/library/yql/tools/yqlrun
+    ydb/library/yql/tests/common/test_framework/udfs_deps
+    ydb/library/yql/udfs/test/test_import
+)
+DATA(
+    arcadia/ydb/library/yql/tests/sql # python files
+    arcadia/ydb/library/yql/mount
+    arcadia/ydb/library/yql/cfg/tests
+)
+PEERDIR(
+    ydb/library/yql/tests/common/test_framework
+    library/python/testing/swag/lib
+)
+
+NO_CHECK_IMPORTS()
+
+REQUIREMENTS(
+    ram:32
+)
+
+IF (SANITIZER_TYPE == "memory")
+    TAG(ya:not_autocheck) # YQL-15385
+ENDIF()
+
+END()

ydb/library/yql/tests/sql/hybrid_file.py

@@ -0,0 +1,63 @@
+import codecs
+import os
+import pytest
+import re
+
+import yatest.common
+
+from yql_utils import replace_vals, yql_binary_path, is_xfail, get_param, \
+    get_gateway_cfg_suffix, normalize_result
+
+from utils import get_config, DATA_PATH
+from file_common import run_file, run_file_no_cache
+
+ASTDIFF_PATH = yql_binary_path('ydb/library/yql/tools/astdiff/astdiff')
+DQRUN_PATH = yql_binary_path('ydb/library/yql/tools/dqrun/dqrun')
+
+def run_test(suite, case, cfg, tmpdir, what, yql_http_file_server):
+    if get_param('SQL_FLAGS'):
+        if what == 'Debug' or what == 'Plan':
+            pytest.skip('SKIP')
+
+    if get_gateway_cfg_suffix() != '' and what != 'Results':
+        pytest.skip('non-trivial gateways.conf')
+
+    config = get_config(suite, case, cfg)
+    xfail = is_xfail(config)
+    if xfail and what != 'Results':
+        pytest.skip('SKIP')
+
+    (res, tables_res) = run_file('hybrid', suite, case, cfg, config, yql_http_file_server, DQRUN_PATH, extra_args=["--emulate-yt", "--analyze-query"])
+
+    to_canonize = []
+
+    if what == 'Results':
+
+        program_sql = os.path.join(DATA_PATH, suite, '%s.sql' % case)
+        with codecs.open(program_sql, encoding='utf-8') as program_file_descr:
+            sql_query = program_file_descr.read()
+
+        # yqlrun run
+        yqlrun_res, yqlrun_tables_res = run_file_no_cache('yt', suite, case, cfg, config, yql_http_file_server)
+        hybrid_result_name = 'HYBRIDFILE'
+        yqlrun_result_name = 'YQLRUN'
+
+        sort = not 'order' in sql_query.lower()
+        hybrid_res_yson = normalize_result(res.results, sort)
+        yqlrun_res_yson = normalize_result(yqlrun_res.results, sort)
+
+        # Compare results
+        assert hybrid_res_yson == yqlrun_res_yson, 'RESULTS_DIFFER\n' \
+            '%(hybrid_result_name)s result:\n %(hybrid_res_yson)s\n\n' \
+            '%(yqlrun_result_name)s result:\n %(yqlrun_res_yson)s\n' % locals()
+        return
+    
+    if what == 'Plan':
+        to_canonize = [yatest.common.canonical_file(res.plan_file)]
+
+    if what == 'Debug':
+        with open(res.opt_file + "_patched", 'w') as f:
+            f.write(re.sub(r"""("?_logical_id"?) '\d+""", r"""\1 '0""", res.opt).encode('utf-8'))
+        to_canonize = [yatest.common.canonical_file(res.opt_file + "_patched", diff_tool=ASTDIFF_PATH)]
+
+    return to_canonize