Method new

datafusion/physical-plan/src/recursive_query.rs:281–307

Create a new recursive query stream

(
        task_context: Arc<TaskContext>,
        work_table: Arc<WorkTable>,
        recursive_term: Arc<dyn ExecutionPlan>,
        static_stream: SendableRecordBatchStream,
        is_distinct: boo

impl RecursiveQueryStream {
    /// Create a new recursive query stream
    fn new(
        task_context: Arc<TaskContext>,
        work_table: Arc<WorkTable>,
        recursive_term: Arc<dyn ExecutionPlan>,
        static_stream: SendableRecordBatchStream,
        is_distinct: bool,
        baseline_metrics: BaselineMetrics,
    ) -> Result<Self> {
        let schema = static_stream.schema();
        let reservation =
            MemoryConsumer::new("RecursiveQuery").register(task_context.memory_pool());
        let distinct_deduplicator = is_distinct
            .then(|| DistinctDeduplicator::new(Arc::clone(&schema), &task_context))
            .transpose()?;
        Ok(Self {
            task_context,
            work_table,
            recursive_term,
            static_stream: Some(static_stream),
            recursive_stream: None,
            schema,
            buffer: vec![],
            reservation,
            distinct_deduplicator,
            baseline_metrics,
        })
    }

    /// Push a clone of the given batch to the in memory buffer, and then return
    /// a poll with it.