feat: make source get API return ordinal optionally (#470)

Author: badmonster0

src/execution/dumper.rs

@@ -167,7 +167,7 @@ impl<'a> Dumper<'a> {
     async fn evaluate_and_dump_for_source(&self, import_op: &AnalyzedImportOp) -> Result<()> {
         let mut keys_by_filename_prefix: IndexMap<String, Vec<value::KeyValue>> = IndexMap::new();
 
-        let mut rows_stream = import_op.executor.list(SourceExecutorListOptions {
+        let mut rows_stream = import_op.executor.list(&SourceExecutorListOptions {
             include_ordinal: false,
         });
         while let Some(rows) = rows_stream.next().await {

src/execution/row_indexer.rs

@@ -13,7 +13,9 @@ use super::stats;
 use crate::base::schema;
 use crate::base::value::{self, FieldValues, KeyValue};
 use crate::builder::plan::*;
-use crate::ops::interface::{ExportTargetMutation, ExportTargetUpsertEntry, Ordinal};
+use crate::ops::interface::{
+    ExportTargetMutation, ExportTargetUpsertEntry, Ordinal, SourceExecutorGetOptions,
+};
 use crate::utils::db::WriteAction;
 use crate::utils::fingerprint::{Fingerprint, Fingerprinter};
 
@@ -460,7 +462,18 @@ pub async fn evaluate_source_entry_with_memory(
         None
     };
     let memory = EvaluationMemory::new(chrono::Utc::now(), stored_info, options);
-    let source_value = match import_op.executor.get_value(key).await? {
+    let source_value = match import_op
+        .executor
+        .get_value(
+            key,
+            &SourceExecutorGetOptions {
+                include_value: true,
+                include_ordinal: false,
+            },
+        )
+        .await?
+        .value
+    {
         Some(d) => d,
         None => return Ok(None),
     };

src/execution/source_indexer.rs

@@ -101,7 +101,17 @@ impl SourceIndexingContext {
                 // also happens for update cases and there's no way to keep them always in sync for many sources.
                 //
                 // We only need source version <= actual version for value.
-                import_op.executor.get_value(&key).await?
+                import_op
+                    .executor
+                    .get_value(
+                        &key,
+                        &interface::SourceExecutorGetOptions {
+                            include_value: true,
+                            include_ordinal: false,
+                        },
+                    )
+                    .await?
+                    .value
             };
             let schema = &self.flow.data_schema;
             let result = row_indexer::update_source_row(
@@ -203,7 +213,7 @@ impl SourceIndexingContext {
         let import_op = &plan.import_ops[self.source_idx];
         let mut rows_stream = import_op
             .executor
-            .list(interface::SourceExecutorListOptions {
+            .list(&interface::SourceExecutorListOptions {
                 include_ordinal: true,
             });
         let mut join_set = JoinSet::new();

src/ops/interface.rs

@@ -66,16 +66,34 @@ pub struct SourceExecutorListOptions {
     pub include_ordinal: bool,
 }
 
+#[derive(Debug, Default)]
+pub struct SourceExecutorGetOptions {
+    pub include_ordinal: bool,
+    pub include_value: bool,
+}
+
+#[derive(Debug, Default)]
+pub struct SourceValue {
+    // None if not exists, or not included in the option.
+    pub value: Option<FieldValues>,
+    // None if unavailable, or not included in the option.
+    pub ordinal: Option<Ordinal>,
+}
+
 #[async_trait]
 pub trait SourceExecutor: Send + Sync {
     /// Get the list of keys for the source.
-    fn list(
-        &self,
-        options: SourceExecutorListOptions,
-    ) -> BoxStream<'_, Result<Vec<SourceRowMetadata>>>;
+    fn list<'a>(
+        &'a self,
+        options: &'a SourceExecutorListOptions,
+    ) -> BoxStream<'a, Result<Vec<SourceRowMetadata>>>;
 
     // Get the value for the given key.
-    async fn get_value(&self, key: &KeyValue) -> Result<Option<FieldValues>>;
+    async fn get_value(
+        &self,
+        key: &KeyValue,
+        options: &SourceExecutorGetOptions,
+    ) -> Result<SourceValue>;
 
     async fn change_stream(&self) -> Result<Option<BoxStream<'async_trait, SourceChange>>> {
         Ok(None)

src/ops/sources/google_drive.rs

@@ -292,21 +292,25 @@ impl<T> ResultExt<T> for google_drive3::Result<T> {
     }
 }
 
+fn optional_modified_time(include_ordinal: bool) -> &'static str {
+    if include_ordinal {
+        ",modifiedTime"
+    } else {
+        ""
+    }
+}
+
 #[async_trait]
 impl SourceExecutor for Executor {
-    fn list(
-        &self,
-        options: SourceExecutorListOptions,
-    ) -> BoxStream<'_, Result<Vec<SourceRowMetadata>>> {
+    fn list<'a>(
+        &'a self,
+        options: &'a SourceExecutorListOptions,
+    ) -> BoxStream<'a, Result<Vec<SourceRowMetadata>>> {
         let mut seen_ids = HashSet::new();
         let mut folder_ids = self.root_folder_ids.clone();
         let fields = format!(
             "files(id,name,mimeType,trashed{})",
-            if options.include_ordinal {
-                ",modifiedTime"
-            } else {
-                ""
-            }
+            optional_modified_time(options.include_ordinal)
         );
         let mut new_folder_ids = Vec::new();
         try_stream! {
@@ -333,20 +337,35 @@ impl SourceExecutor for Executor {
         .boxed()
     }
 
-    async fn get_value(&self, key: &KeyValue) -> Result<Option<FieldValues>> {
+    async fn get_value(
+        &self,
+        key: &KeyValue,
+        options: &SourceExecutorGetOptions,
+    ) -> Result<SourceValue> {
         let file_id = key.str_value()?;
+        let fields = format!(
+            "id,name,mimeType,trashed{}",
+            optional_modified_time(options.include_ordinal)
+        );
         let resp = self
             .drive_hub
             .files()
             .get(file_id)
             .add_scope(Scope::Readonly)
-            .param("fields", "id,name,mimeType,trashed")
+            .param("fields", &fields)
             .doit()
             .await
             .or_not_found()?;
         let file = match resp {
             Some((_, file)) if file.trashed != Some(true) => file,
-            _ => return Ok(None),
+            _ => {
+                return Ok(SourceValue::default());
+            }
+        };
+        let ordinal = if options.include_ordinal {
+            file.modified_time.map(|t| t.try_into()).transpose()?
+        } else {
+            None
         };
         let type_n_body = if let Some(export_mime_type) = file
             .mime_type
@@ -396,7 +415,7 @@ impl SourceExecutor for Executor {
             }
             None => None,
         };
-        Ok(value)
+        Ok(SourceValue { value, ordinal })
     }
 
     async fn change_stream(&self) -> Result<Option<BoxStream<'async_trait, SourceChange>>> {

src/ops/sources/local_file.rs

@@ -40,10 +40,10 @@ impl Executor {
 
 #[async_trait]
 impl SourceExecutor for Executor {
-    fn list(
-        &self,
-        options: SourceExecutorListOptions,
-    ) -> BoxStream<'_, Result<Vec<SourceRowMetadata>>> {
+    fn list<'a>(
+        &'a self,
+        options: &'a SourceExecutorListOptions,
+    ) -> BoxStream<'a, Result<Vec<SourceRowMetadata>>> {
         let root_component_size = self.root_path.components().count();
         let mut dirs = Vec::new();
         dirs.push(Cow::Borrowed(&self.root_path));
@@ -84,24 +84,40 @@ impl SourceExecutor for Executor {
         .boxed()
     }
 
-    async fn get_value(&self, key: &KeyValue) -> Result<Option<FieldValues>> {
+    async fn get_value(
+        &self,
+        key: &KeyValue,
+        options: &SourceExecutorGetOptions,
+    ) -> Result<SourceValue> {
         if !self.is_file_included(key.str_value()?.as_ref()) {
-            return Ok(None);
+            return Ok(SourceValue {
+                value: None,
+                ordinal: None,
+            });
         }
         let path = self.root_path.join(key.str_value()?.as_ref());
-        let value = match std::fs::read(path) {
-            Ok(content) => {
-                let content = if self.binary {
-                    fields_value!(content)
-                } else {
-                    fields_value!(String::from_utf8_lossy(&content).to_string())
-                };
-                Some(content)
+        let ordinal = if options.include_ordinal {
+            Some(path.metadata()?.modified()?.try_into()?)
+        } else {
+            None
+        };
+        let value = if options.include_value {
+            match std::fs::read(path) {
+                Ok(content) => {
+                    let content = if self.binary {
+                        fields_value!(content)
+                    } else {
+                        fields_value!(String::from_utf8_lossy(&content).to_string())
+                    };
+                    Some(content)
+                }
+                Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
+                Err(e) => Err(e)?,
             }
-            Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
-            Err(e) => Err(e)?,
+        } else {
+            None
         };
-        Ok(value)
+        Ok(SourceValue { value, ordinal })
     }
 }
 

src/service/flows.rs

@@ -89,7 +89,7 @@ pub async fn get_keys(
             )
         })?;
 
-    let mut rows_stream = import_op.executor.list(SourceExecutorListOptions {
+    let mut rows_stream = import_op.executor.list(&SourceExecutorListOptions {
         include_ordinal: false,
     });
     let mut keys = Vec::new();