From b2bccad824770a5b6525e38b3a93ba9183f0984e Mon Sep 17 00:00:00 2001 From: Renjie Liu Date: Fri, 1 Sep 2023 12:12:16 +0800 Subject: [PATCH 1/7] feat: Introduce FileIO --- crates/iceberg/Cargo.toml | 26 ++- crates/iceberg/src/error.rs | 12 + crates/iceberg/src/io.rs | 426 ++++++++++++++++++++++++++++++++++++ crates/iceberg/src/lib.rs | 1 + 4 files changed, 455 insertions(+), 10 deletions(-) create mode 100644 crates/iceberg/src/io.rs diff --git a/crates/iceberg/Cargo.toml b/crates/iceberg/Cargo.toml index d025734236..9682b201b8 100644 --- a/crates/iceberg/Cargo.toml +++ b/crates/iceberg/Cargo.toml @@ -29,22 +29,28 @@ keywords = ["iceberg"] [dependencies] anyhow = "1.0.72" apache-avro = "0.15" -bimap = "0.6" -bitvec = "1.0.1" +serde = { version = "^1.0", features = ["rc"] } +serde_bytes = "0.11.8" +serde_json = "^1.0" +serde_derive = "^1.0" +once_cell = "1" +rust_decimal = "1.31.0" chrono = "0.4" +uuid = "1.4.1" +ordered-float = "3.7.0" +bitvec = "1.0.1" derive_builder = "0.12.0" either = "1" itertools = "0.11" lazy_static = "1" -once_cell = "1" -ordered-float = "3.7.0" -rust_decimal = "1.31.0" -serde = { version = "^1.0", features = ["rc"] } -serde_bytes = "0.11.8" -serde_derive = "^1.0" -serde_json = "^1.0" +bimap = "0.6" serde_repr = "0.1.16" -uuid = "1.4.1" +opendal = "0.39" +futures = "0.3" +url = "2" + [dev-dependencies] pretty_assertions = "1.4.0" +tokio = { version = "1", features = ["macros"] } +tempdir = "0.3" diff --git a/crates/iceberg/src/error.rs b/crates/iceberg/src/error.rs index 48bdebc539..e4ae576d82 100644 --- a/crates/iceberg/src/error.rs +++ b/crates/iceberg/src/error.rs @@ -295,6 +295,18 @@ define_from_err!( "Failure in conversion with avro" ); +define_from_err!( + opendal::Error, + ErrorKind::Unexpected, + "Failure in doing io operation" +); + +define_from_err!( + url::ParseError, + ErrorKind::DataInvalid, + "Failed to parse url" +); + /// Helper macro to check arguments. /// /// diff --git a/crates/iceberg/src/io.rs b/crates/iceberg/src/io.rs new file mode 100644 index 0000000000..6f36096562 --- /dev/null +++ b/crates/iceberg/src/io.rs @@ -0,0 +1,426 @@ +// 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. + +//! File io implementation. +//! +//! File io is built on top of [`opendal`](https://docs.rs/opendal/latest/opendal/index.html), which already provided an abstraction over all kinds of storage. +//! +//! # How to build `FileIO` +//! +//! We provided a `FileIOBuilder` to build `FileIO` from scratch. For example: +//! ```rust +//! use iceberg::io::{FileIOBuilder, S3_ARGS_BUCKET, S3_ARGS_REGION}; +//! +//! let file_io = FileIOBuilder::new("s3") +//! .with_arg(S3_ARGS_BUCKET, "test_bucket") +//! .with_arg(S3_ARGS_REGION, "us-east-1") +//! .build() +//! .unwrap(); +//! ``` +//! +//! We also provided a convenient method to build `FileIO` from url: +//! +//! ```rust +//! use iceberg::io::{FileIO, S3_ARGS_REGION}; +//! let file_io = FileIO::build_from_url("s3a://test_bucket/warehouse") +//! .unwrap() +//! .with_arg(S3_ARGS_REGION, "us-east-1") +//! .build() +//! .unwrap(); +//! ``` +//! +//! # How to use `FileIO` +//! +//! Currently `FileIO` provides simple methods for file operations: +//! +//! - `delete`: Delete file. +//! - `is_exist`: Check if file exists. +//! - `new_input`: Create input file for reading. +//! - `new_output`: Create output file for writing. + +use std::{collections::HashMap, sync::Arc}; + +use crate::{error::Result, Error, ErrorKind}; +use futures::{AsyncRead, AsyncSeek, AsyncWrite}; +use opendal::{Operator, Scheme}; +use url::Url; + +/// Following are arguments for s3 operator. +/// s3 root +pub const S3_ARGS_ROOT: &str = "root"; +/// s3 bucket +pub const S3_ARGS_BUCKET: &str = "bucket"; +/// s3 endpoint +pub const S3_ARGS_ENDPOINT: &str = "endpoint"; +/// s3 region +pub const S3_ARGS_REGION: &str = "region"; +/// s3 access key +pub const S3_ARGS_ACCESS_KEY_ID: &str = "access_key_id"; +/// s3 access secret +pub const S3_ARGS_ACCESS_KEY: &str = "secret_access_key"; + +const ROOT_PATH: &str = "/"; +/// FileIO implementation. +/// +/// # Note +/// +/// `FileIO` keep a [`root_uri`](FileIO::root_uri) to indicate the root path of file io. The arguments passed to operations should be in one for following format: +/// * Absolute path which starts with [`root_uri`](FileIO::root_uri). +/// * Relative path which is relative to [`root_uri`](FileIO::root_uri). +/// +/// For example, if `FileIO` is built from `s3a://test_bucket/warehouse`, then [`root_uri`](FileIO::root_uri) is `s3a://test_bucket/`. And then the following paths are valid: +/// +/// * `s3a://test_bucket/warehouse/iceberg/` +/// * `warehouse/iceberg/`, it will be treated same as above. +#[derive(Clone, Debug)] +pub struct FileIO { + root_uri: Arc, + op: Operator, +} + +/// Builder for [`FileIO`]. +/// +/// # Note +/// +/// We should refer to [`opendal`](https://docs.rs/opendal/0.39.0/opendal/) for what args should be passed to operator. +/// The special `root` argument will always be set to `/` for file io. +pub struct FileIOBuilder { + /// This is used to infer scheme of operator. + /// + /// If this is `None`, then [`FileIOBuilder::build`](FileIOBuilder::build) will build a local file io. + scheme_str: Option, + /// Arguments for operator. + args: HashMap, +} + +impl FileIOBuilder { + /// Creates a new builder with scheme. + pub fn new(scheme_str: impl ToString) -> Self { + Self { + scheme_str: Some(scheme_str.to_string()), + args: HashMap::default(), + } + } + + /// Creates a new builder for local file io. + pub fn new_local_file_io() -> Self { + Self { + scheme_str: None, + args: HashMap::default(), + } + } + + /// Add argument for operator. + pub fn with_arg(mut self, key: impl ToString, value: impl ToString) -> Self { + self.args.insert(key.to_string(), value.to_string()); + self + } + + /// Add argument for operator. + pub fn with_args( + mut self, + args: impl IntoIterator, + ) -> Self { + self.args + .extend(args.into_iter().map(|e| (e.0.to_string(), e.1.to_string()))); + self + } + + /// Builds [`FileIO`]. + pub fn build(mut self) -> Result { + let scheme = FileIO::parse_scheme(self.scheme_str.as_deref())?; + // Set root to "/" for file io. + self.args + .insert(S3_ARGS_ROOT.to_string(), ROOT_PATH.to_string()); + let op = Operator::via_map(scheme, self.args)?; + let op_info = op.info(); + + let root_uri = match self.scheme_str.as_deref() { + Some(s) if op_info.name().is_empty() => format!("{s}:{ROOT_PATH}"), + Some(s) => format!("{s}://{}{ROOT_PATH}", op_info.name()), + None => ROOT_PATH.to_string(), + }; + + Ok(FileIO { + root_uri: Arc::from(root_uri), + op, + }) + } +} + +impl FileIO { + /// Creates builder from url. + pub fn build_from_url(path: impl AsRef) -> Result { + let url = Url::parse(path.as_ref())?; + + let scheme = Self::parse_scheme(Some(url.scheme()))?; + + let bucket = url.host_str().ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid s3 url: {}, missing bucket", path.as_ref()), + ) + }); + + match scheme { + Scheme::Fs => Ok(FileIOBuilder::new(url.scheme())), + Scheme::S3 => Ok(FileIOBuilder::new(url.scheme()).with_arg(S3_ARGS_BUCKET, bucket?)), + _ => Err(Error::new( + ErrorKind::FeatureUnsupported, + format!("Constructing file io from scheme: {scheme} not supported now",), + )), + } + } + + /// Parse scheme. + fn parse_scheme(scheme: Option<&str>) -> Result { + match scheme { + Some("file") | None => Ok(Scheme::Fs), + Some("s3") | Some("s3a") => Ok(Scheme::S3), + Some(s) => Ok(s.parse::()?), + } + } + + /// Deletes file. + pub async fn delete(&self, path: impl AsRef) -> Result<()> { + Ok(self.op.delete(self.relative_path(path.as_ref())).await?) + } + + /// Check file exists. + pub async fn is_exist(&self, path: impl AsRef) -> Result { + Ok(self.op.is_exist(path.as_ref()).await?) + } + + /// Creates input file. + pub fn new_input(&self, path: impl AsRef) -> Result { + Ok(InputFile { + op: self.op.clone(), + path: self.relative_path(path.as_ref()).to_string(), + }) + } + + /// Creates output file. + pub fn new_output(&self, path: impl AsRef) -> Result { + Ok(OutputFile { + op: self.op.clone(), + path: self.relative_path(path.as_ref()).to_string(), + }) + } + + fn relative_path<'a>(&self, path: &'a str) -> &'a str { + if path.starts_with(self.root_uri.as_ref()) { + &path[self.root_uri.len()..] + } else { + path + } + } +} + +/// Input file implementation. +pub struct InputFile { + op: Operator, + path: String, +} + +/// Input stream for reading. +pub trait InputStream: AsyncRead + AsyncSeek {} + +impl InputStream for T where T: AsyncRead + AsyncSeek {} + +impl InputFile { + /// Relative path to root uri. + pub fn location(&self) -> &str { + &self.path + } + + /// Check if file exists. + pub async fn exists(&self) -> Result { + Ok(self.op.is_exist(&self.path).await?) + } + + /// Creates [`InputStream`] for reading. + pub async fn reader(&self) -> Result { + Ok(self.op.reader(&self.path).await?) + } +} + +/// Output file implementation. +pub struct OutputFile { + op: Operator, + path: String, +} + +impl OutputFile { + /// Relative path to root uri. + pub fn location(&self) -> &str { + &self.path + } + + /// Checks if file exists. + pub async fn exists(&self) -> Result { + Ok(self.op.is_exist(&self.path).await?) + } + + /// Converts into [`InputFile`]. + pub fn to_input_file(self) -> InputFile { + InputFile { + op: self.op, + path: self.path, + } + } + + /// Creates output file for writing. + pub async fn writer(&self) -> Result { + Ok(self.op.writer(&self.path).await?) + } +} + +#[cfg(test)] +mod tests { + + use std::io::Write; + + use std::{fs::File, path::Path}; + + use futures::io::AllowStdIo; + use futures::{AsyncReadExt, AsyncWriteExt}; + + use tempdir::TempDir; + + use crate::io::S3_ARGS_REGION; + + use super::{FileIO, FileIOBuilder}; + + fn create_local_file_io() -> FileIO { + FileIOBuilder::new_local_file_io().build().unwrap() + } + + fn write_to_file>(s: &str, path: P) { + let mut f = File::create(path).unwrap(); + write!(f, "{s}").unwrap(); + } + + async fn read_from_file>(path: P) -> String { + let mut f = AllowStdIo::new(File::open(path).unwrap()); + let mut s = String::new(); + f.read_to_string(&mut s).await.unwrap(); + s + } + + #[tokio::test] + async fn test_local_input_file() { + let tmp_dir = TempDir::new("test").unwrap(); + + let file_name = "a.txt"; + let content = "Iceberg loves rust."; + + let full_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), file_name); + write_to_file(content, &full_path); + + let file_io = create_local_file_io(); + let input_file = file_io.new_input(&full_path).unwrap(); + + assert!(input_file.exists().await.unwrap()); + // Remove heading slash + assert_eq!(&full_path[1..], input_file.location()); + let read_content = read_from_file(full_path).await; + + assert_eq!(content, &read_content); + } + + #[tokio::test] + async fn test_delete_local_file() { + let tmp_dir = TempDir::new("test").unwrap(); + + let file_name = "a.txt"; + let content = "Iceberg loves rust."; + + let full_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), file_name); + write_to_file(content, &full_path); + + let file_io = create_local_file_io(); + assert!(file_io.is_exist(&full_path).await.unwrap()); + file_io.delete(&full_path).await.unwrap(); + assert!(!file_io.is_exist(&full_path).await.unwrap()); + } + + #[tokio::test] + async fn test_delete_non_exist_file() { + let tmp_dir = TempDir::new("test").unwrap(); + + let file_name = "a.txt"; + let full_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), file_name); + + let file_io = create_local_file_io(); + assert!(!file_io.is_exist(&full_path).await.unwrap()); + assert!(file_io.delete(&full_path).await.is_ok()); + } + + #[tokio::test] + async fn test_local_output_file() { + let tmp_dir = TempDir::new("test").unwrap(); + + let file_name = "a.txt"; + let content = "Iceberg loves rust."; + + let full_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), file_name); + + let file_io = create_local_file_io(); + let output_file = file_io.new_output(&full_path).unwrap(); + + assert!(!output_file.exists().await.unwrap()); + { + let mut writer = output_file.writer().await.unwrap(); + writer.write_all(content.as_bytes()).await.unwrap(); + writer.close().await.unwrap(); + } + + assert_eq!(&full_path[1..], output_file.location()); + + let read_content = read_from_file(full_path).await; + + assert_eq!(content, &read_content); + } + + #[tokio::test] + async fn test_create_s3_file_io() { + let file_io = FileIO::build_from_url("s3a://test_bucket/warehouse") + .unwrap() + .with_arg("root", "xx") + .with_arg(S3_ARGS_REGION, "us-east-1") + .build() + .unwrap(); + + assert_eq!("s3a://test_bucket/", file_io.root_uri.as_ref()); + } + + #[tokio::test] + async fn test_create_local_file_io() { + let file_io = FileIO::build_from_url("file:/warehouse") + .unwrap() + .build() + .unwrap(); + + assert_eq!("file:/", file_io.root_uri.as_ref()); + + let file_io = FileIO::build_from_url("/warehouse"); + assert!(file_io.is_err()); + + let file_io = FileIOBuilder::new_local_file_io().build().unwrap(); + assert_eq!("/", file_io.root_uri.as_ref()); + } +} diff --git a/crates/iceberg/src/lib.rs b/crates/iceberg/src/lib.rs index 5ef9ad300b..93413d75b7 100644 --- a/crates/iceberg/src/lib.rs +++ b/crates/iceberg/src/lib.rs @@ -28,4 +28,5 @@ pub use error::ErrorKind; pub use error::Result; mod avro; +pub mod io; pub mod spec; From f3d6f5a411db923cf1aea2b598701106adc3d3a7 Mon Sep 17 00:00:00 2001 From: Renjie Liu Date: Wed, 13 Sep 2023 16:41:09 +0800 Subject: [PATCH 2/7] Sort --- crates/iceberg/Cargo.toml | 27 +++++++++++++-------------- 1 file changed, 13 insertions(+), 14 deletions(-) diff --git a/crates/iceberg/Cargo.toml b/crates/iceberg/Cargo.toml index 9682b201b8..007c5c2d22 100644 --- a/crates/iceberg/Cargo.toml +++ b/crates/iceberg/Cargo.toml @@ -29,28 +29,27 @@ keywords = ["iceberg"] [dependencies] anyhow = "1.0.72" apache-avro = "0.15" -serde = { version = "^1.0", features = ["rc"] } -serde_bytes = "0.11.8" -serde_json = "^1.0" -serde_derive = "^1.0" -once_cell = "1" -rust_decimal = "1.31.0" -chrono = "0.4" -uuid = "1.4.1" -ordered-float = "3.7.0" +bimap = "0.6" bitvec = "1.0.1" +chrono = "0.4" derive_builder = "0.12.0" either = "1" +futures = "0.3" itertools = "0.11" lazy_static = "1" -bimap = "0.6" -serde_repr = "0.1.16" +once_cell = "1" opendal = "0.39" -futures = "0.3" +ordered-float = "3.7.0" +rust_decimal = "1.31.0" +serde = { version = "^1.0", features = ["rc"] } +serde_bytes = "0.11.8" +serde_derive = "^1.0" +serde_json = "^1.0" +serde_repr = "0.1.16" url = "2" - +uuid = "1.4.1" [dev-dependencies] pretty_assertions = "1.4.0" -tokio = { version = "1", features = ["macros"] } tempdir = "0.3" +tokio = { version = "1", features = ["macros"] } From 5973e89e7ed58db64971a38c4e6716ecff1bbbd8 Mon Sep 17 00:00:00 2001 From: Renjie Liu Date: Mon, 18 Sep 2023 14:30:20 +0800 Subject: [PATCH 3/7] fix --- crates/iceberg/src/io.rs | 327 +++++++++++++++++++++------------------ 1 file changed, 176 insertions(+), 151 deletions(-) diff --git a/crates/iceberg/src/io.rs b/crates/iceberg/src/io.rs index 6f36096562..103ae5221b 100644 --- a/crates/iceberg/src/io.rs +++ b/crates/iceberg/src/io.rs @@ -17,28 +17,14 @@ //! File io implementation. //! -//! File io is built on top of [`opendal`](https://docs.rs/opendal/latest/opendal/index.html), which already provided an abstraction over all kinds of storage. -//! //! # How to build `FileIO` //! //! We provided a `FileIOBuilder` to build `FileIO` from scratch. For example: //! ```rust -//! use iceberg::io::{FileIOBuilder, S3_ARGS_BUCKET, S3_ARGS_REGION}; +//! use iceberg::io::{FileIOBuilder, S3_REGION}; //! //! let file_io = FileIOBuilder::new("s3") -//! .with_arg(S3_ARGS_BUCKET, "test_bucket") -//! .with_arg(S3_ARGS_REGION, "us-east-1") -//! .build() -//! .unwrap(); -//! ``` -//! -//! We also provided a convenient method to build `FileIO` from url: -//! -//! ```rust -//! use iceberg::io::{FileIO, S3_ARGS_REGION}; -//! let file_io = FileIO::build_from_url("s3a://test_bucket/warehouse") -//! .unwrap() -//! .with_arg(S3_ARGS_REGION, "us-east-1") +//! .with_prop(S3_REGION, "us-east-1") //! .build() //! .unwrap(); //! ``` @@ -56,55 +42,51 @@ use std::{collections::HashMap, sync::Arc}; use crate::{error::Result, Error, ErrorKind}; use futures::{AsyncRead, AsyncSeek, AsyncWrite}; +use once_cell::sync::Lazy; use opendal::{Operator, Scheme}; use url::Url; -/// Following are arguments for s3 operator. -/// s3 root -pub const S3_ARGS_ROOT: &str = "root"; -/// s3 bucket -pub const S3_ARGS_BUCKET: &str = "bucket"; -/// s3 endpoint -pub const S3_ARGS_ENDPOINT: &str = "endpoint"; -/// s3 region -pub const S3_ARGS_REGION: &str = "region"; -/// s3 access key -pub const S3_ARGS_ACCESS_KEY_ID: &str = "access_key_id"; -/// s3 access secret -pub const S3_ARGS_ACCESS_KEY: &str = "secret_access_key"; +/// Following are arguments for s3 file io. +/// S3 endopint. +pub const S3_ENDPOINT: &str = "s3.endpoint"; +/// S3 access key id. +pub const S3_ACCESS_KEY_ID: &str = "s3.access-key-id"; +/// S3 secret access key. +pub const S3_SECRET_ACCESS_KEY: &str = "s3.secret-access-key"; +/// S3 region. +pub const S3_REGION: &str = "s3.region"; + +/// A mapping from iceberg s3 configuration key to [`opendal::Operator`] configuration key. +static S3_CONFIG_MAPPING: Lazy> = Lazy::new(|| { + let mut m = HashMap::with_capacity(4); + m.insert(S3_ENDPOINT, "endpoint"); + m.insert(S3_ACCESS_KEY_ID, "access_key_id"); + m.insert(S3_SECRET_ACCESS_KEY, "secret_access_key"); + m.insert(S3_REGION, "region"); + + m +}); const ROOT_PATH: &str = "/"; -/// FileIO implementation. +/// FileIO implementation, used to manipulate files in underlying storage. /// /// # Note /// -/// `FileIO` keep a [`root_uri`](FileIO::root_uri) to indicate the root path of file io. The arguments passed to operations should be in one for following format: -/// * Absolute path which starts with [`root_uri`](FileIO::root_uri). -/// * Relative path which is relative to [`root_uri`](FileIO::root_uri). -/// -/// For example, if `FileIO` is built from `s3a://test_bucket/warehouse`, then [`root_uri`](FileIO::root_uri) is `s3a://test_bucket/`. And then the following paths are valid: -/// -/// * `s3a://test_bucket/warehouse/iceberg/` -/// * `warehouse/iceberg/`, it will be treated same as above. +/// All path passed to `FileIO` must be absolute path starting with scheme string used to construct `FileIO`. +/// For example, if you construct `FileIO` with `s3a` scheme, then all path passed to `FileIO` must start with `s3a://`. #[derive(Clone, Debug)] pub struct FileIO { - root_uri: Arc, - op: Operator, + inner: Arc, } /// Builder for [`FileIO`]. -/// -/// # Note -/// -/// We should refer to [`opendal`](https://docs.rs/opendal/0.39.0/opendal/) for what args should be passed to operator. -/// The special `root` argument will always be set to `/` for file io. pub struct FileIOBuilder { /// This is used to infer scheme of operator. /// /// If this is `None`, then [`FileIOBuilder::build`](FileIOBuilder::build) will build a local file io. scheme_str: Option, /// Arguments for operator. - args: HashMap, + props: HashMap, } impl FileIOBuilder { @@ -112,7 +94,7 @@ impl FileIOBuilder { pub fn new(scheme_str: impl ToString) -> Self { Self { scheme_str: Some(scheme_str.to_string()), - args: HashMap::default(), + props: HashMap::default(), } } @@ -120,120 +102,81 @@ impl FileIOBuilder { pub fn new_local_file_io() -> Self { Self { scheme_str: None, - args: HashMap::default(), + props: HashMap::default(), } } /// Add argument for operator. - pub fn with_arg(mut self, key: impl ToString, value: impl ToString) -> Self { - self.args.insert(key.to_string(), value.to_string()); + pub fn with_prop(mut self, key: impl ToString, value: impl ToString) -> Self { + self.props.insert(key.to_string(), value.to_string()); self } /// Add argument for operator. - pub fn with_args( + pub fn with_props( mut self, args: impl IntoIterator, ) -> Self { - self.args + self.props .extend(args.into_iter().map(|e| (e.0.to_string(), e.1.to_string()))); self } /// Builds [`FileIO`]. - pub fn build(mut self) -> Result { - let scheme = FileIO::parse_scheme(self.scheme_str.as_deref())?; - // Set root to "/" for file io. - self.args - .insert(S3_ARGS_ROOT.to_string(), ROOT_PATH.to_string()); - let op = Operator::via_map(scheme, self.args)?; - let op_info = op.info(); - - let root_uri = match self.scheme_str.as_deref() { - Some(s) if op_info.name().is_empty() => format!("{s}:{ROOT_PATH}"), - Some(s) => format!("{s}://{}{ROOT_PATH}", op_info.name()), - None => ROOT_PATH.to_string(), - }; - + pub fn build(self) -> Result { + let storage = Storage::build(self)?; Ok(FileIO { - root_uri: Arc::from(root_uri), - op, + inner: Arc::new(storage), }) } } impl FileIO { - /// Creates builder from url. - pub fn build_from_url(path: impl AsRef) -> Result { - let url = Url::parse(path.as_ref())?; - - let scheme = Self::parse_scheme(Some(url.scheme()))?; - - let bucket = url.host_str().ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - format!("Invalid s3 url: {}, missing bucket", path.as_ref()), - ) - }); - - match scheme { - Scheme::Fs => Ok(FileIOBuilder::new(url.scheme())), - Scheme::S3 => Ok(FileIOBuilder::new(url.scheme()).with_arg(S3_ARGS_BUCKET, bucket?)), - _ => Err(Error::new( - ErrorKind::FeatureUnsupported, - format!("Constructing file io from scheme: {scheme} not supported now",), - )), - } - } - - /// Parse scheme. - fn parse_scheme(scheme: Option<&str>) -> Result { - match scheme { - Some("file") | None => Ok(Scheme::Fs), - Some("s3") | Some("s3a") => Ok(Scheme::S3), - Some(s) => Ok(s.parse::()?), - } - } - /// Deletes file. pub async fn delete(&self, path: impl AsRef) -> Result<()> { - Ok(self.op.delete(self.relative_path(path.as_ref())).await?) + let (op, relative_path) = self.inner.create_operator(&path)?; + Ok(op.delete(relative_path).await?) } /// Check file exists. pub async fn is_exist(&self, path: impl AsRef) -> Result { - Ok(self.op.is_exist(path.as_ref()).await?) + let (op, relative_path) = self.inner.create_operator(&path)?; + Ok(op.is_exist(relative_path).await?) } /// Creates input file. pub fn new_input(&self, path: impl AsRef) -> Result { + let (op, relative_path) = self.inner.create_operator(&path)?; + let path = path.as_ref().to_string(); + let relative_path_pos = path.len() - relative_path.len(); Ok(InputFile { - op: self.op.clone(), - path: self.relative_path(path.as_ref()).to_string(), + op, + path, + relative_path_pos, }) } /// Creates output file. pub fn new_output(&self, path: impl AsRef) -> Result { + let (op, relative_path) = self.inner.create_operator(&path)?; + let path = path.as_ref().to_string(); + let relative_path_pos = path.len() - relative_path.len(); Ok(OutputFile { - op: self.op.clone(), - path: self.relative_path(path.as_ref()).to_string(), + op, + path, + relative_path_pos, }) } - - fn relative_path<'a>(&self, path: &'a str) -> &'a str { - if path.starts_with(self.root_uri.as_ref()) { - &path[self.root_uri.len()..] - } else { - path - } - } } /// Input file implementation. +#[derive(Debug)] pub struct InputFile { op: Operator, + // Absolution path of file. path: String, + // Relative path of file to uri, starts at [`relative_path_pos`] + relative_path_pos: usize, } /// Input stream for reading. @@ -242,26 +185,33 @@ pub trait InputStream: AsyncRead + AsyncSeek {} impl InputStream for T where T: AsyncRead + AsyncSeek {} impl InputFile { - /// Relative path to root uri. + /// Absolute path to root uri. pub fn location(&self) -> &str { &self.path } /// Check if file exists. pub async fn exists(&self) -> Result { - Ok(self.op.is_exist(&self.path).await?) + Ok(self + .op + .is_exist(&self.path[self.relative_path_pos..]) + .await?) } /// Creates [`InputStream`] for reading. pub async fn reader(&self) -> Result { - Ok(self.op.reader(&self.path).await?) + Ok(self.op.reader(&self.path[self.relative_path_pos..]).await?) } } /// Output file implementation. +#[derive(Debug)] pub struct OutputFile { op: Operator, + // Absolution path of file. path: String, + // Relative path of file to uri, starts at [`relative_path_pos`] + relative_path_pos: usize, } impl OutputFile { @@ -272,7 +222,10 @@ impl OutputFile { /// Checks if file exists. pub async fn exists(&self) -> Result { - Ok(self.op.is_exist(&self.path).await?) + Ok(self + .op + .is_exist(&self.path[self.relative_path_pos..]) + .await?) } /// Converts into [`InputFile`]. @@ -280,12 +233,114 @@ impl OutputFile { InputFile { op: self.op, path: self.path, + relative_path_pos: self.relative_path_pos, } } /// Creates output file for writing. pub async fn writer(&self) -> Result { - Ok(self.op.writer(&self.path).await?) + Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?) + } +} + +// We introduce this becase I don't want to handle unsupported `Scheme` in every method. +#[derive(Debug)] +enum Storage { + LocalFs { + op: Operator, + }, + S3 { + scheme_str: String, + props: HashMap, + }, +} + +impl Storage { + /// Creates operator from path. + /// + /// # Arguments + /// + /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`]. + /// + /// # Returns + /// + /// The return value consists of two parts: + /// + /// * An [`opendal::Operator`] instance used to operate on file. + /// * Relative path to the root uri of [`opendal::Operator`]. + /// + fn create_operator<'a>(&self, path: &'a impl AsRef) -> Result<(Operator, &'a str)> { + let path = path.as_ref(); + match self { + Storage::LocalFs { op } => { + if let Some(stripped) = path.strip_prefix("file:/") { + Ok((op.clone(), stripped)) + } else { + Ok((op.clone(), &path[1..])) + } + } + Storage::S3 { scheme_str, props } => { + let mut props = props.clone(); + let url = Url::parse(path)?; + let bucket = url.host_str().ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid s3 url: {}, missing bucket", path), + ) + })?; + + props.insert("bucket".to_string(), bucket.to_string()); + + let prefix = format!("{}://{}/", scheme_str, bucket); + if path.starts_with(&prefix) { + Ok((Operator::via_map(Scheme::S3, props)?, &path[prefix.len()..])) + } else { + Err(Error::new( + ErrorKind::DataInvalid, + format!("Invalid s3 url: {}, should start with {}", path, prefix), + )) + } + } + } + } + + /// Parse scheme. + fn parse_scheme(scheme: &str) -> Result { + match scheme { + "file" | "" => Ok(Scheme::Fs), + "s3" | "s3a" => Ok(Scheme::S3), + s => Ok(s.parse::()?), + } + } + + /// Convert iceberg config to opendal config. + fn build(file_io_builder: FileIOBuilder) -> Result { + let scheme_str = file_io_builder.scheme_str.unwrap_or("".to_string()); + let scheme = Self::parse_scheme(&scheme_str)?; + let mut new_props = HashMap::default(); + new_props.insert("root".to_string(), ROOT_PATH.to_string()); + + match scheme { + Scheme::Fs => Ok(Self::LocalFs { + op: Operator::via_map(Scheme::Fs, new_props)?, + }), + Scheme::S3 => { + for prop in file_io_builder.props { + if let Some(op_key) = S3_CONFIG_MAPPING.get(prop.0.as_str()) { + new_props.insert(op_key.to_string(), prop.1); + } + } + + Ok(Self::S3 { + scheme_str, + props: new_props, + }) + } + _ => Err(Error::new( + ErrorKind::FeatureUnsupported, + format!("Constructing file io from scheme: {scheme} not supported now",), + )), + } } } @@ -301,8 +356,6 @@ mod tests { use tempdir::TempDir; - use crate::io::S3_ARGS_REGION; - use super::{FileIO, FileIOBuilder}; fn create_local_file_io() -> FileIO { @@ -336,7 +389,7 @@ mod tests { assert!(input_file.exists().await.unwrap()); // Remove heading slash - assert_eq!(&full_path[1..], input_file.location()); + assert_eq!(&full_path, input_file.location()); let read_content = read_from_file(full_path).await; assert_eq!(content, &read_content); @@ -389,38 +442,10 @@ mod tests { writer.close().await.unwrap(); } - assert_eq!(&full_path[1..], output_file.location()); + assert_eq!(&full_path, output_file.location()); let read_content = read_from_file(full_path).await; assert_eq!(content, &read_content); } - - #[tokio::test] - async fn test_create_s3_file_io() { - let file_io = FileIO::build_from_url("s3a://test_bucket/warehouse") - .unwrap() - .with_arg("root", "xx") - .with_arg(S3_ARGS_REGION, "us-east-1") - .build() - .unwrap(); - - assert_eq!("s3a://test_bucket/", file_io.root_uri.as_ref()); - } - - #[tokio::test] - async fn test_create_local_file_io() { - let file_io = FileIO::build_from_url("file:/warehouse") - .unwrap() - .build() - .unwrap(); - - assert_eq!("file:/", file_io.root_uri.as_ref()); - - let file_io = FileIO::build_from_url("/warehouse"); - assert!(file_io.is_err()); - - let file_io = FileIOBuilder::new_local_file_io().build().unwrap(); - assert_eq!("/", file_io.root_uri.as_ref()); - } } From 27888bb9624317da5c6d6c76b290890966a33baa Mon Sep 17 00:00:00 2001 From: Renjie Liu Date: Mon, 18 Sep 2023 15:06:54 +0800 Subject: [PATCH 4/7] Fix typo --- crates/iceberg/src/io.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/iceberg/src/io.rs b/crates/iceberg/src/io.rs index 103ae5221b..cf067b5f89 100644 --- a/crates/iceberg/src/io.rs +++ b/crates/iceberg/src/io.rs @@ -243,7 +243,7 @@ impl OutputFile { } } -// We introduce this becase I don't want to handle unsupported `Scheme` in every method. +// We introduce this because I don't want to handle unsupported `Scheme` in every method. #[derive(Debug)] enum Storage { LocalFs { From a89f67d7c41882a5c59639ee1706926b5f1658c2 Mon Sep 17 00:00:00 2001 From: Renjie Liu Date: Tue, 19 Sep 2023 12:33:53 +0800 Subject: [PATCH 5/7] Fix comments --- crates/iceberg/src/io.rs | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/crates/iceberg/src/io.rs b/crates/iceberg/src/io.rs index cf067b5f89..7f7b02d7fc 100644 --- a/crates/iceberg/src/io.rs +++ b/crates/iceberg/src/io.rs @@ -46,7 +46,7 @@ use once_cell::sync::Lazy; use opendal::{Operator, Scheme}; use url::Url; -/// Following are arguments for s3 file io. +/// Following are arguments for [s3 file io](https://py.iceberg.apache.org/configuration/#s3). /// S3 endopint. pub const S3_ENDPOINT: &str = "s3.endpoint"; /// S3 access key id. @@ -67,7 +67,7 @@ static S3_CONFIG_MAPPING: Lazy> = Lazy::new( m }); -const ROOT_PATH: &str = "/"; +const DEFAULT_ROOT_PATH: &str = "/"; /// FileIO implementation, used to manipulate files in underlying storage. /// /// # Note @@ -99,7 +99,7 @@ impl FileIOBuilder { } /// Creates a new builder for local file io. - pub fn new_local_file_io() -> Self { + pub fn new_fs_io() -> Self { Self { scheme_str: None, props: HashMap::default(), @@ -169,7 +169,7 @@ impl FileIO { } } -/// Input file implementation. +/// Input file is used for reading from files. #[derive(Debug)] pub struct InputFile { op: Operator, @@ -204,7 +204,7 @@ impl InputFile { } } -/// Output file implementation. +/// Output file is used for writing to files.. #[derive(Debug)] pub struct OutputFile { op: Operator, @@ -318,7 +318,7 @@ impl Storage { let scheme_str = file_io_builder.scheme_str.unwrap_or("".to_string()); let scheme = Self::parse_scheme(&scheme_str)?; let mut new_props = HashMap::default(); - new_props.insert("root".to_string(), ROOT_PATH.to_string()); + new_props.insert("root".to_string(), DEFAULT_ROOT_PATH.to_string()); match scheme { Scheme::Fs => Ok(Self::LocalFs { @@ -359,7 +359,7 @@ mod tests { use super::{FileIO, FileIOBuilder}; fn create_local_file_io() -> FileIO { - FileIOBuilder::new_local_file_io().build().unwrap() + FileIOBuilder::new_fs_io().build().unwrap() } fn write_to_file>(s: &str, path: P) { From 88e809e3003e5f9e8ecb208d1de12ad34cdd0b7d Mon Sep 17 00:00:00 2001 From: Renjie Liu Date: Tue, 19 Sep 2023 14:10:33 +0800 Subject: [PATCH 6/7] Rename InputStream to FileRead --- crates/iceberg/src/io.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/iceberg/src/io.rs b/crates/iceberg/src/io.rs index 7f7b02d7fc..245d34e793 100644 --- a/crates/iceberg/src/io.rs +++ b/crates/iceberg/src/io.rs @@ -180,9 +180,9 @@ pub struct InputFile { } /// Input stream for reading. -pub trait InputStream: AsyncRead + AsyncSeek {} +pub trait FileRead: AsyncRead + AsyncSeek {} -impl InputStream for T where T: AsyncRead + AsyncSeek {} +impl FileRead for T where T: AsyncRead + AsyncSeek {} impl InputFile { /// Absolute path to root uri. @@ -199,7 +199,7 @@ impl InputFile { } /// Creates [`InputStream`] for reading. - pub async fn reader(&self) -> Result { + pub async fn reader(&self) -> Result { Ok(self.op.reader(&self.path[self.relative_path_pos..]).await?) } } From 7cac25a6f19ef796965631d592f9d8b58a9a65a1 Mon Sep 17 00:00:00 2001 From: Renjie Liu Date: Tue, 19 Sep 2023 14:13:10 +0800 Subject: [PATCH 7/7] FileWrite trait --- crates/iceberg/src/io.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/crates/iceberg/src/io.rs b/crates/iceberg/src/io.rs index 245d34e793..9d55aac814 100644 --- a/crates/iceberg/src/io.rs +++ b/crates/iceberg/src/io.rs @@ -179,7 +179,7 @@ pub struct InputFile { relative_path_pos: usize, } -/// Input stream for reading. +/// Trait for reading file. pub trait FileRead: AsyncRead + AsyncSeek {} impl FileRead for T where T: AsyncRead + AsyncSeek {} @@ -204,6 +204,11 @@ impl InputFile { } } +/// Trait for writing file. +pub trait FileWrite: AsyncWrite {} + +impl FileWrite for T where T: AsyncWrite {} + /// Output file is used for writing to files.. #[derive(Debug)] pub struct OutputFile { @@ -238,7 +243,7 @@ impl OutputFile { } /// Creates output file for writing. - pub async fn writer(&self) -> Result { + pub async fn writer(&self) -> Result { Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?) } }