fix: Prevent Oracle LOB columns from causing exceptions in Column and Row validation

data_validation/config_manager.py

@@ -66,6 +66,9 @@ def __init__(self, config, source_client=None, target_client=None, verbose=False
         if self.validation_type not in consts.CONFIG_TYPES:
             raise ValueError(f"Unknown Configuration Type: {self.validation_type}")
         self._comparison_max_col_length = None
+        # For some engines we need to know the actual raw data type rather than the Ibis canonical type.
+        self._source_raw_data_types = None
+        self._target_raw_data_types = None
 
     @property
     def config(self):
@@ -842,7 +845,12 @@ def _key_column_needs_casting_to_string(
         )
 
     def build_config_column_aggregates(
-        self, agg_type, arg_value, exclude_cols, supported_types, cast_to_bigint=False
+        self,
+        agg_type: str,
+        arg_value: list,
+        exclude_cols: bool,
+        supported_types: list,
+        cast_to_bigint=False,
     ):
         """Return list of aggregate objects of given agg_type."""
 
@@ -858,13 +866,7 @@ def require_pre_agg_calc_field(
             ]:
                 return True
             elif column_type in ["binary", "!binary"]:
-                if agg_type == "count":
-                    # Oracle BLOB is invalid for use with SQL COUNT function.
-                    return bool(
-                        self.source_client.name == "oracle"
-                        or self.target_client.name == "oracle"
-                    )
-                else:
+                if agg_type != "count":
                     # Convert to length for any min/max/sum on binary columns.
                     return True
             elif cast_to_bigint and column_type in ["int32", "!int32"]:
@@ -946,6 +948,16 @@ def require_pre_agg_calc_field(
                         f"Skipping {agg_type} on {column} due to data type: {column_type}"
                     )
                 continue
+            # Oracle LOB columns are invalid for use with SQL COUNT function.
+            elif agg_type == "count" and self._is_oracle_lob(column):
+                # TODO Eventually we need to allow COUNT(Oracle LOB) by using a CASE expression:
+                #      COUNT(CASE WHEN clob_col IS NULL THEN NULL ELSE 1 END)
+                #      For now we skip them.
+                if self.verbose:
+                    logging.info(
+                        f"Skipping {agg_type} on {column} due to Oracle LOB data type"
+                    )
+                continue
 
             if require_pre_agg_calc_field(
                 column_type, target_column_type, agg_type, cast_to_bigint
@@ -1010,6 +1022,69 @@ def _get_comparison_max_col_length(self) -> int:
             )
         return self._comparison_max_col_length
 
+    def _get_source_raw_data_types(self) -> dict:
+        """Return a dict of column name: raw data type.
+
+        The raw data type is the source/target engine type, for example it might
+        be "NCLOB" when the Ibis type states "string"."""
+        if self._source_raw_data_types is None:
+            if clients.is_oracle_client(self.source_client):
+                raw_data_types = self.source_client.raw_column_metadata(
+                    database=self.source_schema,
+                    table=self.source_table,
+                    query=self.source_query,
+                )
+                self._source_raw_data_types = {
+                    _.casefold(): raw_data_types[_] for _ in raw_data_types
+                }
+            else:
+                self._source_raw_data_types = {}
+        return self._source_raw_data_types
+
+    def _get_target_raw_data_types(self) -> dict:
+        """Return a dict of column name: raw data type.
+
+        The raw data type is the source/target engine type, for example it might
+        be "NCLOB" when the Ibis type states "string"."""
+        if self._target_raw_data_types is None:
+            if clients.is_oracle_client(self.target_client):
+                raw_data_types = self.target_client.raw_column_metadata(
+                    database=self.target_schema,
+                    table=self.target_table,
+                    query=self.target_query,
+                )
+                self._target_raw_data_types = {
+                    _.casefold(): raw_data_types[_] for _ in raw_data_types
+                }
+            else:
+                self._target_raw_data_types = {}
+        return self._target_raw_data_types
+
+    def _is_oracle_lob(self, casefold_column_name: str) -> bool:
+        """Checks if a column in either the source or target database is an Oracle LOB data type.
+
+        Args:
+          casefold_column_name: The case-insensitive name of the column to check.
+        """
+        return bool(
+            self._get_source_raw_data_types()
+            .get(casefold_column_name, "")
+            .endswith("LOB")
+            or self._get_target_raw_data_types()
+            .get(casefold_column_name, "")
+            .endswith("LOB")
+        )
+
+    def _exclude_oracle_lob_columns(self, casefold_column_names: list):
+        """Remove LOB columns from validation to avoid ORA-00932 errors."""
+        oracle_lob_cols = [_ for _ in casefold_column_names if self._is_oracle_lob(_)]
+        if oracle_lob_cols:
+            if self.verbose:
+                logging.info(f"Skipping Oracle LOB columns: {str(oracle_lob_cols)}")
+            return [_ for _ in casefold_column_names if _ not in oracle_lob_cols]
+        else:
+            return casefold_column_names
+
     def _strftime_format(
         self, column_type: Union[dt.Date, dt.Timestamp], client
     ) -> str:
@@ -1124,7 +1199,8 @@ def build_dependent_aliases(
         col_names = []
         for i, calc in enumerate(self._get_order_of_operations(calc_type)):
             if i == 0:
-                previous_level = [x for x in casefold_source_columns.keys()]
+                previous_level = [_ for _ in casefold_source_columns.keys()]
+                previous_level = self._exclude_oracle_lob_columns(previous_level)
             else:
                 previous_level = [k for k, v in column_aliases.items() if v == i - 1]
             if calc in ["concat", "hash"]:

tests/resources/oracle_test_tables.sql

@@ -105,7 +105,13 @@ CREATE TABLE pso_data_validator.dvt_ora2pg_types
 ,   col_blob        BLOB
 ,   col_clob        CLOB
 ,   col_nclob       NCLOB
+,   col_json        CLOB
+,   col_jsonb       CLOB
 );
+ALTER TABLE pso_data_validator.dvt_ora2pg_types
+ADD CONSTRAINT dvt_ora2pg_types_chk1 CHECK (col_json IS JSON) ENABLE;
+ALTER TABLE pso_data_validator.dvt_ora2pg_types
+ADD CONSTRAINT dvt_ora2pg_types_chk2 CHECK (col_jsonb IS JSON) ENABLE;
 COMMENT ON TABLE pso_data_validator.dvt_ora2pg_types IS 'Oracle to PostgreSQL integration test table';
 
 -- Literals below match corresponding table in postgresql_test_tables.sql
@@ -121,6 +127,7 @@ INSERT INTO pso_data_validator.dvt_ora2pg_types VALUES
 ,INTERVAL '1 2:03:44.0' DAY TO SECOND(3)
 ,UTL_RAW.CAST_TO_RAW('DVT'),UTL_RAW.CAST_TO_RAW('DVT')
 ,UTL_RAW.CAST_TO_RAW('DVT'),'DVT A','DVT A'
+,'{"dvt": 123, "status": "abc"}','{"dvt": 123, "status": "abc"}'
 );
 INSERT INTO pso_data_validator.dvt_ora2pg_types VALUES
 (2,2222,123456789,123456789012345678,1234567890123456789012345
@@ -134,6 +141,7 @@ INSERT INTO pso_data_validator.dvt_ora2pg_types VALUES
 ,INTERVAL '2 3:04:55.666' DAY TO SECOND(3)
 ,UTL_RAW.CAST_TO_RAW('DVT'),UTL_RAW.CAST_TO_RAW('DVT DVT')
 ,UTL_RAW.CAST_TO_RAW('DVT DVT'),'DVT B','DVT B'
+,'{"dvt": 234, "status": "def"}','{"dvt": 234, "status": "def"}'
 );
 INSERT INTO pso_data_validator.dvt_ora2pg_types VALUES
 (3,3333,123456789,123456789012345678,1234567890123456789012345
@@ -147,6 +155,7 @@ INSERT INTO pso_data_validator.dvt_ora2pg_types VALUES
 ,INTERVAL '3 4:05:06.7' DAY TO SECOND(3)
 ,UTL_RAW.CAST_TO_RAW('DVT'),UTL_RAW.CAST_TO_RAW('DVT DVT DVT')
 ,UTL_RAW.CAST_TO_RAW('DVT DVT DVT'),'DVT C','DVT C'
+,'{"dvt": 345, "status": "ghi"}','{"dvt": 345, "status": "ghi"}'
 );
 COMMIT;
 

