Fix CHAR/VARCHAR length overflow when writing reconcile intermediate data

When reconciling data from external sources (e.g., Teradata via Lakehouse
Federation), CHAR columns may contain space-padded values that exceed the
declared VARCHAR(n) length limit, causing DELTA_EXCEED_CHAR_VARCHAR_LIMIT
errors when writing intermediate DataFrames to Delta volumes or tables.

Cast all CHAR(n)/VARCHAR(n) columns to STRING (unbounded) before writing
to avoid length constraint violations.

Co-authored-by: Isaac

Author: moomindani

src/databricks/labs/lakebridge/reconcile/recon_capture.py

@@ -8,6 +8,7 @@
 
 from pyspark.sql import DataFrame, SparkSession
 from pyspark.sql.functions import col, collect_list, create_map, lit
+from pyspark.sql.types import StringType
 from pyspark.errors import PySparkException
 from sqlglot import Dialect
 
@@ -87,8 +88,23 @@ def _get_uc_volume_path(self):
             f"{self._metadata_config.volume}"
         )
 
+    @staticmethod
+    def _cast_char_varchar_to_string(df: DataFrame) -> DataFrame:
+        """Cast CHAR(n)/VARCHAR(n) columns to STRING to avoid length constraint violations.
+
+        When reconciling data from external sources (e.g., Teradata via Lakehouse Federation),
+        CHAR columns may contain space-padded values that exceed the declared length limit,
+        causing DELTA_EXCEED_CHAR_VARCHAR_LIMIT errors when writing to Delta.
+        """
+        for field in df.schema.fields:
+            type_name = field.dataType.simpleString().lower()
+            if type_name.startswith("varchar") or type_name.startswith("char"):
+                df = df.withColumn(field.name, col(field.name).cast(StringType()))
+        return df
+
     def _write_df_to_volumes(self, df: DataFrame, path: str) -> None:
         logger.debug(f"Writing DF on {self._format} to path: {path}")
+        df = self._cast_char_varchar_to_string(df)
         df.write.format(self._format).save(path)
         logger.info(f"Wrote DF on {self._format}")
 
@@ -114,6 +130,7 @@ def write_and_read_df_with_volumes(
 
 def _write_df_to_delta(df: DataFrame, table_name: str, mode="append"):
     try:
+        df = ReconIntermediatePersist._cast_char_varchar_to_string(df)
         df.write.mode(mode).saveAsTable(table_name)
         logger.info(f"Data written to {table_name} successfully.")
     except Exception as e:

tests/unit/reconcile/test_recon_capture.py

@@ -0,0 +1,73 @@
+from unittest.mock import MagicMock
+
+import pytest
+from pyspark.sql.types import StructType, StructField, StringType, IntegerType
+
+from databricks.labs.lakebridge.reconcile.recon_capture import ReconIntermediatePersist
+
+
+def _make_df(schema_fields):
+    """Create a mock DataFrame with given schema fields."""
+    schema = StructType(schema_fields)
+    df = MagicMock()
+    df.schema = schema
+    df.columns = [f.name for f in schema_fields]
+
+    def mock_with_column(name, col_expr):
+        # Return a new mock with updated schema (field cast to StringType)
+        new_fields = []
+        for f in df.schema.fields:
+            if f.name == name:
+                new_fields.append(StructField(name, StringType(), f.nullable))
+            else:
+                new_fields.append(f)
+        return _make_df(new_fields)
+
+    df.withColumn = mock_with_column
+    return df
+
+
+def test_cast_char_varchar_to_string_converts_varchar():
+    """VARCHAR(n) columns should be cast to STRING."""
+    df = _make_df([
+        StructField("id", IntegerType(), False),
+        StructField("name", StringType(), True),
+    ])
+    # Simulate VARCHAR(16) by overriding simpleString
+    varchar_type = MagicMock()
+    varchar_type.simpleString.return_value = "varchar(16)"
+    df.schema.fields[1].dataType = varchar_type
+
+    result = ReconIntermediatePersist._cast_char_varchar_to_string(df)
+
+    # The result should have STRING type for the 'name' field
+    assert result.schema.fields[1].dataType == StringType()
+
+
+def test_cast_char_varchar_to_string_converts_char():
+    """CHAR(n) columns should be cast to STRING."""
+    df = _make_df([
+        StructField("id", IntegerType(), False),
+        StructField("code", StringType(), True),
+    ])
+    char_type = MagicMock()
+    char_type.simpleString.return_value = "char(10)"
+    df.schema.fields[1].dataType = char_type
+
+    result = ReconIntermediatePersist._cast_char_varchar_to_string(df)
+
+    assert result.schema.fields[1].dataType == StringType()
+
+
+def test_cast_char_varchar_to_string_leaves_other_types():
+    """Non-CHAR/VARCHAR columns should not be modified."""
+    df = _make_df([
+        StructField("id", IntegerType(), False),
+        StructField("name", StringType(), True),
+    ])
+
+    result = ReconIntermediatePersist._cast_char_varchar_to_string(df)
+
+    # Schema should be unchanged
+    assert result.schema.fields[0].dataType == IntegerType()
+    assert result.schema.fields[1].dataType == StringType()