fix: fix list() interface and use Postgres instead of PostgresDb

Author: badmonster0

examples/postgres_embedding/README.md

@@ -107,7 +107,7 @@ The example uses these environment variables to configure the PostgreSQL source:
 
 The example demonstrates a simple flow:
 
-1. **Read from Source**: Uses `cocoindex.sources.PostgresDb` to read from your existing table
+1. **Read from Source**: Uses `cocoindex.sources.Postgres` to read from your existing table
 2. **Generate Embeddings**: Processes text and creates embeddings using SentenceTransformers
 3. **Store Embeddings**: Exports to the CocoIndex database with automatic table creation
 4. **Search**: Provides interactive semantic search over the stored embeddings

examples/postgres_embedding/main.py

@@ -123,7 +123,7 @@ def postgres_embedding_flow(
         postgres_source_kwargs["ordinal_column"] = ordinal_column
 
     data_scope["documents"] = flow_builder.add_source(
-        cocoindex.sources.PostgresDb(**postgres_source_kwargs)
+        cocoindex.sources.Postgres(**postgres_source_kwargs)
     )
 
     document_embeddings = data_scope.add_collector()

python/cocoindex/sources.py

@@ -69,7 +69,7 @@ class AzureBlob(op.SourceSpec):
     account_access_key: TransientAuthEntryReference[str] | None = None
 
 
-class PostgresDb(op.SourceSpec):
+class Postgres(op.SourceSpec):
     """Import data from a PostgreSQL table."""
 
     _op_category = op.OpCategory.SOURCE

src/ops/registration.rs

@@ -12,7 +12,7 @@ fn register_executor_factories(registry: &mut ExecutorFactoryRegistry) -> Result
     sources::google_drive::Factory.register(registry)?;
     sources::amazon_s3::Factory.register(registry)?;
     sources::azure_blob::Factory.register(registry)?;
-    sources::postgres_db::Factory.register(registry)?;
+    sources::postgres::Factory.register(registry)?;
 
     functions::parse_json::Factory.register(registry)?;
     functions::split_recursively::register(registry)?;

src/ops/sources/mod.rs

@@ -4,4 +4,4 @@ pub mod amazon_s3;
 pub mod azure_blob;
 pub mod google_drive;
 pub mod local_file;
-pub mod postgres_db;
+pub mod postgres;

src/ops/sources/postgres_db.rs renamed to src/ops/sources/postgres.rs

@@ -1,15 +1,8 @@
-use crate::fields_value;
-use async_stream::try_stream;
-use futures::TryStreamExt;
-use log::info;
-use std::sync::Arc;
-
-use crate::base::spec;
 use crate::ops::sdk::*;
+
+use crate::fields_value;
 use crate::settings::DatabaseConnectionSpec;
-use async_trait::async_trait;
-use sqlx::Column;
-use sqlx::{PgPool, Row};
+use sqlx::{Column, PgPool, Row};
 
 #[derive(Debug, Deserialize)]
 pub struct Spec {
@@ -227,11 +220,11 @@ fn convert_pg_value_to_json(
 
 #[async_trait]
 impl SourceExecutor for Executor {
-    fn list<'a>(
-        &'a self,
-        _options: &'a SourceExecutorListOptions,
-    ) -> BoxStream<'a, Result<Vec<PartialSourceRowMetadata>>> {
-        try_stream! {
+    async fn list(
+        &self,
+        _options: &SourceExecutorListOptions,
+    ) -> Result<BoxStream<'async_trait, Result<Vec<PartialSourceRowMetadata>>>> {
+        let stream = try_stream! {
             // Build query to select primary key columns
             let pk_columns: Vec<String> = self.table_schema.primary_key_columns
                 .iter()
@@ -303,6 +296,7 @@ impl SourceExecutor for Executor {
                     key,
                     key_aux_info: serde_json::Value::Null,
                     ordinal: Some(Ordinal::unavailable()),
+                    content_version_fp: None,
                 });
 
                 if batch.len() >= batch_size {
@@ -314,8 +308,8 @@ impl SourceExecutor for Executor {
             if !batch.is_empty() {
                 yield batch;
             }
-        }
-        .boxed()
+        };
+        Ok(stream.boxed())
     }
 
     async fn get_value(
@@ -520,7 +514,11 @@ impl SourceExecutor for Executor {
             None
         };
 
-        Ok(PartialSourceRowData { value, ordinal })
+        Ok(PartialSourceRowData {
+            value,
+            ordinal,
+            content_version_fp: None,
+        })
     }
 }
 
@@ -531,7 +529,7 @@ impl SourceFactoryBase for Factory {
     type Spec = Spec;
 
     fn name(&self) -> &str {
-        "PostgresDb"
+        "Postgres"
     }
 
     async fn get_output_schema(