feat: implement datafusion API using ParquetExec

Cargo.toml

@@ -30,27 +30,27 @@ rust-version = "1.75.0"
 
 [workspace.dependencies]
 # arrow
-arrow = { version = "50", features = ["pyarrow"] }
-arrow-arith = { version = "50" }
-arrow-array = { version = "50", features = ["chrono-tz"] }
-arrow-buffer = { version = "50" }
-arrow-cast = { version = "50" }
-arrow-ipc = { version = "50" }
-arrow-json = { version = "50" }
-arrow-ord = { version = "50" }
-arrow-row = { version = "50" }
-arrow-schema = { version = "50" }
-arrow-select = { version = "50" }
-object_store = { version = "0.9.1" }
-parquet = { version = "50" }
+arrow = { version = "52.0.0", features = ["pyarrow"]}
+arrow-arith = { version = "52.0.0" }
+arrow-array = { version = "52.0.0", features = ["chrono-tz"] }
+arrow-buffer = { version = "52.0.0" }
+arrow-cast = { version = "52.0.0" }
+arrow-ipc = { version = "52.0.0" }
+arrow-json = { version = "52.0.0" }
+arrow-ord = { version = "52.0.0" }
+arrow-row = { version = "52.0.0" }
+arrow-schema = { version = "52.0.0" }
+arrow-select = { version = "52.0.0" }
+object_store = { version = "0.10.1" }
+parquet = { version = "52.0.0" }
 
 # datafusion
-datafusion = { version = "35" }
-datafusion-expr = { version = "35" }
-datafusion-common = { version = "35" }
-datafusion-proto = { version = "35" }
-datafusion-sql = { version = "35" }
-datafusion-physical-expr = { version = "35" }
+datafusion = { version = "39.0.0" }
+datafusion-expr = { version = "39.0.0" }
+datafusion-common = { version = "39.0.0" }
+datafusion-proto = { version = "39.0.0" }
+datafusion-sql = { version = "39.0.0" }
+datafusion-physical-expr = { version = "39.0.0" }
 
 # serde
 serde = { version = "1.0.203", features = ["derive"] }

crates/core/src/config/mod.rs

