Merge pull request #44 from JanKaul/feat/file-catalog

Feat/file catalog

Cargo.toml

@@ -6,6 +6,7 @@ members = [
   "datafusion-iceberg-sql",
   "iceberg-sql-catalog",
   "iceberg-rest-catalog",
+  "iceberg-file-catalog",
 ]
 
 resolver = "2"

datafusion_iceberg/src/planner.rs

@@ -121,7 +121,7 @@ impl ExtensionPlanner for CreateIcebergTablePlanner {
             .iter()
             .enumerate()
             .map(|(i, x)| {
-                let (column, transform) = parse_transform(&x)?;
+                let (column, transform) = parse_transform(x)?;
                 Ok::<_, Error>(PartitionField::new(
                     schema
                         .get_name(&column)
@@ -454,6 +454,8 @@ fn parse_transform(input: &str) -> Result<(String, Transform), Error> {
         .ok_or(Error::InvalidFormat("Partition column".to_owned()))?;
     let arg = args.next();
     match (transform_name.as_str(), column, arg) {
+        ("identity", column, None) => Ok((column, Transform::Identity)),
+        ("void", column, None) => Ok((column, Transform::Void)),
         ("year", column, None) => Ok((column, Transform::Year)),
         ("month", column, None) => Ok((column, Transform::Month)),
         ("day", column, None) => Ok((column, Transform::Day)),

datafusion_iceberg/src/statistics.rs

@@ -5,6 +5,7 @@ use datafusion::{
     physical_plan::{ColumnStatistics, Statistics},
     scalar::ScalarValue,
 };
+use futures::{future, TryFutureExt, TryStreamExt};
 use iceberg_rust::spec::{
     manifest::{ManifestEntry, Status},
     schema::Schema,
@@ -33,16 +34,15 @@ pub(crate) async fn table_statistics(
     table: &Table,
     snapshot_range: &(Option<i64>, Option<i64>),
 ) -> Result<Statistics, Error> {
-    let schema = snapshot_range
+    let schema = &snapshot_range
         .1
         .and_then(|snapshot_id| table.metadata().schema(snapshot_id).ok().cloned())
         .unwrap_or_else(|| table.current_schema(None).unwrap().clone());
     let manifests = table.manifests(snapshot_range.0, snapshot_range.1).await?;
     let datafiles = table.datafiles(&manifests, None).await?;
-    Ok(datafiles
-        .iter()
-        .filter(|manifest| !matches!(manifest.status(), Status::Deleted))
-        .fold(
+    datafiles
+        .try_filter(|manifest| future::ready(!matches!(manifest.status(), Status::Deleted)))
+        .try_fold(
             Statistics {
                 num_rows: Precision::Exact(0),
                 total_byte_size: Precision::Exact(0),
@@ -56,9 +56,9 @@ pub(crate) async fn table_statistics(
                     schema.fields().len()
                 ],
             },
-            |acc, manifest| {
-                let column_stats = column_statistics(&schema, manifest);
-                Statistics {
+            |acc, manifest| async move {
+                let column_stats = column_statistics(schema, &manifest);
+                Ok(Statistics {
                     num_rows: acc.num_rows.add(&Precision::Exact(
                         *manifest.data_file().record_count() as usize,
                     )),
@@ -76,9 +76,11 @@ pub(crate) async fn table_statistics(
                             distinct_count: acc.distinct_count.add(&x.distinct_count),
                         })
                         .collect(),
-                }
+                })
             },
-        ))
+        )
+        .map_err(Error::from)
+        .await
 }
 
 fn column_statistics<'a>(

datafusion_iceberg/src/table.rs

@@ -46,7 +46,10 @@ use crate::{
     statistics::manifest_statistics,
 };
 
-use iceberg_rust::spec::{manifest::Status, util};
+use iceberg_rust::spec::{
+    manifest::{ManifestEntry, Status},
+    util,
+};
 use iceberg_rust::spec::{
     schema::Schema,
     types::{StructField, StructType},
@@ -318,7 +321,7 @@ async fn table_scan(
             .map_err(Into::<Error>::into)?;
 
         // If there is a filter expression on the partition column, the manifest files to read are pruned.
-        let data_files = if let Some(predicate) = partition_predicates {
+        let data_files: Vec<ManifestEntry> = if let Some(predicate) = partition_predicates {
             let physical_partition_predicate = create_physical_expr(
                 &predicate,
                 &arrow_schema.as_ref().clone().try_into()?,
@@ -340,11 +343,17 @@ async fn table_scan(
                 .datafiles(&manifests, Some(manifests_to_prune))
                 .await
                 .map_err(Into::<Error>::into)?
+                .try_collect()
+                .await
+                .map_err(Error::from)?
         } else {
             table
                 .datafiles(&manifests, None)
                 .await
                 .map_err(Into::<Error>::into)?
+                .try_collect()
+                .await
+                .map_err(Error::from)?
         };
 
         let pruning_predicate =
@@ -398,10 +407,13 @@ async fn table_scan(
             .manifests(snapshot_range.0, snapshot_range.1)
             .await
             .map_err(Into::<Error>::into)?;
-        let data_files = table
+        let data_files: Vec<ManifestEntry> = table
             .datafiles(&manifests, None)
             .await
-            .map_err(Into::<Error>::into)?;
+            .map_err(Into::<Error>::into)?
+            .try_collect()
+            .await
+            .map_err(Error::from)?;
         data_files.into_iter().for_each(|manifest| {
             if *manifest.status() != Status::Deleted {
                 let partition_values = manifest

iceberg-file-catalog/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "iceberg-file-catalog"
+version = "0.5.7"
+edition = "2021"
+
+description = "Filesystem catalog for the unofficial Iceberg table format implementation"
+
+license = "Apache-2.0"
+
+repository = "https://github.com/JanKaul/iceberg-rust"
+
+[dependencies]
+async-trait.workspace = true
+futures.workspace = true
+iceberg-rust = { path = "../iceberg-rust", version = "0.5.7" }
+object_store.workspace = true
+serde_json.workspace = true
+thiserror.workspace = true
+url.workspace = true
+uuid = { version = "1.7.0", features = ["v4"] }
+
+[dev-dependencies]
+tokio = "1"

iceberg-file-catalog/src/error.rs

@@ -0,0 +1,18 @@
+use iceberg_rust::error::Error as IcebergError;
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum Error {
+    #[error("{0}")]
+    Text(String),
+    #[error(transparent)]
+    ParseError(#[from] url::ParseError),
+    #[error(transparent)]
+    ParseIntError(#[from] std::num::ParseIntError),
+}
+
+impl From<Error> for IcebergError {
+    fn from(value: Error) -> Self {
+        IcebergError::InvalidFormat(value.to_string())
+    }
+}