TheRakeshPurohit
diff --git a/‎Cargo.lock‎
Lines changed: 5 additions & 3 deletions b/‎Cargo.lock‎
Lines changed: 5 additions & 3 deletions
diff --git a/‎crates/runtime-datafusion-index/src/lib.rs‎
Lines changed: 11 additions & 0 deletions b/‎crates/runtime-datafusion-index/src/lib.rs‎
Lines changed: 11 additions & 0 deletions
diff --git a/‎crates/runtime/Cargo.toml‎
Lines changed: 1 addition & 0 deletions b/‎crates/runtime/Cargo.toml‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎crates/runtime/src/accelerated_table/mod.rs‎
Lines changed: 15 additions & 0 deletions b/‎crates/runtime/src/accelerated_table/mod.rs‎
Lines changed: 15 additions & 0 deletions
diff --git a/‎crates/runtime/src/accelerated_table/sink/table.rs‎
Lines changed: 15 additions & 0 deletions b/‎crates/runtime/src/accelerated_table/sink/table.rs‎
Lines changed: 15 additions & 0 deletions
diff --git a/‎crates/runtime/src/embeddings/connector.rs‎
Lines changed: 104 additions & 1 deletion b/‎crates/runtime/src/embeddings/connector.rs‎
Lines changed: 104 additions & 1 deletion
@@ -49,5 +49,16 @@ pub trait Index: Debug + Send + Sync + 'static {
         Ok(batches)
     }
 
+    /// Called after data has been written via the [`TableSink`] path (full refresh or append).
+    ///
+    /// Default is a no-op. Implementations use this to create or verify persistent structures
+    /// (e.g. a vector HNSW index) after each write. Using `IF NOT EXISTS` semantics makes it
+    /// safe to call on both overwrite (recreates on new table) and append (no-op if index
+    /// already exists). Not called for CDC writes — those maintain indexes automatically via
+    /// `DuckDB` VSS on each insert.
+    async fn on_write_complete(&self) -> Result<()> {
+        Ok(())
+    }
+
     fn as_any(&self) -> &dyn Any;
 }
