risingwave_frontend/optimizer/plan_node/
logical_mssql_query.rs

1// Copyright 2026 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use pretty_xmlish::XmlNode;
16use risingwave_common::bail;
17use risingwave_common::catalog::Schema;
18
19use super::generic::GenericPlanRef;
20use super::utils::{Distill, childless_record};
21use super::{
22    BatchMssqlQuery, ColPrunable, ExprRewritable, Logical, LogicalPlanRef as PlanRef,
23    LogicalProject, PlanBase, PredicatePushdown, ToBatch, ToStream, generic,
24};
25use crate::OptimizerContextRef;
26use crate::error::Result;
27use crate::optimizer::plan_node::expr_visitable::ExprVisitable;
28use crate::optimizer::plan_node::utils::column_names_pretty;
29use crate::optimizer::plan_node::{
30    ColumnPruningContext, LogicalFilter, PredicatePushdownContext, RewriteStreamContext,
31    ToStreamContext,
32};
33use crate::utils::{ColIndexMapping, Condition};
34
35/// Logical plan node for the `mssql_query` table function. The user query
36/// is run exactly once by the batch executor; streaming mode is rejected.
37#[derive(Debug, Clone, PartialEq, Eq, Hash)]
38pub struct LogicalMssqlQuery {
39    pub base: PlanBase<Logical>,
40    pub core: generic::MssqlQuery,
41}
42
43impl LogicalMssqlQuery {
44    /// Build a `LogicalMssqlQuery` from the pre-discovered schema, the
45    /// SQL Server connection parameters, the user query, and the optimizer
46    /// context.
47    ///
48    /// `connection` groups the seven inline connection parameters; the
49    /// 2-arg source-reference form instead reuses connection parameters
50    /// from the named source (and is rewritten by the binder, not here).
51    pub fn new(
52        ctx: OptimizerContextRef,
53        schema: Schema,
54        connection: generic::MssqlConnection,
55        query: String,
56    ) -> Self {
57        let core = generic::MssqlQuery {
58            schema,
59            connection,
60            query,
61            ctx,
62        };
63
64        let base = PlanBase::new_logical_with_core(&core);
65
66        LogicalMssqlQuery { base, core }
67    }
68}
69
70impl_plan_tree_node_for_leaf! { Logical, LogicalMssqlQuery}
71impl Distill for LogicalMssqlQuery {
72    /// Pretty-print the node as `LogicalMssqlQuery { columns: [...] }` for
73    /// `EXPLAIN` output.
74    fn distill<'a>(&self) -> XmlNode<'a> {
75        let fields = vec![("columns", column_names_pretty(self.schema()))];
76        childless_record("LogicalMssqlQuery", fields)
77    }
78}
79
80impl ColPrunable for LogicalMssqlQuery {
81    /// Column pruning wraps the node in a `LogicalProject` that keeps only
82    /// the requested columns. The executor always returns the full schema
83    /// discovered at bind time; pruning happens in the project above.
84    fn prune_col(&self, required_cols: &[usize], _ctx: &mut ColumnPruningContext) -> PlanRef {
85        LogicalProject::with_out_col_idx(self.clone().into(), required_cols.iter().cloned()).into()
86    }
87}
88
89impl ExprRewritable<Logical> for LogicalMssqlQuery {}
90
91impl ExprVisitable for LogicalMssqlQuery {}
92
93impl PredicatePushdown for LogicalMssqlQuery {
94    /// `mssql_query` does not support predicate pushdown — the user
95    /// query is opaque to the optimizer. Wrap with `LogicalFilter` so
96    /// the predicate is applied to the result of the query.
97    fn predicate_pushdown(
98        &self,
99        predicate: Condition,
100        _ctx: &mut PredicatePushdownContext,
101    ) -> PlanRef {
102        // No pushdown.
103        LogicalFilter::create(self.clone().into(), predicate)
104    }
105}
106
107impl ToBatch for LogicalMssqlQuery {
108    /// Lower to a [`BatchMssqlQuery`] plan node that the batch executor
109    /// (`MssqlQueryExecutor`) will consume.
110    fn to_batch(&self) -> Result<crate::optimizer::plan_node::BatchPlanRef> {
111        Ok(BatchMssqlQuery::new(self.core.clone()).into())
112    }
113}
114
115impl ToStream for LogicalMssqlQuery {
116    /// Reject streaming conversion with a clear error — `mssql_query` is
117    /// batch-only. Streaming CDC ingestion should use the
118    /// `sqlserver-cdc` source instead.
119    fn to_stream(
120        &self,
121        _ctx: &mut ToStreamContext,
122    ) -> Result<crate::optimizer::plan_node::StreamPlanRef> {
123        bail!("mssql_query function is not supported in streaming mode")
124    }
125
126    /// `mssql_query` is batch-only; refuse the streaming rewrite as well.
127    fn logical_rewrite_for_stream(
128        &self,
129        _ctx: &mut RewriteStreamContext,
130    ) -> Result<(PlanRef, ColIndexMapping)> {
131        bail!("mssql_query function is not supported in streaming mode")
132    }
133}
risingwave_frontend/optimizer/plan_node/logical_mssql_query.rs

risingwave_frontend/optimizer/plan_node/
logical_mssql_query.rs
