[Iceberg] Add basic iceberg integration

.github/workflows/ci.yml

@@ -118,6 +118,15 @@ jobs:
           key: "data"
       - run: cargo test --features clickbench --test clickbench_plans_test
 
+  iceberg-test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          lfs: true
+      - uses: ./.github/actions/setup
+      - run: cargo test -p datafusion-distributed-iceberg
+
   format-check:
     runs-on: ubuntu-latest
     steps:

iceberg/AGENTS.md

@@ -0,0 +1,12 @@
+# Iceberg crate guide
+
+## Tests
+
+- Reuse `src/test_utils/harness.rs` and `testdata/iceberg/taxi` as much as
+  possible.
+- Prefer integration tests over unit tests when possible. Each isolated feature
+  should have its own dedicated file in `tests/`.
+- Never create long tests with a lot of setup, the pattern should be several
+  tests with a very small body, and have any necessary helper at the bottom
+  of the `mod tests {};` block, below the actual tests.
+- Prefer snapshot testing with `insta` when possible.

iceberg/README.md

@@ -0,0 +1,34 @@
+# DataFusion Distributed Iceberg
+
+Read-only Apache Iceberg tables for DataFusion Distributed.
+
+```rust
+use datafusion::execution::SessionStateBuilder;
+use datafusion::prelude::SessionContext;
+use datafusion_distributed_iceberg::{IcebergExt, IcebergIntegrationOptions};
+
+async fn example() -> datafusion::error::Result<()> {
+    let state = SessionStateBuilder::new()
+        .with_default_features()
+        .with_iceberg_integration(IcebergIntegrationOptions::default())
+        .build();
+    let ctx = SessionContext::new_with_state(state);
+
+    ctx.sql(
+        "CREATE EXTERNAL TABLE taxi STORED AS ICEBERG \
+     LOCATION 's3://warehouse/taxi/metadata/v1.metadata.json'",
+    )
+        .await?
+        .collect()
+        .await?;
+    Ok(())
+}
+```
+
+The default storage factory resolves `file://`, S3 (`s3://`, `s3a://`,
+`s3n://`), and GCS (`gs://`, `gcs://`) URIs. Use
+`IcebergIntegrationOptions` to supply custom storage or an Iceberg runtime.
+
+```bash
+cargo test -p datafusion-distributed-iceberg
+```

iceberg/src/codec.rs

@@ -0,0 +1,65 @@
+use crate::work_unit_feed::FileScanTaskMessage;
+use bytes::{Buf, BufMut};
+use datafusion::common::Result;
+use datafusion::execution::TaskContext;
+use datafusion::physical_plan::ExecutionPlan;
+use datafusion_proto::physical_plan::PhysicalExtensionCodec;
+use prost::encoding::{DecodeContext, WireType};
+use prost::{DecodeError, Message};
+use std::sync::Arc;
+
+#[derive(Debug)]
+pub struct IcebergCodec;
+
+// TODO: Implement protobuf codecs for the IcebergDataSource.
+impl PhysicalExtensionCodec for IcebergCodec {
+    fn try_decode(
+        &self,
+        _buf: &[u8],
+        _inputs: &[Arc<dyn ExecutionPlan>],
+        _ctx: &TaskContext,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        unimplemented!()
+    }
+
+    fn try_encode(&self, _node: Arc<dyn ExecutionPlan>, _buf: &mut Vec<u8>) -> Result<()> {
+        unimplemented!()
+    }
+}
+
+// TODO: Implement serde for FileScanTaskMessage
+//  This message is an individual WorkUnit, but it cannot be serialized yet. During distributed
+//  execution, this will be streamed over the wire from coordinator to workers, but for that to
+//  happen, it will need to be represented as a prost::Message.
+//
+// WARNING: for the ones who end up implementing this. I have no idea if implementing Message here
+//  is really the best option for serialization, it might not be.
+impl Message for FileScanTaskMessage {
+    fn encode_raw(&self, _buf: &mut impl BufMut)
+    where
+        Self: Sized,
+    {
+        unimplemented!()
+    }
+
+    fn merge_field(
+        &mut self,
+        _tag: u32,
+        _wire_type: WireType,
+        _buf: &mut impl Buf,
+        _ctx: DecodeContext,
+    ) -> std::result::Result<(), DecodeError>
+    where
+        Self: Sized,
+    {
+        unimplemented!()
+    }
+
+    fn encoded_len(&self) -> usize {
+        unimplemented!()
+    }
+
+    fn clear(&mut self) {
+        unimplemented!()
+    }
+}

