feat(row-status): expose an API to get row indexing status

Author: badmonster0

src/execution/db_tracking.rs

@@ -211,3 +211,28 @@ impl ListTrackedSourceKeyMetadataState {
         sqlx::query_as(&self.query_str).bind(source_id).fetch(pool)
     }
 }
+
+#[derive(sqlx::FromRow, Debug)]
+pub struct SourceLastProcessedInfo {
+    pub processed_source_ordinal: Option<i64>,
+    pub process_logic_fingerprint: Option<Vec<u8>>,
+    pub process_time_micros: Option<i64>,
+}
+
+pub async fn read_source_last_processed_info(
+    source_id: i32,
+    source_key_json: &serde_json::Value,
+    db_setup: &TrackingTableSetupState,
+    pool: &PgPool,
+) -> Result<Option<SourceLastProcessedInfo>> {
+    let query_str = format!(
+        "SELECT processed_source_ordinal, process_logic_fingerprint, process_time_micros FROM {} WHERE source_id = $1 AND source_key = $2",
+        db_setup.table_name
+    );
+    let last_processed_info = sqlx::query_as(&query_str)
+        .bind(source_id)
+        .bind(source_key_json)
+        .fetch_optional(pool)
+        .await?;
+    Ok(last_processed_info)
+}

src/execution/indexing_status.rs

@@ -0,0 +1,59 @@
+use crate::prelude::*;
+
+use super::db_tracking;
+use super::evaluator;
+use futures::try_join;
+
+#[derive(Debug, Serialize)]
+pub struct SourceRowLastProcessedInfo {
+    pub source_ordinal: Option<interface::Ordinal>,
+    pub processing_time: Option<chrono::DateTime<chrono::Utc>>,
+    pub is_logic_current: bool,
+}
+
+#[derive(Debug, Serialize)]
+pub struct SourceRowInfo {
+    pub ordinal: Option<interface::Ordinal>,
+}
+
+#[derive(Debug, Serialize)]
+pub struct SourceRowIndexingStatus {
+    pub last_processed: Option<SourceRowLastProcessedInfo>,
+    pub current: Option<SourceRowInfo>,
+}
+
+pub async fn get_source_row_indexing_status(
+    src_eval_ctx: &evaluator::SourceRowEvaluationContext<'_>,
+    pool: &sqlx::PgPool,
+) -> Result<SourceRowIndexingStatus> {
+    let source_key_json = serde_json::to_value(src_eval_ctx.key)?;
+    let last_processed_fut = db_tracking::read_source_last_processed_info(
+        src_eval_ctx.import_op.source_id,
+        &source_key_json,
+        &src_eval_ctx.plan.tracking_table_setup,
+        pool,
+    );
+    let current_fut = src_eval_ctx.import_op.executor.get_value(
+        &src_eval_ctx.key,
+        &interface::SourceExecutorGetOptions {
+            include_value: false,
+            include_ordinal: true,
+        },
+    );
+    let (last_processed, current) = try_join!(last_processed_fut, current_fut)?;
+
+    let last_processed = last_processed.map(|l| SourceRowLastProcessedInfo {
+        source_ordinal: l.processed_source_ordinal.map(interface::Ordinal),
+        processing_time: l
+            .process_time_micros
+            .map(chrono::DateTime::<chrono::Utc>::from_timestamp_micros)
+            .flatten(),
+        is_logic_current: Some(src_eval_ctx.plan.logic_fingerprint.0.as_slice())
+            == l.process_logic_fingerprint.as_ref().map(|b| b.as_slice()),
+    });
+    let current = current.map(|c| SourceRowInfo { ordinal: c.ordinal });
+    Ok(SourceRowIndexingStatus {
+        last_processed,
+        current,
+    })
+}

src/execution/mod.rs

@@ -1,6 +1,7 @@
 pub(crate) mod db_tracking_setup;
 pub(crate) mod dumper;
 pub(crate) mod evaluator;
+pub(crate) mod indexing_status;
 pub(crate) mod memoization;
 pub(crate) mod query;
 pub(crate) mod row_indexer;

src/execution/row_indexer.rs

@@ -471,9 +471,10 @@ pub async fn evaluate_source_entry_with_memory(
             },
         )
         .await?
-        .value
     {
-        Some(d) => d,
+        Some(d) => d
+            .value
+            .ok_or_else(|| anyhow::anyhow!("value not returned"))?,
         None => return Ok(None),
     };
     let output = evaluate_source_entry(src_eval_ctx, source_value, &memory).await?;

src/execution/source_indexer.rs

@@ -112,7 +112,8 @@ impl SourceIndexingContext {
                         },
                     )
                     .await?
