datafusion_physical_plan/
empty.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
18//! EmptyRelation with produce_one_row=false execution plan
19
20use std::sync::Arc;
21
22use crate::memory::MemoryStream;
23use crate::{
24    ChildrenPropertiesMode, DisplayAs, PlanProperties, ReplaceChildrenOptions,
25    SendableRecordBatchStream, Statistics,
26};
27use crate::{
28    DisplayFormatType, ExecutionPlan, Partitioning,
29    execution_plan::{Boundedness, EmissionType},
30};
31
32use arrow::datatypes::SchemaRef;
33use arrow::record_batch::RecordBatch;
34use datafusion_common::stats::Precision;
35use datafusion_common::tree_node::TreeNodeRecursion;
36use datafusion_common::{ColumnStatistics, Result, ScalarValue, assert_or_internal_err};
37use datafusion_execution::TaskContext;
38use datafusion_physical_expr::{EquivalenceProperties, PhysicalExpr};
39
40use crate::execution_plan::SchedulingType;
41use crate::statistics::StatisticsArgs;
42use log::trace;
43
44/// Execution plan for empty relation with produce_one_row=false
45#[derive(Debug, Clone)]
46pub struct EmptyExec {
47    /// The schema for the produced row
48    schema: SchemaRef,
49    /// Number of partitions
50    partitions: usize,
51    cache: Arc<PlanProperties>,
52}
53
54impl EmptyExec {
55    /// Create a new EmptyExec
56    pub fn new(schema: SchemaRef) -> Self {
57        let cache = Self::compute_properties(Arc::clone(&schema), 1);
58        EmptyExec {
59            schema,
60            partitions: 1,
61            cache: Arc::new(cache),
62        }
63    }
64
65    /// Create a new EmptyExec with specified partition number
66    pub fn with_partitions(mut self, partitions: usize) -> Self {
67        self.partitions = partitions;
68        // Changing partitions may invalidate output partitioning, so update it:
69        let output_partitioning = Self::output_partitioning_helper(self.partitions);
70        Arc::make_mut(&mut self.cache).partitioning = output_partitioning;
71        self
72    }
73
74    fn data(&self) -> Result<Vec<RecordBatch>> {
75        Ok(vec![])
76    }
77
78    fn output_partitioning_helper(n_partitions: usize) -> Partitioning {
79        Partitioning::UnknownPartitioning(n_partitions)
80    }
81
82    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
83    fn compute_properties(schema: SchemaRef, n_partitions: usize) -> PlanProperties {
84        PlanProperties::new(
85            EquivalenceProperties::new(schema),
86            Self::output_partitioning_helper(n_partitions),
87            EmissionType::Incremental,
88            Boundedness::Bounded,
89        )
90        .with_scheduling_type(SchedulingType::Cooperative)
91    }
92}
93
94impl DisplayAs for EmptyExec {
95    fn fmt_as(
96        &self,
97        t: DisplayFormatType,
98        f: &mut std::fmt::Formatter,
99    ) -> std::fmt::Result {
100        match t {
101            DisplayFormatType::Default | DisplayFormatType::Verbose => {
102                write!(f, "EmptyExec")
103            }
104            DisplayFormatType::TreeRender => {
105                // TODO: collect info
106                write!(f, "")
107            }
108        }
109    }
110}
111
112impl ExecutionPlan for EmptyExec {
113    fn name(&self) -> &'static str {
114        "EmptyExec"
115    }
116
117    /// Return a reference to Any that can be used for downcasting
118    fn properties(&self) -> &Arc<PlanProperties> {
119        &self.cache
120    }
121
122    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
123        vec![]
124    }
125
126    fn apply_expressions(
127        &self,
128        _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
129    ) -> Result<TreeNodeRecursion> {
130        Ok(TreeNodeRecursion::Continue)
131    }
132
133    fn replace_children(
134        self: Arc<Self>,
135        _: Vec<Arc<dyn ExecutionPlan>>,
136        _: ReplaceChildrenOptions,
137    ) -> Result<Arc<dyn ExecutionPlan>> {
138        Ok(self)
139    }
140
141    fn with_new_children(
142        self: Arc<Self>,
143        children: Vec<Arc<dyn ExecutionPlan>>,
144    ) -> Result<Arc<dyn ExecutionPlan>> {
145        self.replace_children(
146            children,
147            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
148        )
149    }
150
151    fn execute(
152        &self,
153        partition: usize,
154        context: Arc<TaskContext>,
155    ) -> Result<SendableRecordBatchStream> {
156        trace!(
157            "Start EmptyExec::execute for partition {} of context session_id {} and task_id {:?}",
158            partition,
159            context.session_id(),
160            context.task_id()
161        );
162
163        assert_or_internal_err!(
164            partition < self.partitions,
165            "EmptyExec invalid partition {} (expected less than {})",
166            partition,
167            self.partitions
168        );
169
170        Ok(Box::pin(MemoryStream::try_new(
171            self.data()?,
172            Arc::clone(&self.schema),
173            None,
174        )?))
175    }
176
177    fn statistics_from_inputs(
178        &self,
179        _input_stats: &[Arc<Statistics>],
180        args: &StatisticsArgs,
181    ) -> Result<Arc<Statistics>> {
182        if let Some(partition) = args.partition() {
183            assert_or_internal_err!(
184                partition < self.partitions,
185                "EmptyExec invalid partition {} (expected less than {})",
186                partition,
187                self.partitions
188            );
189        }
190
191        // Build explicit stats: exact zero rows and bytes, with explicit known column stats
192        let mut stats = Statistics::default()
193            .with_num_rows(Precision::Exact(0))
194            .with_total_byte_size(Precision::Exact(0));
195
196        // Add explicit column stats for each field in schema
197        for _ in self.schema.fields() {
198            stats = stats.add_column_statistics(ColumnStatistics {
199                null_count: Precision::Exact(0),
200                distinct_count: Precision::Exact(0),
201                min_value: Precision::<ScalarValue>::Absent,
202                max_value: Precision::<ScalarValue>::Absent,
203                sum_value: Precision::<ScalarValue>::Absent,
204                byte_size: Precision::Exact(0),
205            });
206        }
207
208        Ok(Arc::new(stats))
209    }
210
211    #[cfg(feature = "proto")]
212    fn try_to_proto(
213        &self,
214        _ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
215    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
216        use datafusion_proto_models::protobuf;
217        let schema = self.schema().as_ref().try_into()?;
218        Ok(Some(protobuf::PhysicalPlanNode {
219            physical_plan_type: Some(
220                protobuf::physical_plan_node::PhysicalPlanType::Empty(
221                    protobuf::EmptyExecNode {
222                        schema: Some(schema),
223                        partitions: self
224                            .properties()
225                            .output_partitioning()
226                            .partition_count() as u32,
227                    },
228                ),
229            ),
230        }))
231    }
232}
233
234#[cfg(feature = "proto")]
235impl EmptyExec {
236    /// Reconstruct an [`EmptyExec`] from its protobuf representation.
237    pub fn try_from_proto(
238        node: &datafusion_proto_models::protobuf::PhysicalPlanNode,
239        _ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>,
240    ) -> Result<Arc<dyn ExecutionPlan>> {
241        use datafusion_proto_models::protobuf;
242        let empty = crate::expect_plan_variant!(
243            node,
244            protobuf::physical_plan_node::PhysicalPlanType::Empty,
245            "EmptyExec",
246        );
247        let schema = empty.schema.as_ref().ok_or_else(|| {
248            datafusion_common::internal_datafusion_err!(
249                "EmptyExec is missing required field 'schema'"
250            )
251        })?;
252        let schema = Arc::new(arrow::datatypes::Schema::try_from(schema)?);
253        // A zero (absent) partition count comes from a plan encoded before the
254        // field existed, which always meant a single partition.
255        let partitions = empty.partitions.max(1) as usize;
256        Ok(Arc::new(EmptyExec::new(schema).with_partitions(partitions)))
257    }
258}
259
260#[cfg(test)]
261mod tests {
262    use super::*;
263    use crate::common;
264    use crate::execution_plan::replace_children_if_necessary;
265    use crate::test;
266
267    #[tokio::test]
268    async fn empty() -> Result<()> {
269        let task_ctx = Arc::new(TaskContext::default());
270        let schema = test::aggr_test_schema();
271
272        let empty = EmptyExec::new(Arc::clone(&schema));
273        assert_eq!(empty.schema(), schema);
274
275        // We should have no results
276        let iter = empty.execute(0, task_ctx)?;
277        let batches = common::collect(iter).await?;
278        assert!(batches.is_empty());
279
280        Ok(())
281    }
282
283    #[test]
284    fn with_new_children() -> Result<()> {
285        let schema = test::aggr_test_schema();
286        let empty = Arc::new(EmptyExec::new(Arc::clone(&schema)));
287
288        let empty2 = replace_children_if_necessary(
289            Arc::clone(&empty) as Arc<dyn ExecutionPlan>,
290            vec![],
291        )?;
292        assert_eq!(empty.schema(), empty2.schema());
293
294        let too_many_kids = vec![empty2];
295        assert!(
296            replace_children_if_necessary(empty, too_many_kids).is_err(),
297            "expected error when providing list of kids"
298        );
299        Ok(())
300    }
301
302    #[tokio::test]
303    async fn invalid_execute() -> Result<()> {
304        let task_ctx = Arc::new(TaskContext::default());
305        let schema = test::aggr_test_schema();
306        let empty = EmptyExec::new(schema);
307
308        // ask for the wrong partition
309        assert!(empty.execute(1, Arc::clone(&task_ctx)).is_err());
310        assert!(empty.execute(20, task_ctx).is_err());
311        Ok(())
312    }
313}
datafusion_physical_plan/empty.rs

datafusion_physical_plan/
empty.rs
