fixed merge conflicts

Author: AdwitaSingh1711

python/cocoindex/flow.py

@@ -486,7 +486,7 @@ class _SourceRefreshOptions:
 class _ExecutionOptions:
     max_inflight_rows: int | None = None
     max_inflight_bytes: int | None = None
-    timeout: int | None = None
+    timeout: datetime.timedelta | None = None
 
 
 class FlowBuilder:

python/cocoindex/op.py

@@ -49,6 +49,7 @@
 )
 from .runtime import to_async_call
 from .index import IndexOptions
+import datetime
 
 
 class OpCategory(Enum):
@@ -165,7 +166,7 @@ class OpArgs:
     batching: bool = False
     max_batch_size: int | None = None
     behavior_version: int | None = None
-    timeout: int | None = None
+    timeout: datetime.timedelta | None = None
     arg_relationship: tuple[ArgRelationship, str] | None = None
 
 
@@ -394,7 +395,7 @@ def enable_cache(self) -> bool:
         def behavior_version(self) -> int | None:
             return op_args.behavior_version
 
-        def timeout(self) -> int | None:
+        def timeout(self) -> datetime.timedelta | None:
             if op_args.timeout is not None:
                 return op_args.timeout
 

rust/cocoindex/src/builder/analyzer.rs

@@ -12,10 +12,10 @@ use crate::{
 };
 use futures::future::{BoxFuture, try_join3};
 use futures::{FutureExt, future::try_join_all};
-use tokio::time::Duration;
+use std::time::Duration;
 use utils::fingerprint::Fingerprinter;
 
