From c6c70c6309b127e27289b5ad3ed1d8df1dce51d5 Mon Sep 17 00:00:00 2001
From: chubei <914745487@qq.com>
Date: Fri, 1 Mar 2024 17:48:06 +0800
Subject: [PATCH] fix: Source state was not correctly collected during DAG
 building

---
 dozer-core/src/builder_dag.rs | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/dozer-core/src/builder_dag.rs b/dozer-core/src/builder_dag.rs
index 0d3f7125de..65dd94cc8d 100644
--- a/dozer-core/src/builder_dag.rs
+++ b/dozer-core/src/builder_dag.rs
@@ -101,10 +101,10 @@ impl BuilderDag {
 
                 let state = sink.get_source_state().map_err(ExecutionError::Sink)?;
                 if let Some(state) = state {
-                    match source_states.entry(handle.clone()) {
+                    match source_states.entry(source.clone()) {
                         Entry::Occupied(entry) => {
                             if entry.get() != &state {
-                                return Err(ExecutionError::SourceStateConflict(handle));
+                                return Err(ExecutionError::SourceStateConflict(source));
                             }
                         }
                         Entry::Vacant(entry) => {
@@ -115,7 +115,7 @@ impl BuilderDag {
 
                 let op_id = sink.get_latest_op_id().map_err(ExecutionError::Sink)?;
                 if let Some(op_id) = op_id {
-                    match source_op_ids.entry(handle.clone()) {
+                    match source_op_ids.entry(source.clone()) {
                         Entry::Occupied(mut entry) => {
                             *entry.get_mut() = op_id.min(*entry.get());
                         }