List all files identified by this [`ListingTableUrl`] for the provided `file_extension`, optionally filtering by a path prefix
(
&'a self,
ctx: &'a dyn Session,
store: &'a dyn ObjectStore,
prefix: Option<Path>,
file_extension: &'a str,
)
| 245 | /// List all files identified by this [`ListingTableUrl`] for the provided `file_extension`, |
| 246 | /// optionally filtering by a path prefix |
| 247 | pub async fn list_prefixed_files<'a>( |
| 248 | &'a self, |
| 249 | ctx: &'a dyn Session, |
| 250 | store: &'a dyn ObjectStore, |
| 251 | prefix: Option<Path>, |
| 252 | file_extension: &'a str, |
| 253 | ) -> Result<BoxStream<'a, Result<ObjectMeta>>> { |
| 254 | let exec_options = &ctx.config_options().execution; |
| 255 | let ignore_subdirectory = exec_options.listing_table_ignore_subdirectory; |
| 256 | |
| 257 | // Build full_prefix for non-cached path and head() calls |
| 258 | let full_prefix = if let Some(ref p) = prefix { |
| 259 | let mut parts = self.prefix.parts().collect::<Vec<_>>(); |
| 260 | parts.extend(p.parts()); |
| 261 | Path::from_iter(parts) |
| 262 | } else { |
| 263 | self.prefix.clone() |
| 264 | }; |
| 265 | |
| 266 | let list: BoxStream<'a, Result<ObjectMeta>> = if self.is_collection() { |
| 267 | list_with_cache( |
| 268 | ctx, |
| 269 | store, |
| 270 | self.table_ref.as_ref(), |
| 271 | &self.prefix, |
| 272 | prefix.as_ref(), |
| 273 | ) |
| 274 | .await? |
| 275 | } else { |
| 276 | match store.head(&full_prefix).await { |
| 277 | Ok(meta) => futures::stream::once(async { Ok(meta) }) |
| 278 | .map_err(|e| DataFusionError::ObjectStore(Box::new(e))) |
| 279 | .boxed(), |
| 280 | // If the head command fails, it is likely that object doesn't exist. |
| 281 | // Retry as though it were a prefix (aka a collection) |
| 282 | Err(object_store::Error::NotFound { .. }) => { |
| 283 | list_with_cache( |
| 284 | ctx, |
| 285 | store, |
| 286 | self.table_ref.as_ref(), |
| 287 | &self.prefix, |
| 288 | prefix.as_ref(), |
| 289 | ) |
| 290 | .await? |
| 291 | } |
| 292 | Err(e) => return Err(e.into()), |
| 293 | } |
| 294 | }; |
| 295 | |
| 296 | Ok(list |
| 297 | .try_filter(move |meta| { |
| 298 | let path = &meta.location; |
| 299 | let extension_match = path.as_ref().ends_with(file_extension); |
| 300 | let glob_match = self.contains(path, ignore_subdirectory); |
| 301 | futures::future::ready(extension_match && glob_match) |
| 302 | }) |
| 303 | .boxed()) |
| 304 | } |