1pub(crate) mod counting;
23pub(crate) mod first_rows_trace;
24pub mod follow;
25pub(crate) mod local_glob;
26pub mod measurements;
27pub(crate) mod open_options;
28pub(crate) mod open_scan;
29pub(crate) mod scan;
30pub mod stdin;
31pub mod tee;
32pub(crate) mod unfinished;
33
34use std::path::{Path, PathBuf};
35use std::sync::Arc;
36use std::sync::atomic::{AtomicU64, Ordering};
37
38use polars::prelude::LazyFrame;
39
40use crate::cloud::download::TempDownload;
41use crate::formats::schema_union::FooterProgress;
42use crate::loading::unfinished::{Unfinished, Writer};
43use crate::table::DataTableState;
44use crate::{CompressionFormat, FileFormat, OpenOptions, cloud::source};
45
46use crate::app::jobs::Hold;
47#[cfg(any(feature = "http", feature = "cloud"))]
48use crate::app::jobs::Jobs;
49
50#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
52pub(crate) struct LoadId(u64);
53
54impl LoadId {
55 #[cfg(test)]
57 pub(crate) fn for_tests(n: u64) -> Self {
58 Self(n)
59 }
60}
61
62#[cfg(any(feature = "http", feature = "cloud"))]
64#[derive(Clone)]
65pub(crate) enum PendingDownload {
66 #[cfg(feature = "http")]
67 Http {
68 url: String,
69 size: Option<u64>,
70 options: OpenOptions,
71 },
72 #[cfg(feature = "cloud")]
73 S3 {
74 url: String,
75 size: Option<u64>,
76 options: OpenOptions,
77 },
78 #[cfg(feature = "cloud")]
79 Gcs {
80 url: String,
81 size: Option<u64>,
82 options: OpenOptions,
83 },
84 #[cfg(feature = "cloud")]
85 Azure {
86 url: String,
87 size: Option<u64>,
88 options: OpenOptions,
89 },
90 #[cfg(feature = "cloud")]
94 Arrow {
95 url: String,
96 objects: Vec<crate::cloud::cloud_arrow::Object>,
97 size: Option<u64>,
98 options: OpenOptions,
99 },
100}
101
102#[cfg(any(feature = "http", feature = "cloud"))]
103impl PendingDownload {
104 pub(crate) fn parts(&self) -> (&str, Option<u64>, &OpenOptions) {
107 match self {
108 #[cfg(feature = "http")]
109 PendingDownload::Http { url, size, options } => (url, *size, options),
110 #[cfg(feature = "cloud")]
111 PendingDownload::S3 { url, size, options } => (url, *size, options),
112 #[cfg(feature = "cloud")]
113 PendingDownload::Gcs { url, size, options } => (url, *size, options),
114 #[cfg(feature = "cloud")]
115 PendingDownload::Azure { url, size, options } => (url, *size, options),
116 #[cfg(feature = "cloud")]
117 PendingDownload::Arrow {
118 url, size, options, ..
119 } => (url, *size, options),
120 }
121 }
122
123 pub(crate) fn arrow_files(&self) -> Option<(usize, usize)> {
126 match self {
127 #[cfg(feature = "cloud")]
128 PendingDownload::Arrow { objects, .. } => {
129 let streams = objects.iter().filter(|o| o.stream).count();
130 Some((streams, objects.len() - streams))
131 }
132 _ => None,
133 }
134 }
135
136 pub(crate) fn asked(mut self) -> Self {
139 match &mut self {
140 #[cfg(feature = "http")]
141 PendingDownload::Http { options, .. } => options.download_unasked = None,
142 #[cfg(feature = "cloud")]
143 PendingDownload::S3 { options, .. }
144 | PendingDownload::Gcs { options, .. }
145 | PendingDownload::Azure { options, .. }
146 | PendingDownload::Arrow { options, .. } => options.download_unasked = None,
147 }
148 self
149 }
150
151 pub(crate) fn with_size(mut self, found: Option<u64>) -> Self {
153 match &mut self {
154 #[cfg(feature = "http")]
155 PendingDownload::Http { size, .. } => *size = found,
156 #[cfg(feature = "cloud")]
157 PendingDownload::S3 { size, .. } => *size = found,
158 #[cfg(feature = "cloud")]
159 PendingDownload::Gcs { size, .. } => *size = found,
160 #[cfg(feature = "cloud")]
161 PendingDownload::Azure { size, .. } => *size = found,
162 #[cfg(feature = "cloud")]
163 PendingDownload::Arrow { size, .. } => *size = found,
164 }
165 self
166 }
167}
168
169#[derive(Clone)]
172pub(crate) struct OpenRequest {
173 pub(crate) paths: Vec<PathBuf>,
174 pub(crate) options: OpenOptions,
175 pub(crate) size: u64,
177 pub(crate) recent: Option<PathBuf>,
180 pub(crate) shown: Option<PathBuf>,
183 pub(crate) warn_in_memory_above: Option<u64>,
186}
187
188impl OpenRequest {
189 pub(crate) fn named(
194 mut paths: Vec<PathBuf>,
195 mut options: OpenOptions,
196 formats: &crate::formats::Registry,
197 ) -> Self {
198 let first = paths[0].clone();
199 let piped = stdin::is_stdin(&first);
200 let local = !piped && matches!(source::input_source(&first), source::InputSource::Local(_));
201 let mut table = None;
202 if local
203 && paths.len() == 1
204 && let Some((db, name)) = crate::formats::members::split(&first)
205 {
206 table = Some(first.clone());
207 options.table = Some(name);
208 paths = vec![db];
209 } else if local
210 && first.is_file()
211 && let Some(name) = options.table.as_deref()
212 && crate::formats::members::holder(&first).is_some()
213 {
214 table = Some(crate::formats::members::place(&first, name));
215 } else if local
216 && paths.len() == 1
217 && options.table.is_none()
218 && let Some((dir, split)) = crate::formats::hf_splits::split_place(&first)
219 {
220 table = Some(first.clone());
222 options.table = Some(split);
223 paths = vec![dir];
224 } else if local
225 && first.is_dir()
226 && let Some(split) = options.table.as_deref()
227 && crate::formats::hf_splits::cache_splits(&first)
228 .iter()
229 .any(|s| s == split)
230 {
231 table = Some(first.join(split));
232 } else if local
233 && paths.len() == 1
234 && options.table.is_none()
235 && let Some((file, variant)) = crate::formats::members::split_variant(&first, formats)
236 {
237 table = Some(first.clone());
239 options.table = Some(variant);
240 paths = vec![file];
241 } else if local
242 && let Some(variant) = options.table.as_deref()
243 && first.is_file()
244 && formats.variants_of(&first).is_some()
245 {
246 table = Some(crate::formats::members::place(&first, variant));
247 }
248 let first = &paths[0];
249 let size = if local {
250 std::fs::metadata(first).map(|m| m.len()).unwrap_or(0)
251 } else {
252 0
253 };
254 let recent = table
256 .clone()
257 .or_else(|| (!piped && (!local || first.exists())).then(|| first.clone()));
258 Self {
259 paths,
260 options,
261 size,
262 recent,
263 shown: table,
264 warn_in_memory_above: None,
265 }
266 }
267}
268
269pub(crate) enum Phase {
272 Starting {
275 label: String,
276 percent: u16,
277 },
278 LookingAtPaths,
280 ReadingSpec,
282 LookingAtDirectory,
284 #[cfg(any(feature = "http", feature = "cloud"))]
286 ReadingHeaders,
287 #[cfg(any(feature = "http", feature = "cloud"))]
290 CheckingSize {
291 note: Option<&'static str>,
292 },
293 ConfirmingRead {
296 scan: Box<Scan>,
297 _hold: Option<Hold>,
298 },
299 #[cfg(any(feature = "http", feature = "cloud"))]
302 Confirming {
303 pending: Box<PendingDownload>,
304 note: Option<&'static str>,
305 _hold: Hold,
306 },
307 #[cfg(any(feature = "http", feature = "cloud"))]
308 Downloading,
309 Spooling {
311 read: Arc<AtomicU64>,
312 },
313 Decompressing,
314 DecompressingRecords,
316 ReadingRecords,
318 Converting {
321 what: Conversion,
322 read: Arc<AtomicU64>,
323 total: u64,
324 },
325 ScanningStrings,
327 CountingFooter,
330 Scanning {
332 downloaded: bool,
333 },
334 ReadingLines,
336 ReadingSchema,
338 FirstRows,
341}
342
343impl Phase {
344 pub(crate) fn label(&self) -> (&str, u16) {
347 match self {
348 Phase::Starting { label, percent } => (label, *percent),
349 Phase::LookingAtPaths => ("Scanning input", 10),
350 Phase::ReadingSpec => ("Reading spec", 5),
351 Phase::LookingAtDirectory => (crate::App::LOOKING_AT_A_DIRECTORY, 5),
352 #[cfg(any(feature = "http", feature = "cloud"))]
353 Phase::ReadingHeaders => ("Reading headers", 20),
354 #[cfg(any(feature = "http", feature = "cloud"))]
355 Phase::CheckingSize { .. } | Phase::Confirming { .. } => ("Checking size", 0),
356 #[cfg(any(feature = "http", feature = "cloud"))]
357 Phase::Downloading => ("Downloading", 20),
358 Phase::Spooling { .. } => ("Reading stdin", 5),
359 Phase::ConfirmingRead { .. } => ("Scanning input", 0),
360 Phase::Decompressing | Phase::DecompressingRecords => ("Decompressing", 30),
361 Phase::ReadingRecords => ("Reading records", 35),
362 Phase::Converting { what, read, total } => {
363 let done = read.load(Ordering::Relaxed).min(*total);
364 let share = (done * 20).checked_div(*total).unwrap_or(0);
366 (what.label(), 10 + share as u16)
367 }
368 Phase::ScanningStrings => ("Scanning string columns", 55),
369 Phase::CountingFooter => (COUNTING_FOOTER, 10),
370 Phase::Scanning { downloaded: false } => ("Scanning input", 10),
371 Phase::Scanning { downloaded: true } => ("Scanning", 30),
372 Phase::ReadingLines => (READING_LINES, 10),
373 Phase::ReadingSchema => ("Reading schema", 40),
374 Phase::FirstRows => ("Loading buffer", 70),
375 }
376 }
377
378 fn asks(&self) -> bool {
380 match self {
381 Phase::ConfirmingRead { .. } => true,
382 #[cfg(any(feature = "http", feature = "cloud"))]
383 Phase::Confirming { .. } => true,
384 _ => false,
385 }
386 }
387
388 fn builds_the_dataset(&self) -> bool {
390 match self {
391 Phase::ReadingSchema | Phase::Decompressing => true,
392 #[cfg(any(feature = "http", feature = "cloud"))]
393 Phase::ReadingHeaders => true,
394 _ => false,
395 }
396 }
397
398 fn starting(&self) -> bool {
401 matches!(
402 self,
403 Phase::Starting { .. } | Phase::LookingAtPaths | Phase::LookingAtDirectory
404 )
405 }
406}
407
408#[derive(Clone)]
411struct Fetched {
412 url: PathBuf,
413 file: TempDownload,
414 arrow: Option<KeptArrow>,
417}
418
419#[derive(Clone)]
420struct KeptArrow {
421 parts: Arc<Vec<crate::formats::ipc_stream::Part>>,
422 splits: Option<Arc<crate::formats::hf_splits::Splits>>,
423 table: Option<String>,
424}
425
426impl Fetched {
427 fn serves(&self, options: &OpenOptions) -> bool {
430 self.arrow
431 .as_ref()
432 .is_none_or(|arrow| arrow.table == options.table)
433 }
434}
435
436pub(crate) struct Load {
438 id: LoadId,
439 from_home: bool,
442 phase: Phase,
443 path: Option<PathBuf>,
446 size: u64,
447 paths: Option<Vec<PathBuf>>,
449 recent: Option<PathBuf>,
450 progress: Arc<FooterProgress>,
452 writer: Writer,
454 download: Option<Fetched>,
455 converted: Vec<TempDownload>,
458 warn_in_memory_above: Option<u64>,
460}
461
462pub(crate) struct Scan {
464 paths: Vec<PathBuf>,
465 options: OpenOptions,
466 display: Option<PathBuf>,
467}
468
469#[derive(Debug, Clone, PartialEq, Eq)]
471pub(crate) struct InMemory {
472 pub(crate) bytes: u64,
474 pub(crate) format: FileFormat,
475 pub(crate) files: usize,
476 pub(crate) name: PathBuf,
478}
479
480impl Load {
481 pub(crate) fn phase(&self) -> &Phase {
482 &self.phase
483 }
484
485 fn made(&self) -> Made {
488 Made {
489 download: self.download.as_ref().map(|fetched| fetched.file.clone()),
490 converted: self.converted.clone(),
491 ..Made::default()
492 }
493 }
494
495 pub(crate) fn path(&self) -> Option<&Path> {
497 self.path.as_deref()
498 }
499
500 pub(crate) fn size(&self) -> u64 {
503 match &self.phase {
504 Phase::Spooling { read } => read.load(Ordering::Relaxed),
505 _ => self.size,
506 }
507 }
508}
509
510pub(crate) enum Step {
512 Nothing,
514 Crash(String),
516 #[cfg(any(feature = "http", feature = "cloud"))]
519 ReadHeaders {
520 url: PathBuf,
521 format: FileFormat,
522 options: OpenOptions,
523 writer: Writer,
524 },
525 #[cfg(any(feature = "http", feature = "cloud"))]
527 Probe(PendingDownload),
528 #[cfg(any(feature = "http", feature = "cloud"))]
530 Ask(PendingDownload),
531 AskRead(InMemory),
533 #[cfg(any(feature = "http", feature = "cloud"))]
536 Download {
537 pending: PendingDownload,
538 writer: Writer,
539 },
540 FetchSpec {
542 url: PathBuf,
543 options: OpenOptions,
544 writer: Writer,
545 },
546 Spool {
548 options: OpenOptions,
549 writer: Writer,
550 read: Arc<AtomicU64>,
551 },
552 Decompress {
555 file: PathBuf,
556 path: PathBuf,
557 options: OpenOptions,
558 writer: Writer,
559 download: Option<TempDownload>,
561 },
562 DecompressRecords {
565 file: PathBuf,
566 path: PathBuf,
567 choice: crate::formats::Choice,
568 options: OpenOptions,
569 writer: Writer,
570 },
571 ReadRecords {
574 copy: PathBuf,
575 path: PathBuf,
576 choice: crate::formats::Choice,
577 options: OpenOptions,
578 },
579 Convert {
582 what: Conversion,
583 files: Vec<PathBuf>,
584 path: Option<PathBuf>,
585 options: OpenOptions,
586 writer: Writer,
587 read: Arc<AtomicU64>,
588 },
589 Scan {
592 paths: Vec<PathBuf>,
593 options: OpenOptions,
594 display: Option<PathBuf>,
595 status: &'static str,
596 },
597 ReadSchema {
599 lf: Box<LazyFrame>,
600 path: Option<PathBuf>,
601 options: OpenOptions,
602 progress: Arc<FooterProgress>,
603 made: Made,
605 },
606 Install(Box<Loaded>),
608 Failed(Failed),
610 Tables(Tables),
613 Hex(Hex),
616}
617
618#[derive(Debug)]
620pub(crate) struct Hex {
621 pub(crate) file: PathBuf,
622 pub(crate) fetched: Option<(PathBuf, TempDownload)>,
624 pub(crate) from_home: bool,
625 pub(crate) asked: bool,
627 pub(crate) record_size: Option<usize>,
629}
630
631#[derive(Debug)]
633pub(crate) struct Tables {
634 pub(crate) database: PathBuf,
636 pub(crate) from_home: bool,
637}
638
639#[derive(Debug, Clone, Copy, PartialEq, Eq)]
641pub(crate) enum Conversion {
642 Streams,
644 Text(FileFormat),
647}
648
649impl Conversion {
650 pub(crate) fn label(self) -> &'static str {
652 match self {
653 Conversion::Streams => "Converting Arrow stream",
654 Conversion::Text(format) => format.conversion().label,
655 }
656 }
657
658 pub(crate) fn status(self) -> &'static str {
660 match self {
661 Conversion::Streams => "Converting Arrow stream...",
662 Conversion::Text(format) => format.conversion().status,
663 }
664 }
665}
666
667pub(crate) enum Converted {
669 Streams {
671 file: TempDownload,
672 parts: Vec<crate::formats::ipc_stream::Part>,
673 },
674 Frame {
677 files: Vec<TempDownload>,
678 lf: Box<LazyFrame>,
679 notes: Vec<crate::notes::Note>,
680 other_tables: Vec<String>,
681 detail: Option<Arc<crate::formats::text_formats::Detail>>,
683 },
684}
685
686#[derive(Default)]
689pub(crate) struct Made {
690 pub(crate) download: Option<TempDownload>,
692 pub(crate) converted: Vec<TempDownload>,
694 pub(crate) notes: Vec<crate::notes::Note>,
695 pub(crate) other_tables: Vec<String>,
696 pub(crate) detail: Option<Arc<crate::formats::text_formats::Detail>>,
697}
698
699#[derive(Debug)]
701pub(crate) struct Failed {
702 pub(crate) message: String,
703 pub(crate) from_home: bool,
704}
705
706pub(crate) struct Loaded {
708 pub(crate) state: DataTableState,
709 pub(crate) path: Option<PathBuf>,
711 pub(crate) options: OpenOptions,
712 pub(crate) debug_label: Option<String>,
713 pub(crate) paths: Option<Vec<PathBuf>>,
715 pub(crate) recent: Option<PathBuf>,
717 pub(crate) from_home: bool,
718 pub(crate) footers: Arc<FooterProgress>,
720}
721
722pub(crate) enum LoadAnswer {
724 Scanned {
726 lf: Box<LazyFrame>,
727 path: Option<PathBuf>,
728 options: OpenOptions,
729 },
730 Compressed {
732 file: PathBuf,
733 path: Option<PathBuf>,
734 options: OpenOptions,
735 },
736 SpecFetched {
738 spec: Arc<crate::formats::Spec>,
739 options: OpenOptions,
740 },
741 CompressedRecords {
744 file: PathBuf,
745 path: Option<PathBuf>,
746 choice: crate::formats::Choice,
747 options: OpenOptions,
748 },
749 DecompressedRecords {
751 copy: TempDownload,
752 path: PathBuf,
753 choice: crate::formats::Choice,
754 options: OpenOptions,
755 },
756 Convert {
759 what: Conversion,
760 files: Vec<PathBuf>,
761 bytes: u64,
762 path: Option<PathBuf>,
763 options: OpenOptions,
764 },
765 Converted {
767 converted: Converted,
768 path: Option<PathBuf>,
769 options: OpenOptions,
770 },
771 Tables {
774 file: PathBuf,
775 tables: Vec<String>,
776 path: Option<PathBuf>,
777 },
778 Hex {
781 file: PathBuf,
782 asked: bool,
783 record_size: Option<usize>,
784 },
785 SchemaRead {
787 state: Box<DataTableState>,
788 path: Option<PathBuf>,
789 options: OpenOptions,
790 debug_label: Option<String>,
791 },
792 #[cfg(any(feature = "http", feature = "cloud"))]
794 NoRanges { options: OpenOptions },
795 #[cfg(any(feature = "http", feature = "cloud"))]
797 Sized(PendingDownload),
798 #[cfg(feature = "cloud")]
802 InPlace { path: PathBuf, options: OpenOptions },
803 #[cfg(any(feature = "http", feature = "cloud"))]
806 PastLimit(PendingDownload),
807 #[cfg(any(feature = "http", feature = "cloud"))]
809 Downloaded {
810 download: TempDownload,
811 options: OpenOptions,
812 },
813 Spooled {
815 download: TempDownload,
816 options: OpenOptions,
817 },
818 Recorded { file: PathBuf, options: OpenOptions },
821}
822
823#[derive(Debug, Clone, Copy)]
825pub(crate) struct Retired {
826 pub(crate) id: LoadId,
827 pub(crate) asking: bool,
829}
830
831#[derive(Default)]
833pub(crate) struct Loader {
834 next_id: u64,
835 load: Option<Load>,
836 kept: Option<Fetched>,
840 unfinished: Unfinished,
843}
844
845impl Loader {
846 pub(crate) fn unfinished(&self) -> &Unfinished {
848 &self.unfinished
849 }
850
851 pub(crate) fn current(&self) -> Option<&Load> {
853 self.load.as_ref()
854 }
855
856 pub(crate) fn id(&self) -> Option<LoadId> {
857 self.load.as_ref().map(|load| load.id)
858 }
859
860 fn current_in(&self, id: LoadId, phase: impl Fn(&Phase) -> bool) -> bool {
861 self.load
862 .as_ref()
863 .is_some_and(|load| load.id == id && phase(&load.phase))
864 }
865
866 pub(crate) fn looking_at_paths(&self, id: LoadId) -> bool {
868 self.current_in(id, |phase| matches!(phase, Phase::LookingAtPaths))
869 }
870
871 pub(crate) fn looking_at_directory(&self, id: LoadId) -> bool {
873 self.current_in(id, |phase| matches!(phase, Phase::LookingAtDirectory))
874 }
875
876 pub(crate) fn awaiting_dataset(&self) -> bool {
879 self.load
880 .as_ref()
881 .is_some_and(|load| !matches!(load.phase, Phase::FirstRows))
882 }
883
884 pub(crate) fn waits(&self) -> bool {
887 self.awaiting_dataset() && !self.asking()
888 }
889
890 pub(crate) fn asking(&self) -> bool {
892 self.load.as_ref().is_some_and(|load| load.phase.asks())
893 }
894
895 pub(crate) fn hold_while_asking(&mut self, hold: Hold) {
897 if let Some(Load {
898 phase: Phase::ConfirmingRead { _hold, .. },
899 ..
900 }) = self.load.as_mut()
901 {
902 *_hold = Some(hold);
903 }
904 }
905
906 #[cfg(any(feature = "http", feature = "cloud"))]
908 pub(crate) fn download_note(&self) -> Option<&'static str> {
909 match self.load.as_ref().map(|load| &load.phase) {
910 Some(Phase::Confirming { note, .. }) => *note,
911 _ => None,
912 }
913 }
914
915 pub(crate) fn progress(&self) -> Option<&Arc<FooterProgress>> {
918 self.load
919 .as_ref()
920 .filter(|load| !matches!(load.phase, Phase::FirstRows))
921 .map(|load| &load.progress)
922 }
923
924 pub(crate) fn retire(&mut self) -> Option<Retired> {
927 let load = self.load.take()?;
928 if !matches!(load.phase, Phase::FirstRows) {
929 load.progress.cancel();
930 }
931 Some(Retired {
932 id: load.id,
933 asking: load.phase.asks(),
934 })
935 }
936
937 pub(crate) fn make_way(&mut self) -> Option<Retired> {
940 if self.load.as_ref().is_some_and(|load| load.phase.starting()) {
941 return None;
942 }
943 self.retire()
944 }
945
946 fn start(&mut self, from_home: bool) -> &mut Load {
948 if self
951 .load
952 .as_ref()
953 .is_some_and(|load| !load.phase.starting())
954 {
955 debug_assert!(false, "an open started without making way");
956 self.retire();
957 }
958 if self.load.is_none() {
959 self.next_id = self.next_id.wrapping_add(1);
960 let progress = Arc::<FooterProgress>::default();
961 self.load = Some(Load {
962 id: LoadId(self.next_id),
963 from_home,
964 phase: Phase::Starting {
965 label: "Loading".to_string(),
966 percent: 0,
967 },
968 path: None,
969 size: 0,
970 paths: None,
971 recent: None,
972 writer: self.unfinished.writer(progress.cancel_flag()),
973 progress,
974 download: None,
975 converted: Vec::new(),
976 warn_in_memory_above: None,
977 });
978 }
979 let load = self.load.as_mut().expect("started just above");
980 load.from_home |= from_home;
981 load
982 }
983
984 pub(crate) fn announce(&mut self, from_home: bool, label: String, percent: u16) -> LoadId {
987 let load = self.start(from_home);
988 load.phase = Phase::Starting { label, percent };
989 load.id
990 }
991
992 #[cfg(test)]
995 pub(crate) fn first_rows_for_tests(&mut self) {
996 self.retire();
997 self.start(false).phase = Phase::FirstRows;
998 }
999
1000 #[cfg(test)]
1002 pub(crate) fn size_for_tests(&mut self, size: u64) {
1003 if let Some(load) = self.load.as_mut() {
1004 load.size = size;
1005 }
1006 }
1007
1008 pub(crate) fn name(&mut self, path: PathBuf) {
1010 if let Some(load) = self
1011 .load
1012 .as_mut()
1013 .filter(|load| !matches!(load.phase, Phase::FirstRows))
1014 {
1015 load.path = Some(stdin::named(&path));
1016 }
1017 }
1018
1019 pub(crate) fn look_at_paths(&mut self) -> LoadId {
1021 let load = self.start(false);
1022 load.phase = Phase::LookingAtPaths;
1023 load.id
1024 }
1025
1026 pub(crate) fn look_at_directory(&mut self, dir: PathBuf) -> LoadId {
1028 let load = self.start(false);
1029 load.phase = Phase::LookingAtDirectory;
1030 load.path = Some(dir);
1031 load.id
1032 }
1033
1034 pub(crate) fn open(&mut self, request: OpenRequest) -> Step {
1037 if !(request.paths.len() == 1
1038 && self
1039 .kept
1040 .as_ref()
1041 .is_some_and(|kept| kept.url == request.paths[0]))
1042 {
1043 self.kept = None;
1044 }
1045 let OpenRequest {
1046 paths,
1047 mut options,
1048 size,
1049 recent,
1050 shown,
1051 warn_in_memory_above,
1052 } = request;
1053 options.arrow_parts = None;
1055 let prepared = options
1056 .prepared
1057 .take()
1058 .and_then(|handoff| handoff.lock().ok()?.take());
1059 let load = self.start(false);
1060 load.path = Some(shown.unwrap_or_else(|| stdin::named(&paths[0])));
1061 load.size = size;
1062 load.recent = recent;
1063 load.paths = Some(paths.clone());
1064 load.warn_in_memory_above = warn_in_memory_above;
1065 match prepared {
1066 Some(prepared) => self.install_prepared(*prepared),
1067 None => self.first_step(paths, options),
1068 }
1069 }
1070
1071 fn install_prepared(&mut self, prepared: crate::home::home_preview::Prepared) -> Step {
1074 let crate::home::home_preview::Prepared {
1075 state,
1076 options,
1077 debug_label,
1078 progress,
1079 } = prepared;
1080 let writer = self.unfinished.writer(progress.cancel_flag());
1081 let load = self.load.as_mut().expect("started by the open");
1082 load.phase = Phase::FirstRows;
1083 load.progress = progress;
1086 load.writer = writer;
1087 Step::Install(Box::new(Loaded {
1088 state: *state,
1089 path: load.path.clone(),
1090 options,
1091 debug_label,
1092 paths: load.paths.clone(),
1093 recent: load.recent.clone(),
1094 from_home: load.from_home,
1095 footers: load.progress.clone(),
1096 }))
1097 }
1098
1099 pub(crate) fn open_frame(&mut self, lf: LazyFrame, options: OpenOptions) -> Step {
1101 let load = self.start(false);
1102 load.path = None;
1103 load.size = 0;
1104 load.paths = None;
1105 load.recent = None;
1106 load.phase = Phase::ReadingSchema;
1107 Step::ReadSchema {
1108 lf: Box::new(lf),
1109 path: None,
1110 options,
1111 progress: load.progress.clone(),
1112 made: Made::default(),
1113 }
1114 }
1115
1116 fn first_step(&mut self, paths: Vec<PathBuf>, options: OpenOptions) -> Step {
1119 let first = paths[0].clone();
1120 let src = source::input_source(&first);
1121 if paths.len() > 1 {
1122 let only_one = match &src {
1123 source::InputSource::S3(_) => {
1124 Some("Only one S3 URL at a time. Open a single s3:// path.")
1125 }
1126 source::InputSource::Gcs(_) => {
1127 Some("Only one GCS URL at a time. Open a single gs:// path.")
1128 }
1129 source::InputSource::Azure(_) => {
1130 Some("Only one Azure URL at a time. Open a single abfss:// path.")
1131 }
1132 source::InputSource::Http(_) => {
1133 Some("Only one HTTP/HTTPS URL at a time. Open a single URL.")
1134 }
1135 source::InputSource::Local(_) => None,
1136 };
1137 if let Some(message) = only_one {
1138 self.load = None;
1139 return Step::Crash(message.to_string());
1140 }
1141 }
1142 if let Some(message) = stdin::refuse(&paths, true) {
1143 self.load = None;
1144 return Step::Crash(message.to_string());
1145 }
1146 if options.follow
1147 && let Some(message) = crate::loading::follow::refuse_paths(&paths, &options)
1148 {
1149 self.load = None;
1150 return Step::Crash(message);
1151 }
1152 if let Some(url) = options
1155 .spec_file
1156 .clone()
1157 .filter(|file| options.spec_fetched.is_none() && source::is_remote_url(file))
1158 {
1159 let load = self.load.as_mut().expect("an open has a load");
1160 load.phase = Phase::ReadingSpec;
1161 return Step::FetchSpec {
1162 url,
1163 options,
1164 writer: load.writer.clone(),
1165 };
1166 }
1167 if stdin::is_stdin(&first) {
1168 if let Some(kept) = self.kept.clone().filter(|kept| {
1170 kept.url == first && kept.file.path().exists() && kept.serves(&options)
1171 }) {
1172 return self.read_download(kept, options);
1173 }
1174 let load = self.load.as_mut().expect("an open has a load");
1175 let read = Arc::<AtomicU64>::default();
1176 load.phase = Phase::Spooling { read: read.clone() };
1177 return Step::Spool {
1178 options,
1179 writer: load.writer.clone(),
1180 read,
1181 };
1182 }
1183 let load = self.load.as_mut().expect("an open has a load");
1184 let compression = options
1185 .compression
1186 .or_else(|| CompressionFormat::from_extension(&first));
1187 let delimited = delimited_format(&first, &options);
1188 if matches!(src, source::InputSource::Local(_))
1189 && paths.len() == 1
1190 && compression.is_some()
1191 && let Some(format) = delimited
1192 {
1193 load.phase = Phase::Decompressing;
1194 return Step::Decompress {
1195 file: first.clone(),
1196 path: first,
1197 options: OpenOptions {
1198 format: Some(format),
1199 ..options
1200 },
1201 writer: load.writer.clone(),
1202 download: None,
1203 };
1204 }
1205 #[cfg(any(feature = "http", feature = "cloud"))]
1207 if let Some(kept) = self
1208 .kept
1209 .clone()
1210 .filter(|kept| paths.len() == 1 && kept.file.path().exists() && kept.serves(&options))
1211 {
1212 return self.read_download(kept, options);
1213 }
1214 #[cfg(any(feature = "http", feature = "cloud"))]
1216 if paths.len() == 1
1217 && let Some(format) = crate::cloud::remote_model::model_format(&first, options.format)
1218 {
1219 let load = self.load.as_mut().expect("an open has a load");
1220 load.phase = Phase::ReadingHeaders;
1221 return Step::ReadHeaders {
1222 url: first,
1223 format,
1224 options,
1225 writer: load.writer.clone(),
1226 };
1227 }
1228 #[cfg(any(feature = "http", feature = "cloud"))]
1229 if let Some(pending) = remote_download(&src, &options) {
1230 let load = self.load.as_mut().expect("an open has a load");
1231 load.phase = Phase::CheckingSize { note: None };
1232 return Step::Probe(pending);
1233 }
1234 let load = self.load.as_mut().expect("an open has a load");
1235 if counts_footer(&paths, &options) {
1236 load.phase = Phase::CountingFooter;
1237 let display = load.path.clone().filter(|shown| *shown != paths[0]);
1238 return Step::Scan {
1239 paths,
1240 options,
1241 display,
1242 status: COUNTING_FOOTER_STATUS,
1243 };
1244 }
1245 if paths.len() == 1 && delimited.is_some() && options.parse_strings.is_some() {
1246 load.phase = Phase::ScanningStrings;
1247 return Step::Scan {
1248 paths,
1249 options,
1250 display: None,
1251 status: "Scanning string columns...",
1252 };
1253 }
1254 let display = load.path.clone().filter(|shown| *shown != paths[0]);
1256 if let Some(limit) = load.warn_in_memory_above
1258 && let Some(read) = in_memory(&paths, &options)
1259 && read.bytes > limit
1260 {
1261 load.phase = Phase::ConfirmingRead {
1262 scan: Box::new(Scan {
1263 paths,
1264 options,
1265 display,
1266 }),
1267 _hold: None,
1268 };
1269 return Step::AskRead(read);
1270 }
1271 if reads_lines(&paths, &options) {
1272 load.phase = Phase::ReadingLines;
1273 return Step::Scan {
1274 paths,
1275 options,
1276 display,
1277 status: READING_LINES_STATUS,
1278 };
1279 }
1280 load.phase = Phase::Scanning { downloaded: false };
1281 Step::Scan {
1282 paths,
1283 options,
1284 display,
1285 status: "Scanning input...",
1286 }
1287 }
1288
1289 fn read_download(&mut self, fetched: Fetched, mut options: OpenOptions) -> Step {
1292 if let Some(arrow) = &fetched.arrow {
1293 options.format = Some(FileFormat::Arrow);
1294 options.hive = false;
1295 options.arrow_parts = Some(arrow.parts.clone());
1296 options.splits = arrow.splits.clone();
1297 }
1298 let load = self.load.as_mut().expect("a download read has a load");
1299 let file = fetched.file.path().to_path_buf();
1300 let url = stdin::named(&fetched.url);
1301 let download = fetched.file.clone();
1302 load.download = Some(fetched);
1303 let compressed = options
1305 .compression
1306 .or_else(|| CompressionFormat::from_extension(&file))
1307 .is_some();
1308 if compressed && let Some(format) = delimited_format(&file, &options) {
1309 load.phase = Phase::Decompressing;
1310 return Step::Decompress {
1311 file,
1312 path: url,
1313 options: OpenOptions {
1314 format: Some(format),
1315 ..options
1316 },
1317 writer: load.writer.clone(),
1318 download: Some(download),
1319 };
1320 }
1321 let paths = vec![file];
1322 let status = if counts_footer(&paths, &options) {
1323 load.phase = Phase::CountingFooter;
1324 COUNTING_FOOTER_STATUS
1325 } else if reads_lines(&paths, &options) {
1326 load.phase = Phase::ReadingLines;
1327 READING_LINES_STATUS
1328 } else {
1329 load.phase = Phase::Scanning { downloaded: true };
1330 "Scanning..."
1331 };
1332 Step::Scan {
1333 paths,
1334 options,
1335 display: Some(url),
1336 status,
1337 }
1338 }
1339
1340 pub(crate) fn answered(
1343 &mut self,
1344 id: LoadId,
1345 answer: LoadAnswer,
1346 #[cfg(any(feature = "http", feature = "cloud"))] jobs: &Jobs,
1347 ) -> Step {
1348 let Some(load) = self.load.as_mut().filter(|load| load.id == id) else {
1349 return Step::Nothing;
1350 };
1351 match (answer, &load.phase) {
1352 (
1353 LoadAnswer::Scanned { lf, path, options },
1354 Phase::Scanning { .. }
1355 | Phase::ScanningStrings
1356 | Phase::CountingFooter
1357 | Phase::ReadingRecords
1358 | Phase::ReadingLines,
1359 ) => {
1360 load.phase = Phase::ReadingSchema;
1361 Step::ReadSchema {
1362 lf,
1363 path,
1364 options,
1365 progress: load.progress.clone(),
1366 made: load.made(),
1367 }
1368 }
1369 (
1370 LoadAnswer::Convert {
1371 what,
1372 files,
1373 bytes,
1374 path,
1375 options,
1376 },
1377 Phase::Scanning { .. }
1378 | Phase::ScanningStrings
1379 | Phase::CountingFooter
1380 | Phase::ReadingLines,
1381 ) if load.converted.is_empty() => {
1382 let read = Arc::<AtomicU64>::default();
1383 load.phase = Phase::Converting {
1384 what,
1385 read: read.clone(),
1386 total: bytes,
1387 };
1388 Step::Convert {
1389 what,
1390 files,
1391 path,
1392 options,
1393 writer: load.writer.clone(),
1394 read,
1395 }
1396 }
1397 (
1398 LoadAnswer::Converted {
1399 converted: Converted::Streams { file, parts },
1400 path,
1401 options,
1402 },
1403 Phase::Converting { .. },
1404 ) => {
1405 let paths = vec![file.path().to_path_buf()];
1406 if let Some(fetched) = load.download.as_mut() {
1409 fetched.file = file.clone();
1410 }
1411 load.converted = vec![file];
1412 load.phase = Phase::Scanning { downloaded: true };
1413 Step::Scan {
1414 paths,
1415 options: OpenOptions {
1416 format: Some(FileFormat::Arrow),
1417 hive: false,
1418 arrow_parts: Some(Arc::new(parts)),
1419 ..options
1420 },
1421 display: path,
1422 status: "Scanning...",
1423 }
1424 }
1425 (
1426 LoadAnswer::Converted {
1427 converted:
1428 Converted::Frame {
1429 files,
1430 lf,
1431 notes,
1432 other_tables,
1433 detail,
1434 },
1435 path,
1436 options,
1437 },
1438 Phase::Converting { .. },
1439 ) => {
1440 load.converted = files;
1441 load.phase = Phase::ReadingSchema;
1442 Step::ReadSchema {
1443 lf,
1444 path,
1445 options,
1446 progress: load.progress.clone(),
1447 made: Made {
1448 notes,
1449 other_tables,
1450 detail,
1451 ..load.made()
1452 },
1453 }
1454 }
1455 (
1456 LoadAnswer::Compressed {
1457 file,
1458 path,
1459 options,
1460 },
1461 Phase::Scanning { .. }
1462 | Phase::ScanningStrings
1463 | Phase::CountingFooter
1464 | Phase::ReadingLines,
1465 ) => {
1466 load.phase = Phase::Decompressing;
1467 Step::Decompress {
1468 path: path.unwrap_or_else(|| file.clone()),
1469 file,
1470 options,
1471 writer: load.writer.clone(),
1472 download: load.download.as_ref().map(|fetched| fetched.file.clone()),
1473 }
1474 }
1475 (LoadAnswer::SpecFetched { spec, options }, Phase::ReadingSpec) => {
1476 let paths = load.paths.clone().unwrap_or_default();
1477 if paths.is_empty() {
1478 return Step::Nothing;
1479 }
1480 self.first_step(
1481 paths,
1482 OpenOptions {
1483 spec_fetched: Some(spec),
1484 ..options
1485 },
1486 )
1487 }
1488 (
1489 LoadAnswer::CompressedRecords {
1490 file,
1491 path,
1492 choice,
1493 options,
1494 },
1495 Phase::Scanning { .. } | Phase::ScanningStrings | Phase::ReadingLines,
1496 ) => {
1497 load.phase = Phase::DecompressingRecords;
1498 Step::DecompressRecords {
1499 path: path.unwrap_or_else(|| file.clone()),
1500 file,
1501 choice,
1502 options,
1503 writer: load.writer.clone(),
1504 }
1505 }
1506 (
1507 LoadAnswer::DecompressedRecords {
1508 copy,
1509 path,
1510 choice,
1511 options,
1512 },
1513 Phase::DecompressingRecords,
1514 ) => {
1515 let file = copy.path().to_path_buf();
1517 load.converted = vec![copy];
1518 load.phase = Phase::ReadingRecords;
1519 Step::ReadRecords {
1520 copy: file,
1521 path,
1522 choice,
1523 options,
1524 }
1525 }
1526 (
1527 LoadAnswer::Hex {
1528 file,
1529 asked,
1530 record_size,
1531 },
1532 Phase::Scanning { .. }
1533 | Phase::ScanningStrings
1534 | Phase::CountingFooter
1535 | Phase::ReadingLines,
1536 ) => {
1537 let from_home = load.from_home;
1538 let fetched = load.download.take();
1541 self.retire();
1542 if let Some(fetched) = &fetched {
1543 self.kept = Some(fetched.clone());
1544 }
1545 Step::Hex(Hex {
1546 file,
1547 fetched: fetched.map(|fetched| (fetched.url, fetched.file)),
1548 from_home,
1549 asked,
1550 record_size,
1551 })
1552 }
1553 (
1554 LoadAnswer::Tables { file, tables, path },
1555 Phase::Scanning { .. }
1556 | Phase::ScanningStrings
1557 | Phase::CountingFooter
1558 | Phase::ReadingLines,
1559 ) => {
1560 let from_home = load.from_home;
1561 let database = path.unwrap_or(file);
1562 let fetched = load.download.is_some();
1565 self.retire();
1566 if fetched {
1567 let shown = tables.iter().take(20).cloned().collect::<Vec<_>>();
1568 let more = tables.len().saturating_sub(shown.len());
1569 let more = match more {
1570 0 => String::new(),
1571 n => format!(" and {n} more"),
1572 };
1573 return Step::Failed(Failed {
1574 message: format!(
1575 "{} holds {} tables: {}{more}. Open one with --table NAME.",
1576 database.display(),
1577 tables.len(),
1578 shown.join(", ")
1579 ),
1580 from_home,
1581 });
1582 }
1583 Step::Tables(Tables {
1584 database,
1585 from_home,
1586 })
1587 }
1588 (
1589 LoadAnswer::SchemaRead {
1590 state,
1591 path,
1592 options,
1593 debug_label,
1594 },
1595 phase,
1596 ) if phase.builds_the_dataset() => {
1597 load.phase = Phase::FirstRows;
1598 if let Some(fetched) = load.download.take() {
1601 self.kept = Some(fetched);
1602 }
1603 let state = *state;
1604 let load = self.load.as_ref().expect("installing its load");
1605 Step::Install(Box::new(Loaded {
1606 state,
1607 path,
1608 options,
1609 debug_label,
1610 paths: load.paths.clone(),
1611 recent: load.recent.clone(),
1612 from_home: load.from_home,
1613 footers: load.progress.clone(),
1614 }))
1615 }
1616 #[cfg(feature = "cloud")]
1618 (
1619 LoadAnswer::Sized(PendingDownload::Arrow {
1620 url,
1621 objects,
1622 options,
1623 ..
1624 }),
1625 Phase::CheckingSize { .. },
1626 ) if crate::cloud::cloud_arrow::in_place(&objects).is_some() => {
1627 load.phase = Phase::Scanning { downloaded: false };
1628 let parts = crate::cloud::cloud_arrow::in_place(&objects).unwrap_or_default();
1629 Step::Scan {
1630 paths: vec![PathBuf::from(url)],
1631 options: OpenOptions {
1632 format: Some(FileFormat::Arrow),
1633 hive: false,
1634 arrow_parts: Some(Arc::new(parts)),
1635 ..options
1636 },
1637 display: None,
1638 status: "Scanning...",
1639 }
1640 }
1641 #[cfg(feature = "cloud")]
1644 (LoadAnswer::InPlace { path, options }, Phase::CheckingSize { .. }) => {
1645 load.phase = Phase::Scanning { downloaded: false };
1646 let display = load.path.clone().filter(|shown| *shown != path);
1647 Step::Scan {
1648 paths: vec![path],
1649 options,
1650 display,
1651 status: "Scanning...",
1652 }
1653 }
1654 #[cfg(any(feature = "http", feature = "cloud"))]
1655 (LoadAnswer::NoRanges { options }, Phase::ReadingHeaders) => {
1656 let first = load.paths.as_ref().and_then(|paths| paths.first().cloned());
1657 match first
1658 .and_then(|first| remote_download(&source::input_source(&first), &options))
1659 {
1660 Some(pending) => {
1661 load.phase = Phase::CheckingSize {
1662 note: Some(NO_RANGES),
1663 };
1664 Step::Probe(pending)
1665 }
1666 None => self.failed(id, NO_RANGES),
1667 }
1668 }
1669 #[cfg(any(feature = "http", feature = "cloud"))]
1672 (LoadAnswer::Sized(pending), Phase::CheckingSize { note: None })
1673 if pending
1674 .parts()
1675 .2
1676 .download_unasked
1677 .is_some_and(|unasked| unasked.covers(pending.parts().1)) =>
1678 {
1679 load.phase = Phase::Downloading;
1680 Step::Download {
1681 pending,
1682 writer: load.writer.clone(),
1683 }
1684 }
1685 #[cfg(any(feature = "http", feature = "cloud"))]
1686 (LoadAnswer::Sized(pending), Phase::CheckingSize { note }) => {
1687 load.phase = Phase::Confirming {
1688 pending: Box::new(pending.clone()),
1689 note: *note,
1690 _hold: jobs.hold(),
1691 };
1692 Step::Ask(pending)
1693 }
1694 #[cfg(any(feature = "http", feature = "cloud"))]
1697 (LoadAnswer::PastLimit(pending), Phase::Downloading) => {
1698 let pending = pending.with_size(None);
1699 load.phase = Phase::Confirming {
1700 pending: Box::new(pending.clone()),
1701 note: Some(PAST_LIMIT),
1702 _hold: jobs.hold(),
1703 };
1704 Step::Ask(pending)
1705 }
1706 #[cfg(any(feature = "http", feature = "cloud"))]
1707 (LoadAnswer::Downloaded { download, options }, Phase::Downloading) => {
1708 let fetched = Fetched {
1709 url: load.path.clone().unwrap_or_default(),
1711 file: download,
1712 arrow: options.arrow_parts.clone().map(|parts| KeptArrow {
1713 parts,
1714 splits: options.splits.clone(),
1715 table: options.table.clone(),
1716 }),
1717 };
1718 self.read_download(fetched, options)
1719 }
1720 (LoadAnswer::Recorded { file, options }, Phase::Spooling { read }) => {
1721 load.size = read.load(Ordering::Relaxed);
1722 self.kept = None;
1724 load.path = Some(file.clone());
1725 load.paths = Some(vec![file.clone()]);
1726 load.recent = Some(file.clone());
1727 let paths = vec![file];
1728 let status = if counts_footer(&paths, &options) {
1729 load.phase = Phase::CountingFooter;
1730 COUNTING_FOOTER_STATUS
1731 } else {
1732 load.phase = Phase::Scanning { downloaded: false };
1733 "Scanning input..."
1734 };
1735 Step::Scan {
1736 paths,
1737 options,
1738 display: None,
1739 status,
1740 }
1741 }
1742 (LoadAnswer::Spooled { download, options }, Phase::Spooling { read }) => {
1743 load.size = read.load(Ordering::Relaxed);
1744 let fetched = Fetched {
1745 url: PathBuf::from(stdin::PATH),
1746 file: download,
1747 arrow: None,
1748 };
1749 self.read_download(fetched, options)
1750 }
1751 _ => Step::Nothing,
1752 }
1753 }
1754
1755 pub(crate) fn failed(&mut self, id: LoadId, message: &str) -> Step {
1758 let Some(load) = self.load.as_ref().filter(|load| load.id == id) else {
1759 return Step::Nothing;
1760 };
1761 if matches!(load.phase, Phase::FirstRows) {
1762 return Step::Nothing;
1763 }
1764 let from_home = load.from_home;
1765 let mut message = match &load.download {
1768 Some(fetched) => crate::error_display::named_by_source(
1769 message,
1770 fetched.file.path(),
1771 &stdin::named(&fetched.url),
1772 ),
1773 None => message.to_string(),
1774 };
1775 if let Some(path) = &load.path {
1777 for converted in &load.converted {
1778 message = crate::error_display::named_by_source(&message, converted.path(), path);
1779 }
1780 }
1781 self.retire();
1782 Step::Failed(Failed { message, from_home })
1783 }
1784
1785 pub(crate) fn confirmed(&mut self) -> Step {
1788 let Some(load) = self.load.as_mut().filter(|load| load.phase.asks()) else {
1789 return Step::Nothing;
1790 };
1791 match std::mem::replace(&mut load.phase, Phase::Scanning { downloaded: false }) {
1794 Phase::ConfirmingRead { scan, .. } => {
1795 let Scan {
1796 paths,
1797 options,
1798 display,
1799 } = *scan;
1800 Step::Scan {
1801 paths,
1802 options,
1803 display,
1804 status: "Scanning input...",
1805 }
1806 }
1807 #[cfg(any(feature = "http", feature = "cloud"))]
1808 Phase::Confirming { pending, .. } => {
1809 load.phase = Phase::Downloading;
1810 Step::Download {
1811 pending: pending.asked(),
1812 writer: load.writer.clone(),
1813 }
1814 }
1815 _ => unreachable!("a phase that asks"),
1816 }
1817 }
1818
1819 pub(crate) fn first_rows_settled(&mut self) {
1822 if self
1823 .load
1824 .as_ref()
1825 .is_some_and(|load| matches!(load.phase, Phase::FirstRows))
1826 {
1827 self.load = None;
1828 }
1829 }
1830}
1831
1832impl Drop for Loader {
1833 fn drop(&mut self) {
1835 self.retire();
1836 }
1837}
1838
1839const READING_LINES: &str = "Reading as lines";
1841const READING_LINES_STATUS: &str = "Reading as lines...";
1842
1843fn reads_lines(paths: &[PathBuf], options: &OpenOptions) -> bool {
1846 options
1847 .format
1848 .or_else(|| match paths {
1849 [one] => FileFormat::from_path(one),
1850 _ => None,
1851 })
1852 .is_some_and(FileFormat::is_lines)
1853}
1854
1855pub(crate) fn in_memory(paths: &[PathBuf], options: &OpenOptions) -> Option<InMemory> {
1859 let mut found: Option<InMemory> = None;
1860 for path in paths {
1861 if !matches!(source::input_source(path), source::InputSource::Local(_)) {
1862 continue;
1863 }
1864 let compression = options
1865 .compression
1866 .or_else(|| CompressionFormat::from_extension(path));
1867 let format = options.format.or_else(|| match compression {
1870 Some(_) => path
1871 .file_stem()
1872 .and_then(|stem| FileFormat::from_path(Path::new(stem))),
1873 None => match FileFormat::from_path(path) {
1874 Some(named) => Some(crate::formats::readers::refined(path, named).unwrap_or(named)),
1875 None => crate::formats::readers::sniff_open(path, None),
1876 },
1877 });
1878 let Some(format) = format else {
1879 continue;
1880 };
1881 let stored = match compression {
1882 Some(_) => crate::Stored::Compressed {
1883 in_memory: options.decompress_in_memory,
1884 },
1885 None => crate::Stored::Plain,
1886 };
1887 if format.read_mode(stored) != Some(crate::ReadMode::InMemory)
1890 || format.http_file() == crate::RemoteRead::InPlace
1891 {
1892 continue;
1893 }
1894 let Some(bytes) = std::fs::metadata(path)
1895 .ok()
1896 .filter(|m| m.is_file())
1897 .map(|m| m.len())
1898 else {
1899 continue;
1900 };
1901 match &mut found {
1902 Some(read) => {
1903 read.bytes += bytes;
1904 read.files += 1;
1905 }
1906 None => {
1907 found = Some(InMemory {
1908 bytes,
1909 format,
1910 files: 1,
1911 name: path.clone(),
1912 })
1913 }
1914 }
1915 }
1916 found
1917}
1918
1919const COUNTING_FOOTER: &str = "Counting rows to skip the footer";
1921const COUNTING_FOOTER_STATUS: &str = "Counting rows to skip the footer...";
1923
1924fn counts_footer(paths: &[PathBuf], options: &OpenOptions) -> bool {
1927 options.skip_tail_rows.is_some_and(|n| n > 0)
1928 && paths.iter().all(|p| delimited_format(p, options).is_some())
1929}
1930
1931pub(crate) fn delimited_format(path: &Path, options: &OpenOptions) -> Option<FileFormat> {
1934 let format = options.format.or_else(|| {
1935 FileFormat::from_path(path).or_else(|| {
1936 CompressionFormat::from_extension(path)?;
1937 FileFormat::from_path(Path::new(path.file_stem()?))
1938 })
1939 })?;
1940 format.decompressed_once().then_some(format)
1941}
1942
1943#[cfg(any(feature = "http", feature = "cloud"))]
1945pub(crate) const PAST_LIMIT: &str =
1946 "This catalog file passed 50 MiB, more than it may download without asking.";
1947
1948#[cfg(any(feature = "http", feature = "cloud"))]
1950pub(crate) const NO_RANGES: &str = "The server does not send byte ranges, so the model's header cannot be read without downloading the whole file.";
1951
1952#[cfg(any(feature = "http", feature = "cloud"))]
1955fn remote_download(src: &source::InputSource, options: &OpenOptions) -> Option<PendingDownload> {
1956 #[cfg(feature = "cloud")]
1957 let spec = options.spec_file.is_some() || options.spec_name.is_some();
1958 #[cfg(feature = "cloud")]
1959 let should_download = |url: &str| {
1960 should_download(url)
1961 || (spec && !source::is_prefix_or_glob(url))
1962 || unnamed_object(url, options)
1963 };
1964 #[cfg(feature = "cloud")]
1965 if !spec && let Some(arrow) = cloud_arrow(src, options) {
1966 return Some(arrow);
1967 }
1968 let options = options.clone();
1969 match src {
1970 #[cfg(feature = "http")]
1971 source::InputSource::Http(url) => Some(PendingDownload::Http {
1972 url: url.clone(),
1973 size: None,
1974 options,
1975 }),
1976 #[cfg(feature = "cloud")]
1977 source::InputSource::S3(url) => {
1978 let full = format!("s3://{url}");
1979 should_download(&full).then_some(PendingDownload::S3 {
1980 url: full,
1981 size: None,
1982 options,
1983 })
1984 }
1985 #[cfg(feature = "cloud")]
1986 source::InputSource::Gcs(url) => {
1987 let full = format!("gs://{url}");
1988 should_download(&full).then_some(PendingDownload::Gcs {
1989 url: full,
1990 size: None,
1991 options,
1992 })
1993 }
1994 #[cfg(feature = "cloud")]
1995 source::InputSource::Azure(url) => should_download(url).then(|| PendingDownload::Azure {
1996 url: url.clone(),
1997 size: None,
1998 options,
1999 }),
2000 _ => None,
2001 }
2002}
2003
2004#[cfg(feature = "cloud")]
2007fn cloud_arrow(src: &source::InputSource, options: &OpenOptions) -> Option<PendingDownload> {
2008 let url = match src {
2009 source::InputSource::S3(url) => format!("s3://{url}"),
2010 source::InputSource::Gcs(url) => format!("gs://{url}"),
2011 source::InputSource::Azure(url) => url.clone(),
2012 _ => return None,
2013 };
2014 if url.contains('*') {
2015 return None;
2016 }
2017 let arrow = if url.ends_with('/') {
2018 options.format == Some(FileFormat::Arrow)
2019 } else {
2020 options.compression.is_none()
2022 && options
2023 .format
2024 .or_else(|| FileFormat::from_path(Path::new(&url)))
2025 == Some(FileFormat::Arrow)
2026 };
2027 arrow.then(|| PendingDownload::Arrow {
2028 url,
2029 objects: Vec::new(),
2030 size: None,
2031 options: options.clone(),
2032 })
2033}
2034
2035#[cfg(feature = "cloud")]
2040pub(crate) fn unnamed_object(url: &str, options: &OpenOptions) -> bool {
2041 options.format.is_none()
2042 && options.spec_file.is_none()
2043 && options.spec_name.is_none()
2044 && !source::is_prefix_or_glob(url)
2045 && source::url_path_extension(url).1.is_none()
2046 && !crate::home::discover::is_parquet_key(url)
2047}
2048
2049#[cfg(feature = "cloud")]
2050fn should_download(url: &str) -> bool {
2051 let (_, ext) = source::url_path_extension(url);
2052 source::cloud_path_should_download(ext.as_deref(), source::is_prefix_or_glob(url))
2053}
2054
2055#[cfg(test)]
2056mod tests;