feat(client): Table.load_columns() REST client for LOAD COLUMNS

Geneva Table.load_columns() parity on the REST-only client. Fills existing
columns from an external Parquet/Lance/IPC source by primary-key join.

- BaseTable::load_columns default (NotSupported) + public Table::load_columns,
  taking a LoadColumnsRequest (source uris/format/storage_options, target/source
  key, (target, source?) column mappings, on_missing, worker/batch/commit knobs).
- Remote impl POSTs to /v1/table/{id}/load_columns with the matching body;
  mock test asserts the request shape.
- PyO3 binding + Python remote Table.load_columns(source, pk, columns, *,
  source_format, source_pk, on_missing, ...) accepting a column list or
  {target: source} dict.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

python/python/lancedb/remote/table.py

@@ -962,6 +962,70 @@ class RemoteTable(Table):
             )
         )
 
+    def load_columns(
+        self,
+        source: Union[str, Iterable[str]],
+        pk: str,
+        columns: Union[Iterable[str], Dict[str, str]],
+        *,
+        source_format: str = "parquet",
+        source_pk: Optional[str] = None,
+        on_missing: str = "carry",
+        source_storage_options: Optional[Dict[str, str]] = None,
+        num_workers: Optional[int] = None,
+        max_workers: Optional[int] = None,
+        batch_size: Optional[int] = None,
+        commit_granularity: Optional[int] = None,
+        priority: Optional[str] = None,
+    ) -> str:
+        """Fill existing columns from an external source by primary-key join.
+
+        The distributed-job equivalent of Geneva's ``Table.load_columns()``:
+        imports precomputed values (e.g. embeddings) from Parquet/Lance/IPC into
+        this table, matching on a primary key. Returns the load job id.
+        Server-backed feature (LanceDB Enterprise / Cloud).
+
+        Parameters
+        ----------
+        source: str | list[str]
+            One source URI or a list of URIs.
+        pk: str
+            Destination primary-key column. Also the source key unless
+            ``source_pk`` is given.
+        columns: list[str] | dict[str, str]
+            Value columns to load. A list loads same-named columns; a dict maps
+            ``{target: source}``.
+        source_format: str
+            ``"parquet"`` (default), ``"lance"``, or ``"ipc"``.
+        source_pk: str, optional
+            Source primary-key column when it differs from ``pk``.
+        on_missing: str
+            Behavior for destination rows with no source match:
+            ``"carry"`` (default, keep existing), ``"null"``, or ``"error"``.
+        """
+        if isinstance(source, str):
+            source = [source]
+        if isinstance(columns, dict):
+            mappings = [(target, src) for target, src in columns.items()]
+        else:
+            mappings = [(c, None) for c in columns]
+        return LOOP.run(
+            self._table.load_columns(
+                list(source),
+                source_format,
+                pk,
+                mappings,
+                source_key=source_pk,
+                source_storage_options=source_storage_options,
+                on_missing=on_missing,
+                num_workers=num_workers,
+                max_workers=max_workers,
+                batch_size=batch_size,
+                commit_granularity=commit_granularity,
+                priority=priority,
+            )
+        )
+
     def alter_columns(
         self, *alterations: Iterable[Dict[str, str]]
     ) -> AlterColumnsResult:

python/src/table.rs

@@ -21,6 +21,7 @@ use lancedb::blob::BlobFile;
 use lancedb::index::scalar::FtsIndexBuilder;
 use lancedb::table::{
     AddDataMode, ColumnAlteration, Duration, FieldMetadataUpdate, FtsToken as LanceDbFtsToken,
+    LoadColumnsRequest,
     NewColumnTransform, OptimizeAction, OptimizeOptions, Ref, Table as LanceDbTable,
 };
 use lancedb::tokenize as lancedb_tokenize;
@@ -1334,6 +1335,43 @@ impl Table {
         })
     }
 