-                    .value
+                    .map(|v| v.value)
+                    .flatten()
             };
             let schema = &self.flow.data_schema;
             let result = row_indexer::update_source_row(

src/ops/interface.rs

@@ -16,7 +16,7 @@ pub struct FlowInstanceContext {
     pub py_exec_ctx: Option<Arc<crate::py::PythonExecutionContext>>,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)]
 pub struct Ordinal(pub i64);
 
 impl From<Ordinal> for i64 {
@@ -72,9 +72,9 @@ pub struct SourceExecutorGetOptions {
     pub include_value: bool,
 }
 
-#[derive(Debug, Default)]
+#[derive(Debug)]
 pub struct SourceValue {
-    // None if not exists, or not included in the option.
+    // None if not included in the option.
     pub value: Option<FieldValues>,
     // None if unavailable, or not included in the option.
     pub ordinal: Option<Ordinal>,
@@ -93,7 +93,7 @@ pub trait SourceExecutor: Send + Sync {
         &self,
         key: &KeyValue,
         options: &SourceExecutorGetOptions,
-    ) -> Result<SourceValue>;
+    ) -> Result<Option<SourceValue>>;
 
     async fn change_stream(&self) -> Result<Option<BoxStream<'async_trait, SourceChange>>> {
         Ok(None)

src/ops/sources/google_drive.rs

@@ -341,7 +341,7 @@ impl SourceExecutor for Executor {
         &self,
         key: &KeyValue,
         options: &SourceExecutorGetOptions,
-    ) -> Result<SourceValue> {
+    ) -> Result<Option<SourceValue>> {
         let file_id = key.str_value()?;
         let fields = format!(
             "id,name,mimeType,trashed{}",
@@ -359,7 +359,7 @@ impl SourceExecutor for Executor {
         let file = match resp {
             Some((_, file)) if file.trashed != Some(true) => file,
             _ => {
-                return Ok(SourceValue::default());
+                return Ok(None);
             }
         };
         let ordinal = if options.include_ordinal {
@@ -415,7 +415,7 @@ impl SourceExecutor for Executor {
             }
             None => None,
         };
-        Ok(SourceValue { value, ordinal })
+        Ok(Some(SourceValue { value, ordinal }))
     }
 
     async fn change_stream(&self) -> Result<Option<BoxStream<'async_trait, SourceChange>>> {

src/ops/sources/local_file.rs

@@ -88,12 +88,9 @@ impl SourceExecutor for Executor {
         &self,
         key: &KeyValue,
         options: &SourceExecutorGetOptions,
-    ) -> Result<SourceValue> {
+    ) -> Result<Option<SourceValue>> {
         if !self.is_file_included(key.str_value()?.as_ref()) {
-            return Ok(SourceValue {
-                value: None,
-                ordinal: None,
-            });
+            return Ok(None);
         }
         let path = self.root_path.join(key.str_value()?.as_ref());
         let ordinal = if options.include_ordinal {
@@ -117,7 +114,7 @@ impl SourceExecutor for Executor {
         } else {
             None
         };
-        Ok(SourceValue { value, ordinal })
+        Ok(Some(SourceValue { value, ordinal }))
     }
 }
 

src/server.rs

@@ -61,6 +61,10 @@ pub async fn init_server(
                     "/flows/:flowInstName/data",
                     routing::get(service::flows::evaluate_data),
                 )
+                .route(
+                    "/flows/:flowInstName/rowStatus",
+                    routing::get(service::flows::get_row_index_status),
+                )
                 .route(
                     "/flows/:flowInstName/update",
                     routing::post(service::flows::update),

src/service/flows.rs

@@ -1,6 +1,6 @@
 use crate::prelude::*;
 
-use crate::execution::{evaluator, memoization, row_indexer, stats};
+use crate::execution::{evaluator, indexing_status, memoization, row_indexer, stats};
 use crate::lib_context::LibContext;
 use crate::{base::schema::FlowSchema, ops::interface::SourceExecutorListOptions};
 use axum::{
@@ -100,7 +100,7 @@ pub async fn get_keys(
 }
 
 #[derive(Deserialize)]
-pub struct EvaluateDataParams {
+pub struct SourceRowKeyParams {
     field: String,
     key: Vec<String>,
 }
@@ -111,56 +111,82 @@ pub struct EvaluateDataResponse {
     data: value::ScopeValue,
 }
 
+struct SourceRowKeyContextHolder<'a> {
+    plan: Arc<plan::ExecutionPlan>,
+    import_op_idx: usize,
+    schema: &'a FlowSchema,
+    key: value::KeyValue,
+}
+
+impl<'a> SourceRowKeyContextHolder<'a> {
+    async fn create(flow_ctx: &'a FlowContext, source_row_key: SourceRowKeyParams) -> Result<Self> {
+        let schema = &flow_ctx.flow.data_schema;
+        let import_op_idx = flow_ctx
+            .flow
+            .flow_instance
+            .import_ops
+            .iter()
+            .position(|op| op.name == source_row_key.field)
+            .ok_or_else(|| {
+                ApiError::new(
+                    &format!("source field not found: {}", source_row_key.field),
+                    StatusCode::BAD_REQUEST,
+                )
+            })?;
+        let plan = flow_ctx.flow.get_execution_plan().await?;
+        let import_op = &plan.import_ops[import_op_idx];
+        let field_schema = &schema.fields[import_op.output.field_idx as usize];
+        let table_schema = match &field_schema.value_type.typ {
+            schema::ValueType::Table(table) => table,
+            _ => api_bail!("field is not a table: {}", source_row_key.field),
+        };
+        let key_field = table_schema
+            .key_field()
+            .ok_or_else(|| api_error!("field {} does not have a key", source_row_key.field))?;
+        let key = value::KeyValue::from_strs(source_row_key.key, &key_field.value_type.typ)?;
+        Ok(Self {
+            plan,
+            import_op_idx,
+            schema,
+            key,
+        })
+    }
+
+    fn as_context<'b>(&'b self) -> evaluator::SourceRowEvaluationContext<'b> {
+        evaluator::SourceRowEvaluationContext {
+            plan: &self.plan,
+            import_op: &self.plan.import_ops[self.import_op_idx],
+            schema: self.schema,
+            key: &self.key,
+        }
+    }
+}
+
 pub async fn evaluate_data(
     Path(flow_name): Path<String>,
-    Query(query): Query<EvaluateDataParams>,
+    Query(query): Query<SourceRowKeyParams>,
     State(lib_context): State<Arc<LibContext>>,
 ) -> Result<Json<EvaluateDataResponse>, ApiError> {
     let flow_ctx = lib_context.get_flow_context(&flow_name)?;
-    let schema = &flow_ctx.flow.data_schema;
-
-    let import_op_idx = flow_ctx
-        .flow
-        .flow_instance
-        .import_ops
-        .iter()
-        .position(|op| op.name == query.field)
-        .ok_or_else(|| {
-            ApiError::new(
-                &format!("source field not found: {}", query.field),
-                StatusCode::BAD_REQUEST,
-            )
-        })?;
-    let plan = flow_ctx.flow.get_execution_plan().await?;
-    let import_op = &plan.import_ops[import_op_idx];
-    let field_schema = &schema.fields[import_op.output.field_idx as usize];
-    let table_schema = match &field_schema.value_type.typ {
-        schema::ValueType::Table(table) => table,
-        _ => api_bail!("field is not a table: {}", query.field),
-    };
-    let key_field = table_schema
-        .key_field()
-        .ok_or_else(|| api_error!("field {} does not have a key", query.field))?;
-    let key = value::KeyValue::from_strs(query.key, &key_field.value_type.typ)?;
-
+    let source_row_key_ctx = SourceRowKeyContextHolder::create(&flow_ctx, query).await?;
     let evaluate_output = row_indexer::evaluate_source_entry_with_memory(
-        &evaluator::SourceRowEvaluationContext {
-            plan: &plan,
-            import_op,
-            schema,
-            key: &key,
-        },
+        &source_row_key_ctx.as_context(),
         memoization::EvaluationMemoryOptions {
             enable_cache: true,
             evaluation_only: true,
         },
         &lib_context.builtin_db_pool,
     )
     .await?
-    .ok_or_else(|| api_error!("value not found for source at the specified key: {key:?}"))?;
+    .ok_or_else(|| {
+        api_error!(
+            "value not found for source at the specified key: {key:?}",
+            key = source_row_key_ctx.key
+        )
+    })?;
 
     Ok(Json(EvaluateDataResponse {
-        schema: schema.clone(),
+        schema: flow_ctx.flow.data_schema.clone(),
         data: evaluate_output.data_scope.into(),
     }))
 }
@@ -182,3 +208,18 @@ pub async fn update(
     live_updater.wait().await?;
     Ok(Json(live_updater.index_update_info()))
 }
+
+pub async fn get_row_index_status(
+    Path(flow_name): Path<String>,
+    Query(query): Query<SourceRowKeyParams>,
+    State(lib_context): State<Arc<LibContext>>,
+) -> Result<Json<indexing_status::SourceRowIndexingStatus>, ApiError> {
+    let flow_ctx = lib_context.get_flow_context(&flow_name)?;
+    let source_row_key_ctx = SourceRowKeyContextHolder::create(&flow_ctx, query).await?;
+    let index_status = indexing_status::get_source_row_indexing_status(
+        &source_row_key_ctx.as_context(),
+        &lib_context.builtin_db_pool,
+    )
+    .await?;
+    Ok(Json(index_status))
+}