Allow empty inputs for stage compilation (#17009)

Author: nikvas0

ydb/core/kqp/ut/query/kqp_query_ut.cpp

@@ -2211,6 +2211,73 @@ Y_UNIT_TEST_SUITE(KqpQuery) {
             UNIT_ASSERT(result.GetStatus() == NYdb::EStatus::OVERLOADED);
         }
     }
+
+    Y_UNIT_TEST(TableSinkWithSubquery) {
+        NKikimrConfig::TAppConfig appConfig;
+        auto settings = TKikimrSettings()
+            .SetAppConfig(appConfig)
+            .SetWithSampleTables(false);
+
+        TKikimrRunner kikimr(settings);
+        Tests::NCommon::TLoggerInit(kikimr).Initialize();
+
+        auto session = kikimr.GetTableClient().CreateSession().GetValueSync().GetSession();
+
+        const TString query = R"(
+            CREATE TABLE `/Root/table1` (
+                p1 Utf8,
+                PRIMARY KEY (p1)
+            )
+            WITH (
+                STORE = ROW
+            );
+
+            CREATE TABLE `/Root/table2` (
+                p1 Utf8,
+                PRIMARY KEY (p1)
+            )
+            WITH (
+                STORE = ROW
+            );
+        )";
+
+        auto result = session.ExecuteSchemeQuery(query).GetValueSync();
+        UNIT_ASSERT_C(result.GetStatus() == NYdb::EStatus::SUCCESS, result.GetIssues().ToString());
+
+        auto client = kikimr.GetQueryClient();
+
+        {
+            auto prepareResult = client.ExecuteQuery(R"(
+                UPSERT INTO `/Root/table1` (p1) VALUES ("a") , ("b"), ("c");
+            )", NYdb::NQuery::TTxControl::BeginTx().CommitTx()).ExtractValueSync();
+            UNIT_ASSERT_C(prepareResult.IsSuccess(), prepareResult.GetIssues().ToString());
+        }
+
+        {
+            auto result = client.ExecuteQuery(R"(
+                $data2 = Cast(AsList() As List<Struct<p1: Utf8>>);
+
+                /* query */
+                SELECT d1.p1 AS p1,
+                FROM `/Root/table1` AS d1
+                CROSS JOIN AS_TABLE($data2) AS d2;
+            )", NYdb::NQuery::TTxControl::BeginTx().CommitTx()).ExtractValueSync();
+            UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
+        }
+
+        {
+            auto result = client.ExecuteQuery(R"(
+                $data2 = Cast(AsList() As List<Struct<p1: Utf8>>);
+
+                /* query */
+                INSERT INTO `/Root/table1`
+                SELECT d1.p1 AS p1,
+                FROM `/Root/table2` AS d1
+                CROSS JOIN AS_TABLE($data2) AS d2;
+            )", NYdb::NQuery::TTxControl::BeginTx().CommitTx()).ExtractValueSync();
+            UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
+        }
+    }
 }
 
 } // namespace NKqp

ydb/library/yql/dq/runtime/dq_tasks_runner.cpp

@@ -608,18 +608,22 @@ class TDqTaskRunner : public IDqTaskRunner {
                 }
             }
 
-            auto entryNode = AllocatedHolder->ProgramParsed.CompGraph->GetEntryPoint(i, true);
-            if (transform) {
-                transform->TransformInput = DqBuildInputValue(inputDesc, transform->TransformInputType, std::move(inputs), holderFactory, {}, Stats->StartTs, InputConsumed, PgBuilder_.get());
-                inputs.clear();
-                inputs.emplace_back(transform->TransformOutput);
-                entryNode->SetValue(AllocatedHolder->ProgramParsed.CompGraph->GetContext(),
-                    CreateInputUnionValue(transform->TransformOutput->GetInputType(), std::move(inputs), holderFactory,
-                        {inputStats, transform->TransformOutputType}, Stats->StartTs, InputConsumed));
+            auto entryNode = AllocatedHolder->ProgramParsed.CompGraph->GetEntryPoint(i, false);
+            if (entryNode) {
+                if (transform) {
+                    transform->TransformInput = DqBuildInputValue(inputDesc, transform->TransformInputType, std::move(inputs), holderFactory, {}, Stats->StartTs, InputConsumed, PgBuilder_.get());
+                    inputs.clear();
+                    inputs.emplace_back(transform->TransformOutput);
+                    entryNode->SetValue(AllocatedHolder->ProgramParsed.CompGraph->GetContext(),
+                        CreateInputUnionValue(transform->TransformOutput->GetInputType(), std::move(inputs), holderFactory,
+                            {inputStats, transform->TransformOutputType}, Stats->StartTs, InputConsumed));
+                } else {
+                    entryNode->SetValue(AllocatedHolder->ProgramParsed.CompGraph->GetContext(),
+                        DqBuildInputValue(inputDesc, entry->InputItemTypes[i], std::move(inputs), holderFactory,
+                            {inputStats, entry->InputItemTypes[i]}, Stats->StartTs, InputConsumed, PgBuilder_.get()));
+                }
             } else {
-                entryNode->SetValue(AllocatedHolder->ProgramParsed.CompGraph->GetContext(),
-                    DqBuildInputValue(inputDesc, entry->InputItemTypes[i], std::move(inputs), holderFactory,
-                        {inputStats, entry->InputItemTypes[i]}, Stats->StartTs, InputConsumed, PgBuilder_.get()));
+                // In some cases we don't need input. For example, for joining EmptyIterator with table.
             }
         }