feat: Use temp table in oracle sink and improve logging in aerospike (#…

…2453)

dozer-ingestion/aerospike/src/connector.rs

@@ -506,7 +506,7 @@ impl Connector for AerospikeConnector {
                                         } else {
                                             FieldType::String
                                         },
-                                        nullable: true,
+                                        nullable: name != "PK",
                                         source: Default::default(),
                                     })
                                     .collect(),

dozer-ingestion/oracle/src/connector/mod.rs

@@ -33,7 +33,7 @@ pub struct Connector {
 
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
-    #[error("oracle error: {0}")]
+    #[error("oracle error: {0:?}")]
     Oracle(#[from] oracle::Error),
     #[error("pdb not found: {0}")]
     PdbNotFound(String),
@@ -264,6 +264,7 @@ impl Connector {
             let columns = table.column_names.join(", ");
             let owner = table.schema.unwrap_or_else(|| self.username.clone());
             let sql = format!("SELECT {} FROM {}.{}", columns, owner, table.name);
+            debug!("{}", sql);
             let rows = self.connection.query(&sql, &[])?;
 
             let mut batch = Vec::with_capacity(self.batch_size);

dozer-ingestion/oracle/src/connector/replicate/log/mod.rs

@@ -1,5 +1,6 @@
 use std::{sync::mpsc::SyncSender, time::Duration};
 
+use dozer_ingestion_connector::dozer_types::log::debug;
 use dozer_ingestion_connector::{
     dozer_types::{
         chrono::{DateTime, Utc},
@@ -72,7 +73,7 @@ fn log_reader_loop(
     let mut last_rba: Option<LastRba> = None;
 
     loop {
-        info!("Listing logs starting from SCN {}", start_scn);
+        debug!(target: "oracle_replication", "Listing logs starting from SCN {}", start_scn);
         let mut logs = match list_logs(connection, start_scn) {
             Ok(logs) => logs,
             Err(e) => {
@@ -95,7 +96,7 @@ fn log_reader_loop(
 
         'replicate_logs: while !logs.is_empty() {
             let log = logs.remove(0);
-            info!(
+            debug!(target: "oracle_replication",
                 "Reading log {} ({}) ({}, {}), starting from {:?}",
                 log.name, log.sequence, log.first_change, log.next_change, last_rba
             );
@@ -145,7 +146,7 @@ fn log_reader_loop(
                 if ingestor.is_closed() {
                     return;
                 }
-                info!("Read all logs, retrying after {:?}", poll_interval);
+                debug!(target: "oracle_replication", "Read all logs, retrying after {:?}", poll_interval);
                 std::thread::sleep(poll_interval);
             } else {
                 // If there are more logs, we need to start from the next log's first change.

dozer-sink-aerospike/src/aerospike.rs

@@ -1,3 +1,4 @@
+use std::time::Instant;
 use std::{
     alloc::{handle_alloc_error, Layout},
     ffi::{c_char, c_void, CStr, CString, NulError},
@@ -9,6 +10,7 @@ use std::{
 use itertools::Itertools;
 
 use aerospike_client_sys::*;
+use dozer_types::log::debug;
 use dozer_types::{
     chrono::{DateTime, NaiveDate},
     geo::{Coord, Point},
@@ -147,7 +149,9 @@ impl Client {
             as_config_init(config.as_mut_ptr());
             config.assume_init()
         };
-        config.policies.batch.base.total_timeout = 10000;
+
+        config.policies.batch.base.total_timeout = 30000;
+        config.policies.batch.base.socket_timeout = 30000;
         config.policies.write.key = as_policy_key_e_AS_POLICY_KEY_SEND;
         config.policies.batch_write.key = as_policy_key_e_AS_POLICY_KEY_SEND;
         unsafe {
@@ -185,6 +189,7 @@ impl Client {
         if let Some(filter) = filter {
             policy.base.filter_exp = filter.as_ptr();
         }
+
         as_try(|err| {
             aerospike_key_put(
                 self.inner.as_ptr(),
@@ -204,6 +209,7 @@ impl Client {
     ) -> Result<(), AerospikeError> {
         let mut policy = self.inner.as_ref().config.policies.write;
         policy.exists = as_policy_exists_e_AS_POLICY_EXISTS_CREATE;
+
         self.put(key, new, policy, filter)
     }
 
@@ -252,7 +258,20 @@ impl Client {
         &self,
         batch: *mut as_batch_records,
     ) -> Result<(), AerospikeError> {
-        as_try(|err| aerospike_batch_write(self.inner.as_ptr(), err, std::ptr::null(), batch))
+        debug!(target: "aerospike_sink", "Writing batch of size {}", batch.as_ref().unwrap().list.size);
+
+        let started = Instant::now();
+        let policy = self.inner.as_ref().config.policies.batch;
+        as_try(|err| {
+            aerospike_batch_write(
+                self.inner.as_ptr(),
+                err,
+                &policy as *const as_policy_batch,
+                batch,
+            )
+        })?;
+        debug!(target: "aerospike_sink", "Batch write took {:?}", started.elapsed());
+        Ok(())
     }
 
     pub(crate) unsafe fn _select(

dozer-sink-aerospike/src/denorm_dag.rs

@@ -10,6 +10,7 @@ use dozer_core::petgraph::visit::{
     EdgeRef, IntoEdgesDirected, IntoNeighborsDirected, IntoNodeReferences,
 };
 use dozer_types::indexmap::IndexMap;
+
 use dozer_types::models::sink::{AerospikeSet, AerospikeSinkTable};
 use dozer_types::thiserror;
 use dozer_types::types::{Field, Record, Schema, TableOperation};

dozer-sink-aerospike/src/lib.rs

@@ -239,7 +239,6 @@ impl Drop for AsRecord<'_> {
 struct AerospikeSink {
     config: AerospikeSinkConfig,
     replication_worker: AerospikeSinkWorker,
-    current_transaction: Option<u64>,
     metadata_namespace: CString,
     metadata_set: CString,
     client: Arc<Client>,
@@ -365,7 +364,6 @@ impl AerospikeSink {
         Ok(Self {
             config,
             replication_worker: worker_instance,
-            current_transaction: None,
             metadata_namespace,
             metadata_set,
             client,
@@ -441,6 +439,7 @@ impl AerospikeSinkWorker {
             .sum();
         // Write denormed tables
         let mut batch = RecordBatch::new(batch_size_est as u32, batch_size_est as u32);
+
         for table in denormalized_tables {
             for (key, record) in table.records {
                 batch.add_write(
@@ -475,17 +474,21 @@ impl Sink for AerospikeSink {
         Ok(())
     }
 
-    fn commit(&mut self, _epoch_details: &dozer_core::epoch::Epoch) -> Result<(), BoxedError> {
-        self.replication_worker
-            .commit(self.current_transaction.take())?;
+    fn commit(&mut self, epoch_details: &dozer_core::epoch::Epoch) -> Result<(), BoxedError> {
+        debug_assert_eq!(epoch_details.common_info.source_states.len(), 1);
+        let txid = epoch_details
+            .common_info
+            .source_states
+            .iter()
+            .next()
+            .and_then(|(_, state)| state.op_id())
+            .map(|op_id| op_id.txid);
+
+        self.replication_worker.commit(txid)?;
         Ok(())
     }
 
     fn process(&mut self, op: TableOperation) -> Result<(), BoxedError> {
-        // Set current transaction before any error can be thrown, so we don't
-        // get stuck in an error loop if this error gets ignored by the caller
-        self.current_transaction = op.id.map(|id| id.txid);
-
         self.replication_worker.process(op)?;
         Ok(())
     }