delta_kernel/
utils.rs

1//! Various utility functions/macros used throughout the kernel
2use std::borrow::Cow;
3use std::ops::Deref;
4use std::path::PathBuf;
5use std::time::{Duration, SystemTime, UNIX_EPOCH};
6
7use delta_kernel_derive::internal_api;
8use url::Url;
9
10use crate::{DeltaResult, Error};
11
12/// convenient way to return an error if a condition isn't true
13macro_rules! require {
14    ( $cond:expr, $err:expr ) => {
15        if !($cond) {
16            return Err($err);
17        }
18    };
19}
20
21pub(crate) use require;
22
23/// Dual of the `FromIterator` trait, similar to how `Into` is the dual of `From`. It is
24/// automatically implemented for any iterable whose items collect into `T`, and can drastically
25/// simplify type bounds. For example, `CollectInto` allows to write this:
26///
27/// ```
28/// # use delta_kernel::CollectInto;
29/// # struct Foo;
30/// fn foo(arg: impl CollectInto<Foo>) -> Foo {
31///     arg.collect_into()
32/// }
33/// ```
34///
35/// instead of the much more verbose:
36///
37/// ```
38/// # struct Foo;
39/// fn foo<T>(arg: impl IntoIterator<Item = T>) -> Foo
40/// where
41///     Foo: FromIterator<T>,
42/// {
43///     Foo::from_iter(arg)
44/// }
45/// ```
46pub trait CollectInto<T>: IntoIterator + Sized {
47    /// Collects this iterable into a `T`
48    fn collect_into(self) -> T;
49}
50
51// blanket impl
52impl<I: IntoIterator, T: FromIterator<I::Item>> CollectInto<T> for I {
53    fn collect_into(self) -> T {
54        T::from_iter(self)
55    }
56}
57
58/// Try to parse string uri into a URL for a table path. This will do it's best to handle things
59/// like `/local/paths`, and even `../relative/paths`.
60#[allow(unused)]
61#[internal_api]
62pub(crate) fn try_parse_uri(uri: impl AsRef<str>) -> DeltaResult<Url> {
63    let uri = uri.as_ref();
64    let uri_type = resolve_uri_type(uri)?;
65    let url = match uri_type {
66        UriType::LocalPath(path) => {
67            if !path.exists() {
68                // When we support writes, create a directory if we can
69                return Err(Error::InvalidTableLocation(format!(
70                    "Path does not exist: {path:?}"
71                )));
72            }
73            if !path.is_dir() {
74                return Err(Error::InvalidTableLocation(format!(
75                    "{path:?} is not a directory"
76                )));
77            }
78            let path = std::fs::canonicalize(path).map_err(|err| {
79                let msg = format!("Invalid table location: {uri} Error: {err:?}");
80                Error::InvalidTableLocation(msg)
81            })?;
82            Url::from_directory_path(path.clone()).map_err(|_| {
83                let msg = format!(
84                    "Could not construct a URL from canonicalized path: {path:?}.\n\
85                     Something must be very wrong with the table path."
86                );
87                Error::InvalidTableLocation(msg)
88            })?
89        }
90        UriType::Url(url) => url,
91    };
92    Ok(url)
93}
94
95#[allow(unused)]
96#[derive(Debug)]
97enum UriType {
98    LocalPath(PathBuf),
99    Url(Url),
100}
101
102/// Utility function to figure out whether string representation of the path is either local path or
103/// some kind or URL.
104///
105/// Will return an error if the path is not valid.
106#[allow(unused)]
107fn resolve_uri_type(table_uri: impl AsRef<str>) -> DeltaResult<UriType> {
108    let table_uri = table_uri.as_ref();
109    let table_uri = if table_uri.ends_with('/') {
110        Cow::Borrowed(table_uri)
111    } else {
112        Cow::Owned(format!("{table_uri}/"))
113    };
114    if let Ok(url) = Url::parse(&table_uri) {
115        let scheme = url.scheme().to_string();
116        if url.scheme() == "file" {
117            Ok(UriType::LocalPath(
118                url.to_file_path()
119                    .map_err(|_| Error::invalid_table_location(table_uri))?,
120            ))
121        } else if scheme.len() == 1 {
122            // NOTE this check is required to support absolute windows paths which may properly
123            // parse as url we assume here that a single character scheme is a windows drive letter
124            Ok(UriType::LocalPath(PathBuf::from(table_uri.as_ref())))
125        } else {
126            Ok(UriType::Url(url))
127        }
128    } else {
129        Ok(UriType::LocalPath(table_uri.deref().into()))
130    }
131}
132
133/// Returns the current time as a Duration since Unix epoch.
134pub(crate) fn current_time_duration() -> DeltaResult<Duration> {
135    SystemTime::now()
136        .duration_since(UNIX_EPOCH)
137        .map_err(|e| Error::generic(format!("System time before Unix epoch: {e}")))
138}
139
140/// Returns the current time in milliseconds since Unix epoch.
141pub(crate) fn current_time_ms() -> DeltaResult<i64> {
142    let duration = current_time_duration()?;
143    i64::try_from(duration.as_millis())
144        .map_err(|_| Error::generic("Current timestamp exceeds i64 millisecond range"))
145}
146
147/// Extension trait for folding zero or one value from an [`Option`] into a base value.
148#[internal_api]
149pub(crate) trait FoldWithOption: Sized {
150    /// Applies an optional fold operation `f` to `self` if `opt` is [`Some`]; otherwise returns
151    /// `self` unchanged.
152    ///
153    /// Similar to `opt.iter().fold(self, |acc, value| f(acc, value))`, but accepting `FnOnce`
154    /// instead of requiring `FnMut`, and with the base value as receiver instead of the option.
155    fn fold_with<U>(self, opt: Option<U>, f: impl FnOnce(Self, U) -> Self) -> Self {
156        match opt {
157            Some(value) => f(self, value),
158            None => self,
159        }
160    }
161
162    /// Fallible [`fold_with`](Self::fold_with): applies `Result`-returning `f` to `self` if `opt`
163    /// is [`Some`], otherwise returns `self` unchanged (wrapped in `Ok`).
164    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
165    fn try_fold_with<U, E>(
166        self,
167        opt: Option<U>,
168        f: impl FnOnce(Self, U) -> Result<Self, E>,
169    ) -> Result<Self, E> {
170        match opt {
171            Some(value) => f(self, value),
172            None => Ok(self),
173        }
174    }
175}
176
177// Blanket impl -- every type can fold_with an Option.
178impl<T: Sized> FoldWithOption for T {}
179
180/// Extension trait for adding completion callbacks to iterators.
181pub(crate) trait IteratorExt: Iterator + Sized {
182    /// Wraps this iterator to call a closure when fully exhausted.
183    ///
184    /// The closure is called only when `next()` returns `None`. If the iterator
185    /// is dropped before exhaustion, a warning is logged but the closure is not called.
186    fn on_complete<F: FnOnce()>(self, f: F) -> OnComplete<Self, F> {
187        OnComplete {
188            inner: self,
189            on_complete: Some(f),
190        }
191    }
192}
193
194impl<I: Iterator> IteratorExt for I {}
195
196/// Iterator adaptor that executes a closure when fully exhausted.
197pub(crate) struct OnComplete<I, F: FnOnce()> {
198    inner: I,
199    on_complete: Option<F>,
200}
201
202impl<I, F: FnOnce()> Drop for OnComplete<I, F> {
203    fn drop(&mut self) {
204        if self.on_complete.is_some() {
205            tracing::debug!(
206                "OnComplete iterator dropped before exhaustion; completion callback not called"
207            );
208        }
209    }
210}
211
212impl<I, F> Iterator for OnComplete<I, F>
213where
214    I: Iterator,
215    F: FnOnce(),
216{
217    type Item = I::Item;
218
219    fn size_hint(&self) -> (usize, Option<usize>) {
220        self.inner.size_hint()
221    }
222
223    fn next(&mut self) -> Option<Self::Item> {
224        match self.inner.next() {
225            Some(item) => Some(item),
226            None => {
227                if let Some(f) = self.on_complete.take() {
228                    f();
229                }
230                None
231            }
232        }
233    }
234}
235
236#[cfg(test)]
237mod tests {
238    use super::*;
239
240    #[test]
241    fn test_path_parsing() {
242        for x in [
243            // windows parsing of file:/// is... odd
244            #[cfg(not(windows))]
245            "file:///foo/bar",
246            #[cfg(not(windows))]
247            "file:///foo/bar/",
248            "/foo/bar",
249            "/foo/bar/",
250            "../foo/bar",
251            "../foo/bar/",
252            "c:/foo/bar",
253            "c:/",
254            "file:///C:/",
255        ] {
256            match resolve_uri_type(x) {
257                Ok(UriType::LocalPath(_)) => {}
258                x => panic!("Should have parsed as a local path {x:?}"),
259            }
260        }
261
262        for x in [
263            "s3://foo/bar",
264            "s3a://foo/bar",
265            "memory://foo/bar",
266            "gs://foo/bar",
267            "https://foo/bar/",
268            "unknown://foo/bar",
269            "s2://foo/bar",
270        ] {
271            match resolve_uri_type(x) {
272                Ok(UriType::Url(_)) => {}
273                x => panic!("Should have parsed as a url {x:?}"),
274            }
275        }
276
277        #[cfg(not(windows))]
278        resolve_uri_type("file://foo/bar").expect_err("file://foo/bar should not have parsed");
279    }
280
281    #[test]
282    fn try_from_uri_without_trailing_slash() {
283        let location = "s3://foo/__unitystorage/catalogs/cid/tables/tid";
284        let url = try_parse_uri(location).unwrap();
285
286        assert_eq!(
287            url.to_string(),
288            "s3://foo/__unitystorage/catalogs/cid/tables/tid/"
289        );
290    }
291
292    mod on_complete_tests {
293        use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
294        use std::sync::Arc;
295
296        use super::*;
297
298        #[test]
299        fn test_calls_on_exhaustion() {
300            let called = Arc::new(AtomicBool::new(false));
301            let called_clone = called.clone();
302            let mut iter = vec![1, 2].into_iter().on_complete(move || {
303                called_clone.store(true, Ordering::SeqCst);
304            });
305            assert_eq!(iter.next(), Some(1));
306            assert!(!called.load(Ordering::SeqCst));
307            assert_eq!(iter.next(), Some(2));
308            assert_eq!(iter.next(), None);
309            assert!(called.load(Ordering::SeqCst));
310        }
311
312        #[test]
313        fn test_does_not_call_on_early_drop() {
314            let called = Arc::new(AtomicBool::new(false));
315            let called_clone = called.clone();
316            {
317                let mut iter = vec![1, 2].into_iter().on_complete(move || {
318                    called_clone.store(true, Ordering::SeqCst);
319                });
320                assert_eq!(iter.next(), Some(1));
321                // Drop without exhausting - callback should NOT be called
322            }
323            assert!(!called.load(Ordering::SeqCst));
324        }
325
326        #[test]
327        fn test_calls_only_once() {
328            let count = Arc::new(AtomicU32::new(0));
329            let count_clone = count.clone();
330            {
331                let mut iter = vec![1].into_iter().on_complete(move || {
332                    count_clone.fetch_add(1, Ordering::SeqCst);
333                });
334                assert_eq!(iter.next(), Some(1));
335                assert_eq!(iter.next(), None); // triggers callback
336                assert_eq!(iter.next(), None); // should not trigger again
337            } // drop should not trigger again
338            assert_eq!(count.load(Ordering::SeqCst), 1);
339        }
340    }
341}
delta_kernel/utils.rs

delta_kernel/
utils.rs
