From e66fae3080e889fbde1a4d8f7ee812f81b944706 Mon Sep 17 00:00:00 2001 From: WenyXu Date: Wed, 9 Sep 2026 13:48:55 +0000 Subject: [PATCH] refactor: add skip_wal fields to internal write requests Signed-off-by: WenyXu --- src/datanode/src/region_server.rs | 1 + src/metric-engine/src/engine.rs | 1 + src/metric-engine/src/engine/bulk_insert.rs | 1 + src/metric-engine/src/engine/flush.rs | 1 + src/metric-engine/src/engine/put.rs | 25 +++++++++++++++++++ src/metric-engine/src/engine/read.rs | 1 + src/metric-engine/src/metadata_region.rs | 1 + src/mito2/src/engine/alter_test.rs | 1 + .../src/engine/apply_staging_manifest_test.rs | 1 + src/mito2/src/engine/basic_test.rs | 1 + src/mito2/src/engine/edit_region_test.rs | 2 ++ src/mito2/src/engine/flush_test.rs | 2 ++ src/mito2/src/engine/open_test.rs | 1 + src/mito2/src/engine/scan_test.rs | 1 + src/mito2/src/engine/set_role_state_test.rs | 2 ++ src/mito2/src/engine/skip_wal_test.rs | 3 +++ src/mito2/src/engine/staging_test.rs | 7 ++++++ src/mito2/src/request.rs | 3 +++ src/mito2/src/test_util.rs | 1 + src/store-api/src/region_request.rs | 4 +++ 20 files changed, 60 insertions(+) diff --git a/src/datanode/src/region_server.rs b/src/datanode/src/region_server.rs index 79d0dd1915..10bcf32b13 100644 --- a/src/datanode/src/region_server.rs +++ b/src/datanode/src/region_server.rs @@ -2018,6 +2018,7 @@ mod tests { assert!(RegionServerInner::is_ingest_request(&RegionRequest::Put( RegionPutRequest { + skip_wal: false, rows: rows(), hint: None, partition_expr_version: None, diff --git a/src/metric-engine/src/engine.rs b/src/metric-engine/src/engine.rs index 6d5dace9c4..18d6ff6f19 100644 --- a/src/metric-engine/src/engine.rs +++ b/src/metric-engine/src/engine.rs @@ -636,6 +636,7 @@ mod test { .handle_request( logical_region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema: row_schema_with_tags(&["job"]), rows: build_rows(1, 5), diff --git a/src/metric-engine/src/engine/bulk_insert.rs b/src/metric-engine/src/engine/bulk_insert.rs index d700d858ac..8472815b14 100644 --- a/src/metric-engine/src/engine/bulk_insert.rs +++ b/src/metric-engine/src/engine/bulk_insert.rs @@ -599,6 +599,7 @@ mod tests { .handle_request( logical_region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: api::v1::Rows { schema, rows }, hint: None, partition_expr_version: None, diff --git a/src/metric-engine/src/engine/flush.rs b/src/metric-engine/src/engine/flush.rs index 21e642de0b..426aa918d0 100644 --- a/src/metric-engine/src/engine/flush.rs +++ b/src/metric-engine/src/engine/flush.rs @@ -90,6 +90,7 @@ mod tests { let schema = row_schema_with_tags(&["job"]); let rows = build_rows(1, 10); let request = RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema, rows }, hint: None, partition_expr_version: None, diff --git a/src/metric-engine/src/engine/put.rs b/src/metric-engine/src/engine/put.rs index 96ed2c3e60..12133e1b69 100644 --- a/src/metric-engine/src/engine/put.rs +++ b/src/metric-engine/src/engine/put.rs @@ -233,6 +233,7 @@ impl MetricEngineInner { } let merged_request = RegionPutRequest { + skip_wal: false, rows: Rows { schema, rows: merged_rows, @@ -291,6 +292,7 @@ impl MetricEngineInner { }; let merged_request = RegionPutRequest { + skip_wal: false, rows: final_rows, hint: None, partition_expr_version: merged_version, @@ -921,6 +923,7 @@ mod tests { ( logical_region_1, RegionPutRequest { + skip_wal: false, rows: Rows { schema: schema_1.clone(), rows: rows_1, @@ -932,6 +935,7 @@ mod tests { ( logical_region_2, RegionPutRequest { + skip_wal: false, rows: Rows { schema: schema_2.clone(), rows: rows_2, @@ -1095,6 +1099,7 @@ mod tests { let schema = test_util::row_schema_with_tags(&["job"]); let rows = test_util::build_rows(1, 5); let request = RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema, rows }, hint: None, partition_expr_version: None, @@ -1170,6 +1175,7 @@ mod tests { let schema = test_util::row_schema_with_tags(columns); let rows = test_util::build_rows(3, 100); let request = RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema, rows }, hint: None, partition_expr_version: None, @@ -1193,6 +1199,7 @@ mod tests { let schema = test_util::row_schema_with_tags(&["abc"]); let rows = test_util::build_rows(1, 100); let request = RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema, rows }, hint: None, partition_expr_version: None, @@ -1214,6 +1221,7 @@ mod tests { let schema = test_util::row_schema_with_tags(&["def"]); let rows = test_util::build_rows(1, 100); let request = RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema, rows }, hint: None, partition_expr_version: None, @@ -1274,6 +1282,7 @@ mod tests { ( logical_region_1, RegionPutRequest { + skip_wal: false, rows: Rows { schema: schema.clone(), rows: rows1, @@ -1285,6 +1294,7 @@ mod tests { ( logical_region_2, RegionPutRequest { + skip_wal: false, rows: Rows { schema: schema.clone(), rows: rows2, @@ -1296,6 +1306,7 @@ mod tests { ( logical_region_3, RegionPutRequest { + skip_wal: false, rows: Rows { schema: schema.clone(), rows: rows3, @@ -1347,6 +1358,7 @@ mod tests { ( logical_region_1, RegionPutRequest { + skip_wal: false, rows: Rows { schema: schema.clone(), rows: test_util::build_rows(1, 3), @@ -1358,6 +1370,7 @@ mod tests { ( nonexistent_region, RegionPutRequest { + skip_wal: false, rows: Rows { schema: schema.clone(), rows: test_util::build_rows(1, 2), @@ -1369,6 +1382,7 @@ mod tests { ( logical_region_2, RegionPutRequest { + skip_wal: false, rows: Rows { schema: schema.clone(), rows: test_util::build_rows(1, 5), @@ -1407,6 +1421,7 @@ mod tests { let requests = vec![( physical_region_id, RegionPutRequest { + skip_wal: false, rows: Rows { schema, rows: test_util::build_rows(1, 1), @@ -1441,6 +1456,7 @@ mod tests { ( logical_region_id, RegionPutRequest { + skip_wal: false, rows: Rows { schema: schema.clone(), rows: test_util::build_rows(1, 1), @@ -1452,6 +1468,7 @@ mod tests { ( physical_region_id, RegionPutRequest { + skip_wal: false, rows: Rows { schema, rows: test_util::build_rows(1, 1), @@ -1487,6 +1504,7 @@ mod tests { let requests = vec![( logical_region_id, RegionPutRequest { + skip_wal: false, rows: Rows { schema, rows: test_util::build_rows(1, 5), @@ -1565,6 +1583,7 @@ mod tests { .handle_request( logical_region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows, hint: None, partition_expr_version: Some(1), @@ -1607,6 +1626,7 @@ mod tests { .handle_request( logical_region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: rows.clone(), hint: None, partition_expr_version: Some(expected_version.wrapping_add(1)), @@ -1621,6 +1641,7 @@ mod tests { .handle_request( logical_region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: rows.clone(), hint: None, partition_expr_version: None, @@ -1635,6 +1656,7 @@ mod tests { .handle_request( logical_region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows, hint: None, partition_expr_version: Some(expected_version), @@ -1703,6 +1725,7 @@ mod tests { .handle_request( logical_region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema, rows }, hint: None, partition_expr_version: None, @@ -1759,6 +1782,7 @@ mod tests { .handle_request( logical_region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema, rows }, hint: None, partition_expr_version: None, @@ -1813,6 +1837,7 @@ mod tests { .handle_request( logical_region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema, rows }, hint: None, partition_expr_version: None, diff --git a/src/metric-engine/src/engine/read.rs b/src/metric-engine/src/engine/read.rs index ee086184c2..dc0ee6706b 100644 --- a/src/metric-engine/src/engine/read.rs +++ b/src/metric-engine/src/engine/read.rs @@ -495,6 +495,7 @@ mod test { let schema = test_util::row_schema_with_tags(&["job"]); let put = |rows| { RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema: schema.clone(), rows: test_util::build_rows(1, rows), diff --git a/src/metric-engine/src/metadata_region.rs b/src/metric-engine/src/metadata_region.rs index ec67ac7010..8d5a4ddb04 100644 --- a/src/metric-engine/src/metadata_region.rs +++ b/src/metric-engine/src/metadata_region.rs @@ -585,6 +585,7 @@ impl MetadataRegion { }; RegionPutRequest { + skip_wal: false, rows, hint: None, partition_expr_version: None, diff --git a/src/mito2/src/engine/alter_test.rs b/src/mito2/src/engine/alter_test.rs index 6bd20100e8..33212e93b5 100644 --- a/src/mito2/src/engine/alter_test.rs +++ b/src/mito2/src/engine/alter_test.rs @@ -3341,6 +3341,7 @@ async fn test_alter_time_index_widen_sparse_compaction() { }; let put_sparse = |rows| { RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows, hint: Some(WriteHint { primary_key_encoding: api::v1::PrimaryKeyEncoding::Sparse.into(), diff --git a/src/mito2/src/engine/apply_staging_manifest_test.rs b/src/mito2/src/engine/apply_staging_manifest_test.rs index 5e7f77257a..e6ac2bc24b 100644 --- a/src/mito2/src/engine/apply_staging_manifest_test.rs +++ b/src/mito2/src/engine/apply_staging_manifest_test.rs @@ -952,6 +952,7 @@ async fn test_apply_staging_manifest_preserves_unflushed_memtable_with_format(fl .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: unflushed_rows, hint: None, partition_expr_version: Some(expected_version), diff --git a/src/mito2/src/engine/basic_test.rs b/src/mito2/src/engine/basic_test.rs index edaa97ff58..0de989f4f9 100644 --- a/src/mito2/src/engine/basic_test.rs +++ b/src/mito2/src/engine/basic_test.rs @@ -732,6 +732,7 @@ async fn test_absent_and_invalid_columns_with_format(flat_format: bool) { .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows, hint: None, partition_expr_version: None, diff --git a/src/mito2/src/engine/edit_region_test.rs b/src/mito2/src/engine/edit_region_test.rs index 992060d75d..de9611ee2d 100644 --- a/src/mito2/src/engine/edit_region_test.rs +++ b/src/mito2/src/engine/edit_region_test.rs @@ -286,6 +286,7 @@ async fn test_write_during_region_editing_is_queued() { .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows, hint: None, partition_expr_version: None, @@ -400,6 +401,7 @@ async fn test_stalled_write_fails_fast_if_region_closed_during_editing() { .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows, hint: None, partition_expr_version: None, diff --git a/src/mito2/src/engine/flush_test.rs b/src/mito2/src/engine/flush_test.rs index b71ce9f9fc..d7155e1052 100644 --- a/src/mito2/src/engine/flush_test.rs +++ b/src/mito2/src/engine/flush_test.rs @@ -740,6 +740,7 @@ async fn test_region_write_buffer_does_not_stall_follower_write() { engine.handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema, rows: build_rows_for_key("follower", 2, 4, 0), @@ -935,6 +936,7 @@ async fn test_region_write_buffer_rejects_only_full_region_queue() { .handle_request( hot_region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema: stalled_schema, rows: build_rows_for_key("hot", 2, 1026, 2), diff --git a/src/mito2/src/engine/open_test.rs b/src/mito2/src/engine/open_test.rs index 55cd92e6ed..c29c9aa391 100644 --- a/src/mito2/src/engine/open_test.rs +++ b/src/mito2/src/engine/open_test.rs @@ -478,6 +478,7 @@ async fn test_engine_open_readonly_with_format(flat_format: bool) { .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: rows.clone(), hint: None, partition_expr_version: None, diff --git a/src/mito2/src/engine/scan_test.rs b/src/mito2/src/engine/scan_test.rs index df5035cdc5..0bd7575665 100644 --- a/src/mito2/src/engine/scan_test.rs +++ b/src/mito2/src/engine/scan_test.rs @@ -1107,6 +1107,7 @@ async fn test_two_phase_series_scan() { }; let put = |rows| { RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows, hint: Some(WriteHint { primary_key_encoding: api::v1::PrimaryKeyEncoding::Sparse.into(), diff --git a/src/mito2/src/engine/set_role_state_test.rs b/src/mito2/src/engine/set_role_state_test.rs index 1d0f3c188b..53aa351145 100644 --- a/src/mito2/src/engine/set_role_state_test.rs +++ b/src/mito2/src/engine/set_role_state_test.rs @@ -114,6 +114,7 @@ async fn test_set_role_state_gracefully_with_format(flat_format: bool) { .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: rows.clone(), hint: None, partition_expr_version: None, @@ -208,6 +209,7 @@ async fn test_write_downgrading_region_with_format(flat_format: bool) { .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: rows.clone(), hint: None, partition_expr_version: None, diff --git a/src/mito2/src/engine/skip_wal_test.rs b/src/mito2/src/engine/skip_wal_test.rs index a564411473..88304777b2 100644 --- a/src/mito2/src/engine/skip_wal_test.rs +++ b/src/mito2/src/engine/skip_wal_test.rs @@ -522,6 +522,7 @@ async fn test_close_region_skip_wal_rejects_writes_queued_after_close() { .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema: rows_schema(&request), rows: build_rows(3, 4), @@ -553,6 +554,7 @@ async fn test_close_region_skip_wal_rejects_writes_queued_after_close() { .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema: rows_schema(&request_cloned), rows: build_rows(4, 5), @@ -579,6 +581,7 @@ async fn test_close_region_skip_wal_rejects_writes_queued_after_close() { .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: Rows { schema: rows_schema(&request_cloned), rows: build_rows(5, 6), diff --git a/src/mito2/src/engine/staging_test.rs b/src/mito2/src/engine/staging_test.rs index 12263976aa..bd82370850 100644 --- a/src/mito2/src/engine/staging_test.rs +++ b/src/mito2/src/engine/staging_test.rs @@ -285,6 +285,7 @@ async fn test_staging_reject_all_writes_rejects_put() { .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows, hint: None, partition_expr_version: None, @@ -355,6 +356,7 @@ async fn test_staging_write_partition_expr_version_with_format(flat_format: bool .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: bad_rows, hint: None, partition_expr_version: Some(origin_version), @@ -376,6 +378,7 @@ async fn test_staging_write_partition_expr_version_with_format(flat_format: bool .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: compat_rows, hint: None, partition_expr_version: None, @@ -393,6 +396,7 @@ async fn test_staging_write_partition_expr_version_with_format(flat_format: bool .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: ok_rows, hint: None, partition_expr_version: Some(expected_version), @@ -440,6 +444,7 @@ async fn test_staging_write_partition_expr_version_with_format(flat_format: bool .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: exit_rows, hint: None, partition_expr_version: Some(origin_version), @@ -457,6 +462,7 @@ async fn test_staging_write_partition_expr_version_with_format(flat_format: bool .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: compat_rows, hint: None, partition_expr_version: None, @@ -483,6 +489,7 @@ async fn test_staging_write_partition_expr_version_with_format(flat_format: bool .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows: commit_rows, hint: None, partition_expr_version: Some(expected_version), diff --git a/src/mito2/src/request.rs b/src/mito2/src/request.rs index 94d442e261..5f354c4603 100644 --- a/src/mito2/src/request.rs +++ b/src/mito2/src/request.rs @@ -76,6 +76,8 @@ pub struct WriteRequest { pub name_to_index: HashMap, /// Whether each column has null. pub has_null: Vec, + /// Whether this insert should skip WAL. Never applies to deletes. + pub skip_wal: bool, /// Write hint. pub hint: Option, /// Region metadata on the time of this request is created. @@ -137,6 +139,7 @@ impl WriteRequest { name_to_index, has_null, hint: None, + skip_wal: false, region_metadata, partition_expr_version: None, }) diff --git a/src/mito2/src/test_util.rs b/src/mito2/src/test_util.rs index 8b0302a365..aa03e4bc84 100644 --- a/src/mito2/src/test_util.rs +++ b/src/mito2/src/test_util.rs @@ -1351,6 +1351,7 @@ pub async fn put_rows(engine: &MitoEngine, region_id: RegionId, rows: Rows) { .handle_request( region_id, RegionRequest::Put(RegionPutRequest { + skip_wal: false, rows, hint: None, partition_expr_version: None, diff --git a/src/store-api/src/region_request.rs b/src/store-api/src/region_request.rs index 7cbf5b1a7f..d3690a6281 100644 --- a/src/store-api/src/region_request.rs +++ b/src/store-api/src/region_request.rs @@ -217,6 +217,7 @@ fn make_region_puts(inserts: InsertRequests) -> Result, + /// Skip WAL for this insert without changing region options. + /// Metadata writes must not inherit this option from user inserts. + pub skip_wal: bool, /// Partition expression version for the region. pub partition_expr_version: Option, }