| 158 | } |
| 159 | |
| 160 | async fn scan( |
| 161 | &self, |
| 162 | state: &dyn Session, |
| 163 | projection: Option<&Vec<usize>>, |
| 164 | filters: &[Expr], |
| 165 | limit: Option<usize>, |
| 166 | ) -> DFResult<Arc<dyn ExecutionPlan>> { |
| 167 | // Plan splits eagerly so we know partition count upfront. |
| 168 | let filter_analysis = analyze_filters(filters, self.table.schema().fields()); |
| 169 | let mut read_builder = self.table.new_read_builder(); |
| 170 | if let Some(filter) = filter_analysis.pushed_predicate.clone() { |
| 171 | read_builder.with_filter(filter); |
| 172 | } |
| 173 | let pushed_limit = limit.filter(|_| !filter_analysis.has_untranslated_residual); |
| 174 | if let Some(limit) = pushed_limit { |
| 175 | read_builder.with_limit(limit); |
| 176 | } |
| 177 | let scan = read_builder.new_scan(); |
| 178 | // DataFusion's Python FFI may poll `TableProvider::scan()` without an active |
| 179 | // Tokio runtime. `scan.plan()` can reach OpenDAL/Tokio filesystem calls while |
| 180 | // reading Paimon metadata, so we must provide a runtime here instead of |
| 181 | // assuming the caller already entered one. |
| 182 | let plan = await_with_runtime(scan.plan()) |
| 183 | .await |
| 184 | .map_err(to_datafusion_error)?; |
| 185 | |
| 186 | let target = state.config_options().execution.target_partitions; |
| 187 | let filter_exact = !filter_analysis.has_untranslated_residual |
| 188 | && filter_analysis |
| 189 | .pushed_predicate |
| 190 | .as_ref() |
| 191 | .is_none_or(|p| read_builder.is_exact_filter_pushdown(p)); |
| 192 | PaimonScanBuilder { |
| 193 | table: &self.table, |
| 194 | schema: &self.schema, |
| 195 | plan: &plan, |
| 196 | projection, |
| 197 | pushed_predicate: filter_analysis.pushed_predicate, |
| 198 | limit: pushed_limit, |
| 199 | target_partitions: target, |
| 200 | filter_exact, |
| 201 | } |
| 202 | .build() |
| 203 | } |
| 204 | |
| 205 | async fn insert_into( |
| 206 | &self, |