Merge 62c7ab3 into d227d31

Author: GrigoriyPA

ydb/library/yql/cfg/tests/kqprun_config.conf

@@ -0,0 +1,24 @@
+FeatureFlags {
+  EnableExternalDataSources: true
+  EnableScriptExecutionOperations: true
+}
+
+QueryServiceConfig {
+  FileStorage {
+    MaxFiles: 1000
+    MaxSizeMb: 512
+    RetryCount: 3
+    Threads: 2
+  }
+
+  Yt {
+    DefaultSettings {
+      Name: "InferSchema"
+      Value: "1"
+    }
+    DefaultSettings {
+      Name: "_EnableYtPartitioning"
+      Value: "true"
+    }
+  }
+}

ydb/library/yql/cfg/tests/kqprun_scheme.sql

@@ -0,0 +1,8 @@
+CREATE OBJECT yt_token (TYPE SECRET) WITH (value = "token");
+
+CREATE EXTERNAL DATA SOURCE plato WITH (
+    SOURCE_TYPE="YT",
+    LOCATION="localhost",
+    AUTH_METHOD="TOKEN",
+    TOKEN_SECRET_NAME="yt_token"
+);

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

@@ -0,0 +1,84 @@
+import os
+
+import yatest.common
+
+import yql_utils
+
+
+class KqpRun(object):
+    def __init__(self, udfs_dir=None):
+        self.kqprun_binary = yql_utils.yql_binary_path('ydb/tests/tools/kqprun/kqprun')
+
+        self.config_file = yql_utils.yql_source_path(os.path.join('ydb/library/yql/cfg/tests', 'kqprun_config.conf'))
+        self.scheme_file = yql_utils.yql_source_path(os.path.join('ydb/library/yql/cfg/tests', 'kqprun_scheme.sql'))
+
+        self.res_dir = yql_utils.get_yql_dir(prefix='kqprun_')
+
+        if udfs_dir is None:
+            self.udfs_dir = yql_utils.get_udfs_path()
+        else:
+            self.udfs_dir = udfs_dir
+
+    def __res_file_path(self, name):
+        return os.path.join(self.res_dir, name)
+
+    def yql_exec(self, program=None, program_file=None, verbose=False, check_error=True, tables=None):
+        udfs_dir = self.udfs_dir
+
+        config_file = self.config_file
+        program_file = yql_utils.prepare_program(program, program_file, self.res_dir, ext='sql')[1]
+        scheme_file = self.scheme_file
+
+        results_file = self.__res_file_path('results.txt')
+        log_file = self.__res_file_path('log.txt')
+
+        cmd = self.kqprun_binary + ' '
+
+        cmd += '--emulate-yt ' \
+            '--clear-execution ' \
+            '--app-config=%(config_file)s ' \
+            '--script-query=%(program_file)s ' \
+            '--scheme-query=%(scheme_file)s ' \
+            '--result-file=%(results_file)s ' \
+            '--log-file=%(log_file)s ' \
+            '--udfs-dir=%(udfs_dir)s ' \
+            '--result-rows-limit 0 ' % locals()
+
+        if tables is not None:
+            for table in tables:
+                cmd += '--table=yt.Root/%s@%s ' % (table.full_name, table.yqlrun_file)
+        
+        proc_result = yatest.common.process.execute(cmd.strip().split(), check_exit_code=False, cwd=self.res_dir)
+        if proc_result.exit_code != 0 and check_error:
+            assert 0, \
+                'Command\n%(command)s\n finished with exit code %(code)d, stderr:\n\n%(stderr)s\n\nlog file:\n%(log_file)s' % {
+                    'command': cmd,
+                    'code': proc_result.exit_code,
+                    'stderr': proc_result.std_err,
+                    'log_file': yql_utils.read_res_file(log_file)[1]
+                }
+
+        results, log_results = yql_utils.read_res_file(results_file)
+        err, log_err = yql_utils.read_res_file(log_file)
+
+        if verbose:
+            yql_utils.log('PROGRAM:')
+            yql_utils.log(program)
+            yql_utils.log('RESULTS:')
+            yql_utils.log(log_results)
+            yql_utils.log('ERROR:')
+            yql_utils.log(log_err)
+
+        return yql_utils.YQLExecResult(
+            proc_result.std_out,
+            yql_utils.normalize_source_code_path(err.replace(self.res_dir, '<tmp_path>')),
+            results,
+            results_file,
+            None,
+            None,
+            None,
+            None,
+            program,
+            proc_result,
+            None
+        )

ydb/library/yql/tests/common/test_framework/ya.make