-const TIMEOUT_THRESHOLD: u64 = 1800;
+const TIMEOUT_THRESHOLD: Duration = Duration::from_secs(1800);
 
 #[derive(Debug)]
 pub(super) enum ValueTypeBuilder {
@@ -818,7 +818,7 @@ impl AnalyzerContext {
                             let behavior_version = executor.behavior_version();
                             let timeout = executor.timeout()
                                 .or(execution_options_timeout)
-                                .or(Some(Duration::from_secs(TIMEOUT_THRESHOLD)));
+                                .or(Some(TIMEOUT_THRESHOLD));
                             trace!("Finished building executor for transform op `{op_name}`, enable cache: {enable_cache}, behavior version: {behavior_version:?}");
                             let function_exec_info = AnalyzedFunctionExecInfo {
                                 enable_cache,

rust/cocoindex/src/builder/plan.rs

@@ -3,8 +3,8 @@ use crate::base::spec::FieldName;
 use crate::prelude::*;
 
 use crate::ops::interface::*;
-use utils::fingerprint::{Fingerprint, Fingerprinter};
 use std::time::Duration;
+use utils::fingerprint::{Fingerprint, Fingerprinter};
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 pub struct AnalyzedLocalFieldReference {

rust/cocoindex/src/execution/evaluator.rs

@@ -13,8 +13,8 @@ use utils::immutable::RefList;
 
 use super::memoization::{EvaluationMemory, EvaluationMemoryOptions, evaluate_with_cell};
 
-const TIMEOUT_THRESHOLD: u64 = 1800;
-const WARNING_THRESHOLD: u64 = 30;
+const TIMEOUT_THRESHOLD: Duration = Duration::from_secs(1800);
+const WARNING_THRESHOLD: Duration = Duration::from_secs(30);
 
 #[derive(Debug)]
 pub struct ScopeValueBuilder {
@@ -390,7 +390,7 @@ where
             _ = tokio::time::sleep(warn_duration), if !warned => {
                 warn!(
                     "Function '{}' ({}) is taking longer than {}s",
-                    op_kind, op_name, WARNING_THRESHOLD
+                    op_kind, op_name, WARNING_THRESHOLD.as_secs()
                 );
                 warned = true;
             }
@@ -408,21 +408,20 @@ async fn evaluate_op_scope(
     for reactive_op in op_scope.reactive_ops.iter() {
         match reactive_op {
             AnalyzedReactiveOp::Transform(op) => {
-                let transform_key = format!("transform/{}{}", op_scope.scope_qualifier, op.name);
-
+                // Track transform operation start
                 if let Some(ref op_stats) = operation_in_process_stats {
+                    let transform_key =
+                        format!("transform/{}{}", op_scope.scope_qualifier, op.name);
                     op_stats.start_processing(&transform_key, 1);
                 }
 
                 let mut input_values = Vec::with_capacity(op.inputs.len());
                 for value in assemble_input_values(&op.inputs, scoped_entries) {
                     input_values.push(value?);
                 }
-                let timeout_duration = op
-                    .function_exec_info
-                    .timeout
-                    .unwrap_or(Duration::from_secs(TIMEOUT_THRESHOLD));
-                let warn_duration = Duration::from_secs(WARNING_THRESHOLD);
+
+                let timeout_duration = op.function_exec_info.timeout.unwrap_or(TIMEOUT_THRESHOLD);
+                let warn_duration = WARNING_THRESHOLD;
 
                 let op_name_for_warning = op.name.clone();
                 let op_kind_for_warning = op.op_kind.clone();
@@ -468,7 +467,10 @@ async fn evaluate_op_scope(
                     head_scope.define_field(&op.output, &v)
                 };
 
+                // Track transform operation completion
                 if let Some(ref op_stats) = operation_in_process_stats {
+                    let transform_key =
+                        format!("transform/{}{}", op_scope.scope_qualifier, op.name);
                     op_stats.finish_processing(&transform_key, 1);
                 }
 
@@ -574,6 +576,43 @@ async fn evaluate_op_scope(
                 let collector_entry = scoped_entries
                     .headn(op.collector_ref.scope_up_level as usize)
                     .ok_or_else(|| anyhow::anyhow!("Collector level out of bound"))?;
+
+                // Assemble input values
+                let input_values: Vec<value::Value> =
+                    assemble_input_values(&op.input.fields, scoped_entries)
+                        .collect::<Result<Vec<_>>>()?;
+
+                // Create field_values vector for all fields in the merged schema
+                let mut field_values = op
+                    .field_index_mapping
+                    .iter()
+                    .map(|idx| {
+                        idx.map_or(value::Value::Null, |input_idx| {
+                            input_values[input_idx].clone()
+                        })
+                    })
+                    .collect::<Vec<_>>();
+
+                // Handle auto_uuid_field (assumed to be at position 0 for efficiency)
+                if op.has_auto_uuid_field {
+                    if let Some(uuid_idx) = op.collector_schema.auto_uuid_field_idx {
+                        let uuid = memory.next_uuid(
+                            op.fingerprinter
+                                .clone()
+                                .with(
+                                    &field_values
+                                        .iter()
+                                        .enumerate()
+                                        .filter(|(i, _)| *i != uuid_idx)
+                                        .map(|(_, v)| v)
+                                        .collect::<Vec<_>>(),
+                                )?
+                                .into_fingerprint(),
+                        )?;
+                        field_values[uuid_idx] = value::Value::Basic(value::BasicValue::Uuid(uuid));
+                    }
+                }
+
                 {
                     let mut collected_records = collector_entry.collected_values
                         [op.collector_ref.local.collector_idx as usize]

rust/cocoindex/src/execution/live_updater.rs

@@ -172,7 +172,6 @@ impl SourceUpdateTask {
                 let mut change_stream = change_stream;
                 let retry_options = retryable::RetryOptions {
                     retry_timeout: None,
-                    per_call_timeout: None,
                     initial_backoff: std::time::Duration::from_secs(5),
                     max_backoff: std::time::Duration::from_secs(60),
                 };

rust/cocoindex/src/ops/interface.rs

@@ -186,7 +186,7 @@ pub trait SimpleFunctionExecutor: Send + Sync {
         None
     }
 
-    /// Returns None to use the default timeout (300s)
+    /// Returns None to use the default timeout (1800s)
     fn timeout(&self) -> Option<std::time::Duration> {
         None
     }

rust/cocoindex/src/ops/py_factory.rs

@@ -281,7 +281,7 @@ impl interface::SimpleFunctionFactory for PyFunctionFactory {
                         Ok((
                             prepare_fut,
                             enable_cache,
-                            behavior_version, 
+                            behavior_version,
                             timeout,
                             batching_options,
                         ))

rust/utils/src/retryable.rs

@@ -109,7 +109,6 @@ pub fn Ok<T>(value: T) -> Result<T> {
 
 pub struct RetryOptions {
     pub retry_timeout: Option<Duration>,
-    pub per_call_timeout: Option<Duration>,
     pub initial_backoff: Duration,
     pub max_backoff: Duration,
 }
@@ -118,7 +117,6 @@ impl Default for RetryOptions {
     fn default() -> Self {
         Self {
             retry_timeout: Some(DEFAULT_RETRY_TIMEOUT),
-            per_call_timeout: None,
             initial_backoff: Duration::from_millis(100),
             max_backoff: Duration::from_secs(10),
         }
@@ -127,7 +125,6 @@ impl Default for RetryOptions {
 
 pub static HEAVY_LOADED_OPTIONS: RetryOptions = RetryOptions {
     retry_timeout: Some(DEFAULT_RETRY_TIMEOUT),
-    per_call_timeout: Some(Duration::from_secs(300)),
     initial_backoff: Duration::from_secs(1),
     max_backoff: Duration::from_secs(60),
 };