feat: support blob v2 compaction workflow

- update lance submodule to include blob v2 compaction and zero-copy support

- add sf-cli db commands for blob compaction e2e testing and audio verification

- enable songs table compaction in backend table compactor

- move blob storage optimization writeup to docs and add compaction deep dive

Author: acking-you

backend/src/state.rs

@@ -1,5 +1,5 @@
 use std::{
-    collections::HashMap,
+    collections::{HashMap, HashSet},
     env,
     sync::Arc,
     time::{Duration, Instant},
@@ -406,9 +406,7 @@ fn spawn_table_compactor(
     let config = CompactConfig {
         fragment_threshold: threshold,
         prune_older_than_hours: 2,
-        // songs table uses blob v2 encoding (data_storage_version=2.2) which
-        // the current lance version cannot compact yet.
-        skip_tables: ["songs"].iter().map(|s| s.to_string()).collect(),
+        skip_tables: HashSet::new(),
     };
 
     tokio::spawn(async move {

cli/src/cli.rs

@@ -524,4 +524,22 @@ pub enum DbCommands {
         /// Target version number.
         version: u64,
     },
+    /// Run blob v2 compaction e2e test with synthetic data.
+    TestBlobCompact {
+        /// Number of synthetic songs to insert (default: 5).
+        #[arg(long, default_value = "5")]
+        count: usize,
+        /// Size of each synthetic audio blob in bytes (default: 5MB).
+        #[arg(long, default_value = "5242880")]
+        blob_size: usize,
+    },
+    /// Verify audio data retrieval for songs in the database.
+    VerifyAudio {
+        /// Only verify specific song IDs (comma-separated).
+        #[arg(long)]
+        ids: Option<String>,
+        /// Maximum number of songs to verify (default: all).
+        #[arg(long)]
+        limit: Option<usize>,
+    },
 }

cli/src/commands/db_manage.rs

@@ -1215,3 +1215,234 @@ fn downcast_timestamp_ms<'a>(
         .downcast_ref::<TimestampMillisecondArray>()
         .ok_or_else(|| anyhow!("column `{column}` is not TimestampMillisecondArray"))
 }
+
+// ---------------------------------------------------------------------------
+// Blob V2 Compaction E2E Test
+// ---------------------------------------------------------------------------
+
+pub async fn test_blob_compact(db_path: &Path, count: usize, blob_size: usize) -> Result<()> {
+    use std::time::Instant;
+
+    use static_flow_shared::music_store::{MusicDataStore, SongRecord};
+
+    if count == 0 {
+        bail!("count must be at least 1");
+    }
+
+    let test_dir = db_path.join("_test_blob_compact");
+    if test_dir.exists() {
+        std::fs::remove_dir_all(&test_dir)?;
+    }
+    std::fs::create_dir_all(&test_dir)?;
+    let db_uri = test_dir.to_string_lossy().to_string();
+
+    tracing::info!(
+        "=== Blob V2 Compaction E2E Test ===\n  songs: {count}\n  blob_size: {} bytes\n  test_db: \
+         {db_uri}",
+        blob_size
+    );
+
+    let store = MusicDataStore::connect(&db_uri).await?;
+    let now_ms = chrono::Utc::now().timestamp_millis();
+
+    // Step 1: insert N songs (each as separate fragment)
+    let start = Instant::now();
+    for i in 0..count {
+        let record = SongRecord {
+            id: format!("test-song-{i:04}"),
+            title: format!("Test Song {i}"),
+            artist: "CompactionTest".into(),
+            album: "BlobV2Test".into(),
+            album_id: None,
+            cover_image: None,
+            duration_ms: 180_000,
+            format: "mp3".into(),
+            bitrate: 320,
+            lyrics_lrc: None,
+            lyrics_translation: None,
+            audio_data: vec![(i as u8).wrapping_add(42); blob_size],
+            source: "test".into(),
+            source_id: None,
+            tags: "test,compaction".into(),
+            searchable_text: format!("Test Song {i} CompactionTest"),
+            vector_en: None,
+            vector_zh: None,
+            created_at: now_ms,
+            updated_at: now_ms,
+        };
+        store.upsert_song(&record).await?;
+    }
+    let insert_ms = start.elapsed().as_millis();
+    tracing::info!("Inserted {count} songs in {insert_ms}ms");
+
+    // Step 2: count fragments before compaction
+    let table = store.connection().open_table("songs").execute().await?;
+    let ds_before = table.dataset().context("no dataset")?;
+    let frags_before = ds_before.get().await?.get_fragments().len();
+    tracing::info!("Fragments before compact: {frags_before}");
+
+    // Step 3: compact
+    let compact_start = Instant::now();
+    table.optimize(OptimizeAction::All).await?;
+    let compact_ms = compact_start.elapsed().as_millis();
+    tracing::info!("Compact completed in {compact_ms}ms");
+
+    // Step 4: fragments after
+    let ds_after = table.dataset().context("no dataset")?;
+    let frags_after = ds_after.get().await?.get_fragments().len();
+    tracing::info!("Fragments after compact: {frags_after}");
+
+    // Step 5: prune
+    table
+        .optimize(OptimizeAction::Prune {
+            older_than: Some(ChronoDuration::zero()),
+            delete_unverified: Some(true),
+            error_if_tagged_old_versions: Some(false),
+        })
+        .await?;
+    tracing::info!("Prune completed");
+
+    // Step 6: verify audio data integrity
+    let mut pass = 0usize;
+    let mut fail = 0usize;
+    for i in 0..count {
+        let id = format!("test-song-{i:04}");
+        match store.get_song_audio(&id).await {
+            Ok(Some((data, fmt))) => {
+                let expected_byte = (i as u8).wrapping_add(42);
+                let size_ok = data.len() == blob_size;
+                let content_ok = data[0] == expected_byte && data[data.len() - 1] == expected_byte;
+                let fmt_ok = fmt == "mp3";
+                if size_ok && content_ok && fmt_ok {
+                    pass += 1;
+                } else {
+                    fail += 1;
+                    tracing::error!(
+                        "FAIL {id}: size={}/{blob_size} first_byte={}/{expected_byte} format={fmt}",
+                        data.len(),
+                        data[0]
+                    );
+                }
+            },
+            Ok(None) => {
+                fail += 1;
+                tracing::error!("FAIL {id}: audio not found");
+            },
+            Err(err) => {
+                fail += 1;
+                tracing::error!("FAIL {id}: {err}");
+            },
+        }
+    }
+
+    // Step 7: verify metadata
+    for i in 0..count {
+        let id = format!("test-song-{i:04}");
+        match store.get_song(&id).await {
+            Ok(Some(detail)) => {
+                if detail.title != format!("Test Song {i}") || detail.artist != "CompactionTest" {
+                    fail += 1;
+                    tracing::error!(
+                        "FAIL {id}: metadata mismatch title={} artist={}",
+                        detail.title,
+                        detail.artist
+                    );
+                }
+            },
+            Ok(None) => {
+                fail += 1;
+                tracing::error!("FAIL {id}: metadata not found");
+            },
+            Err(err) => {
+                fail += 1;
+                tracing::error!("FAIL {id}: metadata error {err}");
+            },
+        }
+    }
+
+    // Cleanup
+    if let Err(err) = std::fs::remove_dir_all(&test_dir) {
+        tracing::warn!("Failed to cleanup test dir: {err}");
+    }
+
+    let total_ms = start.elapsed().as_millis();
+    tracing::info!(
+        "\n=== Result ===\n  pass: {pass}/{count}\n  fail: {fail}\n  fragments: {frags_before} -> \
+         {frags_after}\n  total: {total_ms}ms"
+    );
+
+    if fail > 0 {
+        bail!("{fail} verification(s) failed");
+    }
+    Ok(())
+}
+
+// ---------------------------------------------------------------------------
+// Verify Audio Data Retrieval
+// ---------------------------------------------------------------------------
+
+pub async fn verify_audio(db_path: &Path, ids: Option<String>, limit: Option<usize>) -> Result<()> {
+    use static_flow_shared::music_store::MusicDataStore;
+
+    let db_uri = db_path.to_string_lossy().to_string();
+    let store = MusicDataStore::connect(&db_uri).await?;
+
+    let target_ids: Vec<String> = match ids {
+        Some(ref csv) => csv.split(',').map(|s| s.trim().to_string()).collect(),
+        None => {
+            // Query all song IDs from the table
+            let db = connect_db(db_path).await?;
+            let table = open_table(&db, "songs").await?;
+            let mut query = table.query();
+            if let Some(lim) = limit {
+                query = query.limit(lim);
+            }
+            let stream = query
+                .select(Select::columns(&["id"]))
+                .execute()
+                .await
+                .map_err(|err| anyhow!("failed to query songs: {err}"))?;
+            let batches = stream.try_collect::<Vec<_>>().await?;
+            let mut all_ids = Vec::new();
+            for batch in &batches {
+                let id_col = downcast_string(batch, "id")?;
+                for row in 0..batch.num_rows() {
+                    all_ids.push(id_col.value(row).to_string());
+                }
+            }
+            all_ids
+        },
+    };
+
+    if target_ids.is_empty() {
+        tracing::info!("No songs to verify.");
+        return Ok(());
+    }
+
+    tracing::info!("Verifying audio for {} song(s)...", target_ids.len());
+
+    let mut ok = 0usize;
+    let mut err_count = 0usize;
+    for id in &target_ids {
+        match store.get_song_audio(id).await {
+            Ok(Some((data, fmt))) => {
+                tracing::info!("  ✓ {id}: {} bytes, format={fmt}", data.len());
+                ok += 1;
+            },
+            Ok(None) => {
+                tracing::error!("  ✗ {id}: audio not found");
+                err_count += 1;
+            },
+            Err(err) => {
+                tracing::error!("  ✗ {id}: {err}");
+                err_count += 1;
+            },
+        }
+    }
+
+    tracing::info!("{ok}/{} songs verified OK", target_ids.len());
+    if err_count > 0 {
+        bail!("{err_count} song(s) failed audio verification");
+    }
+    Ok(())
+}

cli/src/commands/mod.rs

@@ -290,6 +290,14 @@ pub async fn run(cli: Cli) -> Result<()> {
                 table,
                 version,
             } => db_manage::restore_table(&db_path, &table, version).await,
+            DbCommands::TestBlobCompact {
+                count,
+                blob_size,
+            } => db_manage::test_blob_compact(&db_path, count, blob_size).await,
+            DbCommands::VerifyAudio {
+                ids,
+                limit,
+            } => db_manage::verify_audio(&db_path, ids, limit).await,
         },
     }
 }

cli/src/db.rs

@@ -1,7 +1,10 @@
 use std::{collections::HashSet, path::Path, sync::Arc};
 
 use anyhow::{Context, Result};
-use arrow_array::{new_null_array, Array, ArrayRef, RecordBatch, RecordBatchIterator, RecordBatchReader, StringArray};
+use arrow_array::{
+    new_null_array, Array, ArrayRef, RecordBatch, RecordBatchIterator, RecordBatchReader,
+    StringArray,
+};
 use arrow_schema::{DataType, Schema};
 use futures::TryStreamExt;
 use lancedb::{
@@ -33,7 +36,9 @@ pub async fn ensure_table(db: &Connection, name: &str, schema: Arc<Schema>) -> R
         Err(_) => {
             let batch = RecordBatch::new_empty(schema.clone());
             let batches = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema.clone());
-            db.create_table(name, Box::new(batches) as Box<dyn RecordBatchReader + Send>).execute().await?;
+            db.create_table(name, Box::new(batches) as Box<dyn RecordBatchReader + Send>)
+                .execute()
+                .await?;
             db.open_table(name).execute().await?
         },
     };