alamb commented on code in PR #25194: URL: https://github.com/apache/datafusion/pull/25194#discussion_r4087020951
########## datafusion/storage/src/path/mod.rs: ########## @@ -0,0 +1,815 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Backend-independent file paths. Adapted from Apache Arrow ObjectStore. Review Comment: yeah this is unfortunate but we probably do need somthing like this ########## datafusion/storage/README.md: ########## @@ -0,0 +1,159 @@ +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# DataFusion Storage + +The file stack uses two backend-independent traits: `Storage` for a namespace's Review Comment: file stack? ########## datafusion/storage/src/read.rs: ########## @@ -0,0 +1,53 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::Result; +use async_trait::async_trait; +use bytes::Bytes; +use futures::stream::BoxStream; +use std::{fmt::Debug, ops::Range, sync::Arc}; + +/// A bounded range or a suffix read, as used by existing file formats. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ReadRange { + Bounded(Range<u64>), + Suffix(u64), +} + +impl From<Range<u64>> for ReadRange { + fn from(range: Range<u64>) -> Self { + Self::Bounded(range) + } +} + +/// A reusable reader for one file. Review Comment: This APi makes sense to me ########## datafusion/storage/src/lib.rs: ########## @@ -0,0 +1,177 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Backend-independent file operations and storage registration. + +pub mod path; +mod read; +mod registry; + +pub use read::{FileReader, ReadRange}; +pub use registry::{StorageBinding, StorageRegistry, StorageUrl}; + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use futures::{StreamExt, stream::BoxStream}; +use path::Path; +use std::{fmt::Debug, sync::Arc}; +use tokio::io::AsyncWrite; +use tokio_util::sync::CancellationToken; + +/// An error independent of the selected storage SDK. +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("File not found: {0}")] + NotFound(String), + #[error("Storage operation not supported: {0}")] + NotSupported(String), + #[error("Invalid file access: {0}")] + InvalidInput(String), + #[error(transparent)] + Path(#[from] path::Error), + #[error(transparent)] + Url(#[from] url::ParseError), + #[error(transparent)] + Io(#[from] std::io::Error), + #[error("{backend}: {source}")] + Backend { + backend: &'static str, + #[source] + source: Box<dyn std::error::Error + Send + Sync>, + }, +} + +pub type Result<T> = std::result::Result<T, Error>; + +/// File metadata relative to a storage namespace. +/// +/// ETags and versions are observations returned by the backend. Reading a file +/// does not automatically turn these fields into conditional requests. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct FileInfo { + pub location: Path, + pub size: u64, + pub last_modified: DateTime<Utc>, + pub e_tag: Option<String>, + pub version: Option<String>, +} + +impl FileInfo { + /// Describe an explicitly supplied file without performing metadata I/O. + pub fn new(location: Path, size: u64) -> Self { + Self { + location, + size, + last_modified: DateTime::UNIX_EPOCH, + e_tag: None, + version: None, + } + } +} + +/// Access state scoped to one planning operation or query execution. +/// Backends may use it to associate their work with the calling query. +#[derive(Clone, Debug, Default)] +pub struct FileAccessContext { + pub query_id: Arc<str>, + pub cancellation: CancellationToken, Review Comment: what is the usecase of this token? Is it to cancel a particular request? ########## datafusion/storage/src/lib.rs: ########## @@ -0,0 +1,177 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Backend-independent file operations and storage registration. + +pub mod path; +mod read; +mod registry; + +pub use read::{FileReader, ReadRange}; +pub use registry::{StorageBinding, StorageRegistry, StorageUrl}; + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use futures::{StreamExt, stream::BoxStream}; +use path::Path; +use std::{fmt::Debug, sync::Arc}; +use tokio::io::AsyncWrite; +use tokio_util::sync::CancellationToken; + +/// An error independent of the selected storage SDK. +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("File not found: {0}")] + NotFound(String), + #[error("Storage operation not supported: {0}")] + NotSupported(String), + #[error("Invalid file access: {0}")] + InvalidInput(String), + #[error(transparent)] + Path(#[from] path::Error), + #[error(transparent)] + Url(#[from] url::ParseError), + #[error(transparent)] + Io(#[from] std::io::Error), + #[error("{backend}: {source}")] + Backend { + backend: &'static str, + #[source] + source: Box<dyn std::error::Error + Send + Sync>, + }, +} + +pub type Result<T> = std::result::Result<T, Error>; + +/// File metadata relative to a storage namespace. +/// +/// ETags and versions are observations returned by the backend. Reading a file +/// does not automatically turn these fields into conditional requests. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct FileInfo { + pub location: Path, + pub size: u64, + pub last_modified: DateTime<Utc>, + pub e_tag: Option<String>, Review Comment: I recommend using Arc<str> for these if possible ########## datafusion/storage/src/registry.rs: ########## @@ -0,0 +1,171 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::{Error, Result, Storage}; +use parking_lot::RwLock; +use std::{ + collections::HashMap, + fmt::{Display, Formatter}, + sync::Arc, +}; +use url::{Position, Url}; + +/// A storage namespace, normalized independently of the chosen backend. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct StorageUrl(Url); +impl StorageUrl { + pub fn parse(value: impl AsRef<str>) -> Result<Self> { + Self::new(&Url::parse(value.as_ref())?) + } + pub fn new(url: &Url) -> Result<Self> { + let key = format!( + "{}://{}", + url.scheme(), + &url[Position::BeforeHost..Position::AfterPort] + ); + let mut url = Url::parse(&key)?; + url.set_path("/"); + Ok(Self(url)) + } + pub fn local_filesystem() -> Self { + Self::parse("file://").unwrap() + } + pub fn as_str(&self) -> &str { + self.0.as_str() + } +} +impl AsRef<Url> for StorageUrl { + fn as_ref(&self) -> &Url { + &self.0 + } +} +impl Display for StorageUrl { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + self.0.fmt(f) + } +} + +/// A single authoritative map of storage registrations. +#[derive(Debug, Default)] +pub struct StorageRegistry { + storages: RwLock<HashMap<StorageUrl, Arc<StorageBinding>>>, +} +impl StorageRegistry { + pub fn register( + &self, + url: &Url, + storage: Arc<dyn Storage>, + ) -> Result<Option<Arc<StorageBinding>>> { + let key = StorageUrl::new(url)?; + let storage = Arc::new(StorageBinding::new(key.clone(), storage)); + Ok(self.storages.write().insert(key, storage)) + } + /// Install a default without replacing an application registration. + pub fn register_default(&self, url: &Url, storage: Arc<dyn Storage>) -> Result<()> { + let key = StorageUrl::new(url)?; + self.storages + .write() + .entry(key.clone()) + .or_insert_with(|| Arc::new(StorageBinding::new(key, storage))); + Ok(()) + } + pub fn get(&self, url: &Url) -> Result<Arc<StorageBinding>> { + let key = StorageUrl::new(url)?; + self.storages + .read() + .get(&key) + .cloned() + .ok_or_else(|| Error::NotFound(format!("storage registration {key}"))) + } + pub fn deregister(&self, url: &Url) -> Result<Arc<StorageBinding>> { + let key = StorageUrl::new(url)?; + self.storages + .write() + .remove(&key) + .ok_or_else(|| Error::NotFound(format!("storage registration {key}"))) + } +} + +impl AsRef<str> for StorageUrl { + fn as_ref(&self) -> &str { + self.as_str() + } +} + +/// An immutable registration: a namespace and its backend. +/// Plans retain this binding across replacement or deregistration. +#[derive(Debug)] +pub struct StorageBinding { + id: u64, Review Comment: what is this id? I don't fully understand the need ########## datafusion/storage/src/path/mod.rs: ########## @@ -0,0 +1,815 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Backend-independent file paths. Adapted from Apache Arrow ObjectStore. + +use percent_encoding::percent_decode; +use std::fmt::Formatter; +#[cfg(not(target_arch = "wasm32"))] +use url::Url; + +/// The delimiter to separate object namespaces, creating a directory structure. +pub const DELIMITER: &str = "/"; + +/// The path delimiter as a single byte +pub const DELIMITER_BYTE: u8 = DELIMITER.as_bytes()[0]; + +/// The path delimiter as a single char +pub const DELIMITER_CHAR: char = DELIMITER_BYTE as char; + +mod parts; + +pub use parts::{InvalidPart, PathPart, PathParts}; + +/// Error returned by [`Path::parse`] +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum Error { + /// Error when there's an empty segment between two slashes `/` in the path + #[error("Path \"{}\" contained empty path segment", path)] + EmptySegment { + /// The source path + path: String, + }, + + /// Error when an invalid segment is encountered in the given path + #[error("Error parsing Path \"{}\": {}", path, source)] + BadSegment { + /// The source path + path: String, + /// The part containing the error + source: Box<InvalidPart>, + }, + + /// Error when path cannot be canonicalized + #[error("Failed to canonicalize path \"{}\": {}", path.display(), source)] + Canonicalize { + /// The source path + path: std::path::PathBuf, + /// The underlying error + source: std::io::Error, + }, + + /// Error when the path is not a valid URL + #[error("Unable to convert path \"{}\" to URL", path.display())] + InvalidPath { + /// The source path + path: std::path::PathBuf, + }, + + /// Error when a path contains non-unicode characters + #[error("Path \"{}\" contained non-unicode characters: {}", path, source)] + NonUnicode { + /// The source path + path: String, + /// The underlying `UTF8Error` + source: std::str::Utf8Error, + }, + + /// Error when the a path doesn't start with given prefix + #[error("Path {} does not start with prefix {}", path, prefix)] + PrefixMismatch { + /// The source path + path: String, + /// The mismatched prefix + prefix: String, + }, +} + +/// A parsed path representation that can be safely written to object storage +/// +/// A [`Path`] maintains the following invariants: +/// +/// * Paths are delimited by `/` +/// * Paths do not contain leading or trailing `/` +/// * Paths do not contain relative path segments, i.e. `.` or `..` +/// * Paths do not contain empty path segments +/// * Paths do not contain any ASCII control characters +/// +/// There are no enforced restrictions on path length, however, it should be noted that most +/// object stores do not permit paths longer than 1024 bytes, and many filesystems do not +/// support path segments longer than 255 bytes. +/// +/// # Encode +/// +/// In theory object stores support any UTF-8 character sequence, however, certain character +/// sequences cause compatibility problems with some applications and protocols. Additionally +/// some filesystems may impose character restrictions, see [`LocalFileSystem`]. As such the +/// naming guidelines for [S3], [GCS] and [Azure Blob Storage] all recommend sticking to a +/// limited character subset. +/// +/// [S3]: https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-keys.html +/// [GCS]: https://cloud.google.com/storage/docs/naming-objects +/// [Azure Blob Storage]: https://docs.microsoft.com/en-us/rest/api/storageservices/Naming-and-Referencing-Containers--Blobs--and-Metadata#blob-names +/// +/// A string containing potentially problematic path segments can therefore be encoded to a [`Path`] +/// using [`Path::from`] or [`Path::from_iter`]. This will percent encode any problematic +/// segments according to [RFC 1738]. +/// +/// ``` +/// # use datafusion_storage::path::Path; +/// assert_eq!(Path::from("foo/bar").as_ref(), "foo/bar"); +/// assert_eq!(Path::from("foo//bar").as_ref(), "foo/bar"); +/// assert_eq!(Path::from("foo/../bar").as_ref(), "foo/%2E%2E/bar"); +/// assert_eq!(Path::from("/").as_ref(), ""); +/// assert_eq!(Path::from_iter(["foo", "foo/bar"]).as_ref(), "foo/foo%2Fbar"); +/// ``` +/// +/// Note: if provided with an already percent encoded string, this will encode it again +/// +/// ``` +/// # use datafusion_storage::path::Path; +/// assert_eq!(Path::from("foo/foo%2Fbar").as_ref(), "foo/foo%252Fbar"); +/// ``` +/// +/// # Parse +/// +/// Alternatively a [`Path`] can be parsed from an existing string, returning an +/// error if it is invalid. Unlike the encoding methods above, this will permit +/// arbitrary unicode, including percent encoded sequences. +/// +/// ``` +/// # use datafusion_storage::path::Path; +/// assert_eq!(Path::parse("/foo/foo%2Fbar").unwrap().as_ref(), "foo/foo%2Fbar"); +/// Path::parse("..").unwrap_err(); // Relative path segments are disallowed +/// Path::parse("/foo//").unwrap_err(); // Empty path segments are disallowed +/// Path::parse("\x00").unwrap_err(); // ASCII control characters are disallowed +/// ``` +/// +/// [RFC 1738]: https://www.ietf.org/rfc/rfc1738.txt +/// [`LocalFileSystem`]: https://docs.rs/object_store/latest/object_store/local/struct.LocalFileSystem.html +#[derive(Debug, Clone, Default, PartialEq, Eq, Hash, Ord, PartialOrd)] +pub struct Path { + /// The raw path with no leading or trailing delimiters + raw: String, Review Comment: I recommend using `Arc<str>` if possible as these get copied around a bunch ########## datafusion/storage/src/lib.rs: ########## @@ -0,0 +1,177 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Backend-independent file operations and storage registration. + +pub mod path; +mod read; +mod registry; + +pub use read::{FileReader, ReadRange}; +pub use registry::{StorageBinding, StorageRegistry, StorageUrl}; + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use futures::{StreamExt, stream::BoxStream}; +use path::Path; +use std::{fmt::Debug, sync::Arc}; +use tokio::io::AsyncWrite; +use tokio_util::sync::CancellationToken; + +/// An error independent of the selected storage SDK. +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("File not found: {0}")] + NotFound(String), + #[error("Storage operation not supported: {0}")] + NotSupported(String), + #[error("Invalid file access: {0}")] + InvalidInput(String), + #[error(transparent)] + Path(#[from] path::Error), + #[error(transparent)] + Url(#[from] url::ParseError), + #[error(transparent)] + Io(#[from] std::io::Error), + #[error("{backend}: {source}")] + Backend { + backend: &'static str, + #[source] + source: Box<dyn std::error::Error + Send + Sync>, + }, +} + +pub type Result<T> = std::result::Result<T, Error>; + +/// File metadata relative to a storage namespace. +/// +/// ETags and versions are observations returned by the backend. Reading a file +/// does not automatically turn these fields into conditional requests. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct FileInfo { + pub location: Path, + pub size: u64, + pub last_modified: DateTime<Utc>, + pub e_tag: Option<String>, + pub version: Option<String>, +} + +impl FileInfo { + /// Describe an explicitly supplied file without performing metadata I/O. + pub fn new(location: Path, size: u64) -> Self { + Self { + location, + size, + last_modified: DateTime::UNIX_EPOCH, + e_tag: None, + version: None, + } + } +} + +/// Access state scoped to one planning operation or query execution. +/// Backends may use it to associate their work with the calling query. +#[derive(Clone, Debug, Default)] +pub struct FileAccessContext { + pub query_id: Arc<str>, + pub cancellation: CancellationToken, +} + +impl FileAccessContext { + pub fn new(query_id: impl Into<Arc<str>>) -> Self { + Self { + query_id: query_id.into(), + cancellation: CancellationToken::new(), + } + } +} + +/// Immediate files and common directory prefixes returned by delimiter listing. +#[derive(Debug, Default)] +pub struct DirectoryListing { + pub files: Vec<FileInfo>, + pub directories: Vec<Path>, +} + +/// Existing output buffering configuration, interpreted by the backend writer. +#[derive(Debug, Clone, Copy, Default)] +pub struct WriterOptions { + pub buffer_size: Option<usize>, +} + +/// Output consumed by the existing format encoders and compression wrappers. +/// `shutdown` completes the output using the backend's normal write semantics. +pub type FileOutput = Box<dyn AsyncWrite + Send + Unpin>; Review Comment: I think the writers should also take `Bytes` if possible -- I think `AsyncWrite` forces a copy as I recall (as it gets `&[u8]`? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
