Merge 44e0a8e into d425c83

vladl2802 · web-flow · commit 8dffd6181db3 · 2024-08-26T11:20:58.000Z
diff --git a/ydb/library/benchmarks/runner/run_tests/run_tests.py b/ydb/library/benchmarks/runner/run_tests/run_tests.py
@@ -0,0 +1,116 @@
+import argparse
+import subprocess
+import pathlib
+import os
+from sys import stderr
+
+def parse_args(passed=None):
+    YDB_ROOT = "../../../"
+    
+    def variant(string):
+        if string not in ["h", "ds"]:
+            raise ValueError("variant must be h or ds")
+        return string
+    
+    parser = argparse.ArgumentParser()
+    
+    parser.add_argument('--datasize', type=int, default=1)
+    parser.add_argument('--variant', type=variant, default='h')
+    parser.add_argument('--tasks', type=int, default=1)
+    parser.add_argument('--ydb-root', type=lambda path: pathlib.Path(path).resolve(), default=YDB_ROOT)
+    parser.add_argument('-o', '--output', default="./results")
+    parser.add_argument('--clean-old', action="store_true", default=False)
+    parser.add_argument('--query-filter', action="append", default=[])
+    
+    return parser.parse_args(passed)
+
+class Runner:
+    def prepare_queries_dir(self, custom_pragmas):
+        print("Preparing queries...")
+        self.queries_dir.mkdir(parents=True, exist_ok=True)
+        cmd = [self.args.gen_queries]
+        cmd += ["--output", f"{self.queries_dir}"]
+        cmd += ["--variant", f"{self.args.variant}"]
+        cmd += ["--syntax", "yql"]
+        cmd += ["--dataset-size", f"{self.args.datasize}"]
+        for it in custom_pragmas:
+            cmd += ["--pragma", it]
+        subprocess.run(cmd)
+    
+    def prepare_tpc_dir(self):
+        print("Preparing tpc...")
+        cmd = [f"{self.args.downloaders_dir}/download_files_{self.args.variant}_{self.args.datasize}.sh"]
+        subprocess.run(cmd)
+        os.symlink(f"{self.args.downloaders_dir}/tpc", f"{pathlib.Path("./tpc")}", target_is_directory=True)
+    
+    def __init__(self, args, enable_spilling):
+        self.args = args
+        self.enable_spilling = enable_spilling
+        
+        self.queries_dir = pathlib.Path(f"queries{"+" if self.enable_spilling else "-"}spilling-{args.datasize}-{args.tasks}")
+        if self.args.clean_old or not self.queries_dir.exists():
+            self.prepare_queries_dir([
+                f"dq.MaxTasksPerStage={self.args.tasks}",
+                "dq.OptLLVM=ON"
+            ] + [
+                "dq.UseFinalizeByKey=true",
+                "dq.EnableSpillingNodes=All",
+            ] if self.enable_spilling else [])
+        
+        self.tpc_dir = pathlib.Path(f"{self.args.downloaders_dir}/tpc/{self.args.variant}/{self.args.datasize}").resolve()
+        if self.args.clean_old or not self.tpc_dir.exists():
+            self.prepare_tpc_dir()
+        
+        self.result_dir = pathlib.Path(f"{self.args.output}/{"with" if self.enable_spilling else "no"}-spilling/{args.variant}-{args.datasize}-{args.tasks}").resolve()
+        self.result_dir.mkdir(parents=True, exist_ok=True)
+
+    def run(self):
+        cmd = ["/usr/bin/time", f"{self.args.runner_path}"]
+        cmd += ["--perf"]
+        for it in self.args.query_filter:
+            cmd += ["--query-filter", it]
+        cmd += ["--query-dir", f"{self.queries_dir}/{self.args.variant}"]
+        cmd += ["--bindings", f"{self.queries_dir}/{self.args.variant}/bindings.json"]
+        cmd += ["--result-dir", f"{self.result_dir}"]
+        cmd += [f"{self.args.dqrun}", "-s"]
+        cmd += ["--enable-spilling"] if self.enable_spilling else []
+        cmd += ["--udfs-dir", f"{self.args.udfs_dir}"]
+        cmd += ["--fs-cfg", f"{self.args.fs_cfg}"]
+        cmd += ["--gateways-cfg", f"{self.args.gateways_cfg}"]
+        subprocess.run(cmd)
+        
+        return self.result_dir
+
+def result_compare(args, to_compare):
+    cmd = [f"{args.result_compare}"]
+    cmd += ["-v"]
+    cmd += to_compare
+    print(cmd, file=stderr)
+    with open(f"{args.output}/result-{args.variant}-{args.datasize}-{args.tasks}.htm", "w") as result_table:
+        subprocess.run(cmd, stdout=result_table)
+
+def main(passed=None):
+    args, _ = parse_args(passed)
+    args.dqrun = args.ydb_root / "library" / "yql" / "tools" / "dqrun" / "dqrun"
+    args.gen_queries = args.ydb_root / "library" / "benchmarks" / "gen_queries" / "gen_queries"
+    args.downloaders_dir = args.ydb_root / "library" / "benchmarks" / "runner"
+    args.udfs_dir = args.ydb_root / "library" / "yql" / "udfs" / "common"
+    args.fs_cfg = args.ydb_root / "library" / "yql" / "tools" / "dqrun" / "examples" / "fs.conf"
+    args.result_compare = args.ydb_root / "library" / "benchmarks" / "runner" / "result_compare" / "result_compare"
+    args.gateways_cfg = args.ydb_root / "library" / "benchmarks" / "runner" / "runner" / "test-gateways.conf"
+    args.runner_path = args.ydb_root / "library" / "benchmarks" / "runner" / "runner" / "runner"
+    
+    print(args.query_filter)
+    
+    results = []
+    print("With spilling...", file=stderr)
+    results.append(Runner(args, True).run())
+    print("No spilling...", file=stderr)
+    results.append(Runner(args, False).run())
+    
+    print(results, file=stderr)
+    
+    result_compare(args, results)
+
+if __name__ == "__main__":
+    main()
diff --git a/ydb/library/benchmarks/runner/run_tests/ya.make b/ydb/library/benchmarks/runner/run_tests/ya.make
@@ -0,0 +1,10 @@
+PY3_PROGRAM()
+
+PY_SRCS(
+    MAIN run_tests.py
+)
+
+PEERDIR(
+)
+
+END()
diff --git a/ydb/library/benchmarks/runner/runner/runner.py b/ydb/library/benchmarks/runner/runner/runner.py
@@ -65,25 +65,25 @@ def run(argv, out, err, timeout=30*60, hard_timeout=5):
 
 
 def main():