+    #[allow(clippy::too_many_arguments)]
+    #[pyo3(signature = (source_uris, source_format, target_key, columns, source_key=None, source_storage_options=None, on_missing=None, num_workers=None, max_workers=None, batch_size=None, commit_granularity=None, priority=None))]
+    pub fn load_columns(
+        self_: PyRef<'_, Self>,
+        source_uris: Vec<String>,
+        source_format: String,
+        target_key: String,
+        columns: Vec<(String, Option<String>)>,
+        source_key: Option<String>,
+        source_storage_options: Option<std::collections::HashMap<String, String>>,
+        on_missing: Option<String>,
+        num_workers: Option<u32>,
+        max_workers: Option<u32>,
+        batch_size: Option<u32>,
+        commit_granularity: Option<u32>,
+        priority: Option<String>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        let request = LoadColumnsRequest {
+            source_uris,
+            source_format,
+            source_storage_options,
+            target_key,
+            source_key,
+            columns,
+            on_missing,
+            num_workers,
+            max_workers,
+            batch_size,
+            commit_granularity,
+            priority,
+        };
+        future_into_py(self_.py(), async move {
+            inner.load_columns(request).await.infer_error()
+        })
+    }
+
     pub fn add_columns(
         self_: PyRef<'_, Self>,
         definitions: Vec<(String, String)>,

rust/lancedb/src/remote/table.rs

@@ -2484,6 +2484,64 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
         Ok(body.job_id)
     }
 
