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

buoyant_kernel/
utils.rs