iceberg/src/common/error.rs

@@ -0,0 +1,105 @@
+use datafusion::error::DataFusionError;
+use iceberg::{Error, ErrorKind};
+
+/// Converts a datafusion error into an iceberg error.
+pub fn iceberg_err(error: DataFusionError) -> Error {
+    let fallback_message = error.to_string();
+    let DataFusionError::Context(ctx, err) = error else {
+        return Error::new(
+            ErrorKind::Unexpected,
+            format!("DataFusion execution failed: {fallback_message}"),
+        );
+    };
+
+    let Some(kind) = parse_iceberg_error_kind(&ctx) else {
+        return Error::new(
+            ErrorKind::Unexpected,
+            format!("DataFusion execution failed: {fallback_message}"),
+        );
+    };
+    let DataFusionError::Execution(message) = err.as_ref() else {
+        return Error::new(
+            ErrorKind::Unexpected,
+            format!("DataFusion execution failed: {fallback_message}"),
+        );
+    };
+
+    Error::new(kind, strip_error_kind(kind, message))
+}
+
+/// Converts an Iceberg error into a DataFusion error.
+pub fn df_err(error: Error) -> DataFusionError {
+    DataFusionError::Context(
+        format!("IcebergError({})", error.kind()),
+        Box::new(DataFusionError::Execution(error.to_string())),
+    )
+}
+
+fn parse_iceberg_error_kind(context: &str) -> Option<ErrorKind> {
+    let kind = context.strip_prefix("IcebergError(")?.strip_suffix(')')?;
+
+    match kind {
+        "PreconditionFailed" => Some(ErrorKind::PreconditionFailed),
+        "Unexpected" => Some(ErrorKind::Unexpected),
+        "DataInvalid" => Some(ErrorKind::DataInvalid),
+        "NamespaceAlreadyExists" => Some(ErrorKind::NamespaceAlreadyExists),
+        "TableAlreadyExists" => Some(ErrorKind::TableAlreadyExists),
+        "NamespaceNotFound" => Some(ErrorKind::NamespaceNotFound),
+        "TableNotFound" => Some(ErrorKind::TableNotFound),
+        "FeatureUnsupported" => Some(ErrorKind::FeatureUnsupported),
+        "CatalogCommitConflicts" => Some(ErrorKind::CatalogCommitConflicts),
+        _ => None,
+    }
+}
+
+fn strip_error_kind(kind: ErrorKind, message: &str) -> String {
+    let kind = kind.into_static();
+    if message == kind {
+        String::new()
+    } else {
+        message
+            .strip_prefix(&format!("{kind} => "))
+            .unwrap_or(message)
+            .to_string()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use datafusion::error::DataFusionError;
+    use iceberg::{Error, ErrorKind};
+
+    use super::{df_err, iceberg_err};
+
+    #[test]
+    fn roundtrips_iceberg_error_kind_and_message() {
+        let error = Error::new(ErrorKind::DataInvalid, "invalid manifest");
+        let roundtripped = iceberg_err(df_err(error));
+
+        assert_eq!(roundtripped.kind(), ErrorKind::DataInvalid);
+        assert_eq!(roundtripped.to_string(), "DataInvalid => invalid manifest");
+    }
+
+    #[test]
+    fn encodes_iceberg_errors_with_native_datafusion_variants() {
+        let error = df_err(Error::new(ErrorKind::DataInvalid, "invalid manifest"));
+
+        assert!(matches!(
+            error,
+            DataFusionError::Context(context, inner)
+                if context == "IcebergError(DataInvalid)"
+                    && matches!(inner.as_ref(), DataFusionError::Execution(message) if message == "DataInvalid => invalid manifest")
+        ));
+    }
+
+    #[test]
+    fn maps_non_iceberg_datafusion_errors_to_unexpected() {
+        let error = iceberg_err(DataFusionError::Execution("worker failed".to_string()));
+
+        assert_eq!(error.kind(), ErrorKind::Unexpected);
+        assert_eq!(
+            error.to_string(),
+            "Unexpected => DataFusion execution failed: Execution error: worker failed"
+        );
+    }
+}