YQ-2738 add integration tests for YT in KQP (ydb-platform#3150)

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,85 @@
+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 ' \
+            '--exclude-linked-udfs ' \
+            '--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

@@ -41,40 +41,55 @@ 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 check_provider(provider, config):
     if provider not in get_supported_providers(config):
         pytest.skip('%s provider is not supported here' % provider)
 
+
+def get_sql_query(provider, suite, case, config):
     pragmas = get_pragmas(config)
 
     if get_param('TARGET_PLATFORM'):
         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):
+    check_provider(provider, config)
+
+    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 +104,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')
     )

ydb/tests/fq/ya.make

@@ -9,4 +9,5 @@ RECURSE_FOR_TESTS(
     restarts
     s3
     yds
+    yt
 )

ydb/tests/fq/yt/kqp_yt_file.make

@@ -0,0 +1,36 @@
+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/cfg/tests
+    arcadia/ydb/library/yql/tests/sql
+    arcadia/ydb/tests/fq/yt
+)
+
+PEERDIR(
+    ydb/library/yql/tests/common/test_framework
+)
+
+NO_CHECK_IMPORTS()
+
+REQUIREMENTS(ram:20)
+
+END()