[dq] Fix the parallel precompute execution hangup in some cases (#5046)

ydb-platform · May 31, 2024 · 4b60d4d · 4b60d4d
1 parent 9a1058e
commit 4b60d4d
Show file tree

Hide file tree

Showing 9 changed files with 191 additions and 69 deletions.
diff --git a/ydb/library/yql/providers/dq/provider/exec/ya.make b/ydb/library/yql/providers/dq/provider/exec/ya.make
@@ -9,6 +9,7 @@ PEERDIR(
     library/cpp/yson/node
     library/cpp/svnversion
     library/cpp/digest/md5
+    library/cpp/threading/future
     ydb/public/lib/yson_value
     ydb/public/sdk/cpp/client/ydb_driver
     ydb/library/yql/core

diff --git a/ydb/library/yql/providers/dq/provider/exec/yql_dq_exectransformer.cpp b/ydb/library/yql/providers/dq/provider/exec/yql_dq_exectransformer.cpp
diff --git a/ydb/library/yql/tests/sql/dq_file/part10/canondata/result.json b/ydb/library/yql/tests/sql/dq_file/part10/canondata/result.json
@@ -653,6 +653,39 @@
         }
     ],
     "test.test[csee-same_l1-default.txt-Results]": [],
+    "test.test[dq-precompute_parallel_indep--Analyze]": [
+        {
+            "checksum": "4f6978cd58bf614cf032ab65c9ccd92e",
+            "size": 16119,
+            "uri": "https://{canondata_backend}/1942525/68adc93267fab0086b1faf825d05122058d5f469/resource.tar.gz#test.test_dq-precompute_parallel_indep--Analyze_/plan.txt"
+        }
+    ],
+    "test.test[dq-precompute_parallel_indep--Debug]": [
+        {
+            "checksum": "60e05e19e13cece105b65405a56027f7",
+            "size": 5127,
+            "uri": "https://{canondata_backend}/1942525/68adc93267fab0086b1faf825d05122058d5f469/resource.tar.gz#test.test_dq-precompute_parallel_indep--Debug_/opt.yql_patched"
+        }
+    ],
+    "test.test[dq-precompute_parallel_indep--Plan]": [
+        {
+            "checksum": "f0c8afc03280179f72fde524e4bbeaf1",
+            "size": 16363,
+            "uri": "https://{canondata_backend}/1942525/68adc93267fab0086b1faf825d05122058d5f469/resource.tar.gz#test.test_dq-precompute_parallel_indep--Plan_/plan.txt"
+        }
+    ],
+    "test.test[dq-precompute_parallel_indep--Results]": [
+        {
+            "checksum": "0c4292db52668dfefd7453df371175ee",
+            "size": 78,
+            "uri": "https://{canondata_backend}/1942525/68adc93267fab0086b1faf825d05122058d5f469/resource.tar.gz#test.test_dq-precompute_parallel_indep--Results_/Output.txt"
+        },
+        {
+            "checksum": "778fa10c4a890f9bfdaf2e98a8f09462",
+            "size": 377,
+            "uri": "https://{canondata_backend}/1942525/68adc93267fab0086b1faf825d05122058d5f469/resource.tar.gz#test.test_dq-precompute_parallel_indep--Results_/Output.yqlrun.txt.attr"
+        }
+    ],
     "test.test[expr-expr_add_literal_nulls-default.txt-Analyze]": [
         {
             "checksum": "b4dd508a329723c74293d80f0278c705",

diff --git a/ydb/library/yql/tests/sql/dq_file/part2/canondata/result.json b/ydb/library/yql/tests/sql/dq_file/part2/canondata/result.json
@@ -772,6 +772,49 @@
         }
     ],
     "test.test[distinct-distinct_count_and_full_count-default.txt-Results]": [],
+    "test.test[dq-precompute_parallel_mix--Analyze]": [
+        {
+            "checksum": "191f1cfae918330a0d2b2b7c5880565e",
+            "size": 10762,
+            "uri": "https://{canondata_backend}/1937027/06ef7ef6dee3ee697013fd133a8e8a843e5f5de9/resource.tar.gz#test.test_dq-precompute_parallel_mix--Analyze_/plan.txt"
+        }
+    ],
+    "test.test[dq-precompute_parallel_mix--Debug]": [
+        {
+            "checksum": "d3976ed532b22b1acd4ebdf94a6a2676",
+            "size": 3920,
+            "uri": "https://{canondata_backend}/1937027/06ef7ef6dee3ee697013fd133a8e8a843e5f5de9/resource.tar.gz#test.test_dq-precompute_parallel_mix--Debug_/opt.yql_patched"
+        }
+    ],
+    "test.test[dq-precompute_parallel_mix--Plan]": [
+        {
+            "checksum": "1e3854603cf82672f83af3c5bf63caf8",
+            "size": 10884,
+            "uri": "https://{canondata_backend}/1937027/06ef7ef6dee3ee697013fd133a8e8a843e5f5de9/resource.tar.gz#test.test_dq-precompute_parallel_mix--Plan_/plan.txt"
+        }
+    ],
+    "test.test[dq-precompute_parallel_mix--Results]": [
+        {
+            "checksum": "2c10838daac18c57801f0c309c433310",
+            "size": 62,
+            "uri": "https://{canondata_backend}/1937027/06ef7ef6dee3ee697013fd133a8e8a843e5f5de9/resource.tar.gz#test.test_dq-precompute_parallel_mix--Results_/Output1.txt"
+        },
+        {
+            "checksum": "99a23a80241cdd0057583910f9155f61",
+            "size": 600,
+            "uri": "https://{canondata_backend}/1937027/06ef7ef6dee3ee697013fd133a8e8a843e5f5de9/resource.tar.gz#test.test_dq-precompute_parallel_mix--Results_/Output1.yqlrun.txt.attr"
+        },
+        {
+            "checksum": "cb0e24eef504f21bcf257bd15077a828",
+            "size": 190,
+            "uri": "https://{canondata_backend}/1937027/06ef7ef6dee3ee697013fd133a8e8a843e5f5de9/resource.tar.gz#test.test_dq-precompute_parallel_mix--Results_/Output2.txt"
+        },
+        {
+            "checksum": "99a23a80241cdd0057583910f9155f61",
+            "size": 600,
+            "uri": "https://{canondata_backend}/1937027/06ef7ef6dee3ee697013fd133a8e8a843e5f5de9/resource.tar.gz#test.test_dq-precompute_parallel_mix--Results_/Output2.yqlrun.txt.attr"
+        }
+    ],
     "test.test[expr-as_table_emptylist2-default.txt-Analyze]": [
         {
             "checksum": "55515ae638f317612d048052be489bfd",

diff --git a/ydb/library/yql/tests/sql/sql2yql/canondata/result.json b/ydb/library/yql/tests/sql/sql2yql/canondata/result.json
@@ -5046,6 +5046,20 @@
             "uri": "https://{canondata_backend}/1937027/973c239492ba32946806ddc66cf0af4b38c06ae8/resource.tar.gz#test_sql2yql.test_dq-precompute_parallel_/sql.yql"
         }
     ],