+    async fn load_columns(&self, request: crate::table::LoadColumnsRequest) -> Result<String> {
+        let columns: Vec<serde_json::Value> = request
+            .columns
+            .iter()
+            .map(|(target, source)| {
+                serde_json::json!({
+                    "target": target,
+                    "source": source.clone().unwrap_or_else(|| target.clone()),
+                })
+            })
+            .collect();
+        let mut source = serde_json::json!({
+            "uris": request.source_uris,
+            "format": request.source_format,
+        });
+        if let Some(opts) = request.source_storage_options {
+            source["storage_options"] = serde_json::to_value(opts).unwrap_or_default();
+        }
+        let mut body = serde_json::json!({
+            "columns": columns,
+            "source": source,
+            "target_key": request.target_key,
+        });
+        if let Some(k) = request.source_key {
+            body["source_key"] = serde_json::Value::String(k);
+        }
+        if let Some(m) = request.on_missing {
+            body["on_missing"] = serde_json::Value::String(m);
+        }
+        if let Some(n) = request.num_workers {
+            body["num_workers"] = n.into();
+        }
+        if let Some(n) = request.max_workers {
+            body["max_workers"] = n.into();
+        }
+        if let Some(n) = request.batch_size {
+            body["batch_size"] = n.into();
+        }
+        if let Some(n) = request.commit_granularity {
+            body["commit_granularity"] = n.into();
+        }
+        if let Some(p) = request.priority {
+            body["priority"] = serde_json::Value::String(p);
+        }
+        let http_request = self
+            .client
+            .post(&format!("/v1/table/{}/load_columns", self.identifier))
+            .json(&body);
+        let (request_id, response) = self.send(http_request, true).await?;
+        let response = self.check_table_response(&request_id, response).await?;
+        #[derive(serde::Deserialize)]
+        struct LoadColumnsResponse {
+            job_id: String,
+        }
+        let body: LoadColumnsResponse = response.json().await.err_to_http(request_id)?;
+        Ok(body.job_id)
+    }
+
     async fn add_columns(
         &self,
         transforms: NewColumnTransform,
@@ -3002,6 +3060,51 @@ mod tests {
         assert_eq!(job_id, "j-9");
     }
 
+    #[tokio::test]
+    async fn test_load_columns() {
+        let table = Table::new_with_handler("my_table", |request| {
+            assert_eq!(request.method(), "POST");
+            assert_eq!(request.url().path(), "/v1/table/my_table/load_columns");
+            let body: serde_json::Value =
+                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
+            assert_eq!(
+                body["columns"],
+                serde_json::json!([{"target": "embedding", "source": "emb"}])
+            );
+            assert_eq!(body["source"]["format"], "parquet");
+            assert_eq!(
+                body["source"]["uris"],
+                serde_json::json!(["s3://b/x.parquet"])
+            );
+            assert_eq!(body["target_key"], "document_id");
+            assert_eq!(body["source_key"], "doc_id");
+            assert_eq!(body["on_missing"], "null");
+            assert_eq!(body["num_workers"], 4);
+
+            http::Response::builder()
+                .status(202)
+                .body(r#"{"job_id":"lc-7"}"#)
+                .unwrap()
+        });
+
+        let request = crate::table::LoadColumnsRequest {
+            source_uris: vec!["s3://b/x.parquet".to_string()],
+            source_format: "parquet".to_string(),
+            source_storage_options: None,
+            target_key: "document_id".to_string(),
+            source_key: Some("doc_id".to_string()),
+            columns: vec![("embedding".to_string(), Some("emb".to_string()))],
+            on_missing: Some("null".to_string()),
+            num_workers: Some(4),
+            max_workers: None,
+            batch_size: None,
+            commit_granularity: None,
+            priority: None,
+        };
+        let job_id = table.load_columns(request).await.unwrap();
+        assert_eq!(job_id, "lc-7");
+    }
+
     #[tokio::test]
     async fn test_version() {
         let table = Table::new_with_handler("my_table", |request| {

rust/lancedb/src/table.rs

@@ -502,6 +502,33 @@ pub fn tokenize(query: &str, params: &InvertedIndexParams) -> Result<Vec<FtsToke
         .collect())
 }
 
+/// Request to fill existing table columns from an external source by
+/// primary-key join (Geneva `Table.load_columns()` parity). Server-backed
+/// feature (LanceDB Enterprise / Cloud).
+#[derive(Debug, Clone)]
+pub struct LoadColumnsRequest {
+    /// External source URIs.
+    pub source_uris: Vec<String>,
+    /// Source format: "parquet" | "lance" | "ipc".
+    pub source_format: String,
+    /// Source-only storage options (e.g. cloud credentials).
+    pub source_storage_options: Option<HashMap<String, String>>,
+    /// Destination primary-key column.
+    pub target_key: String,
+    /// Source primary-key column. Defaults to `target_key` when None.
+    pub source_key: Option<String>,
+    /// Value column mappings as `(target, source)`; a None source defaults to
+    /// the target name.
+    pub columns: Vec<(String, Option<String>)>,
+    /// Missing-row policy: "carry" (default) | "null" | "error".
+    pub on_missing: Option<String>,
+    pub num_workers: Option<u32>,
+    pub max_workers: Option<u32>,
+    pub batch_size: Option<u32>,
+    pub commit_granularity: Option<u32>,
+    pub priority: Option<String>,
+}
+
 /// A trait for anything "table-like".  This is used for both native tables (which target
 /// Lance datasets) and remote tables (which target LanceDB cloud)
 ///
@@ -694,6 +721,14 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
             message: "refresh_column is not supported by this table".into(),
         })
     }
+    /// Fill existing columns from an external source by primary-key join
+    /// (Geneva `load_columns`). Returns the load job id. Server-backed feature;
+    /// the default returns NotSupported.
+    async fn load_columns(&self, _request: LoadColumnsRequest) -> Result<String> {
+        Err(Error::NotSupported {
+            message: "load_columns is not supported by this table".into(),
+        })
+    }
     /// Alter columns in the table.
     async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result<AlterColumnsResult>;
     /// Drop columns from the table.
@@ -1571,6 +1606,12 @@ impl Table {
             .await
     }
 
+    /// Fill existing columns from an external Parquet/Lance/IPC source by
+    /// primary-key join (Geneva `Table.load_columns()`). Returns the job id.
+    pub async fn load_columns(&self, request: LoadColumnsRequest) -> Result<String> {
+        self.inner.load_columns(request).await
+    }
+
     /// Change a column's name or nullability.
     pub async fn alter_columns(
         &self,