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):
@@ -840,7 +843,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."""
 
@@ -856,13 +864,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"]:
@@ -944,6 +946,15 @@ def require_pre_agg_calc_field(
                         f"Skipping {agg_type} on {column} due to data type: {column_type}"
                     )
                 continue
+            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
@@ -1008,6 +1019,66 @@ 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_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_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:
+        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
+
+        Not using set() - set() because we want to retain the order of the list."""
+        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:
@@ -1122,7 +1193,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/common_functions.py

@@ -166,8 +166,8 @@ def schema_validation_test(
     tables="pso_data_validator.dvt_core_types",
     tc="bq-conn",
     exclusion_columns="id",
-    allow_list=None,
-    allow_list_file=None,
+    allow_list: str = None,
+    allow_list_file: str = None,
 ):
     """Generic schema validation test.
 

tests/system/data_sources/test_oracle.py

@@ -99,6 +99,8 @@
     "col_blob",
     "col_clob",
     "col_nclob",
+    "col_json",
+    "col_jsonb",
 ]
 
 
@@ -237,8 +239,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",
     )
 
@@ -381,9 +385,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",
@@ -599,9 +600,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/unit/test_config_manager.py

@@ -152,6 +152,22 @@ def table(self, table, database=None):
         return MockIbisTable()
 
 
+class MockOracleClient(object):
+    _source_type = "Oracle"
+    name = "oracle"
+
+    def table(self, table, database=None):
+        return MockIbisTable()
+
+    def raw_metadata(self, database: str = None, table: str = None, query: str = None):
+        return {
+            "a": "NUMBER",
+            "b": "VARCHAR2",
+            "c": "DATE",
+            "d": "NCLOB",
+        }
+
+
 class MockIbisTable(object):
     def __init__(self):
         self.columns = ["a", "b", "c", "d"]
@@ -634,3 +650,19 @@ def test_build_comp_fields(module_under_test):
         True,
     )
     assert comparison_fields == {"b": "b", "d": "d"}
+
+
+@mock.patch(
+    "data_validation.config_manager.ConfigManager.get_target_ibis_calculated_table",
+    new=lambda x: MockIbisTable(),
+)
+def test_is_oracle_lob(module_under_test):
+    config_manager = module_under_test.ConfigManager(
+        SAMPLE_CONFIG, MockOracleClient(), MockOracleClient(), verbose=False
+    )
+    assert not config_manager._is_oracle_lob(
+        "a"
+    ), "Column a is not a LOB therefore this should be False"
+    assert config_manager._is_oracle_lob(
+        "d"
+    ), "Column d is a LOB therefore this should be True"

third_party/ibis/ibis_oracle/__init__.py

@@ -163,3 +163,25 @@ def list_primary_key_columns(self, database: str, table: str) -> list:
                 list_pk_col_sql, parameters=(database.upper(), table.upper())
             )
             return [_[0] for _ in result.cursor.fetchall()]
+
+    def raw_metadata(
+        self, database: str = None, table: str = None, query: str = None
+    ) -> dict:
+        def strip_prefix(s: str):
+            if s.startswith("DB_TYPE_"):
+                return s[8:]
+            else:
+                return s
+
+        assert (database and table) or query, "We should never receive all args=None"
+        if database and table:
+            source = f'"{database}"."{table}"'.upper()
+        elif query:
+            source = f"({query})"
+
+        with self.begin() as con:
+            result = con.exec_driver_sql(f"SELECT * FROM {source} t0 WHERE ROWNUM <= 1")
+            cursor = result.cursor
+            return {
+                column[0]: strip_prefix(column[1].name) for column in cursor.description
+            }