fix incorrect subscribe mode check for internal subquery

src/Interpreters/InterpreterSelectWithUnionQuery.cpp

@@ -400,7 +400,7 @@ BlockIO InterpreterSelectWithUnionQuery::execute()
         BuildQueryPipelineSettings::fromContext(context), context);
 
     /// proton : starts, setup execute mode
-    builder->setExecuteMode(queryExecuteMode(query_plan.isStreaming(), context->getSettingsRef()));
+    builder->setExecuteMode(queryExecuteMode(query_plan.isStreaming(), options.is_subquery, context->getSettingsRef()));
     /// proton : ends
 
     res.pipeline = QueryPipelineBuilder::getPipeline(std::move(*builder));

src/Processors/QueryPlan/QueryExecuteMode.cpp

@@ -10,7 +10,7 @@ extern const int INVALID_SETTING_VALUE;
 extern const int NOT_IMPLEMENTED;
 }
 
-ExecuteMode queryExecuteMode(bool is_streaming, const Settings & settings)
+ExecuteMode queryExecuteMode(bool is_streaming, bool is_subquery, const Settings & settings)
 {
     switch (settings.exec_mode)
     {
@@ -20,7 +20,13 @@ ExecuteMode queryExecuteMode(bool is_streaming, const Settings & settings)
         case ExecuteMode::UNSUBSCRIBE:
         case ExecuteMode::RECOVER:
             if (!is_streaming)
+            {
+                /// Assume it's a historical subquery in streaming query, we only execute it in normal mode.
+                if (is_subquery)
+                    return ExecuteMode::NORMAL;
+
                 throw Exception(ErrorCodes::NOT_IMPLEMENTED, "SUBSCRIBE can only work with streaming query");
+            }
             break;
     }
 

src/Processors/QueryPlan/QueryExecuteMode.h

@@ -5,5 +5,5 @@
 namespace DB
 {
 struct Settings;
-ExecuteMode queryExecuteMode(bool is_streaming, const Settings & settings);
+ExecuteMode queryExecuteMode(bool is_streaming, bool is_subquery, const Settings & settings);
 }

src/Processors/QueryPlan/QueryPlan.cpp

@@ -198,8 +198,6 @@ QueryPipelineBuilderPtr QueryPlan::buildQueryPipeline(
     last_pipeline->addResources(std::move(resources));
 
     /// proton : starts
-    last_pipeline->setExecuteMode(queryExecuteMode(isStreaming(), query_context->getSettingsRef()));
-
     if (isStreaming() != last_pipeline->isStreaming())
         throw Exception(
             ErrorCodes::LOGICAL_ERROR,

tests/stream/test_stream_smoke/0099_fixed_issues.json

@@ -668,6 +668,33 @@
             ]
         }
       ]
+    },
+    {
+      "id": 23,
+      "tags": ["in set", "checkpoint"],
+      "name": "#498",
+      "description": "mv over versioned_kv with where in set",
+      "steps":[
+        {
+          "statements": [
+            {"client":"python", "query_type": "table", "query": "drop view if exists test_mv_498"},
+            {"client":"python", "query_type": "table", "query": "drop stream if exists test_dim_498"},
+            {"client":"python", "query_type": "table", "query": "drop stream if exists test_vk_498"},
+            {"client":"python", "query_type": "table", "exist": "test_dim_498", "exist_wait":2, "wait":1, "query": "create stream test_dim_498(id int, type string)"},
+            {"client":"python", "query_type": "table", "exist": "test_vk_498", "exist_wait":2, "wait":1, "query": "create stream test_vk_498(id int, value float) primary key id settings mode='versioned_kv'"},
+            {"client":"python", "query_type": "table", "depends_on_stream": "test_dim_498", "wait": 1, "query": "insert into test_dim_498(id, type) values(1, 'old')(2, 'old')(3, 'new');"},
+            {"client":"python", "query_type": "table", "wait":2, "query":"create materialized view test_mv_498 as with list as (select id from table(test_dim_498)) select id, value from test_vk_498 where id in list;"},
+            {"client":"python", "query_type": "stream", "depends_on_stream":"test_mv_498", "wait": 1, "query_id":"9923", "query":"select id, value from test_mv_498;"},
+            {"client":"python", "query_type": "table", "depends_on":"9923", "wait": 1, "kill":"9923", "kill_wait":2, "drop_view":"test_mv_498", "drop_view_wait":2, "query": "insert into test_vk_498(id, value) values(3, 20)(3, 30)(1, 10)(4, 40)(5, 50);"}
+          ]
+        }
+      ],
+      "expected_results": [
+        {
+            "query_id":"9923",
+            "expected_results":[[1, 10], [3, 20], [3, 30]]
+        }
+      ]
     }
   ]
 }