fix: Oracle redo log can have NULL REDO_SQL (#2418)

getdozer · Feb 22, 2024 · f7803fc · f7803fc
1 parent c0f6df0
commit f7803fc
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 10 deletions.
diff --git a/dozer-ingestion/oracle/src/connector/mod.rs b/dozer-ingestion/oracle/src/connector/mod.rs
@@ -400,6 +400,9 @@ impl Connector {
                     let content = match content {
                         Ok(content) => content,
                         Err(e) => {
+                            if ingestor.is_closed() {
+                                return Ok(());
+                            }
                             error!("Error during log mining: {}. Retrying.", e);
                             break 'replicate_logs;
                         }

diff --git a/dozer-ingestion/oracle/src/connector/replicate/log_miner/listing.rs b/dozer-ingestion/oracle/src/connector/replicate/log_miner/listing.rs
@@ -10,7 +10,7 @@ pub struct LogManagerContent {
     pub operation_code: u8,
     pub seg_owner: Option<String>,
     pub table_name: Option<String>,
-    pub sql_redo: String,
+    pub sql_redo: Option<String>,
 }
 
 impl LogManagerContent {
@@ -25,7 +25,7 @@ impl LogManagerContent {
             u8,
             Option<String>,
             Option<String>,
-            String,
+            Option<String>,
         );
         let rows = if let Some(con_id) = con_id {
             let sql = "SELECT COMMIT_SCN, COMMIT_TIMESTAMP, OPERATION_CODE, SEG_OWNER, TABLE_NAME, SQL_REDO FROM V$LOGMNR_CONTENTS WHERE COMMIT_SCN >= :start_scn AND SRC_CON_ID = :con_id";

diff --git a/dozer-ingestion/oracle/src/connector/replicate/log_miner/parse/mod.rs b/dozer-ingestion/oracle/src/connector/replicate/log_miner/parse/mod.rs
@@ -88,18 +88,23 @@ impl Parser {
         };
 
         let op = match content.operation_code {
-            OP_CODE_INSERT => {
-                ParsedOperation::Insert(self.insert_parser.parse(&content.sql_redo, &table_pair)?)
-            }
-            OP_CODE_DELETE => {
-                ParsedOperation::Delete(self.delete_parser.parse(&content.sql_redo, &table_pair)?)
-            }
+            OP_CODE_INSERT => ParsedOperation::Insert(self.insert_parser.parse(
+                &content.sql_redo.expect("insert must have redo"),
+                &table_pair,
+            )?),
+            OP_CODE_DELETE => ParsedOperation::Delete(self.delete_parser.parse(
+                &content.sql_redo.expect("delete must have redo"),
+                &table_pair,
+            )?),
             OP_CODE_UPDATE => {
-                let (old, new) = self.update_parser.parse(&content.sql_redo, &table_pair)?;
+                let (old, new) = self.update_parser.parse(
+                    &content.sql_redo.expect("update must have redo"),
+                    &table_pair,
+                )?;
                 ParsedOperation::Update { old, new }
             }
             OP_CODE_DDL => {
-                warn!("Ignoring DDL operation: {}", content.sql_redo);
+                warn!("Ignoring DDL operation: {:?}", content.sql_redo);
                 return Ok(None);
             }
             _ => {