fix: stabilize expanded compaction correctness randomized lane

Author: belveryin

.github/workflows/rust.yml

@@ -68,7 +68,6 @@ jobs:
     name: compaction correctness (optional expanded randomized sst+reopen)
     runs-on: ubuntu-latest
     needs: build
-    continue-on-error: true
     env:
       TONBO_COMPACTION_MODEL_SST: "1"
       TONBO_COMPACTION_REOPEN: "1"

src/db/tests/core/compaction_correctness.rs

@@ -278,11 +278,11 @@ impl ScenarioHarness {
     }
 
     async fn compact_l0(
-        &self,
+        &mut self,
         sst_ids: Vec<u64>,
         target_level: u32,
     ) -> Result<crate::compaction::executor::CompactionOutcome, Box<dyn std::error::Error>> {
-        let start_id = 10_000;
+        let start_id = self.next_sst_id.max(10_000);
         let outcome = compact_merge_l0(
             self.db.inner().as_ref(),
             sst_ids,
@@ -291,6 +291,14 @@ impl ScenarioHarness {
             start_id,
         )
         .await?;
+        let next_generated_id = outcome
+            .add_ssts
+            .iter()
+            .map(|entry| entry.sst_id().raw())
+            .max()
+            .map(|id| id.saturating_add(1))
+            .unwrap_or_else(|| start_id.saturating_add(1));
+        self.next_sst_id = self.next_sst_id.max(next_generated_id);
         Ok(outcome)
     }
 
@@ -636,6 +644,7 @@ struct ModelRunner {
     oracle: MvccOracle,
     l0_ssts: Vec<u64>,
     active_snapshot_ts: Option<u64>,
+    active_snapshot: Option<TxSnapshot>,
     allow_reopen: bool,
     eager_flush: bool,
     allow_sst: bool,
@@ -659,6 +668,7 @@ impl ModelRunner {
             oracle: MvccOracle::default(),
             l0_ssts: Vec::new(),
             active_snapshot_ts: None,
+            active_snapshot: None,
             allow_reopen,
             eager_flush,
             allow_sst,
@@ -673,6 +683,11 @@ impl ModelRunner {
         }
     }
 
+    fn clear_active_snapshot(&mut self) {
+        self.active_snapshot_ts = None;
+        self.active_snapshot = None;
+    }
+
     async fn run(&mut self) -> Result<(), Box<dyn std::error::Error>> {
         for _ in 0..MODEL_OPS_PER_SEED {
             let op_kind = self.pick_op();
@@ -703,6 +718,7 @@ impl ModelRunner {
     async fn apply_op(&mut self, op_kind: OpKind) -> Result<(), Box<dyn std::error::Error>> {
         match op_kind {
             OpKind::Put => {
+                self.clear_active_snapshot();
                 let key = self.pick_key();
                 let value = self.rng.next_i64(10_000);
                 self.trace.push(Op::Put {
@@ -720,6 +736,7 @@ impl ModelRunner {
                 }
             }
             OpKind::Delete => {
+                self.clear_active_snapshot();
                 let key = self.pick_key();
                 self.trace.push(Op::Delete { key: key.clone() });
                 self.harness.ingest_delete(&key, &mut self.oracle).await?;
@@ -731,6 +748,7 @@ impl ModelRunner {
                 }
             }
             OpKind::Flush => {
+                self.clear_active_snapshot();
                 self.trace.push(Op::Flush);
                 if self.allow_sst {
                     if let Some(sst_id) = self.harness.try_flush_immutables_to_l0().await? {
@@ -764,18 +782,7 @@ impl ModelRunner {
                 if outcome.add_ssts.is_empty() {
                     return Err("compaction produced no output sst".into());
                 }
-                let snapshot_ts = match self.active_snapshot_ts {
-                    Some(ts) => ts,
-                    None => {
-                        let snapshot = self.harness.db.begin_snapshot().await?;
-                        snapshot.read_view().read_ts().get()
-                    }
-                };
-                let snapshot = self
-                    .harness
-                    .db
-                    .snapshot_at(Timestamp::new(snapshot_ts))
-                    .await?;
+                let (snapshot, snapshot_ts) = self.read_snapshot().await?;
                 let ctx = self.failure_context(Some(snapshot_ts));
                 assert_oracle_matches(
                     "model_based_compaction",
@@ -836,6 +843,7 @@ impl ModelRunner {
                 let snapshot = self.harness.db.begin_snapshot().await?;
                 let snapshot_ts = snapshot.read_view().read_ts().get();
                 self.active_snapshot_ts = Some(snapshot_ts);
+                self.active_snapshot = Some(snapshot.clone());
                 self.trace.push(Op::Snapshot { snapshot_ts });
                 let ctx = self.failure_context(Some(snapshot_ts));
                 assert_oracle_matches(
@@ -868,15 +876,23 @@ impl ModelRunner {
                         Some(&ctx),
                     )
                     .await?;
+                    self.active_snapshot = Some(snapshot);
+                } else {
+                    self.active_snapshot = None;
                 }
             }
         }
         Ok(())
     }
 
     async fn read_snapshot(&mut self) -> Result<(TxSnapshot, u64), Box<dyn std::error::Error>> {
-        if let Some(ts) = self.active_snapshot_ts {
+        if let Some(snapshot) = self.active_snapshot.as_ref()
+            && let Some(ts) = self.active_snapshot_ts
+        {
+            Ok((snapshot.clone(), ts))
+        } else if let Some(ts) = self.active_snapshot_ts {
             let snapshot = self.harness.db.snapshot_at(Timestamp::new(ts)).await?;
+            self.active_snapshot = Some(snapshot.clone());
             Ok((snapshot, ts))
         } else {
             let snapshot = self.harness.db.begin_snapshot().await?;
@@ -1236,9 +1252,6 @@ async fn compaction_correctness_reopen_snapshot_durability()
     let snapshot_ts = snapshot.read_view().read_ts().get();
     assert_oracle_matches(scenario, snapshot_ts, &snapshot, &oracle, &harness.db, None).await?;
 
-    // Note: avoid compaction here because some SSTs can be missing Parquet page indexes,
-    // and the compaction read path currently requires them. Re-enable compaction coverage
-    // once page index emission is consistent for all SSTs.
     harness.reopen().await?;
     let reopened_snapshot = harness.db.snapshot_at(Timestamp::new(snapshot_ts)).await?;
     assert_oracle_matches(
@@ -1254,6 +1267,85 @@ async fn compaction_correctness_reopen_snapshot_durability()
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn compaction_correctness_delete_only_sst_reopen_and_compact()
+-> Result<(), Box<dyn std::error::Error>> {
+    let scenario = "delete_only_sst_reopen_and_compact";
+    let mut harness = ScenarioHarness::new("compaction-correctness-delete-only-sst").await?;
+    let mut oracle = MvccOracle::default();
+
+    let _ts0 = harness.ingest_delete("k01", &mut oracle).await?;
+    let sst0 = harness.flush_immutables_to_l0().await?;
+
+    let _ts1 = harness.ingest_put("k02", 20, &mut oracle).await?;
+    let sst1 = harness.flush_immutables_to_l0().await?;
+
+    let snapshot = harness.db.begin_snapshot().await?;
+    let snapshot_ts = snapshot.read_view().read_ts().get();
+    assert_oracle_matches(scenario, snapshot_ts, &snapshot, &oracle, &harness.db, None).await?;
+    assert_range_matches(
+        scenario,
+        snapshot_ts,
+        &snapshot,
+        &oracle,
+        &harness.db,
+        "k01",
+        "k02",
+        None,
+    )
+    .await?;
+
+    harness.reopen().await?;
+    let reopened = harness.db.snapshot_at(Timestamp::new(snapshot_ts)).await?;
+    assert_oracle_matches(scenario, snapshot_ts, &reopened, &oracle, &harness.db, None).await?;
+    assert_range_matches(
+        scenario,
+        snapshot_ts,
+        &reopened,
+        &oracle,
+        &harness.db,
+        "k01",
+        "k02",
+        None,
+    )
+    .await?;
+
+    let outcome = harness.compact_l0(vec![sst0, sst1], 1).await?;
+    assert_eq!(
+        outcome.remove_ssts.len(),
+        2,
+        "scenario={scenario} expected compaction to remove 2 SSTs"
+    );
+    assert!(
+        !outcome.add_ssts.is_empty(),
+        "scenario={scenario} expected compaction to add SSTs"
+    );
+
+    let post_snapshot = harness.db.snapshot_at(Timestamp::new(snapshot_ts)).await?;
+    assert_oracle_matches(
+        scenario,
+        snapshot_ts,
+        &post_snapshot,
+        &oracle,
+        &harness.db,
+        None,
+    )
+    .await?;
+    assert_range_matches(
+        scenario,
+        snapshot_ts,
+        &post_snapshot,
+        &oracle,
+        &harness.db,
+        "k01",
+        "k02",
+        None,
+    )
+    .await?;
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn compaction_correctness_iterator_seek_stability() -> Result<(), Box<dyn std::error::Error>>
 {

src/db/tests/core/scan.rs

@@ -217,6 +217,44 @@ async fn plan_scan_missing_page_indexes_is_error() {
     );
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn plan_scan_empty_sst_without_page_indexes_is_allowed() {
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("id", DataType::Utf8, false),
+        Field::new("v", DataType::Int32, false),
+    ]));
+    let db = db_with_schema(schema.clone()).await;
+
+    let sst_root = Path::from("scan-empty-no-page-index");
+    db.fs.create_dir_all(&sst_root).await.expect("create dir");
+    let data_path = sst_root.child("000.parquet");
+    let empty_batch = rows_with_commit_ts(0, 0, Timestamp::MIN.get());
+    write_parquet_data_missing_page_index(Arc::clone(&db.fs), data_path.clone(), empty_batch).await;
+
+    let sst_entry = SstEntry::new(SsTableId::new(12), None, None, data_path, None);
+    db.manifest
+        .apply_version_edits(
+            db.manifest_table,
+            &[VersionEdit::AddSsts {
+                level: 0,
+                entries: vec![sst_entry],
+            }],
+        )
+        .await
+        .expect("add sst");
+
+    let snapshot = db.begin_snapshot().await.expect("snapshot");
+    let plan = snapshot
+        .plan_scan(&db, &Expr::True, None, None)
+        .await
+        .expect("empty sst without page indexes should plan");
+    assert_eq!(plan.sst_selections.len(), 1);
+
+    let stream = db.execute_scan(plan).await.expect("execute");
+    let batches = stream.try_collect::<Vec<_>>().await.expect("collect");
+    assert!(collect_ids(&batches).is_empty());
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn plan_scan_prunes_sst_row_groups_and_pages() {
     let schema = Arc::new(Schema::new(vec![

src/ondisk/sstable.rs

@@ -1487,6 +1487,36 @@ pub(crate) fn validate_page_indexes(
     metadata: &ParquetMetaData,
 ) -> Result<(), SsTableError> {
     let path = path.to_string();
+    let row_groups = metadata.num_row_groups();
+
+    // Empty SST data files (e.g. delete-only runs) have no pages, so parquet metadata may omit
+    // page-index sections entirely. Treat that representation as valid.
+    if row_groups == 0 {
+        if let Some(column_index) = metadata.column_index()
+            && !column_index.is_empty()
+        {
+            return Err(SsTableError::MissingPageIndex {
+                path: path.clone(),
+                reason: format!(
+                    "column index row group count mismatch: expected {row_groups}, got {}",
+                    column_index.len()
+                ),
+            });
+        }
+        if let Some(offset_index) = metadata.offset_index()
+            && !offset_index.is_empty()
+        {
+            return Err(SsTableError::MissingPageIndex {
+                path: path.clone(),
+                reason: format!(
+                    "offset index row group count mismatch: expected {row_groups}, got {}",
+                    offset_index.len()
+                ),
+            });
+        }
+        return Ok(());
+    }
+
     let column_index = metadata
         .column_index()
         .ok_or_else(|| SsTableError::MissingPageIndex {
@@ -1499,8 +1529,6 @@ pub(crate) fn validate_page_indexes(
             path: path.clone(),
             reason: "offset index missing".to_string(),
         })?;
-
-    let row_groups = metadata.num_row_groups();
     if column_index.len() != row_groups {
         return Err(SsTableError::MissingPageIndex {
             path: path.clone(),
@@ -2393,4 +2421,61 @@ mod tests {
             other => panic!("unexpected error: {other:?}"),
         }
     }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn open_parquet_empty_without_page_indexes_is_allowed() {
+        use arrow_array::StringArray;
+
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Utf8, false)]));
+        let values: ArrayRef = Arc::new(StringArray::from(Vec::<&str>::new()));
+        let batch = RecordBatch::try_new(Arc::clone(&schema), vec![values]).expect("batch");
+
+        let fs: Arc<dyn DynFs> = Arc::new(LocalFs {});
+        let tempdir = tempfile::tempdir().expect("tempdir");
+        let path = Path::from(
+            tempdir
+                .path()
+                .join("empty-no-page-index.parquet")
+                .to_string_lossy()
+                .to_string(),
+        );
+        let file = fs
+            .open_options(
+                &path,
+                OpenOptions::default()
+                    .create(true)
+                    .write(true)
+                    .truncate(true),
+            )
+            .await
+            .expect("open file");
+
+        let properties = WriterProperties::builder()
+            .set_statistics_enabled(EnabledStatistics::None)
+            .set_offset_index_disabled(true)
+            .build();
+        let mut writer = AsyncArrowWriter::try_new(
+            AsyncWriter::new(file, NoopExecutor),
+            Arc::clone(&schema),
+            Some(properties),
+        )
+        .expect("writer");
+        writer.write(&batch).await.expect("write");
+        writer.close().await.expect("close");
+
+        let (mut stream, stream_schema) = open_parquet_stream_with_schema(
+            Arc::clone(&fs),
+            path,
+            None,
+            None,
+            None,
+            None,
+            NoopExecutor,
+        )
+        .await
+        .expect("empty parquet without page indexes should be readable");
+
+        assert_eq!(stream_schema.fields().len(), 1);
+        assert!(stream.next().await.is_none());
+    }
 }