+    "test_sql2yql.test[dq-precompute_parallel_indep]": [
+        {
+            "checksum": "decf219a1091da410ee315ad657f3fd4",
+            "size": 3402,
+            "uri": "https://{canondata_backend}/1903885/e05277820455e95dc3443a43a34b1956971f88cf/resource.tar.gz#test_sql2yql.test_dq-precompute_parallel_indep_/sql.yql"
+        }
+    ],
+    "test_sql2yql.test[dq-precompute_parallel_mix]": [
+        {
+            "checksum": "c557f52fe0852d86a73c69134c80e332",
+            "size": 4230,
+            "uri": "https://{canondata_backend}/1937027/08c3c9c6866171c02a77702fcc07f26f028f4001/resource.tar.gz#test_sql2yql.test_dq-precompute_parallel_mix_/sql.yql"
+        }
+    ],
     "test_sql2yql.test[dq-precompute_result]": [
         {
             "checksum": "043145e8db2d44a158f122c3a724e034",
@@ -23673,6 +23687,20 @@
             "uri": "https://{canondata_backend}/1880306/64654158d6bfb1289c66c626a8162239289559d0/resource.tar.gz#test_sql_format.test_dq-precompute_parallel_/formatted.sql"
         }
     ],
+    "test_sql_format.test[dq-precompute_parallel_indep]": [
+        {
+            "checksum": "5ee71b61a11efee1059774aa4478b4fc",
+            "size": 215,
+            "uri": "https://{canondata_backend}/1903885/e05277820455e95dc3443a43a34b1956971f88cf/resource.tar.gz#test_sql_format.test_dq-precompute_parallel_indep_/formatted.sql"
+        }
+    ],
+    "test_sql_format.test[dq-precompute_parallel_mix]": [
+        {
+            "checksum": "ef7b7571431f5c4e198b4df24fab006a",
+            "size": 322,
+            "uri": "https://{canondata_backend}/1937027/08c3c9c6866171c02a77702fcc07f26f028f4001/resource.tar.gz#test_sql_format.test_dq-precompute_parallel_mix_/formatted.sql"
+        }
+    ],
     "test_sql_format.test[dq-precompute_result]": [
         {
             "checksum": "e9aad7ef908ef2e5d99e7fd838fdc4e9",

diff --git a/ydb/library/yql/tests/sql/suites/dq/precompute_parallel_indep.cfg b/ydb/library/yql/tests/sql/suites/dq/precompute_parallel_indep.cfg
@@ -0,0 +1,5 @@
+providers dq
+in Input1 input.txt
+in Input2 input.txt
+in Input3 input.txt
+out Output output.txt
diff --git a/ydb/library/yql/tests/sql/suites/dq/precompute_parallel_indep.sql b/ydb/library/yql/tests/sql/suites/dq/precompute_parallel_indep.sql
@@ -0,0 +1,5 @@
+use plato;
+
+insert into Output select sum(cast(key as int32)) from Input1;
+insert into Output select sum(cast(key as int32)) from Input2;
+insert into Output select sum(cast(key as int32)) from Input3;
diff --git a/ydb/library/yql/tests/sql/suites/dq/precompute_parallel_mix.cfg b/ydb/library/yql/tests/sql/suites/dq/precompute_parallel_mix.cfg
@@ -0,0 +1,4 @@
+providers dq
+in Input input.txt
+out Output1 output1.txt
+out Output2 output2.txt
diff --git a/ydb/library/yql/tests/sql/suites/dq/precompute_parallel_mix.sql b/ydb/library/yql/tests/sql/suites/dq/precompute_parallel_mix.sql
@@ -0,0 +1,7 @@
+use plato;
+
+$a = select key from Input order by key limit 1;
+$b = select key from Input order by key limit 1 offset 1;
+
+insert into Output1 select * from Input where key <= $a;
+insert into Output2 select * from Input where key >= $a and key != $b;