-
     parser = argparse.ArgumentParser()
     parser.add_argument('--query-dir', type=str, default='q/scalar')
     parser.add_argument('--bindings', type=str, default='bindings.json')
-    parser.add_argument('--result-dir', type=str, default="result-{:%Y%m%dT%H%M%S}".format(datetime.datetime.now()))
+    parser.add_argument('--result-dir', type=Path, default="result-{:%Y%m%dT%H%M%S}".format(datetime.datetime.now()))
     parser.add_argument('--timeout', type=int, default=30*60)
     parser.add_argument('--perf', action='store_true')
     parser.add_argument('--arc-path', type=str, default='{}/arcadia'.format(os.environ['HOME']))
     parser.add_argument('--include-q', default=[], action='append')
     parser.add_argument('--exclude-q', default=[], action='append')
+    parser.add_argument('--query-filter', action="append", default=[])
+            
     args, argv = parser.parse_known_intermixed_args()
     qdir = args.query_dir
     bindings = args.bindings
     outdir = args.result_dir
     assert len(argv)
     querydir = Path(qdir)
-    os.makedirs(outdir + '/' + qdir, exist_ok=True)
-    with open(outdir + '/' + qdir + "/summary.tsv", "w") as outf, \
-         open(outdir + '/' + qdir + "/summary.json", "w") as outj:
+    with open(outdir / "summary.tsv", "w") as outf, \
+         open(outdir / "summary.json", "w") as outj:
         print(' '.join(argv + ['-p', qdir, '--bindings-file', bindings]), file=outf)
         print(json.dumps({
             'cmdline': argv,
@@ -92,8 +92,13 @@ def main():
             'version': 100
         }), file=outj)
         for query in sorted(querydir.glob('**/*.sql'), key=lambda x: tuple(map(lambda y: int(y) if re.match(RE_DIGITS, y) else y, re.split(RE_DIGITS, str(x))))):
-            q = str(query)
-            name = outdir + '/' + q
+            q = str(query.stem)
+            # q<num>.sql
+            num = q[1:-4]
+            if args.query_filter != [] and num not in args.query_filter:
+                continue
+            print(f"{q}", end="", flush=True)
+            name = str(outdir / q)
             if len(args.include_q):
                 include = False
                 for r in args.include_q:
@@ -112,6 +117,7 @@ def main():
                     continue
             print(q, end='\t', file=outf)
             outname = name + '-result.yson'
