Release 0.14.1

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.14.0
+current_version = 0.14.1
 commit = False
 tag = False
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)

CHANGELOG.md

@@ -17,6 +17,20 @@ and this project adheres to [semantic versioning](https://semver.org/spec/v2.0.0
 
 ### Removed
 
+## [0.14.1] - 2025-08-13
+
+### Added
+
+### Changed
+- Updated `insert_df_to_hive_table` function in `cdp/io/output.py`; enhanced schema
+  mismatch error logging to show specific column differences.
+
+### Deprecated
+
+### Fixed
+
+### Removed
+
 ## [0.14.0] - 2025-08-06
 
 ### Added
@@ -886,6 +900,8 @@ and this project adheres to [semantic versioning](https://semver.org/spec/v2.0.0
 > due to bugs in the GitHub Action `deploy_pypi.yaml`, which deploys to PyPI
 > and GitHub Releases.
 
+- rdsa-utils v0.14.1: [GitHub Release](https://github.com/ONSdigital/rdsa-utils/releases/tag/v0.14.1) |
+  [PyPI](https://pypi.org/project/rdsa-utils/0.14.1/)
 - rdsa-utils v0.14.0: [GitHub Release](https://github.com/ONSdigital/rdsa-utils/releases/tag/v0.14.0) |
   [PyPI](https://pypi.org/project/rdsa-utils/0.14.0/)
 - rdsa-utils v0.13.3: [GitHub Release](https://github.com/ONSdigital/rdsa-utils/releases/tag/v0.13.3) |

rdsa_utils/__init__.py

@@ -1 +1 @@
-__version__ = "0.14.0"
+__version__ = "0.14.1"

rdsa_utils/cdp/io/output.py

@@ -162,22 +162,39 @@ def insert_df_to_hive_table(
         msg = f"Cannot write an empty SparkDF to {table_name}"
         raise DataframeEmptyError(msg)
 
-    # Handle missing columns if specified
-    if fill_missing_cols and table_exists:
-        missing_columns = list(set(table_columns) - set(df.columns))
-        for col in missing_columns:
-            column_type = [
-                field.dataType for field in table_schema if field.name == col
-            ][0]
-            df = df.withColumn(col, F.lit(None).cast(column_type))
-    elif not fill_missing_cols and table_exists:
-        # Validate schema before writing
-        if set(table_columns) != set(df.columns):
-            msg = (
-                f"SparkDF schema does not match table {table_name} "
-                f"schema and 'fill_missing_cols' is False."
-            )
-            raise ValueError(msg)
+    # Handle schema validation and alignment
+    if table_exists:
+        df_cols_set = set(df.columns)
+        table_cols_set = set(table_columns)
+
+        if df_cols_set != table_cols_set:
+            if fill_missing_cols:
+                missing_in_df = list(table_cols_set - df_cols_set)
+                logger.info(f"Adding missing columns with null values: {missing_in_df}")
+                for col_name in missing_in_df:
+                    # Find the correct data type from the target table's schema
+                    col_type = next(
+                        field.dataType
+                        for field in table_schema
+                        if field.name == col_name
+                    )
+                    df = df.withColumn(col_name, F.lit(None).cast(col_type))
+            else:
+                missing_in_df = sorted(table_cols_set - df_cols_set)
+                extra_in_df = sorted(df_cols_set - table_cols_set)
+
+                error_msg = [
+                    f"Schema mismatch for table '{table_name}' with "
+                    "'fill_missing_cols=False'.",
+                ]
+                if missing_in_df:
+                    error_msg.append(
+                        f"  - Columns missing from DataFrame: {missing_in_df}",
+                    )
+                if extra_in_df:
+                    error_msg.append(f"  - Extra columns in DataFrame: {extra_in_df}")
+
+                raise ValueError("\n".join(error_msg))
 
     # Ensure column order
     df = df.select(table_columns) if table_exists else df

tests/cdp/io/test_cdsw_output.py

@@ -1,5 +1,6 @@
 """Tests for the cdp/io/output.py module."""
 
+import re
 from typing import Callable
 from unittest.mock import MagicMock, Mock, patch
 
@@ -129,25 +130,62 @@ def test_insert_df_to_hive_table_with_missing_columns(
         # Compare the exact column expression (the expected one, not the mock)
         mock_with_column.assert_any_call("address", expected_column)
 
+    @patch("rdsa_utils.cdp.io.output.is_df_empty", return_value=False)
     @patch("pyspark.sql.DataFrameReader.table")
-    def test_insert_df_to_hive_table_without_missing_columns(
+    def test_insert_df_to_hive_table_schema_mismatch(
         self,
         mock_table,
+        mock_is_empty,
         spark_session: SparkSession,
-        test_df: SparkDF,
+        caplog,
+        test_df,
     ) -> None:
-        """Test that insert_df_to_hive_table raises a ValueError when
-        'fill_missing_cols' is False and DataFrame schema doesn't match with the
-        table schema.
+        """Test ValueError for schema mismatch with 'fill_missing_cols=False'.
+
+        This test verifies that the function correctly identifies and reports a
+        schema mismatch between the source DataFrame and the target Hive table
+        when automatic column filling is disabled.
+
+        The specific mismatch scenario is configured as follows:
+        - DataFrame columns: ['id', 'name', 'age']
+        - Mock Hive table columns: ['id', 'name', 'address', 'status']
+
+        This creates a difference where:
+        - Columns 'address' and 'status' are missing from the DataFrame.
+        - Column 'age' is an extra column in the DataFrame that is not
+          present in the target table.
+
+        The test asserts that a `ValueError` is raised and that its message
+        accurately details these specific column differences.
         """
-        table_name = "test_table"
-        # Mock the table columns
-        mock_table.return_value.columns = ["id", "name", "age", "address"]
-        with pytest.raises(ValueError):
+        table_name = "test_db.mismatched_table"
+
+        mock_table_schema = T.StructType(
+            [
+                T.StructField("id", T.IntegerType()),
+                T.StructField("name", T.StringType()),
+                T.StructField("address", T.StringType()),
+                T.StructField("status", T.StringType()),
+            ],
+        )
+        mock_hive_df = MagicMock()
+        mock_hive_df.schema = mock_table_schema
+        mock_hive_df.columns = ["id", "name", "address", "status"]
+        mock_table.return_value = mock_hive_df
+
+        expected_error_pattern = re.escape(
+            f"Schema mismatch for table '{table_name}' with 'fill_missing_cols=False'.\n"
+            f"  - Columns missing from DataFrame: ['address', 'status']\n"
+            f"  - Extra columns in DataFrame: ['age']",
+        )
+
+        caplog.set_level(logging.INFO)
+
+        with pytest.raises(ValueError, match=expected_error_pattern):
             insert_df_to_hive_table(
-                spark_session,
-                test_df,
-                table_name,
+                spark=spark_session,
+                df=test_df,
+                table_name=table_name,
                 fill_missing_cols=False,
             )