perf: tune the condition about when to list with values

Author: georgeh0

src/execution/live_updater.rs

@@ -242,7 +242,9 @@ impl SourceUpdateTask {
             let live_mode = self.options.live_mode;
             async move {
                 let update_stats = Arc::new(stats::UpdateStats::default());
-                source_context.update(&pool, &update_stats).await?;
+                source_context
+                    .update(&pool, &update_stats, /*expect_little_diff=*/ false)
+                    .await?;
                 if update_stats.has_any_change() {
                     status_tx.send_modify(|update| {
                         update.source_updates_num[source_idx] += 1;
@@ -260,7 +262,9 @@ impl SourceUpdateTask {
                         interval.tick().await;
 
                         let update_stats = Arc::new(stats::UpdateStats::default());
-                        source_context.update(&pool, &update_stats).await?;
+                        source_context
+                            .update(&pool, &update_stats, /*expect_little_diff=*/ true)
+                            .await?;
                         if update_stats.has_any_change() {
                             status_tx.send_modify(|update| {
                                 update.source_updates_num[source_idx] += 1;

src/execution/source_indexer.rs

@@ -367,6 +367,7 @@ impl SourceIndexingContext {
         self: &Arc<Self>,
         pool: &PgPool,
         update_stats: &Arc<stats::UpdateStats>,
+        expect_little_diff: bool,
     ) -> Result<()> {
         let pending_update_fut = {
             let mut pending_update = self.pending_update.lock().unwrap();
@@ -383,7 +384,8 @@ impl SourceIndexingContext {
                             let mut pending_update = slf.pending_update.lock().unwrap();
                             *pending_update = None;
                         }
-                        slf.update_once(&pool, &update_stats).await?;
+                        slf.update_once(&pool, &update_stats, expect_little_diff)
+                            .await?;
                     }
                     anyhow::Ok(())
                 });
@@ -406,17 +408,18 @@ impl SourceIndexingContext {
         self: &Arc<Self>,
         pool: &PgPool,
         update_stats: &Arc<stats::UpdateStats>,
+        expect_little_diff: bool,
     ) -> Result<()> {
         let plan = self.flow.get_execution_plan().await?;
         let import_op = &plan.import_ops[self.source_idx];
-        let rows_stream = import_op
-            .executor
-            .list(&interface::SourceExecutorReadOptions {
-                include_ordinal: true,
-                include_content_version_fp: true,
-                include_value: true,
-            })
-            .await?;
+        let read_options = interface::SourceExecutorReadOptions {
+            include_ordinal: true,
+            include_content_version_fp: true,
+            // When only a little diff is expected and the source provides ordinal, we don't fetch values during `list()` by default,
+            // as there's a high chance that we don't need the values at all
+            include_value: !(expect_little_diff && import_op.executor.provides_ordinal()),
+        };
+        let rows_stream = import_op.executor.list(&read_options).await?;
         self.update_with_stream(import_op, rows_stream, pool, update_stats)
             .await
     }

src/ops/interface.rs

@@ -154,6 +154,8 @@ pub trait SourceExecutor: Send + Sync {
     ) -> Result<Option<BoxStream<'async_trait, Result<SourceChangeMessage>>>> {
         Ok(None)
     }
+
+    fn provides_ordinal(&self) -> bool;
 }
 
 #[async_trait]

src/ops/sources/amazon_s3.rs

@@ -188,6 +188,10 @@ impl SourceExecutor for Executor {
         };
         Ok(Some(stream.boxed()))
     }
+
+    fn provides_ordinal(&self) -> bool {
+        true
+    }
 }
 
 #[derive(Debug, Deserialize)]

src/ops/sources/azure_blob.rs

@@ -166,6 +166,10 @@ impl SourceExecutor for Executor {
         // Azure Blob Storage doesn't have built-in change notifications like S3+SQS
         Ok(None)
     }
+
+    fn provides_ordinal(&self) -> bool {
+        true
+    }
 }
 
 pub struct Factory;

src/ops/sources/google_drive.rs

@@ -435,6 +435,10 @@ impl SourceExecutor for Executor {
         };
         Ok(Some(stream.boxed()))
     }
+
+    fn provides_ordinal(&self) -> bool {
+        true
+    }
 }
 
 pub struct Factory;

src/ops/sources/local_file.rs

@@ -115,6 +115,10 @@ impl SourceExecutor for Executor {
             content_version_fp: None,
         })
     }
+
+    fn provides_ordinal(&self) -> bool {
+        true
+    }
 }
 
 pub struct Factory;

src/ops/sources/postgres.rs

@@ -479,6 +479,10 @@ impl SourceExecutor for Executor {
 
         Ok(data)
     }
+
+    fn provides_ordinal(&self) -> bool {
+        self.table_schema.ordinal_field_schema.is_some()
+    }
 }
 
 pub struct Factory;