datafusion_python/sql/
logical.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::sync::Arc;
19
20use datafusion::logical_expr::{DdlStatement, LogicalPlan, Statement};
21use datafusion_proto::logical_plan::AsLogicalPlan;
22use prost::Message;
23use pyo3::exceptions::PyRuntimeError;
24use pyo3::prelude::*;
25use pyo3::types::PyBytes;
26
27use crate::codec::PythonLogicalCodec;
28use crate::context::PySessionContext;
29use crate::errors::PyDataFusionResult;
30use crate::expr::aggregate::PyAggregate;
31use crate::expr::analyze::PyAnalyze;
32use crate::expr::copy_to::PyCopyTo;
33use crate::expr::create_catalog::PyCreateCatalog;
34use crate::expr::create_catalog_schema::PyCreateCatalogSchema;
35use crate::expr::create_external_table::PyCreateExternalTable;
36use crate::expr::create_function::PyCreateFunction;
37use crate::expr::create_index::PyCreateIndex;
38use crate::expr::create_memory_table::PyCreateMemoryTable;
39use crate::expr::create_view::PyCreateView;
40use crate::expr::describe_table::PyDescribeTable;
41use crate::expr::distinct::PyDistinct;
42use crate::expr::dml::PyDmlStatement;
43use crate::expr::drop_catalog_schema::PyDropCatalogSchema;
44use crate::expr::drop_function::PyDropFunction;
45use crate::expr::drop_table::PyDropTable;
46use crate::expr::drop_view::PyDropView;
47use crate::expr::empty_relation::PyEmptyRelation;
48use crate::expr::explain::PyExplain;
49use crate::expr::extension::PyExtension;
50use crate::expr::filter::PyFilter;
51use crate::expr::join::PyJoin;
52use crate::expr::limit::PyLimit;
53use crate::expr::logical_node::LogicalNode;
54use crate::expr::projection::PyProjection;
55use crate::expr::recursive_query::PyRecursiveQuery;
56use crate::expr::repartition::PyRepartition;
57use crate::expr::sort::PySort;
58use crate::expr::statement::{
59    PyDeallocate, PyExecute, PyPrepare, PyResetVariable, PySetVariable, PyTransactionEnd,
60    PyTransactionStart,
61};
62use crate::expr::subquery::PySubquery;
63use crate::expr::subquery_alias::PySubqueryAlias;
64use crate::expr::table_scan::PyTableScan;
65use crate::expr::union::PyUnion;
66use crate::expr::unnest::PyUnnest;
67use crate::expr::values::PyValues;
68use crate::expr::window::PyWindowExpr;
69
70#[pyclass(
71    from_py_object,
72    frozen,
73    name = "LogicalPlan",
74    module = "datafusion",
75    subclass,
76    eq
77)]
78#[derive(Debug, Clone, PartialEq, Eq)]
79pub struct PyLogicalPlan {
80    pub(crate) plan: Arc<LogicalPlan>,
81}
82
83impl PyLogicalPlan {
84    /// creates a new PyLogicalPlan
85    pub fn new(plan: LogicalPlan) -> Self {
86        Self {
87            plan: Arc::new(plan),
88        }
89    }
90
91    pub fn plan(&self) -> Arc<LogicalPlan> {
92        self.plan.clone()
93    }
94}
95
96#[pymethods]
97impl PyLogicalPlan {
98    /// Return the specific logical operator
99    pub fn to_variant<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
100        match self.plan.as_ref() {
101            LogicalPlan::Aggregate(plan) => PyAggregate::from(plan.clone()).to_variant(py),
102            LogicalPlan::Analyze(plan) => PyAnalyze::from(plan.clone()).to_variant(py),
103            LogicalPlan::Distinct(plan) => PyDistinct::from(plan.clone()).to_variant(py),
104            LogicalPlan::EmptyRelation(plan) => PyEmptyRelation::from(plan.clone()).to_variant(py),
105            LogicalPlan::Explain(plan) => PyExplain::from(plan.clone()).to_variant(py),
106            LogicalPlan::Extension(plan) => PyExtension::from(plan.clone()).to_variant(py),
107            LogicalPlan::Filter(plan) => PyFilter::from(plan.clone()).to_variant(py),
108            LogicalPlan::Join(plan) => PyJoin::from(plan.clone()).to_variant(py),
109            LogicalPlan::Limit(plan) => PyLimit::from(plan.clone()).to_variant(py),
110            LogicalPlan::Projection(plan) => PyProjection::from(plan.clone()).to_variant(py),
111            LogicalPlan::Sort(plan) => PySort::from(plan.clone()).to_variant(py),
112            LogicalPlan::TableScan(plan) => PyTableScan::from(plan.clone()).to_variant(py),
113            LogicalPlan::Subquery(plan) => PySubquery::from(plan.clone()).to_variant(py),
114            LogicalPlan::SubqueryAlias(plan) => PySubqueryAlias::from(plan.clone()).to_variant(py),
115            LogicalPlan::Unnest(plan) => PyUnnest::from(plan.clone()).to_variant(py),
116            LogicalPlan::Window(plan) => PyWindowExpr::from(plan.clone()).to_variant(py),
117            LogicalPlan::Repartition(plan) => PyRepartition::from(plan.clone()).to_variant(py),
118            LogicalPlan::Union(plan) => PyUnion::from(plan.clone()).to_variant(py),
119            LogicalPlan::Statement(plan) => match plan {
120                Statement::TransactionStart(plan) => {
121                    PyTransactionStart::from(plan.clone()).to_variant(py)
122                }
123                Statement::TransactionEnd(plan) => {
124                    PyTransactionEnd::from(plan.clone()).to_variant(py)
125                }
126                Statement::SetVariable(plan) => PySetVariable::from(plan.clone()).to_variant(py),
127                Statement::ResetVariable(plan) => {
128                    PyResetVariable::from(plan.clone()).to_variant(py)
129                }
130                Statement::Prepare(plan) => PyPrepare::from(plan.clone()).to_variant(py),
131                Statement::Execute(plan) => PyExecute::from(plan.clone()).to_variant(py),
132                Statement::Deallocate(plan) => PyDeallocate::from(plan.clone()).to_variant(py),
133            },
134            LogicalPlan::Values(plan) => PyValues::from(plan.clone()).to_variant(py),
135            LogicalPlan::Dml(plan) => PyDmlStatement::from(plan.clone()).to_variant(py),
136            LogicalPlan::Ddl(plan) => match plan {
137                DdlStatement::CreateExternalTable(plan) => {
138                    PyCreateExternalTable::from(plan.clone()).to_variant(py)
139                }
140                DdlStatement::CreateMemoryTable(plan) => {
141                    PyCreateMemoryTable::from(plan.clone()).to_variant(py)
142                }
143                DdlStatement::CreateView(plan) => PyCreateView::from(plan.clone()).to_variant(py),
144                DdlStatement::CreateCatalogSchema(plan) => {
145                    PyCreateCatalogSchema::from(plan.clone()).to_variant(py)
146                }
147                DdlStatement::CreateCatalog(plan) => {
148                    PyCreateCatalog::from(plan.clone()).to_variant(py)
149                }
150                DdlStatement::CreateIndex(plan) => PyCreateIndex::from(plan.clone()).to_variant(py),
151                DdlStatement::DropTable(plan) => PyDropTable::from(plan.clone()).to_variant(py),
152                DdlStatement::DropView(plan) => PyDropView::from(plan.clone()).to_variant(py),
153                DdlStatement::DropCatalogSchema(plan) => {
154                    PyDropCatalogSchema::from(plan.clone()).to_variant(py)
155                }
156                DdlStatement::CreateFunction(plan) => {
157                    PyCreateFunction::from(plan.clone()).to_variant(py)
158                }
159                DdlStatement::DropFunction(plan) => {
160                    PyDropFunction::from(plan.clone()).to_variant(py)
161                }
162            },
163            LogicalPlan::Copy(plan) => PyCopyTo::from(plan.clone()).to_variant(py),
164            LogicalPlan::DescribeTable(plan) => PyDescribeTable::from(plan.clone()).to_variant(py),
165            LogicalPlan::RecursiveQuery(plan) => {
166                PyRecursiveQuery::from(plan.clone()).to_variant(py)
167            }
168        }
169    }
170
171    /// Get the inputs to this plan
172    fn inputs(&self) -> Vec<PyLogicalPlan> {
173        let mut inputs = vec![];
174        for input in self.plan.inputs() {
175            inputs.push(input.to_owned().into());
176        }
177        inputs
178    }
179
180    fn __repr__(&self) -> PyResult<String> {
181        Ok(format!("{:?}", self.plan))
182    }
183
184    fn display(&self) -> String {
185        format!("{}", self.plan.display())
186    }
187
188    fn display_indent(&self) -> String {
189        format!("{}", self.plan.display_indent())
190    }
191
192    fn display_indent_schema(&self) -> String {
193        format!("{}", self.plan.display_indent_schema())
194    }
195
196    fn display_graphviz(&self) -> String {
197        format!("{}", self.plan.display_graphviz())
198    }
199
200    #[pyo3(signature = (ctx=None))]
201    pub fn to_bytes<'py>(
202        &'py self,
203        py: Python<'py>,
204        ctx: Option<PySessionContext>,
205    ) -> PyDataFusionResult<Bound<'py, PyBytes>> {
206        let default_codec;
207        let codec: &dyn datafusion_proto::logical_plan::LogicalExtensionCodec = match ctx {
208            Some(ref ctx) => ctx.logical_codec().as_ref(),
209            None => {
210                default_codec = PythonLogicalCodec::default();
211                &default_codec
212            }
213        };
214        let proto =
215            datafusion_proto::protobuf::LogicalPlanNode::try_from_logical_plan(&self.plan, codec)?;
216
217        let bytes = proto.encode_to_vec();
218        Ok(PyBytes::new(py, &bytes))
219    }
220
221    #[staticmethod]
222    pub fn from_bytes(
223        ctx: PySessionContext,
224        proto_msg: Bound<'_, PyBytes>,
225    ) -> PyDataFusionResult<Self> {
226        let bytes: &[u8] = proto_msg.extract().map_err(Into::<PyErr>::into)?;
227        let proto_plan =
228            datafusion_proto::protobuf::LogicalPlanNode::decode(bytes).map_err(|e| {
229                PyRuntimeError::new_err(format!(
230                    "Unable to decode logical node from serialized bytes: {e}"
231                ))
232            })?;
233
234        let codec = ctx.logical_codec();
235        let plan = proto_plan.try_into_logical_plan(&ctx.ctx.task_ctx(), codec.as_ref())?;
236        Ok(Self::new(plan))
237    }
238}
239
240impl From<PyLogicalPlan> for LogicalPlan {
241    fn from(logical_plan: PyLogicalPlan) -> LogicalPlan {
242        logical_plan.plan.as_ref().clone()
243    }
244}
245
246impl From<LogicalPlan> for PyLogicalPlan {
247    fn from(logical_plan: LogicalPlan) -> PyLogicalPlan {
248        PyLogicalPlan {
249            plan: Arc::new(logical_plan),
250        }
251    }
252}
datafusion_python/sql/logical.rs

datafusion_python/sql/
logical.rs
