feat: basic read operation for Iceberg

Cargo.toml

@@ -104,6 +104,7 @@ bytes = "1.10.1"
 indexmap = "2.11.4"
 pin-project-lite = "0.2.16"
 ordered-float = { version = "5.1.0", features = ["serde"] }
+apache-avro = { version = "0.20.0" }
 
 ######
 # The versions of the following dependencies are managed manually.

crates/sail-data-source/Cargo.toml

@@ -10,6 +10,7 @@ workspace = true
 sail-common = { path = "../sail-common" }
 sail-common-datafusion = { path = "../sail-common-datafusion" }
 sail-delta-lake = { path = "../sail-delta-lake" }
+sail-iceberg = { path = "../sail-iceberg" }
 
 async-trait = { workspace = true }
 serde = { workspace = true }

crates/sail-data-source/src/formats/iceberg.rs

@@ -0,0 +1,45 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use datafusion::catalog::{Session, TableProvider};
+use datafusion::common::Result;
+use datafusion::physical_plan::ExecutionPlan;
+use sail_common_datafusion::datasource::{SinkInfo, SourceInfo, TableFormat};
+use sail_iceberg::IcebergTableFormat;
+
+/// Iceberg table format implementation that delegates to sail-iceberg
+#[derive(Debug)]
+pub struct IcebergDataSourceFormat {
+    inner: IcebergTableFormat,
+}
+
+impl Default for IcebergDataSourceFormat {
+    fn default() -> Self {
+        Self {
+            inner: IcebergTableFormat,
+        }
+    }
+}
+
+#[async_trait]
+impl TableFormat for IcebergDataSourceFormat {
+    fn name(&self) -> &str {
+        self.inner.name()
+    }
+
+    async fn create_provider(
+        &self,
+        ctx: &dyn Session,
+        info: SourceInfo,
+    ) -> Result<Arc<dyn TableProvider>> {
+        self.inner.create_provider(ctx, info).await
+    }
+
+    async fn create_writer(
+        &self,
+        ctx: &dyn Session,
+        info: SinkInfo,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        self.inner.create_writer(ctx, info).await
+    }
+}

crates/sail-data-source/src/formats/mod.rs

@@ -4,6 +4,7 @@ pub mod binary;
 pub mod console;
 pub mod csv;
 pub mod delta;
+pub mod iceberg;
 pub mod json;
 pub mod listing;
 pub mod parquet;

crates/sail-data-source/src/options/data/iceberg_read.yaml

@@ -0,0 +1,32 @@
+# Options for reading from an Apache Iceberg table.
+
+- key: use_ref
+  aliases:
+    - ref
+    - branch
+    - tag
+    - iceberg.ref
+  description: |
+    Select a snapshot reference (tag or branch) to time-travel when reading.
+    If unset, the current snapshot is used.
+  supported: true
+  rust_type: String
+
+- key: snapshot_id
+  aliases:
+    - snapshot-id
+    - snapshotId
+  description: |
+    Select a specific snapshot id to time-travel when reading.
+  supported: true
+  rust_type: i64
+  rust_deserialize_with: crate::options::serde::deserialize_i64
+
+- key: timestamp_as_of
+  aliases:
+    - timestampAsOf
+  description: |
+    Select snapshot as of the given timestamp. Accepts RFC3339 or 'yyyy-MM-dd HH:mm:ss.SSS'.
+  supported: true
+  rust_type: String
+

crates/sail-data-source/src/options/data/iceberg_write.yaml

@@ -0,0 +1,22 @@
+# Options for writing to an Apache Iceberg table.
+
+- key: overwrite_schema
+  aliases:
+    - overwriteSchema
+  description: |
+    If `true`, allows overwriting the schema of the table when using overwrite mode.
+  default: "false"
+  supported: true
+  rust_type: bool
+  rust_deserialize_with: crate::options::serde::deserialize_bool
+
+- key: merge_schema
+  aliases:
+    - mergeSchema
+  description: |
+    If `true`, allows automatic schema merging during an append or overwrite operation.
+  default: "false"
+  supported: true
+  rust_type: bool
+  rust_deserialize_with: crate::options::serde::deserialize_bool
+

crates/sail-data-source/src/registry.rs

@@ -11,6 +11,7 @@ use crate::formats::binary::BinaryTableFormat;
 use crate::formats::console::ConsoleTableFormat;
 use crate::formats::csv::CsvTableFormat;
 use crate::formats::delta::DeltaTableFormat;
+use crate::formats::iceberg::IcebergDataSourceFormat;
 use crate::formats::json::JsonTableFormat;
 use crate::formats::parquet::ParquetTableFormat;
 use crate::formats::rate::RateTableFormat;
@@ -42,6 +43,7 @@ impl TableFormatRegistry {
         registry.register_format(Arc::new(BinaryTableFormat::default()));
         registry.register_format(Arc::new(CsvTableFormat::default()));
         registry.register_format(Arc::new(DeltaTableFormat));
+        registry.register_format(Arc::new(IcebergDataSourceFormat::default()));
         registry.register_format(Arc::new(JsonTableFormat::default()));
         registry.register_format(Arc::new(ParquetTableFormat::default()));
         registry.register_format(Arc::new(TextTableFormat::default()));

crates/sail-iceberg/Cargo.toml

@@ -0,0 +1,39 @@
+[package]
+name = "sail-iceberg"
+version = { workspace = true }
+edition = { workspace = true }
+
+[dependencies]
+sail-common-datafusion = { path = "../sail-common-datafusion" }
+
+# DataFusion dependencies
+datafusion = { workspace = true }
+datafusion-common = { workspace = true }
+
+# Arrow dependencies
+arrow-schema = { workspace = true }
+
+# Essential utilities
+async-trait = { workspace = true }
+object_store = { workspace = true }
+chrono = { workspace = true }
+serde_json = { workspace = true }
+url = { workspace = true }
+futures = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+uuid = { workspace = true }
+# parquet = { workspace = true }
+# bytes = { workspace = true }
+# indexmap = { workspace = true }
+log = { workspace = true }
+# itertools = { workspace = true }
+percent-encoding = { workspace = true }
+once_cell = { workspace = true }
+ordered-float = { workspace = true }
+apache-avro = { workspace = true }
+num-bigint = { workspace = true }
+num-traits = { workspace = true }
+# base64 = { workspace = true }
+
+[lints]
+workspace = true