tests/resources/postgresql_test_tables.sql

@@ -79,6 +79,8 @@ CREATE TABLE pso_data_validator.dvt_ora2pg_types
 ,   col_blob        bytea
 ,   col_clob        text
 ,   col_nclob       text
+,   col_json        json
+,   col_jsonb       jsonb
 );
 COMMENT ON TABLE pso_data_validator.dvt_ora2pg_types IS 'Oracle to PostgreSQL integration test table';
 
@@ -94,7 +96,8 @@ INSERT INTO pso_data_validator.dvt_ora2pg_types VALUES
 ,TIMESTAMP WITH TIME ZONE'1970-01-01 00:00:01.123456 +00:00'
 ,INTERVAL '1 2:03:44.0' DAY TO SECOND(3)
 ,CAST('DVT' AS BYTEA),CAST('DVT' AS BYTEA)
-,CAST('DVT' AS BYTEA),'DVT A','DVT A')
+,CAST('DVT' AS BYTEA),'DVT A','DVT A'
+,'{"dvt": 123, "status": "abc"}','{"dvt": 123, "status": "abc"}')
 ,(2,2222,123456789,123456789012345678,1234567890123456789012345
 ,123.12,123.11
 --,123400,0.002
@@ -105,7 +108,8 @@ INSERT INTO pso_data_validator.dvt_ora2pg_types VALUES
 ,TIMESTAMP WITH TIME ZONE'1970-01-02 00:00:02.123456 -02:00'
 ,INTERVAL '2 3:04:55.666' DAY TO SECOND(3)
 ,CAST('DVT' AS BYTEA),CAST('DVT DVT' AS BYTEA)
-,CAST('DVT DVT' AS BYTEA),'DVT B','DVT B')
+,CAST('DVT DVT' AS BYTEA),'DVT B','DVT B'
+,'{"dvt": 234, "status": "def"}','{"dvt": 234, "status": "def"}')
 ,(3,3333,123456789,123456789012345678,1234567890123456789012345
 ,123.123,123.11
 --,123400,0.003
@@ -117,6 +121,7 @@ INSERT INTO pso_data_validator.dvt_ora2pg_types VALUES
 ,INTERVAL '3 4:05:06.7' DAY TO SECOND(3)
 ,CAST('DVT' AS BYTEA),CAST('DVT DVT DVT' AS BYTEA)
 ,CAST('DVT DVT DVT' AS BYTEA),'DVT C','DVT C'
+,'{"dvt": 345, "status": "ghi"}','{"dvt": 345, "status": "ghi"}'
 );
 
  /* Following table used for validating generating table partitions */

