Propagate dlt column descriptions to Iceberg docs

src/dlt_iceberg/schema_converter.py

@@ -72,6 +72,7 @@ def convert_dlt_to_iceberg_schema(
             name=col_name,
             field_type=iceberg_type,
             required=required,
+            doc=dlt_col.get("description"),
         )
         fields.append(field)
         field_id += 1

src/dlt_iceberg/schema_evolution.py

@@ -138,6 +138,21 @@ def compare_schemas(
     return added_fields, type_changes, list(dropped_names)
 
 
+def _find_doc_updates(
+    existing_schema: Schema,
+    new_schema: Schema,
+) -> List[Tuple[str, str]]:
+    """Return non-null incoming field docs that differ from stored docs."""
+    existing_fields = {field.name: field for field in existing_schema.fields}
+    return [
+        (field.name, field.doc)
+        for field in new_schema.fields
+        if field.name in existing_fields
+        and field.doc is not None
+        and field.doc != existing_fields[field.name].doc
+    ]
+
+
 def validate_schema_changes(
     added_fields: List[NestedField],
     type_changes: List[Tuple[str, IcebergType, IcebergType]],
@@ -177,6 +192,7 @@ def apply_schema_evolution(
     added_fields: List[NestedField],
     type_changes: List[Tuple[str, IcebergType, IcebergType]],
     dropped_fields: Optional[List[str]] = None,
+    doc_updates: Optional[List[Tuple[str, str]]] = None,
 ) -> None:
     """
     Apply schema evolution changes to an Iceberg table.
@@ -186,15 +202,17 @@ def apply_schema_evolution(
         added_fields: New fields to add
         type_changes: Type promotions to apply
         dropped_fields: Fields to remove from the schema
+        doc_updates: Non-null column documentation updates to apply
     """
-    if not added_fields and not type_changes and not dropped_fields:
+    if not added_fields and not type_changes and not dropped_fields and not doc_updates:
         logger.info("No schema changes to apply")
         return
 
     logger.info(
         f"Applying schema evolution: "
         f"{len(added_fields)} new columns, {len(type_changes)} type promotions, "
-        f"{len(dropped_fields or [])} dropped columns"
+        f"{len(dropped_fields or [])} dropped columns, "
+        f"{len(doc_updates or [])} documentation updates"
     )
 
     # Apply changes using update_schema transaction
@@ -217,6 +235,11 @@ def apply_schema_evolution(
                 field_type=new_type
             )
 
+        # Apply documentation updates only when an incoming doc was provided.
+        for field_name, doc in (doc_updates or []):
+            logger.info(f"  Updating documentation for column: {field_name}")
+            update.update_column(path=field_name, doc=doc)
+
         # Delete dropped columns
         for field_name in (dropped_fields or []):
             logger.info(f"  Dropping column: {field_name}")
@@ -252,6 +275,7 @@ def evolve_schema_if_needed(
     added_fields, type_changes, dropped_fields = compare_schemas(
         existing_schema, new_schema
     )
+    doc_updates = _find_doc_updates(existing_schema, new_schema)
     missing_required_fields = (
         _required_dropped_fields(existing_schema, dropped_fields)
         if not allow_column_drops
@@ -263,6 +287,8 @@ def evolve_schema_if_needed(
         logger.info(f"Detected {len(added_fields)} new columns: {[f.name for f in added_fields]}")
     if type_changes:
         logger.info(f"Detected {len(type_changes)} type changes: {[(name, str(old), str(new)) for name, old, new in type_changes]}")
+    if doc_updates:
+        logger.info(f"Detected documentation updates for columns: {[name for name, _ in doc_updates]}")
     if dropped_fields:
         if allow_column_drops:
             logger.info(f"Detected {len(dropped_fields)} columns to drop: {dropped_fields}")
@@ -278,7 +304,7 @@ def evolve_schema_if_needed(
             )
 
     # If no changes, nothing to do
-    if not added_fields and not type_changes and not dropped_fields:
+    if not added_fields and not type_changes and not dropped_fields and not doc_updates:
         logger.debug("No schema changes detected")
         return False
 
@@ -293,13 +319,14 @@ def evolve_schema_if_needed(
 
     # When allow_column_drops=False and only dropped fields were detected,
     # the table schema is already correct — no evolution needed.
-    if not allow_column_drops and not added_fields and not type_changes:
+    if not allow_column_drops and not added_fields and not type_changes and not doc_updates:
         return False
 
     # Apply evolution, passing dropped_fields only when allow_column_drops=True
     apply_schema_evolution(
         table, added_fields, type_changes,
         dropped_fields=dropped_fields if allow_column_drops else None,
+        doc_updates=doc_updates,
     )
 
     return True

tests/test_schema_converter.py

@@ -37,7 +37,11 @@ def test_convert_dlt_to_iceberg_schema():
         "name": "test_table",
         "columns": {
             "id": {"data_type": "bigint", "nullable": False},
-            "name": {"data_type": "text", "nullable": True},
+            "name": {
+                "data_type": "text",
+                "nullable": True,
+                "description": "Display name for the account",
+            },
             "active": {"data_type": "bool", "nullable": True},
             "created_at": {"data_type": "timestamp", "nullable": False},
         },
@@ -77,9 +81,11 @@ def test_convert_dlt_to_iceberg_schema():
     name_field = [f for f in iceberg_schema.fields if f.name == "name"][0]
     assert isinstance(name_field.field_type, StringType)
     assert not name_field.required  # Nullable
+    assert name_field.doc == "Display name for the account"
 
     active_field = [f for f in iceberg_schema.fields if f.name == "active"][0]
     assert isinstance(active_field.field_type, BooleanType)
+    assert active_field.doc is None
 
     created_field = [f for f in iceberg_schema.fields if f.name == "created_at"][0]
     assert isinstance(created_field.field_type, TimestampType)

tests/test_schema_evolution.py

@@ -413,5 +413,112 @@ def update_schema(self):
     assert not table.update_schema_called
 
 
+class RecordingSchemaUpdate:
+    def __init__(self):
+        self.added_columns = []
+        self.updated_columns = []
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        return False
+
+    def add_column(self, **kwargs):
+        self.added_columns.append(kwargs)
+
+    def update_column(self, **kwargs):
+        self.updated_columns.append(kwargs)
+
+    def delete_column(self, path):
+        raise AssertionError("no column should be deleted")
+
+
+class RecordingTable:
+    def __init__(self, schema):
+        self._schema = schema
+        self.schema_update = RecordingSchemaUpdate()
+        self.update_schema_calls = 0
+
+    def schema(self):
+        return self._schema
+
+    def update_schema(self):
+        self.update_schema_calls += 1
+        return self.schema_update
+
+
+def test_evolve_schema_updates_non_null_changed_doc():
+    """A changed incoming description updates the existing Iceberg field doc."""
+    from dlt_iceberg.schema_evolution import evolve_schema_if_needed
+
+    table = RecordingTable(
+        Schema(NestedField(1, "name", StringType(), required=False, doc="Old comment"))
+    )
+    incoming_schema = Schema(
+        NestedField(1, "name", StringType(), required=False, doc="New comment")
+    )
+
+    assert evolve_schema_if_needed(table, incoming_schema)
+    assert table.update_schema_calls == 1
+    assert table.schema_update.updated_columns == [
+        {"path": "name", "doc": "New comment"}
+    ]
+
+
+def test_evolve_schema_adds_column_with_doc():
+    """A documented incoming column keeps its doc when added to Iceberg."""
+    from dlt_iceberg.schema_evolution import evolve_schema_if_needed
+
+    table = RecordingTable(
+        Schema(NestedField(1, "id", LongType(), required=True))
+    )
+    incoming_schema = Schema(
+        NestedField(1, "id", LongType(), required=True),
+        NestedField(
+            2,
+            "name",
+            StringType(),
+            required=False,
+            doc="Display name for the account",
+        ),
+    )
+
+    assert evolve_schema_if_needed(table, incoming_schema)
+    assert table.schema_update.added_columns == [
+        {
+            "path": "name",
+            "field_type": StringType(),
+            "required": False,
+            "doc": "Display name for the account",
+        }
+    ]
+
+
+@pytest.mark.parametrize("incoming_doc", [None, "Existing comment"])
+def test_evolve_schema_preserves_doc_without_changed_description(incoming_doc):
+    """Missing or unchanged descriptions do not overwrite an Iceberg field doc."""
+    from dlt_iceberg.schema_evolution import evolve_schema_if_needed
+
+    table = RecordingTable(
+        Schema(
+            NestedField(
+                1,
+                "name",
+                StringType(),
+                required=False,
+                doc="Existing comment",
+            )
+        )
+    )
+    incoming_schema = Schema(
+        NestedField(1, "name", StringType(), required=False, doc=incoming_doc)
+    )
+
+    assert not evolve_schema_if_needed(table, incoming_schema)
+    assert table.update_schema_calls == 0
+    assert table.schema_update.updated_columns == []
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-v", "-s"])