@@ -2,6 +2,7 @@ PY23_LIBRARY()
 
 PY_SRCS(
     TOP_LEVEL
+    kqprun.py
     yql_utils.py
     yql_ports.py
     yqlrun.py

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

@@ -11,6 +11,7 @@
 from google.protobuf import text_format
 from yql_utils import execute_sql, get_supported_providers, get_tables, get_files, get_http_files, \
     get_pragmas, log, KSV_ATTR, is_xfail, get_param, YQLExecResult, yql_binary_path
+from kqprun import KqpRun
 from yqlrun import YQLRun
 
 from utils import get_config, get_parameters_json, DATA_PATH
@@ -41,7 +42,11 @@ def get_gateways_config(http_files, yql_http_file_server, force_blocks=False, is
     return config
 
 
-def run_file_no_cache(provider, suite, case, cfg, config, yql_http_file_server, yqlrun_binary=None, extra_args=[], force_blocks=False):
+def is_hybrid(provider):
+    return provider == 'hybrid'
+
+
+def get_sql_query(provider, suite, case, config):
     if provider not in get_supported_providers(config):
         pytest.skip('%s provider is not supported here' % provider)
 
@@ -51,30 +56,37 @@ def run_file_no_cache(provider, suite, case, cfg, config, yql_http_file_server,
         if "yson" in case or "regexp" in case or "match" in case:
             pytest.skip('yson/match/regexp is not supported on non-default target platform')
 
-    xfail = is_xfail(config)
-    if get_param('TARGET_PLATFORM') and xfail:
+    if get_param('TARGET_PLATFORM') and is_xfail(config):
         pytest.skip('xfail is not supported on non-default target platform')
 
-    in_tables, out_tables = get_tables(suite, config, DATA_PATH, def_attr=KSV_ATTR)
-    files = get_files(suite, config, DATA_PATH)
-    http_files = get_http_files(suite, config, DATA_PATH)
-    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) or (is_hybrid and ('ytfile can not' in sql_query)):
+        if (provider + 'file can not' in sql_query) or (is_hybrid(provider) and ('ytfile can not' in sql_query)):
             pytest.skip(provider + ' can not execute this')
 
     pragmas.append(sql_query)
     sql_query = ';\n'.join(pragmas)
     if 'Python' in sql_query or 'Javascript' in sql_query:
         pytest.skip('ScriptUdf')
+    
+    return sql_query
+
+
+def run_file_no_cache(provider, suite, case, cfg, config, yql_http_file_server, yqlrun_binary=None, extra_args=[], force_blocks=False):
+    sql_query = get_sql_query(provider, suite, case, config)
+
+    xfail = is_xfail(config)
+
+    in_tables, out_tables = get_tables(suite, config, DATA_PATH, def_attr=KSV_ATTR)
+    files = get_files(suite, config, DATA_PATH)
+    http_files = get_http_files(suite, config, DATA_PATH)
+    http_files_urls = yql_http_file_server.register_files({}, http_files)
+
     for table in in_tables:
         if cyson.loads(table.attr).get("type") == "document":
             content = table.content
@@ -89,7 +101,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, is_hybrid=is_hybrid),
+        gateway_config=get_gateways_config(http_files, yql_http_file_server, force_blocks=force_blocks, is_hybrid=is_hybrid(provider)),
         extra_args=extra_args,
         udfs_dir=yql_binary_path('ydb/library/yql/tests/common/test_framework/udfs_deps')
     )
@@ -129,11 +141,40 @@ def run_file_no_cache(provider, suite, case, cfg, config, yql_http_file_server,
     return fixed_result, tables_res
 
 
+def run_file_kqp_no_cache(suite, case, cfg):
+    config = get_config(suite, case, cfg)
+
+    if is_xfail(config):
+        pytest.skip('skip fail tests')
+
+    sql_query = get_sql_query('yt', suite, case, config)
+    in_tables = get_tables(suite, config, DATA_PATH, def_attr=KSV_ATTR)[0]
+
+    kqprun = KqpRun(
+        udfs_dir=yql_binary_path('ydb/library/yql/tests/common/test_framework/udfs_deps')
+    )
+
+    return kqprun.yql_exec(
+        program=sql_query,
+        verbose=True,
+        check_error=True,
+        tables=in_tables
+    )
+
+
 def run_file(provider, suite, case, cfg, config, yql_http_file_server, yqlrun_binary=None, extra_args=[], force_blocks=False):
     if (suite, case, cfg) not in run_file.cache:
         run_file.cache[(suite, case, cfg)] = run_file_no_cache(provider, suite, case, cfg, config, yql_http_file_server, yqlrun_binary, extra_args, force_blocks=force_blocks)
 
     return run_file.cache[(suite, case, cfg)]
 
 
+def run_file_kqp(suite, case, cfg):
+    if (suite, case, cfg) not in run_file_kqp.cache:
+        run_file_kqp.cache[(suite, case, cfg)] = run_file_kqp_no_cache(suite, case, cfg)
+
+    return run_file_kqp.cache[(suite, case, cfg)]
+
+
 run_file.cache = {}
+run_file_kqp.cache = {}

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

@@ -0,0 +1,35 @@
+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()
+
+DEPENDS(
+    ydb/library/yql/tests/common/test_framework/udfs_deps
+    ydb/tests/tools/kqprun
+)
+
+DATA(
+    arcadia/ydb/library/yql/tests/sql  # python files
+    arcadia/ydb/library/yql/cfg/tests
+)
+
+PEERDIR(
+    ydb/library/yql/tests/common/test_framework
+)
+
+NO_CHECK_IMPORTS()
+
+REQUIREMENTS(ram:20)
+
+END()