1use std::borrow::Cow;
2use std::collections::VecDeque;
3use std::path::{Path, PathBuf};
4use std::sync::{Arc, LazyLock};
5
6use polars_buffer::Buffer;
7use polars_core::config;
8use polars_core::error::{PolarsResult, polars_bail, to_compute_err};
9use polars_utils::aliases::PlHashMap;
10use polars_utils::pl_path::{CloudScheme, PlRefPath};
11use polars_utils::pl_str::PlSmallStr;
12
13#[cfg(feature = "cloud")]
14mod hugging_face;
15
16use crate::cloud::CloudOptions;
17
18#[allow(clippy::bind_instead_of_map)]
19pub static POLARS_TEMP_DIR_BASE_PATH: LazyLock<Box<Path>> = LazyLock::new(|| {
20 (|| {
21 let verbose = config::verbose();
22
23 let path = if let Ok(v) = std::env::var("POLARS_TEMP_DIR").map(PathBuf::from) {
24 if verbose {
25 eprintln!("init_temp_dir: sourced from POLARS_TEMP_DIR")
26 }
27 v
28 } else if cfg!(target_family = "unix") {
29 let id = std::env::var("USER")
30 .inspect(|_| {
31 if verbose {
32 eprintln!("init_temp_dir: sourced $USER")
33 }
34 })
35 .or_else(|_e| {
36 #[cfg(feature = "file_cache")]
39 {
40 std::env::var("HOME")
41 .inspect(|_| {
42 if verbose {
43 eprintln!("init_temp_dir: sourced $HOME")
44 }
45 })
46 .map(|x| blake3::hash(x.as_bytes()).to_hex()[..32].to_string())
47 }
48 #[cfg(not(feature = "file_cache"))]
49 {
50 Err(_e)
51 }
52 });
53
54 if let Ok(v) = id {
55 std::env::temp_dir().join(format!("polars-{v}/"))
56 } else {
57 return Err(std::io::Error::other(
58 "could not load $USER or $HOME environment variables",
59 ));
60 }
61 } else if cfg!(target_family = "windows") {
62 std::env::temp_dir().join("polars/")
66 } else {
67 std::env::temp_dir().join("polars/")
68 }
69 .into_boxed_path();
70
71 let perm_result = create_dir_owner_only(path.as_ref());
72
73 if std::env::var("POLARS_ALLOW_UNSECURED_TEMP_DIR").as_deref() != Ok("1") {
74 perm_result?;
75 }
76
77 std::io::Result::Ok(path)
78 })()
79 .map_err(|e| {
80 std::io::Error::new(
81 e.kind(),
82 format!(
83 "error initializing temporary directory: {e} \
84 consider explicitly setting POLARS_TEMP_DIR"
85 ),
86 )
87 })
88 .unwrap()
89});
90
91pub fn create_dir_owner_only(path: &Path) -> std::io::Result<()> {
93 std::fs::create_dir_all(path)?;
94
95 #[cfg(target_family = "unix")]
96 {
97 use std::os::unix::fs::PermissionsExt;
98
99 std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700))?;
100 let perms = std::fs::metadata(path)?.permissions();
101
102 if (perms.mode() % 0o1000) != 0o700 {
103 return Err(std::io::Error::other(format!(
104 "error setting directory permissions: permission mismatch: {perms:?} (path = {path:?})"
105 )));
106 }
107 }
108
109 Ok(())
110}
111
112pub fn resolve_homedir<'a, S: AsRef<Path> + ?Sized>(path: &'a S) -> Cow<'a, Path> {
114 return inner(path.as_ref());
115
116 fn inner(path: &Path) -> Cow<'_, Path> {
117 if path.starts_with("~") {
118 #[cfg(not(target_family = "wasm"))]
119 if let Some(homedir) = std::env::home_dir() {
120 return Cow::Owned(homedir.join(path.strip_prefix("~").unwrap()));
121 }
122 }
123
124 Cow::Borrowed(path)
125 }
126}
127
128fn has_glob(path: &[u8]) -> bool {
129 return get_glob_start_idx(path).is_some();
130
131 fn get_glob_start_idx(path: &[u8]) -> Option<usize> {
133 memchr::memchr3(b'*', b'?', b'[', path)
134 }
135}
136
137fn is_file_uri_with_escape(path: &PlRefPath) -> bool {
139 path.scheme().is_some_and(|s| s.is_file()) && path.strip_scheme().contains('%')
140}
141
142pub fn decode_file_uri_paths(paths: &[PlRefPath], glob: bool) -> Cow<'_, [PlRefPath]> {
157 if !paths.iter().any(is_file_uri_with_escape) {
159 return Cow::Borrowed(paths);
160 }
161
162 Cow::Owned(
163 paths
164 .iter()
165 .map(|path| {
166 if is_file_uri_with_escape(path)
167 && let Some(decoded) = decode_file_uri_path(path.strip_scheme(), glob)
168 {
169 PlRefPath::new(decoded)
170 } else {
171 path.clone()
174 }
175 })
176 .collect(),
177 )
178}
179
180fn decode_file_uri_path(path: &str, glob: bool) -> Option<String> {
184 let decoded = percent_encoding::percent_decode_str(path)
185 .decode_utf8()
186 .ok()?;
187 let path = strip_windows_drive_slash(&decoded);
188 Some(if glob {
189 glob::Pattern::escape(path)
190 } else {
191 path.to_owned()
192 })
193}
194
195pub fn local_path(path: &PlRefPath) -> Option<Cow<'_, Path>> {
197 let scheme = path.scheme();
198 if scheme.is_some_and(|s| !s.is_file()) {
199 return None;
200 }
201 let ("", local_path) = path.strip_scheme_split_authority()? else {
202 return None;
203 };
204 Some(if scheme.is_some() {
205 Cow::Owned(decode_file_uri_path(local_path, false)?.into())
206 } else {
207 resolve_homedir(local_path)
208 })
209}
210
211fn strip_windows_drive_slash(path: &str) -> &str {
215 #[cfg(target_family = "windows")]
216 {
217 let b = path.as_bytes();
218 if b.len() >= 3 && b[0] == b'/' && b[1].is_ascii_alphabetic() && b[2] == b':' {
219 return &path[1..];
220 }
221 }
222 path
223}
224
225pub fn expanded_from_single_directory(paths: &[PlRefPath], expanded_paths: &[PlRefPath]) -> bool {
227 paths.len() == 1 && !has_glob(paths[0].strip_scheme().as_bytes())
229 && {
231 (
232 !paths[0].has_scheme() && paths[0].as_std_path().is_dir()
234 )
235 || (
236 expanded_paths.is_empty() || (paths[0] != expanded_paths[0])
239 )
240 }
241}
242
243pub async fn expand_paths(
245 paths: &[PlRefPath],
246 glob: bool,
247 hidden_file_prefix: &[PlSmallStr],
248 #[allow(unused_variables)] cloud_options: &mut Option<CloudOptions>,
249) -> PolarsResult<Buffer<PlRefPath>> {
250 expand_paths_hive(paths, glob, hidden_file_prefix, cloud_options, false)
251 .await
252 .map(|x| x.0)
253}
254
255pub type BytesPerSource = Option<Arc<[u64]>>;
260
261struct HiveIdxTracker<'a> {
262 idx: usize,
263 paths: &'a [PlRefPath],
264 check_directory_level: bool,
265}
266
267impl HiveIdxTracker<'_> {
268 fn update(&mut self, i: usize, path_idx: usize) -> PolarsResult<()> {
269 let check_directory_level = self.check_directory_level;
270 let paths = self.paths;
271
272 if check_directory_level
273 && ![usize::MAX, i].contains(&self.idx)
274 && (path_idx > 0 && paths[path_idx].parent() != paths[path_idx - 1].parent())
276 {
277 polars_bail!(
278 InvalidOperation:
279 "attempted to read from different directory levels with hive partitioning enabled: \
280 first path: {}, second path: {}",
281 &paths[path_idx - 1],
282 &paths[path_idx],
283 )
284 } else {
285 self.idx = std::cmp::min(self.idx, i);
286 Ok(())
287 }
288 }
289}
290
291#[cfg(feature = "cloud")]
292async fn expand_path_cloud(
293 path: PlRefPath,
294 cloud_options: Option<&CloudOptions>,
295 glob: bool,
296 first_path_has_scheme: bool,
297) -> PolarsResult<(usize, Vec<(PlRefPath, Option<u64>)>)> {
298 let format_path = |scheme: &str, bucket: &str, location: &str| {
299 if first_path_has_scheme {
300 format!("{scheme}://{bucket}/{location}")
301 } else {
302 format!("/{location}")
303 }
304 };
305
306 use polars_utils::io::_limit_path_len_io_err;
307
308 use crate::cloud::object_path_from_str;
309 let path_str = path.as_str();
310
311 let (cloud_location, store) =
312 crate::cloud::build_object_store(path.clone(), cloud_options, glob).await?;
313 let prefix = object_path_from_str(&cloud_location.prefix)?;
314
315 let out = if !path_str.ends_with("/") && (!glob || cloud_location.expansion.is_none()) && {
316 path.has_scheme() || path.as_std_path().is_file()
321 } {
322 (
323 0,
325 vec![(
326 PlRefPath::new(format_path(
327 cloud_location.scheme,
328 &cloud_location.bucket,
329 prefix.as_ref(),
330 )),
331 None,
332 )],
333 )
334 } else {
335 use futures::TryStreamExt;
336
337 if !path.has_scheme() {
338 path.as_std_path()
343 .metadata()
344 .map_err(|err| _limit_path_len_io_err(path.as_std_path(), err))?;
345 }
346
347 let cloud_location = &cloud_location;
348 let prefix_ref = &prefix;
349
350 let mut paths = store
351 .exec_with_rebuild_retry_on_err(|s| async move {
352 let out = s
353 .list(Some(prefix_ref))
354 .try_filter_map(|x| async move {
355 let out = (x.size > 0).then(|| {
357 (
358 PlRefPath::new({
359 format_path(
360 cloud_location.scheme,
361 &cloud_location.bucket,
362 x.location.as_ref(),
363 )
364 }),
365 Some(x.size),
366 )
367 });
368 Ok(out)
369 })
370 .try_collect::<Vec<_>>()
371 .await?;
372
373 Ok(out)
374 })
375 .await?;
376
377 let mut prefix = prefix.to_string();
380 if path_str.ends_with('/') && !prefix.ends_with('/') {
381 prefix.push('/')
382 };
383
384 paths.sort_unstable();
385
386 (
387 format_path(
388 cloud_location.scheme,
389 &cloud_location.bucket,
390 prefix.as_ref(),
391 )
392 .len(),
393 paths,
394 )
395 };
396
397 PolarsResult::Ok(out)
398}
399
400pub async fn expand_paths_hive(
404 paths: &[PlRefPath],
405 glob: bool,
406 hidden_file_prefix: &[PlSmallStr],
407 #[allow(unused_variables)] cloud_options: &mut Option<CloudOptions>,
408 check_directory_level: bool,
409) -> PolarsResult<(Buffer<PlRefPath>, usize, BytesPerSource)> {
410 let Some(first_path) = paths.first() else {
411 return Ok((vec![].into(), 0, None));
412 };
413
414 let first_path_has_scheme = first_path.has_scheme();
415
416 let is_hidden_file = move |path: &PlRefPath| {
417 path.file_name()
418 .and_then(|x| x.to_str())
419 .is_some_and(|file_name| {
420 hidden_file_prefix
421 .iter()
422 .any(|x| file_name.starts_with(x.as_str()))
423 })
424 };
425
426 let mut out_paths = OutPaths {
427 paths: vec![],
428 exts: [None, None],
429 is_hidden_file: &is_hidden_file,
430 };
431
432 let mut sizes_by_path: PlHashMap<PlRefPath, u64> = PlHashMap::default();
435
436 let mut hive_idx_tracker = HiveIdxTracker {
437 idx: usize::MAX,
438 paths,
439 check_directory_level,
440 };
441
442 if first_path_has_scheme || {
443 cfg!(not(target_family = "windows")) && polars_config::config().force_async()
444 } {
445 #[cfg(feature = "cloud")]
446 {
447 if first_path.scheme() == Some(CloudScheme::Hf) {
448 let (expand_start_idx, paths) = hugging_face::expand_paths_hf(
449 paths,
450 check_directory_level,
451 cloud_options,
452 glob,
453 )
454 .await?;
455
456 return Ok((paths.into(), expand_start_idx, None));
458 }
459
460 for (path_idx, path) in paths.iter().enumerate() {
461 use std::borrow::Cow;
462
463 let mut path = Cow::Borrowed(path);
464
465 if matches!(path.scheme(), Some(CloudScheme::Http | CloudScheme::Https)) {
466 let mut rewrite_aws = false;
467
468 #[cfg(feature = "aws")]
469 if let Some(p) = (|| {
470 use crate::cloud::CloudConfig;
471
472 let after_scheme = path.strip_scheme();
475
476 let bucket_end = after_scheme.find(".s3.")?;
477 let offset = bucket_end + 4;
478 let region_end = offset + after_scheme[offset..].find(".amazonaws.com/")?;
480
481 if after_scheme[..region_end].contains('/') || after_scheme.contains('?') {
483 return None;
484 }
485
486 let bucket = &after_scheme[..bucket_end];
487 let region = &after_scheme[bucket_end + 4..region_end];
488 let key = &after_scheme[region_end + 15..];
489
490 if let CloudConfig::Aws(configs) = cloud_options
491 .get_or_insert_default()
492 .config
493 .get_or_insert_with(|| CloudConfig::Aws(Vec::with_capacity(1)))
494 {
495 use object_store::aws::AmazonS3ConfigKey;
496
497 if !matches!(configs.last(), Some((AmazonS3ConfigKey::Region, _))) {
498 configs.push((AmazonS3ConfigKey::Region, region.into()))
499 }
500 }
501
502 Some(format!("s3://{bucket}/{key}"))
503 })() {
504 path = Cow::Owned(PlRefPath::new(p));
505 rewrite_aws = true;
506 }
507
508 if !rewrite_aws {
509 out_paths.push(path.into_owned());
510 hive_idx_tracker.update(0, path_idx)?;
511 continue;
512 }
513 }
514
515 let sort_start_idx = out_paths.paths.len();
516
517 if glob && has_glob(path.as_bytes()) {
518 hive_idx_tracker.update(0, path_idx)?;
519
520 let iter = crate::async_glob(path.into_owned(), cloud_options.as_ref()).await?;
521
522 for (url, size) in iter {
523 let p = if first_path_has_scheme {
526 PlRefPath::new(url)
527 } else {
528 PlRefPath::new(&url[7..])
529 };
530 sizes_by_path.insert(p.clone(), size);
531 out_paths.push(p);
532 }
533 } else {
534 let (expand_start_idx, paths) = expand_path_cloud(
535 path.into_owned(),
536 cloud_options.as_ref(),
537 glob,
538 first_path_has_scheme,
539 )
540 .await?;
541 for (p, size) in paths {
542 if let Some(size) = size {
543 sizes_by_path.insert(p.clone(), size);
544 }
545 out_paths.push(p);
546 }
547 hive_idx_tracker.update(expand_start_idx, path_idx)?;
548 };
549
550 if let Some(mut_slice) = out_paths.paths.get_mut(sort_start_idx..) {
551 <[PlRefPath]>::sort_unstable(mut_slice);
552 }
553 }
554 }
555 #[cfg(not(feature = "cloud"))]
556 panic!("Feature `cloud` must be enabled to use globbing patterns with cloud urls.")
557 } else {
558 let mut stack: VecDeque<Cow<'_, Path>> = VecDeque::new();
559 let mut paths_scratch: Vec<PathBuf> = vec![];
560
561 for (path_idx, path) in paths.iter().enumerate() {
562 stack.clear();
563 let sort_start_idx = out_paths.paths.len();
564
565 if path.as_std_path().is_dir() {
566 let i = path.as_str().len();
567
568 hive_idx_tracker.update(i, path_idx)?;
569
570 stack.push_back(Cow::Borrowed(path.as_std_path()));
571
572 while let Some(dir) = stack.pop_front() {
573 let mut last_err = Ok(());
574
575 paths_scratch.clear();
576 paths_scratch.extend(std::fs::read_dir(dir)?.map_while(|x| {
577 match x.map(|x| x.path()) {
578 Ok(v) => Some(v),
579 Err(e) => {
580 last_err = Err(e);
581 None
582 },
583 }
584 }));
585
586 last_err?;
587
588 for path in paths_scratch.drain(..) {
589 let md = path.metadata()?;
590
591 if md.is_dir() {
592 stack.push_back(Cow::Owned(path));
593 } else if md.len() > 0 {
594 let p = PlRefPath::try_from_path(&path)?;
595 sizes_by_path.insert(p.clone(), md.len());
596 out_paths.push(p);
597 }
598 }
599 }
600 } else if glob && has_glob(path.as_bytes()) {
601 hive_idx_tracker.update(0, path_idx)?;
602
603 let Ok(paths) = glob::glob(path.as_str()) else {
604 polars_bail!(ComputeError: "invalid glob pattern given")
605 };
606
607 for path in paths {
608 let path = path.map_err(to_compute_err)?;
609 let md = path.metadata()?;
610 if !md.is_dir() && md.len() > 0 {
611 let p = PlRefPath::try_from_path(&path)?;
612 sizes_by_path.insert(p.clone(), md.len());
613 out_paths.push(p);
614 }
615 }
616 } else {
617 hive_idx_tracker.update(0, path_idx)?;
618 out_paths.push(path.clone());
619 };
620
621 if let Some(mut_slice) = out_paths.paths.get_mut(sort_start_idx..) {
622 <[PlRefPath]>::sort_unstable(mut_slice);
623 }
624 }
625 }
626
627 if expanded_from_single_directory(paths, out_paths.paths.as_slice()) {
628 if let [Some((_, p1)), Some((_, p2))] = out_paths.exts {
629 polars_bail!(
630 InvalidOperation: "directory contained paths with different file extensions: \
631 first path: {}, second path: {}. Please use a glob pattern to explicitly specify \
632 which files to read (e.g. 'dir/**/*', 'dir/**/*.parquet')",
633 &p1, &p2
634 )
635 }
636 }
637
638 let bytes_per_source: BytesPerSource = out_paths
640 .paths
641 .iter()
642 .map(|p| sizes_by_path.get(p).copied())
643 .collect::<Option<Vec<u64>>>()
644 .map(Arc::from);
645
646 return Ok((
647 out_paths.paths.into(),
648 hive_idx_tracker.idx,
649 bytes_per_source,
650 ));
651
652 struct OutPaths<'a, F: Fn(&PlRefPath) -> bool> {
655 paths: Vec<PlRefPath>,
656 exts: [Option<(PlSmallStr, PlRefPath)>; 2],
657 is_hidden_file: &'a F,
658 }
659
660 impl<F> OutPaths<'_, F>
661 where
662 F: Fn(&PlRefPath) -> bool,
663 {
664 fn push(&mut self, value: PlRefPath) {
665 if (self.is_hidden_file)(&value) {
666 return;
667 }
668
669 let exts = &mut self.exts;
670 Self::update_ext_status(exts, &value);
671
672 self.paths.push(value)
673 }
674
675 fn update_ext_status(exts: &mut [Option<(PlSmallStr, PlRefPath)>; 2], value: &PlRefPath) {
676 let ext = value
677 .extension()
678 .map_or(PlSmallStr::EMPTY, PlSmallStr::from);
679
680 if exts[0].is_none() {
681 exts[0] = Some((ext, value.clone()));
682 } else if exts[1].is_none() && ext != exts[0].as_ref().unwrap().0 {
683 exts[1] = Some((ext, value.clone()));
684 }
685 }
686 }
687}
688
689#[cfg(feature = "file_cache")]
691pub(crate) fn ensure_directory_init(path: &Path) -> std::io::Result<()> {
692 let result = std::fs::create_dir_all(path);
693
694 if path.is_dir() { Ok(()) } else { result }
695}
696
697#[cfg(test)]
698mod tests {
699 use std::path::PathBuf;
700
701 use polars_core::runtime::ASYNC;
702 use polars_utils::pl_path::PlRefPath;
703
704 use super::resolve_homedir;
705
706 #[cfg(not(target_os = "windows"))]
707 #[test]
708 fn test_resolve_homedir() {
709 let paths: Vec<PathBuf> = vec![
710 "~/dir1/dir2/test.csv".into(),
711 "/abs/path/test.csv".into(),
712 "rel/path/test.csv".into(),
713 "/".into(),
714 "~".into(),
715 ];
716
717 let resolved: Vec<PathBuf> = paths
718 .iter()
719 .map(resolve_homedir)
720 .map(|x| x.into_owned())
721 .collect();
722
723 assert_eq!(resolved[0].file_name(), paths[0].file_name());
724 assert!(resolved[0].is_absolute());
725 assert_eq!(resolved[1], paths[1]);
726 assert_eq!(resolved[2], paths[2]);
727 assert_eq!(resolved[3], paths[3]);
728 assert!(resolved[4].is_absolute());
729 }
730
731 #[cfg(target_os = "windows")]
732 #[test]
733 fn test_resolve_homedir_windows() {
734 let paths: Vec<PathBuf> = vec![
735 r#"c:\Users\user1\test.csv"#.into(),
736 r#"~\user1\test.csv"#.into(),
737 "~".into(),
738 ];
739
740 let resolved: Vec<PathBuf> = paths
741 .iter()
742 .map(resolve_homedir)
743 .map(|x| x.into_owned())
744 .collect();
745
746 assert_eq!(resolved[0], paths[0]);
747 assert_eq!(resolved[1].file_name(), paths[1].file_name());
748 assert!(resolved[1].is_absolute());
749 assert!(resolved[2].is_absolute());
750 }
751
752 #[test]
753 fn test_http_path_with_query_parameters_is_not_expanded_as_glob() {
754 use super::expand_paths;
758
759 let path = "https://pola.rs/test.csv?token=bear";
760 let paths = &[PlRefPath::new(path)];
761 let out = ASYNC
762 .block_on(expand_paths(paths, true, &[], &mut None))
763 .unwrap();
764 assert_eq!(out.as_ref(), paths);
765 }
766}