tests/system/data_sources/test_db2.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import os
+import pytest
 from unittest import mock
 
 from data_validation import cli_tools
@@ -137,6 +138,26 @@ def test_column_validation_core_types_to_bigquery():
     )
 
 
+@mock.patch(
+    "data_validation.state_manager.StateManager.get_connection_config",
+    new=mock_get_connection_config,
+)
+def test_column_validation_dvt_binary_to_bigquery():
+    """DB2 to BigQuery dvt_binary column validation"""
+    # Remove skip of this test when working on issue-1354
+    pytest.skip(
+        "Skipping test_column_validation_dvt_binary_to_bigquery until issue-1354 is actioned"
+    )
+    column_validation_test(
+        tc="bq-conn",
+        tables="db2inst1.dvt_binary=pso_data_validator.dvt_binary",
+        count_cols="*",
+        sum_cols="*",
+        min_cols="*",
+        max_cols="*",
+    )
+
+
 @mock.patch(
     "data_validation.state_manager.StateManager.get_connection_config",
     new=mock_get_connection_config,

tests/system/data_sources/test_mysql.py

@@ -246,6 +246,22 @@ def test_column_validation_core_types_to_bigquery():
     )
 
 
+@mock.patch(
+    "data_validation.state_manager.StateManager.get_connection_config",
+    new=mock_get_connection_config,
+)
+def test_column_validation_dvt_binary_to_bigquery():
+    """MySQL to BigQuery dvt_binary column validation"""
+    column_validation_test(
+        tc="bq-conn",
+        tables="pso_data_validator.dvt_binary",
+        count_cols="*",
+        sum_cols="*",
+        min_cols="*",
+        max_cols="*",
+    )
+
+
 @mock.patch(
     "data_validation.state_manager.StateManager.get_connection_config",
     new=mock_get_connection_config,

tests/system/data_sources/test_oracle.py

@@ -100,6 +100,8 @@
     "col_blob",
     "col_clob",
     "col_nclob",
+    "col_json",
+    "col_jsonb",
 ]
 
 