+            print(".", end="", flush=True)
             exitcode, rusage, elapsed, iostat = run(
                 argv + [
                     '--result-file', outname,
@@ -120,7 +126,7 @@ def main():
                     '--err-file', name + '-err.txt',
                     '--expr-file', name + '-expr.txt',
                     '--stat', name + '-stat.yson',
-                    '-p', q
+                    '-p', str(query)
                 ],
                 name + '-stdout.txt',
                 name + '-stderr.txt',
@@ -164,25 +170,27 @@ def main():
                 }
             }), file=outj)
             outj.flush()
+            print(".", end="", flush=True)
             if args.perf:
                 exitcode, rusage, elapsed, iostat = run(
-                    ['{}/ya'.format(args.arc_path), 'tool', 'perf', 'record', '-F250', '-g', '--call-graph', 'dwarf', '-o', '{}/perf.data'.format(outdir), '--'] +
+                    ['/usr/bin/perf', 'record', '-F250', '-g', '--call-graph', 'dwarf', '-o', '{}/perf.data'.format(outdir), '--'] +
                     argv + [
                         '--result-file', '/dev/null',
                         '--bindings-file', bindings,
                         '--plan-file', '/dev/null',
                         '--err-file', '/dev/null',
                         '--expr-file', '/dev/null',
-                        '-p', q
+                        '-p', str(query)
                     ],
                     name + '-stdout-perf.txt',
                     name + '-stderr-perf.txt',
                     timeout=args.timeout)
                 os.system('''
-                {0}/ya tool perf script -i {2}/perf.data --header |
+                /usr/bin/perf script -i {2}/perf.data --header |
                 {0}/contrib/tools/flame-graph/stackcollapse-perf.pl |
                 {0}/contrib/tools/flame-graph/flamegraph.pl > {1}.svg
                 '''.format(args.arc_path, name, outdir))
+            print(".", flush=True)
 
 
 if __name__ == "__main__":
diff --git a/ydb/library/benchmarks/runner/runner/ya.make b/ydb/library/benchmarks/runner/runner/ya.make
@@ -5,6 +5,13 @@ PY_SRCS(
 )
 
 PEERDIR(
+    ydb/library/yql/tools/dqrun
+    ydb/library/benchmarks/gen_queries
+
+    ydb/library/yql/udfs/common/set
+    ydb/library/yql/udfs/common/url_base
+    ydb/library/yql/udfs/common/datetime2
+    ydb/library/yql/udfs/common/re2
 )
 
 END()
diff --git a/ydb/library/benchmarks/runner/tpc_tests.py b/ydb/library/benchmarks/runner/tpc_tests.py
@@ -0,0 +1,16 @@
+import run_tests.run_tests as run_tests
+
+
+def wrapped_run(variant, datasize, tasks):
+    cmd = []
+    cmd += ["--variant", f"{variant}"]
+    cmd += ["--datasize", f"{datasize}"]
+    cmd += ["--tasks", f"{tasks}"]
+    cmd += ["--query-filter", "1"]
+    cmd += ["--ydb-root", "/home/vladluk/ydbwork/ydb/ydb"]
+    cmd += ["-o", "/home/vladluk/ydbwork/ydb/ydb/library/benchmarks/runner/results"]
+    run_tests.main(cmd)
+
+
+def test_tpc_h_1_1():
+    wrapped_run("h", 1, 1)
diff --git a/ydb/library/benchmarks/runner/ya.make b/ydb/library/benchmarks/runner/ya.make
@@ -1,4 +1,19 @@
+PY3TEST()
+
+SIZE(MEDIUM)
+
+PY_SRCS(
+    run_tests/run_tests.py
+)
+
+TEST_SRCS(
+    tpc_tests.py
+)
+
+END()
+
 RECURSE(
+    run_tests
     runner
     result_convert
     result_compare

Original file line number	Diff line number	Diff line change
`@@ -5,6 +5,13 @@ PY_SRCS(`
`5`	`5`	`)`
`6`	`6`
`7`	`7`	`PEERDIR(`
	`8`	`+ ydb/library/yql/tools/dqrun`
	`9`	`+ ydb/library/benchmarks/gen_queries`
	`10`	`+`
	`11`	`+ ydb/library/yql/udfs/common/set`
	`12`	`+ ydb/library/yql/udfs/common/url_base`
	`13`	`+ ydb/library/yql/udfs/common/datetime2`
	`14`	`+ ydb/library/yql/udfs/common/re2`
`8`	`15`	`)`
`9`	`16`
`10`	`17`	`END()`