refactor: expose fragment publication API

Cargo.lock

@@ -5473,6 +5473,7 @@ dependencies = [
  "serde_json",
  "serde_with",
  "serial_test",
+ "sha2 0.10.9",
  "snafu 0.8.9",
  "tempfile",
  "test-log",

Cargo.toml

@@ -69,6 +69,7 @@ regex = "1.10"
 semver = "1.0.25"
 serde = "1"
 serde_json = "1"
+sha2 = "0.10.9"
 tempfile = "3.5.0"
 tokio = { version = "1.23", features = ["rt-multi-thread", "sync"] }
 uuid = { version = "1.7.0", features = ["v4"] }

rust/lancedb/Cargo.toml

@@ -92,6 +92,7 @@ candle-transformers = { version = "0.9.1", optional = true }
 candle-nn = { version = "0.9.1", optional = true }
 tokenizers = { version = "0.19.1", optional = true }
 semver = { workspace = true }
+sha2 = { workspace = true }
 
 [dev-dependencies]
 anyhow = "1"

rust/lancedb/src/table.rs

@@ -73,6 +73,7 @@ mod create_index;
 pub mod datafusion;
 pub(crate) mod dataset;
 pub mod delete;
+pub mod fragment_publication;
 pub mod lsm_stats;
 pub mod merge;
 pub mod optimize;

rust/lancedb/src/table/refresh.rs

@@ -83,7 +83,10 @@ pub(crate) async fn execute_refresh_column(
         }
         rows_filled += gained;
         let values = fill_stream(&dataset, &fragment, bound.clone(), column).await?;
-        replacements.push(fragment.write_columns(values, &column_schema).await?);
+        replacements.push(
+            super::fragment_publication::stage_fragment_columns(&fragment, values, &column_schema)
+                .await?,
+        );
     }
 
     if replacements.is_empty() {
@@ -140,7 +143,7 @@ pub(crate) async fn execute_refresh_column_async(table: &NativeTable, column: &s
 /// Refresh enumerates base fragments, and a write spec keeps visible rows in
 /// un-compacted MemWAL tiers it cannot reach -- success would silently omit
 /// readable rows.
-async fn ensure_no_lsm_write_spec(table: &NativeTable) -> Result<()> {
+pub(crate) async fn ensure_no_lsm_write_spec(table: &NativeTable) -> Result<()> {
     // The catch-up flag outlives unset and marks retained SSTable rows.
     let catchup = table.dataset.get().await?.manifest().reader_feature_flags
         & lance_table::feature_flags::FLAG_MEM_WAL_INDEX_CATCHUP