feat: plumbing source name to SourceFactory (#953)

Author: georgeh0

src/builder/analyzer.rs

@@ -657,14 +657,15 @@ impl AnalyzerContext {
         let source_factory = get_source_factory(&import_op.spec.source.kind)?;
         let (output_type, executor) = source_factory
             .build(
+                &import_op.name,
                 serde_json::Value::Object(import_op.spec.source.spec),
                 self.flow_ctx.clone(),
             )
             .await?;
 
-        let op_name = import_op.name.clone();
+        let op_name = import_op.name;
         let primary_key_schema = Box::from(output_type.typ.key_schema());
-        let output = op_scope.add_op_output(import_op.name, output_type)?;
+        let output = op_scope.add_op_output(op_name.clone(), output_type)?;
 
         let concur_control_options = import_op
             .spec

src/ops/factory_bases.rs

@@ -218,6 +218,7 @@ pub trait SourceFactoryBase: SourceFactory + Send + Sync + 'static {
 
     async fn build_executor(
         self: Arc<Self>,
+        source_name: &str,
         spec: Self::Spec,
         context: Arc<FlowInstanceContext>,
     ) -> Result<Box<dyn SourceExecutor>>;
@@ -237,6 +238,7 @@ pub trait SourceFactoryBase: SourceFactory + Send + Sync + 'static {
 impl<T: SourceFactoryBase> SourceFactory for T {
     async fn build(
         self: Arc<Self>,
+        source_name: &str,
         spec: serde_json::Value,
         context: Arc<FlowInstanceContext>,
     ) -> Result<(
@@ -245,8 +247,9 @@ impl<T: SourceFactoryBase> SourceFactory for T {
     )> {
         let spec: T::Spec = serde_json::from_value(spec)?;
         let output_schema = self.get_output_schema(&spec, &context).await?;
-        let executor = self.build_executor(spec, context);
-        Ok((output_schema, executor))
+        let source_name = source_name.to_string();
+        let executor = async move { self.build_executor(&source_name, spec, context).await };
+        Ok((output_schema, Box::pin(executor)))
     }
 }
 

src/ops/interface.rs

@@ -162,6 +162,7 @@ pub trait SourceExecutor: Send + Sync {
 pub trait SourceFactory {
     async fn build(
         self: Arc<Self>,
+        source_name: &str,
         spec: serde_json::Value,
         context: Arc<FlowInstanceContext>,
     ) -> Result<(

src/ops/sources/amazon_s3.rs

@@ -331,6 +331,7 @@ impl SourceFactoryBase for Factory {
 
     async fn build_executor(
         self: Arc<Self>,
+        _source_name: &str,
         spec: Spec,
         _context: Arc<FlowInstanceContext>,
     ) -> Result<Box<dyn SourceExecutor>> {

src/ops/sources/azure_blob.rs

@@ -213,6 +213,7 @@ impl SourceFactoryBase for Factory {
 
     async fn build_executor(
         self: Arc<Self>,
+        _source_name: &str,
         spec: Spec,
         context: Arc<FlowInstanceContext>,
     ) -> Result<Box<dyn SourceExecutor>> {

src/ops/sources/google_drive.rs

@@ -494,6 +494,7 @@ impl SourceFactoryBase for Factory {
 
     async fn build_executor(
         self: Arc<Self>,
+        _source_name: &str,
         spec: Spec,
         _context: Arc<FlowInstanceContext>,
     ) -> Result<Box<dyn SourceExecutor>> {

src/ops/sources/local_file.rs

@@ -163,6 +163,7 @@ impl SourceFactoryBase for Factory {
 
     async fn build_executor(
         self: Arc<Self>,
+        _source_name: &str,
         spec: Spec,
         _context: Arc<FlowInstanceContext>,
     ) -> Result<Box<dyn SourceExecutor>> {

src/ops/sources/postgres.rs

@@ -774,6 +774,7 @@ impl SourceFactoryBase for Factory {
 
     async fn build_executor(
         self: Arc<Self>,
+        _source_name: &str,
         spec: Spec,
         context: Arc<FlowInstanceContext>,
     ) -> Result<Box<dyn SourceExecutor>> {