rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-09-13 00:42:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: add skip_wal fields to internal write requests

Browse Source 
Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
WenyXu
2026-09-09 13:48:55 +00:00
parent c65a4e545e
commit e66fae3080
20 changed files with 60 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/datanode/src/region_server.rs
+1
View File
@@ -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,
src/metric-engine/src/engine.rs
+1
View File
@@ -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),
src/metric-engine/src/engine/bulk_insert.rs
+1
View File
@@ -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,
src/metric-engine/src/engine/flush.rs
+1
View File
@@ -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,
src/metric-engine/src/engine/put.rs
+25
View File
@@ -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,
src/metric-engine/src/engine/read.rs
+1
View File
@@ -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),
src/metric-engine/src/metadata_region.rs
+1
View File
@@ -585,6 +585,7 @@ impl MetadataRegion {
};
RegionPutRequest {
skip_wal: false,
rows,
hint: None,
partition_expr_version: None,
src/mito2/src/engine/alter_test.rs
+1
View File
@@ -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(),
src/mito2/src/engine/apply_staging_manifest_test.rs
+1
View File
@@ -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),
src/mito2/src/engine/basic_test.rs
+1
View File
@@ -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,
src/mito2/src/engine/edit_region_test.rs
+2
View File
@@ -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,
src/mito2/src/engine/flush_test.rs
+2
View File
@@ -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),
src/mito2/src/engine/open_test.rs
+1
View File
@@ -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,
src/mito2/src/engine/scan_test.rs
+1
View File
@@ -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(),
src/mito2/src/engine/set_role_state_test.rs
+2
View File
@@ -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,
src/mito2/src/engine/skip_wal_test.rs
+3
View File
@@ -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),
src/mito2/src/engine/staging_test.rs
+7
View File
@@ -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),
src/mito2/src/request.rs
+3
View File
@@ -76,6 +76,8 @@ pub struct WriteRequest {
pub name_to_index: HashMap<String, usize>,
/// Whether each column has null.
pub has_null: Vec<bool>,
/// Whether this insert should skip WAL. Never applies to deletes.
pub skip_wal: bool,
/// Write hint.
pub hint: Option<WriteHint>,
/// 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,
})
src/mito2/src/test_util.rs
+1
View File
@@ -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,
src/store-api/src/region_request.rs
+4
View File
@@ -217,6 +217,7 @@ fn make_region_puts(inserts: InsertRequests) -> Result<Vec<(RegionId, RegionRequ
(
region_id,
RegionRequest::Put(RegionPutRequest {
skip_wal: false,
rows,
hint: None,
partition_expr_version: r.partition_expr_version.map(|v| v.value),
@@ -522,6 +523,9 @@ pub struct RegionPutRequest {
pub rows: Rows,
/// Write hint.
pub hint: Option<WriteHint>,
/// 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<u64>,
}