Refactor ExportTargetFactory::build to make return data more clear.

Author: badmonster0

src/builder/analyzer.rs

@@ -886,7 +886,7 @@ impl AnalyzerContext<'_> {
             }
         };
 
-        let ((setup_key, desired_state), executor_fut) = export_factory.clone().build(
+        let setup_output = export_factory.clone().build(
             export_op.name.clone(),
             spec,
             key_fields_schema,
@@ -895,7 +895,7 @@ impl AnalyzerContext<'_> {
             self.flow_ctx.clone(),
         )?;
         let resource_id = ResourceIdentifier {
-            key: setup_key.clone(),
+            key: setup_output.setup_key.clone(),
             target_kind: export_target.kind.clone(),
         };
         let existing_target_states = existing_target_states.get(&resource_id);
@@ -904,8 +904,10 @@ impl AnalyzerContext<'_> {
                 let mut compatible_target_ids = HashSet::<Option<i32>>::new();
                 let mut reusable_schema_version_ids = HashSet::<Option<i32>>::new();
                 for existing_state in existing_target_states.iter().flat_map(|v| v.iter()) {
-                    let compatibility = export_factory
-                        .check_state_compatibility(&desired_state, &existing_state.state)?;
+                    let compatibility = export_factory.check_state_compatibility(
+                        &setup_output.desired_setup_state,
+                        &existing_state.state,
+                    )?;
                     let compatible_target_id =
                         if compatibility != SetupStateCompatibility::NotCompatible {
                             reusable_schema_version_ids.insert(
@@ -946,10 +948,7 @@ impl AnalyzerContext<'_> {
                 } else {
                     max_schema_version_id + 1
                 };
-                match setup_state.targets.entry(ResourceIdentifier {
-                    key: setup_key,
-                    target_kind: export_target.kind.clone(),
-                }) {
+                match setup_state.targets.entry(resource_id) {
                     indexmap::map::Entry::Occupied(entry) => {
                         api_bail!(
                             "Target resource already exists: kind = {}, key = {}",
@@ -964,7 +963,7 @@ impl AnalyzerContext<'_> {
                                 schema_version_id,
                                 max_schema_version_id: max_schema_version_id.max(schema_version_id),
                             },
-                            state: desired_state,
+                            state: setup_output.desired_setup_state,
                         });
                     }
                 }
@@ -980,7 +979,8 @@ impl AnalyzerContext<'_> {
             .unwrap_or(false);
         Ok(async move {
             trace!("Start building executor for export op `{}`", export_op.name);
-            let (executor, query_target) = executor_fut
+            let (executor, query_target) = setup_output
+                .executor
                 .await
                 .with_context(|| format!("Analyzing export op: {}", export_op.name))?;
             trace!(

src/ops/factory_bases.rs

@@ -1,21 +1,13 @@
-use std::collections::HashMap;
+use crate::prelude::*;
 use std::fmt::Debug;
 use std::hash::Hash;
-use std::sync::Arc;
-
-use anyhow::Result;
-use axum::async_trait;
-use futures::future::BoxFuture;
-use serde::de::DeserializeOwned;
-use serde::Serialize;
 
 use super::interface::*;
 use super::registry::*;
 use crate::api_bail;
 use crate::api_error;
 use crate::base::schema::*;
 use crate::base::spec::*;
-use crate::base::value;
 use crate::builder::plan::AnalyzedValueMapping;
 use crate::setup;
 // SourceFactoryBase
@@ -272,6 +264,13 @@ impl<T: SimpleFunctionFactoryBase> SimpleFunctionFactory for T {
     }
 }
 
+pub struct ExportTargetBuildOutput<F: StorageFactoryBase + ?Sized> {
+    pub executor:
+        BoxFuture<'static, Result<(Arc<dyn ExportTargetExecutor>, Option<Arc<dyn QueryTarget>>)>>,
+    pub setup_key: F::Key,
+    pub desired_setup_state: F::SetupState,
+}
+
 pub trait StorageFactoryBase: ExportTargetFactory + Send + Sync + 'static {
     type Spec: DeserializeOwned + Send + Sync;
     type Key: Debug + Clone + Serialize + DeserializeOwned + Eq + Hash + Send + Sync;
@@ -287,11 +286,9 @@ pub trait StorageFactoryBase: ExportTargetFactory + Send + Sync + 'static {
         value_fields_schema: Vec<FieldSchema>,
         storage_options: IndexOptions,
         context: Arc<FlowInstanceContext>,
-    ) -> Result<(
-        (Self::Key, Self::SetupState),
-        BoxFuture<'static, Result<(Arc<dyn ExportTargetExecutor>, Option<Arc<dyn QueryTarget>>)>>,
-    )>;
+    ) -> Result<ExportTargetBuildOutput<Self>>;
 
+    /// This is only called for non-user-setup targets.
     fn check_setup_status(
         &self,
         key: Self::Key,
@@ -387,12 +384,9 @@ impl<T: StorageFactoryBase> ExportTargetFactory for T {
         value_fields_schema: Vec<FieldSchema>,
         storage_options: IndexOptions,
         context: Arc<FlowInstanceContext>,
-    ) -> Result<(
-        (serde_json::Value, serde_json::Value),
-        BoxFuture<'static, Result<(Arc<dyn ExportTargetExecutor>, Option<Arc<dyn QueryTarget>>)>>,
-    )> {
+    ) -> Result<interface::ExportTargetBuildOutput> {
         let spec: T::Spec = serde_json::from_value(spec)?;
-        let ((setup_key, setup_state), executors) = StorageFactoryBase::build(
+        let build_output = StorageFactoryBase::build(
             self,
             name,
             spec,
@@ -401,13 +395,11 @@ impl<T: StorageFactoryBase> ExportTargetFactory for T {
             storage_options,
             context,
         )?;
-        Ok((
-            (
-                serde_json::to_value(setup_key)?,
-                serde_json::to_value(setup_state)?,
-            ),
-            executors,
-        ))
+        Ok(interface::ExportTargetBuildOutput {
+            executor: build_output.executor,
+            setup_key: serde_json::to_value(build_output.setup_key)?,
+            desired_setup_state: serde_json::to_value(build_output.desired_setup_state)?,
+        })
     }
 
     fn check_setup_status(

src/ops/interface.rs

@@ -154,6 +154,13 @@ pub enum SetupStateCompatibility {
     NotCompatible,
 }
 
+pub struct ExportTargetBuildOutput {
+    pub executor:
+        BoxFuture<'static, Result<(Arc<dyn ExportTargetExecutor>, Option<Arc<dyn QueryTarget>>)>>,
+    pub setup_key: serde_json::Value,
+    pub desired_setup_state: serde_json::Value,
+}
+
 pub trait ExportTargetFactory {
     // The first field of the `input_schema` is the primary key field.
     // If it has struct type, it should be converted to composite primary key.
@@ -165,10 +172,7 @@ pub trait ExportTargetFactory {
         value_fields_schema: Vec<FieldSchema>,
         storage_options: IndexOptions,
         context: Arc<FlowInstanceContext>,
-    ) -> Result<(
-        (serde_json::Value, serde_json::Value),
-        BoxFuture<'static, Result<(Arc<dyn ExportTargetExecutor>, Option<Arc<dyn QueryTarget>>)>>,
-    )>;
+    ) -> Result<ExportTargetBuildOutput>;
 
     fn check_setup_status(
         &self,

src/ops/sdk.rs

@@ -10,6 +10,8 @@ pub use crate::base::schema::*;
 pub use crate::base::spec::*;
 pub use crate::base::value::*;
 
+// Disambiguate the ExportTargetBuildOutput type.
+pub use super::factory_bases::ExportTargetBuildOutput;
 /// Defined for all types convertible to ValueType, to ease creation for ValueType in various operation factories.
 pub trait TypeCore {
     fn into_type(self) -> ValueType;

src/ops/storages/postgres.rs

@@ -921,10 +921,7 @@ impl StorageFactoryBase for Arc<Factory> {
         value_fields_schema: Vec<FieldSchema>,
         storage_options: IndexOptions,
         context: Arc<FlowInstanceContext>,
-    ) -> Result<(
-        (TableId, SetupState),
-        BoxFuture<'static, Result<(Arc<dyn ExportTargetExecutor>, Option<Arc<dyn QueryTarget>>)>>,
-    )> {
+    ) -> Result<ExportTargetBuildOutput<Self>> {
         let table_id = TableId {
             database_url: spec.database_url.clone(),
             table_name: spec
@@ -951,7 +948,11 @@ impl StorageFactoryBase for Arc<Factory> {
                 Some(query_target as Arc<dyn QueryTarget>),
             ))
         };
-        Ok(((table_id, setup_state), executors.boxed()))
+        Ok(ExportTargetBuildOutput {
+            executor: executors.boxed(),
+            setup_key: table_id,
+            desired_setup_state: setup_state,
+        })
     }
 
     fn check_setup_status(

src/prelude.rs

@@ -7,7 +7,7 @@ pub(crate) use futures::{future::BoxFuture, prelude::*, stream::BoxStream};
 pub(crate) use futures::{FutureExt, StreamExt};
 pub(crate) use indexmap::{IndexMap, IndexSet};
 pub(crate) use itertools::Itertools;
-pub(crate) use serde::{Deserialize, Serialize};
+pub(crate) use serde::{de::DeserializeOwned, Deserialize, Serialize};
 pub(crate) use std::borrow::Cow;
 pub(crate) use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 pub(crate) use std::sync::{Arc, LazyLock, Mutex, OnceLock, RwLock, Weak};