[BUG] skip metadata check for field equality

requirements-dev.txt

@@ -39,7 +39,7 @@ ray[data, client]==2.7.1; python_version >= '3.8'
 
 
 #Iceberg
-pyiceberg==0.5.1; python_version >= '3.8'
+pyiceberg==0.6.0; python_version >= '3.8'
 tenacity==8.2.3; python_version >= '3.8'
 
 # Delta Lake

src/daft-core/src/datatypes/field.rs

@@ -1,4 +1,5 @@
 use std::fmt::{Display, Formatter, Result};
+use std::hash::Hash;
 use std::sync::Arc;
 
 use arrow2::datatypes::Field as ArrowField;
@@ -10,7 +11,7 @@ use serde::{Deserialize, Serialize};
 
 pub type Metadata = std::collections::BTreeMap<String, String>;
 
-#[derive(Clone, Debug, PartialEq, Eq, Deserialize, Serialize, Hash)]
+#[derive(Clone, Debug, Eq, Deserialize, Serialize)]
 pub struct Field {
     pub name: String,
     pub dtype: DataType,
@@ -24,6 +25,21 @@ pub struct FieldID {
     pub id: Arc<str>,
 }
 
+impl PartialEq for Field {
+    fn eq(&self, other: &Self) -> bool {
+        // Skips metadata check
+        self.name == other.name && self.dtype == other.dtype
+    }
+}
+
+impl Hash for Field {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.name.hash(state);
+        self.dtype.hash(state);
+        // Skip hashing metadata
+    }
+}
+
 impl FieldID {
     pub fn new<S: Into<Arc<str>>>(id: S) -> Self {
         Self { id: id.into() }

tests/integration/iceberg/test_pyiceberg_written_table_load.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import pytest
+
+pyiceberg = pytest.importorskip("pyiceberg")
+
+import contextlib
+
+import pyarrow as pa
+
+import daft
+from tests.conftest import assert_df_equals
+
+
+@contextlib.contextmanager
+def table_written_by_pyiceberg(local_iceberg_catalog):
+    schema = pa.schema([("col", pa.int64()), ("mapCol", pa.map_(pa.int32(), pa.string()))])
+
+    data = {"col": [1, 2, 3], "mapCol": [[(1, "foo"), (2, "bar")], [(3, "baz")], [(4, "foobar")]]}
+    arrow_table = pa.Table.from_pydict(data, schema=schema)
+    try:
+        table = local_iceberg_catalog.create_table("pyiceberg.map_table", schema=schema)
+        table.append(arrow_table)
+        yield table
+    except Exception as e:
+        raise e
+    finally:
+        local_iceberg_catalog.drop_table("pyiceberg.map_table")
+
+
+@pytest.mark.integration()
+def test_localdb_catalog(local_iceberg_catalog):
+    with table_written_by_pyiceberg(local_iceberg_catalog) as catalog_table:
+        df = daft.read_iceberg(catalog_table)
+        daft_pandas = df.to_pandas()
+        iceberg_pandas = catalog_table.scan().to_arrow().to_pandas()
+        assert_df_equals(daft_pandas, iceberg_pandas, sort_key=[])