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

* refactor: pack `SourceEvalContext` for common args for processing/eval

* 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/dumper.rs

@@ -11,6 +11,7 @@ use std::collections::BTreeMap;
 use std::path::{Path, PathBuf};
 use yaml_rust2::YamlEmitter;
 
+use super::evaluator::SourceRowEvaluationContext;
 use super::memoization::EvaluationMemoryOptions;
 use super::row_indexer;
 use crate::base::{schema, value};
@@ -77,10 +78,12 @@ impl<'a> Dumper<'a> {
         'a: 'b,
     {
         let data_builder = row_indexer::evaluate_source_entry_with_memory(
-            self.plan,
-            import_op,
-            self.schema,
-            key,
+            &SourceRowEvaluationContext {
+                plan: self.plan,
+                import_op,
+                schema: self.schema,
+                key,
+            },
             EvaluationMemoryOptions {
                 enable_cache: self.options.use_cache,
                 evaluation_only: true,

src/execution/evaluator.rs

@@ -446,30 +446,34 @@ async fn evaluate_op_scope(
     Ok(())
 }
 
+pub struct SourceRowEvaluationContext<'a> {
+    pub plan: &'a ExecutionPlan,
+    pub import_op: &'a AnalyzedImportOp,
+    pub schema: &'a schema::FlowSchema,
+    pub key: &'a value::KeyValue,
+}
+
 #[derive(Debug)]
 pub struct EvaluateSourceEntryOutput {
     pub data_scope: ScopeValueBuilder,
     pub collected_values: Vec<Vec<value::FieldValues>>,
 }
 
 pub async fn evaluate_source_entry(
-    plan: &ExecutionPlan,
-    import_op: &AnalyzedImportOp,
-    schema: &schema::FlowSchema,
-    key: &value::KeyValue,
+    src_eval_ctx: &SourceRowEvaluationContext<'_>,
     source_value: value::FieldValues,
     memory: &EvaluationMemory,
 ) -> Result<EvaluateSourceEntryOutput> {
-    let root_schema = &schema.schema;
+    let root_schema = &src_eval_ctx.schema.schema;
     let root_scope_value = ScopeValueBuilder::new(root_schema.fields.len());
     let root_scope_entry = ScopeEntry::new(
         ScopeKey::None,
         &root_scope_value,
         root_schema,
-        &plan.op_scope,
+        &src_eval_ctx.plan.op_scope,
     );
 
-    let table_schema = match &root_schema.fields[import_op.output.field_idx as usize]
+    let table_schema = match &root_schema.fields[src_eval_ctx.import_op.output.field_idx as usize]
         .value_type
         .typ
     {
@@ -482,12 +486,12 @@ pub async fn evaluate_source_entry(
     let scope_value =
         ScopeValueBuilder::augmented_from(&value::ScopeValue(source_value), table_schema)?;
     root_scope_entry.define_field_w_builder(
-        &import_op.output,
-        value::Value::KTable(BTreeMap::from([(key.clone(), scope_value)])),
+        &src_eval_ctx.import_op.output,
+        value::Value::KTable(BTreeMap::from([(src_eval_ctx.key.clone(), scope_value)])),
     );
 
     evaluate_op_scope(
-        &plan.op_scope,
+        &src_eval_ctx.plan.op_scope,
         RefList::Nil.prepend(&root_scope_entry),
         memory,
     )

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;