From a6afd776a5cee0a5d290b79329695c8be3651c5b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Karolis=20Gudi=C5=A1kis?= <karolisgud@gmail.com>
Date: Mon, 19 Feb 2024 16:22:33 +0200
Subject: [PATCH] Dont use batch insert

---
 dozer-cli/src/pipeline/dummy_sink.rs       |  2 +-
 dozer-ingestion/aerospike/src/connector.rs | 30 ++++++++++++----------
 2 files changed, 17 insertions(+), 15 deletions(-)

diff --git a/dozer-cli/src/pipeline/dummy_sink.rs b/dozer-cli/src/pipeline/dummy_sink.rs
index f5d07ddca2..9607278bad 100644
--- a/dozer-cli/src/pipeline/dummy_sink.rs
+++ b/dozer-cli/src/pipeline/dummy_sink.rs
@@ -44,7 +44,7 @@ impl SinkFactory for DummySinkFactory {
             .next()
             .and_then(|schema| {
                 schema.fields.into_iter().enumerate().find(|(_, field)| {
-                    field.name == "inserted_at" && field.typ == FieldType::Timestamp
+                    field.name.to_lowercase() == "inserted_at" && field.typ == FieldType::Timestamp
                 })
             })
             .map(|(index, _)| index);
diff --git a/dozer-ingestion/aerospike/src/connector.rs b/dozer-ingestion/aerospike/src/connector.rs
index 7256e615a2..80463cc7fe 100644
--- a/dozer-ingestion/aerospike/src/connector.rs
+++ b/dozer-ingestion/aerospike/src/connector.rs
@@ -6,7 +6,7 @@ use dozer_ingestion_connector::dozer_types::models::ingestion_types::{
     IngestionMessage, TransactionInfo,
 };
 use dozer_ingestion_connector::dozer_types::node::{NodeHandle, OpIdentifier, SourceState};
-use dozer_ingestion_connector::dozer_types::types::Operation::{BatchInsert, Insert};
+use dozer_ingestion_connector::dozer_types::types::Operation::Insert;
 use dozer_ingestion_connector::dozer_types::types::{
     Field, FieldDefinition, FieldType, Record, Schema,
 };
@@ -362,31 +362,33 @@ async fn batch_event_request_handler(
     let events = json.into_inner();
     let state = data.into_inner();
 
-    let mut mapped_events: HashMap<usize, Vec<Record>> = HashMap::new();
     for event in events {
-        match map_record(event, state.tables_index_map.clone()).await {
-            Ok(Some((table_index, record))) => {
-                mapped_events.entry(table_index).or_default().push(record);
+        match map_events(event, state.tables_index_map.clone()).await {
+            Ok(Some(message)) => {
+                if let Err(e) = state.ingestor.handle_message(message).await {
+                    error!("Aerospike ingestion message send error: {:?}", e);
+                    return HttpResponse::InternalServerError().finish();
+                }
             }
             Ok(None) => {}
             Err(e) => return map_error(e),
         }
     }
 
-    for (table_index, records) in mapped_events {
-        let event = IngestionMessage::OperationEvent {
-            table_index,
-            op: BatchInsert { new: records },
+    let transaction_ingestion_result = state
+        .ingestor
+        .handle_message(IngestionMessage::TransactionInfo(TransactionInfo::Commit {
             id: None,
-        };
+        }))
+        .await;
 
-        if let Err(e) = state.ingestor.handle_message(event).await {
+    match transaction_ingestion_result {
+        Ok(_) => HttpResponse::Ok().finish(),
+        Err(e) => {
             error!("Aerospike ingestion message send error: {:?}", e);
-            return HttpResponse::InternalServerError().finish();
+            HttpResponse::InternalServerError().finish()
         }
     }
-
-    HttpResponse::Ok().finish()
 }
 
 #[derive(Clone, Debug)]