@@ -238,8 +240,10 @@ def test_schema_validation_not_null_vs_nullable():
 def test_schema_validation_oracle_to_postgres():
     """Oracle to PostgreSQL schema validation"""
     schema_validation_test(
-        tables="pso_data_validator.dvt_core_types",
+        tables="pso_data_validator.dvt_ora2pg_types",
         tc="pg-conn",
+        # TODO We should be able to remove allow_list below when actioning issue-1338.
+        allow_list="string:json",
         allow_list_file="samples/allow_list/oracle_to_postgres.yaml",
     )
 
@@ -313,6 +317,22 @@ def test_column_validation_oracle_to_postgres():
     )
 
 
+@mock.patch(
+    "data_validation.state_manager.StateManager.get_connection_config",
+    new=mock_get_connection_config,
+)
+def test_column_validation_dvt_binary_to_bigquery():
+    """Oracle to BigQuery dvt_binary column validation"""
+    column_validation_test(
+        tc="bq-conn",
+        tables="pso_data_validator.dvt_binary",
+        count_cols="*",
+        sum_cols="*",
+        min_cols="*",
+        max_cols="*",
+    )
+
+
 @mock.patch(
     "data_validation.state_manager.StateManager.get_connection_config",
     new=mock_get_connection_config,
@@ -382,9 +402,6 @@ def test_row_validation_oracle_to_postgres():
             for _ in ORA2PG_COLUMNS
             if _
             not in (
-                "col_blob",
-                "col_clob",
-                "col_nclob",
                 "col_raw",
                 "col_long_raw",
                 "col_float32",
@@ -600,9 +617,6 @@ def test_custom_query_row_validation_oracle_to_postgres():
             for _ in ORA2PG_COLUMNS
             if _
             not in (
-                "col_blob",
-                "col_clob",
-                "col_nclob",
                 "col_raw",
                 "col_long_raw",
                 "col_float32",

tests/system/data_sources/test_postgres.py

@@ -663,6 +663,22 @@ def test_column_validation_core_types_to_bigquery():
     )
 
 
+@mock.patch(
+    "data_validation.state_manager.StateManager.get_connection_config",
+    new=mock_get_connection_config,
+)
+def test_column_validation_dvt_binary_to_bigquery():
+    """PostgreSQL to BigQuery dvt_binary column validation"""
+    column_validation_test(
+        tc="bq-conn",
+        tables="pso_data_validator.dvt_binary",
+        count_cols="*",
+        sum_cols="*",
+        min_cols="*",
+        max_cols="*",
+    )
+
+
 @mock.patch(
     "data_validation.state_manager.StateManager.get_connection_config",
     new=mock_get_connection_config,

tests/system/data_sources/test_sql_server.py

@@ -337,6 +337,22 @@ def test_column_validation_core_types_to_bigquery():
     )
 
 
+@mock.patch(
+    "data_validation.state_manager.StateManager.get_connection_config",
+    new=mock_get_connection_config,
+)
+def test_column_validation_dvt_binary_to_bigquery():
+    """SQL Server to BigQuery dvt_binary column validation"""
+    column_validation_test(
+        tc="bq-conn",
+        tables="pso_data_validator.dvt_binary",
+        count_cols="*",
+        sum_cols="*",
+        min_cols="*",
+        max_cols="*",
+    )
+
+
 @mock.patch(
     "data_validation.state_manager.StateManager.get_connection_config",
     new=mock_get_connection_config,