@@ -294,6 +294,7 @@ duckdb = [
     "data_components/duckdb",
     "datafusion-optimizer-rules/duckdb",
     "runtime-acceleration/duckdb",
+    "search/duckdb",
 ]
 dynamodb = [
     "dep:aws-sdk-dynamodb",
 
@@ -181,6 +181,12 @@ pub enum Error {
     #[snafu(display("Failed to construct data for the accelerated dataset: {source}"))]
     FailedToBuildRecordBatch { source: ArrowError },
 
+    #[snafu(display("Failed to process upsert batch for dataset {dataset_name}: {reason}"))]
+    InvalidUpsertPrimaryKeys {
+        dataset_name: String,
+        reason: String,
+    },
+
     #[snafu(display("No primary keys defined for dataset {dataset_name}"))]
     NoPrimaryKeysDefined { dataset_name: String },
 }
@@ -1083,6 +1089,15 @@ impl AcceleratedTable {
         Arc::clone(&self.accelerator)
     }
 
+    #[must_use]
+    pub(crate) fn get_accelerator_ref(&self) -> &Arc<dyn TableProvider> {
+        &self.accelerator
+    }
+
+    pub(crate) fn set_accelerator(&mut self, accelerator: Arc<dyn TableProvider>) {
+        self.accelerator = accelerator;
+    }
+
     /// Add a child accelerator that should receive cached data when this parent stores new cache entries.
     /// This is used for localpod caching synchronization.
     pub async fn add_synchronized_child(&self, child_accelerator: Arc<dyn TableProvider>) {
 
@@ -22,6 +22,7 @@ use datafusion::{
     physical_plan::collect, prelude::SessionContext,
 };
 use runtime_datafusion::execution_plan::schema_cast::SchemaCastScanExec;
+use runtime_datafusion_index::IndexedTableProvider;
 use runtime_table_partition::provider::PartitionTableProvider;
 use util::RetryError;
 
@@ -123,6 +124,20 @@ impl TableSink {
                     // Don't fail the write - data was successfully written, index rebuild is best-effort
                 }
             }
+
+            // Call on_write_complete on every Index in an IndexedTableProvider.
+            // Uses IF NOT EXISTS semantics: creates index after overwrite (new table),
+            // no-op after append (index already exists). CDC skips this path entirely.
+            if let Some(indexed) = provider.as_any().downcast_ref::<IndexedTableProvider>() {
+                for index in indexed.get_all_indexes() {
+                    if let Err(e) = index.on_write_complete().await {
+                        tracing::warn!(
+                            "TableSink: on_write_complete failed for index '{}': {e}. Index may be stale until next refresh.",
+                            index.name()
+                        );
+                    }
+                }
+            }
         }
 
         tracing::debug!(
 
@@ -18,6 +18,8 @@ use crate::changes::Indexes;
 use crate::changes::index_change_envelope;
 use crate::component::ComponentInitialization;
 use crate::component::dataset::Dataset;
+#[cfg(feature = "duckdb")]
+use crate::component::dataset::acceleration::Engine;
 use crate::component::metrics::MetricsProvider;
 use crate::dataconnector::{DataConnector, DataConnectorError, DataConnectorResult};
 use crate::embeddings::execution_plan::{
@@ -37,6 +39,8 @@ use runtime_datafusion_index::IndexedTableProvider;
 use search::generation::text_search::index::FullTextDatabaseIndex;
 use search::index::VectorScanTableProvider;
 use spicepod::component::embeddings::ColumnEmbeddingConfig;
+#[cfg(feature = "duckdb")]
+use spicepod::{semantic::ColumnLevelEmbeddingConfig, vector::VectorStore};
 use std::any::Any;
 use std::sync::Arc;
 use tokio::sync::{Mutex, RwLock};
@@ -93,6 +97,21 @@ impl EmbeddingConnector {
         if let Some(vector_engine) = &dataset.vectors
             && vector_engine.enabled
         {
+            #[cfg(feature = "duckdb")]
+            if vector_engine.engine.as_deref() == Some("duckdb")
+                && !dataset.acceleration.as_ref().is_some_and(|acceleration| {
+                    acceleration.engine.to_unpartitioned() == Engine::DuckDB
+                })
+            {
+                return Err(DataConnectorError::InvalidConfigurationSourceOnly {
+                    dataconnector: dataset.source().to_string(),
+                    connector_component: dataset.into(),
+                    source: Box::<dyn std::error::Error + Send + Sync>::from(
+                        "DuckDB vector engine requires DuckDB acceleration. Configure the dataset with `acceleration.engine: duckdb`.",
+                    ),
+                });
+            }
+
             return wrap_table_as_index(
                 &dataset.runtime().datafusion().ctx,
                 &self.embedding_models,
@@ -248,7 +267,30 @@ impl DataConnector for EmbeddingConnector {
     ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         self.inner_connector
             .on_accelerated_table_registration(dataset, accelerated_table)
-            .await
+            .await?;
+
+        #[cfg(feature = "duckdb")]
+        if let Some(vector_engine) = duckdb_vector_store_for_accelerated_table(dataset) {
+            let embedding_columns = duckdb_embedding_columns(dataset);
+            if embedding_columns.is_empty() {
+                return Ok(());
+            }
+
+            let accelerator = accelerated_table.get_accelerator();
+            let indexed_accelerator =
+                crate::embeddings::index::duckdb::wrap_accelerator_with_duckdb_vector_indexes(
+                    &dataset.name,
+                    embedding_columns,
+                    &vector_engine,
+                    accelerator,
+                    Arc::clone(&self.embedding_models),
+                    Arc::clone(&self.secrets),
+                )
+                .await?;
+            accelerated_table.set_accelerator(indexed_accelerator);
+        }
+
+        Ok(())
     }
 
     fn supports_changes_stream(&self) -> bool {
@@ -389,6 +431,67 @@ impl DataConnector for EmbeddingConnector {
     }
 }
 
+#[cfg(feature = "duckdb")]
+fn duckdb_vector_store_for_accelerated_table(dataset: &Dataset) -> Option<VectorStore> {
+    if let Some(vector_engine) = &dataset.vectors
+        && vector_engine.enabled
+        && vector_engine.engine.as_deref() == Some("duckdb")
+    {
+        return Some(vector_engine.clone());
+    }
+
+    if !dataset.has_embeddings()
+        || !dataset
+            .acceleration
+            .as_ref()
+            .is_some_and(|acceleration| acceleration.engine.to_unpartitioned() == Engine::DuckDB)
+    {
+        return None;
+    }
+
+    let acceleration = dataset.acceleration.as_ref()?;
+    crate::embeddings::index::duckdb::vector_store_from_embedding_params(&acceleration.params)
+}
+
+#[cfg(feature = "duckdb")]
+fn duckdb_embedding_columns(dataset: &Dataset) -> Vec<(String, ColumnLevelEmbeddingConfig)> {
+    let mut embedding_columns = dataset
+        .embeddings
+        .iter()
+        .map(|embedding| {
+            (
+                embedding.column.clone(),
+                ColumnLevelEmbeddingConfig {
+                    model: embedding.model.clone(),
+                    chunking: embedding.chunking.clone(),
+                    row_ids: embedding.primary_keys.clone(),
+                    vector_size: embedding.vector_size,
+                    aggregation: embedding.aggregation,
+                    max_elements_per_row: embedding.max_elements_per_row,
+                },
+            )
+        })
+        .collect::<Vec<_>>();
+
+    for column in &dataset.columns {
+        // Must be `last()` to mimic what model `EmbeddingTable`'s HashMap ends up with.
+        let Some(embedding) = column.embeddings.last() else {
+            continue;
+        };
+
+        if let Some((_, existing)) = embedding_columns
+            .iter_mut()
+            .find(|(column_name, _)| column_name == &column.name)
+        {
+            *existing = embedding.clone();
+        } else {
+            embedding_columns.push((column.name.clone(), embedding.clone()));
+        }
+    }
+
+    embedding_columns
+}
+
 fn underlying_federated_table_for_indexed_table(
     src_table_provider: &Arc<dyn TableProvider>,
 ) -> Option<Arc<FederatedTable>> {
Original file line number	Diff line number	Diff line change
`@@ -294,6 +294,7 @@ duckdb = [`
`294`	`294`	`"data_components/duckdb",`
`295`	`295`	`"datafusion-optimizer-rules/duckdb",`
`296`	`296`	`"runtime-acceleration/duckdb",`
	`297`	`+ "search/duckdb",`
`297`	`298`	`]`
`298`	`299`	`dynamodb = [`
`299`	`300`	`"dep:aws-sdk-dynamodb",`