@@ -0,0 +1,118 @@
+/*
+ * 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 std::collections::HashMap;
+
+use anyhow::{anyhow, Context, Result};
+
+pub trait OptionsParser {
+    type Output;
+
+    fn parse_value(&self, options: &HashMap<String, String>) -> Result<Self::Output>;
+
+    fn parse_value_or_default(&self, options: &HashMap<String, String>) -> Self::Output;
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub enum HudiConfig {
+    ReadInputPartitions,
+}
+
+#[derive(Debug)]
+pub enum HudiConfigValue {
+    Integer(isize),
+}
+
+impl HudiConfigValue {
+    pub fn cast<T: 'static + TryFrom<isize> + TryFrom<usize> + std::fmt::Debug>(&self) -> T {
+        match self {
+            HudiConfigValue::Integer(value) => T::try_from(*value).unwrap_or_else(|_| {
+                panic!("Failed to convert isize to {}", std::any::type_name::<T>())
+            }),
+        }
+    }
+}
+
+impl HudiConfig {
+    fn default_value(&self) -> Option<HudiConfigValue> {
+        match self {
+            Self::ReadInputPartitions => Some(HudiConfigValue::Integer(0)),
+        }
+    }
+}
+
+impl AsRef<str> for HudiConfig {
+    fn as_ref(&self) -> &str {
+        match self {
+            Self::ReadInputPartitions => "hoodie.read.input.partitions",
+        }
+    }
+}
+
+impl OptionsParser for HudiConfig {
+    type Output = HudiConfigValue;
+
+    fn parse_value(&self, options: &HashMap<String, String>) -> Result<Self::Output> {
+        match self {
+            HudiConfig::ReadInputPartitions => options.get(self.as_ref()).map_or_else(
+                || Err(anyhow!("Config '{}' not found", self.as_ref())),
+                |v| {
+                    v.parse::<isize>()
+                        .map(HudiConfigValue::Integer)
+                        .with_context(|| {
+                            format!("Failed to parse '{}' for config '{}'", v, self.as_ref())
+                        })
+                },
+            ),
+        }
+    }
+
+    fn parse_value_or_default(&self, options: &HashMap<String, String>) -> Self::Output {
+        self.parse_value(options).unwrap_or_else(|_| {
+            self.default_value()
+                .unwrap_or_else(|| panic!("No default value for config '{}'", self.as_ref()))
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::config::HudiConfig::ReadInputPartitions;
+    use crate::config::OptionsParser;
+    use std::collections::HashMap;
+
+    #[test]
+    fn parse_invalid_config_value() {
+        let options =
+            HashMap::from([(ReadInputPartitions.as_ref().to_string(), "foo".to_string())]);
+        let value = ReadInputPartitions.parse_value(&options);
+        assert_eq!(
+            value.err().unwrap().to_string(),
+            format!(
+                "Failed to parse 'foo' for config '{}'",
+                ReadInputPartitions.as_ref()
+            )
+        );
+        assert_eq!(
+            ReadInputPartitions
+                .parse_value_or_default(&options)
+                .cast::<isize>(),
+            0
+        );
+    }
+}

crates/core/src/lib.rs

@@ -22,7 +22,8 @@ use crate::table::Table;
 pub mod file_group;
 pub mod table;
 pub type HudiTable = Table;
-mod storage;
+pub mod config;
+pub mod storage;
 
 pub fn crate_version() -> &'static str {
     env!("CARGO_PKG_VERSION")

crates/core/src/storage/mod.rs

@@ -37,9 +37,9 @@ use url::Url;
 use crate::storage::file_info::FileInfo;
 use crate::storage::utils::join_url_segments;
 
-pub(crate) mod file_info;
-pub(crate) mod file_stats;
-pub(crate) mod utils;
+pub mod file_info;
+pub mod file_stats;
+pub mod utils;
 
 #[allow(dead_code)]
 #[derive(Clone, Debug)]
@@ -351,7 +351,7 @@ mod tests {
         assert_eq!(file_info.name, "a.parquet");
         assert_eq!(
             file_info.uri,
-            storage.base_url.join("a.parquet").unwrap().to_string()
+            storage.base_url.join("a.parquet").unwrap().as_ref()
         );
         assert_eq!(file_info.size, 866);
     }

crates/core/src/storage/utils.rs

@@ -17,7 +17,8 @@
  * under the License.
  */
 
-use std::path::Path;
+use std::path::{Path, PathBuf};
+use std::str::FromStr;
 
 use anyhow::{anyhow, Result};
 use url::{ParseError, Url};
@@ -40,6 +41,24 @@ pub fn split_filename(filename: &str) -> Result<(String, String)> {
     Ok((stem, extension))
 }
 
