diff --git a/.bumpversion.cfg b/.bumpversion.cfg index 58b09366..efdb1af0 100644 --- a/.bumpversion.cfg +++ b/.bumpversion.cfg @@ -1,5 +1,5 @@ [bumpversion] -current_version = 0.14.0 +current_version = 0.14.1 commit = False tag = False parse = (?P\d+)\.(?P\d+)\.(?P\d+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 99379fb7..363c5d11 100644 --- a/CHANGELOG.md +++ b/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) | diff --git a/rdsa_utils/__init__.py b/rdsa_utils/__init__.py index 9e78220f..f075dd36 100644 --- a/rdsa_utils/__init__.py +++ b/rdsa_utils/__init__.py @@ -1 +1 @@ -__version__ = "0.14.0" +__version__ = "0.14.1" diff --git a/rdsa_utils/cdp/io/output.py b/rdsa_utils/cdp/io/output.py index 529af6f6..4b42f7bb 100644 --- a/rdsa_utils/cdp/io/output.py +++ b/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 diff --git a/tests/cdp/io/test_cdsw_output.py b/tests/cdp/io/test_cdsw_output.py index aefac2e0..9f2cd241 100644 --- a/tests/cdp/io/test_cdsw_output.py +++ b/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, )