Try to create a new [`StreamingTableExec`] returning an error if the schema is incorrect
(
schema: SchemaRef,
partitions: Vec<Arc<dyn PartitionStream>>,
projection: Option<&Vec<usize>>,
projected_output_ordering: impl IntoIterator<Item = LexOrdering>,
| 73 | impl StreamingTableExec { |
| 74 | /// Try to create a new [`StreamingTableExec`] returning an error if the schema is incorrect |
| 75 | pub fn try_new( |
| 76 | schema: SchemaRef, |
| 77 | partitions: Vec<Arc<dyn PartitionStream>>, |
| 78 | projection: Option<&Vec<usize>>, |
| 79 | projected_output_ordering: impl IntoIterator<Item = LexOrdering>, |
| 80 | infinite: bool, |
| 81 | limit: Option<usize>, |
| 82 | ) -> Result<Self> { |
| 83 | for x in partitions.iter() { |
| 84 | let partition_schema = x.schema(); |
| 85 | if !schema.eq(partition_schema) { |
| 86 | debug!( |
| 87 | "Target schema does not match with partition schema. \ |
| 88 | Target_schema: {schema:?}. Partition Schema: {partition_schema:?}" |
| 89 | ); |
| 90 | return plan_err!("Mismatch between schema and batches"); |
| 91 | } |
| 92 | } |
| 93 | |
| 94 | let projected_schema = match projection { |
| 95 | Some(p) => Arc::new(schema.project(p)?), |
| 96 | None => schema, |
| 97 | }; |
| 98 | let projected_output_ordering = |
| 99 | projected_output_ordering.into_iter().collect::<Vec<_>>(); |
| 100 | let cache = Self::compute_properties( |
| 101 | Arc::clone(&projected_schema), |
| 102 | projected_output_ordering.clone(), |
| 103 | &partitions, |
| 104 | infinite, |
| 105 | ); |
| 106 | Ok(Self { |
| 107 | partitions, |
| 108 | projected_schema, |
| 109 | projection: projection.cloned().map(Into::into), |
| 110 | projected_output_ordering, |
| 111 | infinite, |
| 112 | limit, |
| 113 | cache: Arc::new(cache), |
| 114 | metrics: ExecutionPlanMetricsSet::new(), |
| 115 | }) |
| 116 | } |
| 117 | |
| 118 | pub fn partitions(&self) -> &Vec<Arc<dyn PartitionStream>> { |
| 119 | &self.partitions |