+pub fn parse_uri(uri: &str) -> Result<Url> {
+    let mut url = Url::parse(uri)
+        .or(Url::from_file_path(PathBuf::from_str(uri)?))
+        .map_err(|_| anyhow!("Failed to parse uri: {}", uri))?;
+
+    if url.path().ends_with('/') {
+        url.path_segments_mut()
+            .map_err(|_| anyhow!("Failed to parse uri: {}", uri))?
+            .pop();
+    }
+
+    Ok(url)
+}
+
+pub fn get_scheme_authority(url: &Url) -> String {
+    format!("{}://{}", url.scheme(), url.authority())
+}
+
 pub fn join_url_segments(base_url: &Url, segments: &[&str]) -> Result<Url> {
     let mut url = base_url.clone();
 
@@ -63,7 +82,31 @@ mod tests {
 
     use url::Url;
 
-    use crate::storage::utils::join_url_segments;
+    use crate::storage::utils::{join_url_segments, parse_uri};
+
+    #[test]
+    fn parse_valid_uri_in_various_forms() {
+        let urls = vec![
+            parse_uri("/foo/").unwrap(),
+            parse_uri("file:/foo/").unwrap(),
+            parse_uri("file:///foo/").unwrap(),
+            parse_uri("hdfs://foo/").unwrap(),
+            parse_uri("s3://foo").unwrap(),
+            parse_uri("s3://foo/").unwrap(),
+            parse_uri("s3a://foo/bar/").unwrap(),
+            parse_uri("gs://foo/").unwrap(),
+            parse_uri("wasb://foo/bar").unwrap(),
+            parse_uri("wasbs://foo/").unwrap(),
+        ];
+        let schemes = vec![
+            "file", "file", "file", "hdfs", "s3", "s3", "s3a", "gs", "wasb", "wasbs",
+        ];
+        let paths = vec![
+            "/foo", "/foo", "/foo", "/", "", "/", "/bar", "/", "/bar", "/",
+        ];
+        assert_eq!(urls.iter().map(|u| u.scheme()).collect::<Vec<_>>(), schemes);
+        assert_eq!(urls.iter().map(|u| u.path()).collect::<Vec<_>>(), paths);
+    }
 
     #[test]
     fn join_base_url_with_segments() {

crates/core/src/table/mod.rs

@@ -19,7 +19,6 @@
 
 use std::collections::HashMap;
 use std::io::{BufRead, BufReader};
-use std::path::PathBuf;
 use std::str::FromStr;
 use std::sync::Arc;
 
@@ -29,6 +28,7 @@ use arrow_schema::Schema;
 use url::Url;
 
 use crate::file_group::FileSlice;
+use crate::storage::utils::parse_uri;
 use crate::storage::Storage;
 use crate::table::config::BaseFileFormat;
 use crate::table::config::{ConfigKey, TableType};
@@ -52,9 +52,7 @@ pub struct Table {
 
 impl Table {
     pub async fn new(base_uri: &str, storage_options: HashMap<String, String>) -> Result<Self> {
-        let base_url = Url::from_file_path(PathBuf::from(base_uri))
-            .map_err(|_| anyhow!("Failed to create table URL: {}", base_uri))?;
-        let base_url = Arc::new(base_url);
+        let base_url = Arc::new(parse_uri(base_uri)?);
         let storage_options = Arc::new(storage_options);
 
         let props = Self::load_properties(base_url.clone(), storage_options.clone())
@@ -114,6 +112,17 @@ impl Table {
         self.timeline.get_latest_schema().await
     }
 
+    pub async fn split_file_slices(&self, n: usize) -> Result<Vec<Vec<FileSlice>>> {
+        let n = std::cmp::max(1, n);
+        let file_slices = self.get_file_slices().await?;
+        let chunk_size = (file_slices.len() + n - 1) / n;
+
+        Ok(file_slices
+            .chunks(chunk_size)
+            .map(|chunk| chunk.to_vec())
+            .collect())
+    }
+
     pub async fn get_file_slices(&self) -> Result<Vec<FileSlice>> {
         if let Some(timestamp) = self.timeline.get_latest_commit_timestamp() {
             self.get_file_slices_as_of(timestamp).await

crates/datafusion/Cargo.toml

@@ -56,18 +56,14 @@ serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 
 # async
+async-trait = { workspace = true }
+futures = { workspace = true }
 tokio = { workspace = true }
 
 # "stdlib"
 anyhow = { workspace = true }
 bytes = { workspace = true }
 chrono = { workspace = true, default-features = false, features = ["clock"] }
-hashbrown = "0.14.3"
 regex = { workspace = true }
 uuid = { workspace = true, features = ["serde", "v4"] }
 url = { workspace = true }
-
-# test
-tempfile = "3.10.1"
-zip-extract = "0.1.3"
-async-trait = "0.1.79"