diff --git a/Cargo.lock b/Cargo.lock index b3e8a6d8cd5..9f87145c4b4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6123,7 +6123,7 @@ dependencies = [ [[package]] name = "greptime-proto" version = "0.1.0" -source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=b74667d17827481dbb6222fa0326163627211557#b74667d17827481dbb6222fa0326163627211557" +source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=0c4beffaa2b3968034915867593a717c6ab0c6b4#0c4beffaa2b3968034915867593a717c6ab0c6b4" dependencies = [ "prost 0.14.1", "prost-types 0.14.1", diff --git a/Cargo.toml b/Cargo.toml index 862f93e3032..9dc78442cd6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -159,7 +159,7 @@ fs2 = "0.4" fst = "0.4.7" futures = "0.3" futures-util = "0.3" -greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "b74667d17827481dbb6222fa0326163627211557" } +greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "0c4beffaa2b3968034915867593a717c6ab0c6b4" } hex = "0.4" hostname = "0.4.0" http = "1" diff --git a/src/client/src/database.rs b/src/client/src/database.rs index 881730cfaaa..e97fff6db00 100644 --- a/src/client/src/database.rs +++ b/src/client/src/database.rs @@ -825,7 +825,17 @@ impl Database { /// Ingest a stream of [RecordBatch]es that belong to a table, using Arrow Flight's "`DoPut`" /// method. The return value is also a stream, produces [DoPutResponse]s. pub async fn do_put(&self, stream: FlightDataStream) -> Result { + self.do_put_with_hints(stream, &[]).await + } + + /// Ingest a stream of [RecordBatch]es using Arrow Flight's `DoPut` with request hints. + pub async fn do_put_with_hints( + &self, + stream: FlightDataStream, + hints: &[(&str, &str)], + ) -> Result { let mut request = tonic::Request::new(stream); + Self::put_hints(request.metadata_mut(), hints)?; if let Some(AuthHeader { auth_scheme: Some(AuthScheme::Basic(Basic { username, password })), diff --git a/src/frontend/src/instance/grpc.rs b/src/frontend/src/instance/grpc.rs index 7d4f0ee7959..2d89b5489df 100644 --- a/src/frontend/src/instance/grpc.rs +++ b/src/frontend/src/instance/grpc.rs @@ -453,6 +453,7 @@ impl Instance { request.flight_data, request.record_batch, request.schema_bytes, + ctx.skip_wal(), ) .await .context(TableOperationSnafu)?; diff --git a/src/metric-engine/src/engine/bulk_insert.rs b/src/metric-engine/src/engine/bulk_insert.rs index 87b6679c61c..370493199ab 100644 --- a/src/metric-engine/src/engine/bulk_insert.rs +++ b/src/metric-engine/src/engine/bulk_insert.rs @@ -124,6 +124,7 @@ impl MetricEngineInner { let aligned_schema_version = Some(self.physical_schema_version(data_region_id).await?); let request = RegionBulkInsertsRequest { + skip_wal: request.skip_wal, region_id: data_region_id, payload: modified_batch, raw_data: ArrowIpc { @@ -245,14 +246,20 @@ mod tests { use datatypes::arrow::datatypes::{DataType, Field, Schema as ArrowSchema, TimeUnit}; use datatypes::arrow::record_batch::RecordBatch; use mito2::config::MitoConfig; - use store_api::metric_engine_consts::PRIMARY_KEY_ENCODING; + use store_api::metric_engine_consts::{ + METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY, PRIMARY_KEY_ENCODING, + }; use store_api::path_utils::table_dir; use store_api::region_engine::RegionEngine; - use store_api::region_request::{RegionBulkInsertsRequest, RegionPutRequest, RegionRequest}; + use store_api::region_request::{ + PathType, RegionBulkInsertsRequest, RegionCloseRequest, RegionOpenRequest, + RegionPutRequest, RegionRequest, + }; use store_api::storage::{RegionId, ScanRequest}; use super::record_batch_to_ipc; use crate::batch_modifier::{TagColumnInfo, modify_batch_sparse}; + use crate::engine::MetricEngine; use crate::error::Error; use crate::test_util::{self, TestEnv}; @@ -287,9 +294,14 @@ mod tests { .unwrap() } - fn build_bulk_request(logical_region_id: RegionId, batch: RecordBatch) -> RegionRequest { + fn build_bulk_request( + logical_region_id: RegionId, + batch: RecordBatch, + skip_wal: bool, + ) -> RegionRequest { let (schema, data_header, payload) = record_batch_to_ipc(&batch).unwrap(); RegionRequest::BulkInserts(RegionBulkInsertsRequest { + skip_wal, region_id: logical_region_id, payload: batch, raw_data: ArrowIpc { @@ -332,6 +344,7 @@ mod tests { let batch = build_logical_batch(0, 0); let request = RegionRequest::BulkInserts(RegionBulkInsertsRequest { + skip_wal: false, region_id: logical_region_id, payload: batch, raw_data: ArrowIpc::default(), @@ -348,17 +361,21 @@ mod tests { #[tokio::test] async fn test_bulk_insert_physical_region_passthrough() { - check_bulk_insert_physical_region_passthrough(false).await; + check_bulk_insert_physical_region_passthrough(false, false).await; + check_bulk_insert_physical_region_passthrough(true, false).await; } #[tokio::test] async fn test_physical_bulk_partition_columns_preserve_stored_rows() { - let stripped = check_bulk_insert_physical_region_passthrough(false).await; - let retained = check_bulk_insert_physical_region_passthrough(true).await; - assert_eq!(retained, stripped); + for skip_wal in [false, true] { + let stripped = check_bulk_insert_physical_region_passthrough(skip_wal, false).await; + let retained = check_bulk_insert_physical_region_passthrough(skip_wal, true).await; + assert_eq!(retained, stripped); + } } async fn check_bulk_insert_physical_region_passthrough( + skip_wal: bool, retain_partition_columns: bool, ) -> String { // Use flat format so that BulkMemtable is used (supports write_bulk). @@ -368,12 +385,13 @@ mod tests { }; let env = TestEnv::with_mito_config("", mito_config, Default::default()).await; env.init_metric_region().await; + env.metric().inner.flush_task.stop().await.unwrap(); let physical_region_id = env.default_physical_region_id(); let logical_region_id = env.default_logical_region_id(); // First, do a normal logical bulk insert so we can compare results. let logical_batch = build_logical_batch(0, 3); - let logical_request = build_bulk_request(logical_region_id, logical_batch.clone()); + let logical_request = build_bulk_request(logical_region_id, logical_batch, skip_wal); let response = env .metric() .handle_request(logical_region_id, logical_request) @@ -406,7 +424,7 @@ mod tests { physical_batch.column_by_name("job").is_some(), retain_partition_columns ); - let request = build_bulk_request(physical_region_id, physical_batch); + let request = build_bulk_request(physical_region_id, physical_batch, skip_wal); let response = env .metric() .handle_request(physical_region_id, request) @@ -422,7 +440,50 @@ mod tests { .unwrap(); let batches = RecordBatches::try_collect(stream).await.unwrap(); assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 6); - batches.pretty_print().unwrap() + let stored_rows = batches.pretty_print().unwrap(); + + // Closing without a flush must recover only WAL-backed data. Recreate + // the wrapper too, so its metadata cache cannot hide missing metadata. + let stat = env.mito().region_statistic(physical_region_id).unwrap(); + assert_eq!(stat.sst_num, 0); + env.metric() + .handle_request( + physical_region_id, + RegionRequest::Close(RegionCloseRequest { + flush_on_close: false, + }), + ) + .await + .unwrap(); + let reopened = MetricEngine::try_new(env.mito(), Default::default()).unwrap(); + reopened.inner.flush_task.stop().await.unwrap(); + reopened + .handle_request( + physical_region_id, + RegionRequest::Open(RegionOpenRequest { + engine: METRIC_ENGINE_NAME.to_string(), + table_dir: TestEnv::default_table_dir(), + path_type: PathType::Bare, + options: [(PHYSICAL_TABLE_METADATA_KEY.to_string(), String::new())] + .into_iter() + .collect(), + skip_wal_replay: false, + checkpoint: None, + requirements: Default::default(), + }), + ) + .await + .unwrap(); + let stream = reopened + .scan_to_stream(logical_region_id, ScanRequest::default()) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + assert_eq!( + batches.iter().map(|b| b.num_rows()).sum::(), + if skip_wal { 0 } else { 6 }, + ); + stored_rows } #[tokio::test] @@ -437,7 +498,7 @@ mod tests { let physical_region_id = env.default_physical_region_id(); let batch = build_logical_batch(0, 0); - let request = build_bulk_request(physical_region_id, batch); + let request = build_bulk_request(physical_region_id, batch, false); let response = env .metric() .handle_request(physical_region_id, request) @@ -471,7 +532,7 @@ mod tests { ) .unwrap(); - let request = build_bulk_request(logical_region_id, batch); + let request = build_bulk_request(logical_region_id, batch, false); let err = env .metric() .handle_request(logical_region_id, request) @@ -521,7 +582,7 @@ mod tests { ) .unwrap(); - let request = build_bulk_request(logical_region_id, batch); + let request = build_bulk_request(logical_region_id, batch, false); let response = env .metric() .handle_request(logical_region_id, request) @@ -544,7 +605,7 @@ mod tests { env.init_metric_region().await; let logical_region_id = env.default_logical_region_id(); - let request = build_bulk_request(logical_region_id, build_logical_batch(0, 3)); + let request = build_bulk_request(logical_region_id, build_logical_batch(0, 3), false); let response = env .metric() .handle_request(logical_region_id, request) @@ -552,7 +613,7 @@ mod tests { .unwrap(); assert_eq!(response.affected_rows, 3); - let request = build_bulk_request(logical_region_id, build_logical_batch(3, 5)); + let request = build_bulk_request(logical_region_id, build_logical_batch(3, 5), false); let response = env .metric() .handle_request(logical_region_id, request) @@ -575,7 +636,7 @@ mod tests { env.init_metric_region().await; let logical_region_id = env.default_logical_region_id(); - let request = build_bulk_request(logical_region_id, build_logical_batch(0, 4)); + let request = build_bulk_request(logical_region_id, build_logical_batch(0, 4), false); let response = env .metric() .handle_request(logical_region_id, request) @@ -597,7 +658,7 @@ mod tests { let env = TestEnv::new().await; let logical_region_id = init_dense_metric_region(&env).await; - let request = build_bulk_request(logical_region_id, build_logical_batch(0, 2)); + let request = build_bulk_request(logical_region_id, build_logical_batch(0, 2), false); let err = env .metric() .handle_request(logical_region_id, request) @@ -639,7 +700,7 @@ mod tests { let env_bulk = TestEnv::new().await; env_bulk.init_metric_region().await; - let request = build_bulk_request(logical_region_id, build_logical_batch(0, 5)); + let request = build_bulk_request(logical_region_id, build_logical_batch(0, 5), false); env_bulk .metric() .handle_request(logical_region_id, request) diff --git a/src/mito2/src/engine/alter_test.rs b/src/mito2/src/engine/alter_test.rs index 916424622b9..fd74c4b498a 100644 --- a/src/mito2/src/engine/alter_test.rs +++ b/src/mito2/src/engine/alter_test.rs @@ -3003,6 +3003,7 @@ fn build_schema_v0_bulk_request(region_id: RegionId) -> RegionBulkInsertsRequest let raw_data = encode_arrow_ipc(&payload); RegionBulkInsertsRequest { + skip_wal: false, region_id, payload, raw_data, diff --git a/src/mito2/src/engine/edit_region_test.rs b/src/mito2/src/engine/edit_region_test.rs index de9611ee2df..39e688c5c4d 100644 --- a/src/mito2/src/engine/edit_region_test.rs +++ b/src/mito2/src/engine/edit_region_test.rs @@ -700,6 +700,7 @@ fn build_bulk_insert_request( let (schema, record_batch) = encode_to_flight_data(payload.clone()); RegionBulkInsertsRequest { + skip_wal: false, region_id, payload, raw_data: ArrowIpc { diff --git a/src/mito2/src/engine/scan_test.rs b/src/mito2/src/engine/scan_test.rs index 8c21e04541c..025531cbea8 100644 --- a/src/mito2/src/engine/scan_test.rs +++ b/src/mito2/src/engine/scan_test.rs @@ -2387,6 +2387,7 @@ fn build_bulk_insert_request( let (schema, record_batch) = encode_to_flight_data(payload.clone()); RegionBulkInsertsRequest { + skip_wal: false, region_id, payload, raw_data: ArrowIpc { @@ -2399,6 +2400,139 @@ fn build_bulk_insert_request( } } +#[tokio::test] +async fn test_bulk_skip_wal_recovery() { + check_bulk_skip_wal_recovery(false, false).await; + check_bulk_skip_wal_recovery(false, true).await; + check_bulk_skip_wal_recovery(true, false).await; + check_bulk_skip_wal_recovery(true, true).await; +} + +async fn check_bulk_skip_wal_recovery(skip_wal: bool, flush: bool) { + let mut env = TestEnv::new().await; + let engine = env.create_engine(MitoConfig::default()).await; + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("memtable.type", "bulk") + .build(); + let table_dir = request.table_dir.clone(); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + let mut request = build_bulk_insert_request(region_id, 0, 4); + request.skip_wal = skip_wal; + let affected = engine + .handle_request(region_id, RegionRequest::BulkInserts(request)) + .await + .unwrap(); + assert_eq!(affected.affected_rows, 4); + let region = engine.get_region(region_id).unwrap(); + let current = region.version_control.current(); + assert_eq!(current.committed_sequence, 4); + assert_eq!(current.last_entry_id, u64::from(!skip_wal)); + assert_eq!(current.version.flushed_entry_id, 0); + let stream = engine + .scan_to_stream(region_id, ScanRequest::default()) + .await + .unwrap(); + assert_eq!( + RecordBatches::try_collect(stream) + .await + .unwrap() + .iter() + .map(|b| b.num_rows()) + .sum::(), + 4 + ); + if flush { + test_util::flush_region(&engine, region_id, None).await; + let current = region.version_control.current(); + assert_eq!(current.last_entry_id, u64::from(!skip_wal)); + assert_eq!(current.version.flushed_sequence, 4); + assert_eq!( + engine + .region_statistic(region_id) + .unwrap() + .manifest + .data_flushed_entry_id(), + u64::from(!skip_wal) + ); + } + reopen_region(&engine, region_id, table_dir, false, HashMap::new()).await; + let stream = engine + .scan_to_stream(region_id, ScanRequest::default()) + .await + .unwrap(); + assert_eq!( + RecordBatches::try_collect(stream) + .await + .unwrap() + .iter() + .map(|b| b.num_rows()) + .sum::(), + if skip_wal && !flush { 0 } else { 4 } + ); +} + +#[tokio::test] +async fn test_bulk_skip_wal_flush_watermarks() { + let mut env = TestEnv::new().await; + let engine = env.create_engine(MitoConfig::default()).await; + let region_id = RegionId::new(1, 1); + engine + .handle_request( + region_id, + RegionRequest::Create( + CreateRequestBuilder::new() + .insert_option("memtable.type", "bulk") + .build(), + ), + ) + .await + .unwrap(); + // Establish a persisted baseline, skip one bulk write, then resume WAL. + for (round, skip_wal) in [false, true, false].into_iter().enumerate() { + let region = engine.get_region(region_id).unwrap(); + let before = region.version_control.current(); + let mut request = build_bulk_insert_request(region_id, round * 4, (round + 1) * 4); + request.skip_wal = skip_wal; + let affected = engine + .handle_request(region_id, RegionRequest::BulkInserts(request)) + .await + .unwrap(); + assert_eq!(affected.affected_rows, 4); + let expected_entry_id = before.last_entry_id + u64::from(!skip_wal); + let after = region.version_control.current(); + assert_eq!(after.last_entry_id, expected_entry_id); + assert_eq!(after.committed_sequence, before.committed_sequence + 4); + assert_eq!( + after.version.flushed_sequence, + before.version.flushed_sequence + ); + assert_eq!( + engine + .region_statistic(region_id) + .unwrap() + .manifest + .data_flushed_entry_id(), + before.version.flushed_entry_id + ); + test_util::flush_region(&engine, region_id, None).await; + let after = region.version_control.current(); + assert_eq!(after.last_entry_id, expected_entry_id); + assert_eq!(after.version.flushed_sequence, (round as u64 + 1) * 4); + assert_eq!( + engine + .region_statistic(region_id) + .unwrap() + .manifest + .data_flushed_entry_id(), + expected_entry_id + ); + } +} + /// Bulk parts fold per-part sequences (commit-unit granularity): an exact range /// keeps or drops each whole part. The same set must hold before and after flush. #[tokio::test] diff --git a/src/mito2/src/engine/staging_test.rs b/src/mito2/src/engine/staging_test.rs index 6493c5e2ad2..1b2a7ffee13 100644 --- a/src/mito2/src/engine/staging_test.rs +++ b/src/mito2/src/engine/staging_test.rs @@ -309,6 +309,7 @@ async fn test_staging_reject_all_writes_rejects_put_and_bulk() { partition_expr_version: None, }), RegionRequest::BulkInserts(RegionBulkInsertsRequest { + skip_wal: false, region_id, payload, raw_data: ArrowIpc { diff --git a/src/mito2/src/flush.rs b/src/mito2/src/flush.rs index b7c8513afc7..cf7c6c02851 100644 --- a/src/mito2/src/flush.rs +++ b/src/mito2/src/flush.rs @@ -1728,6 +1728,7 @@ mod tests { ( SenderBulkRequest { + skip_wal: false, sender: OptionOutputTx::from(sender), region_id, request: converter.convert().unwrap(), diff --git a/src/mito2/src/region/opener.rs b/src/mito2/src/region/opener.rs index 67574e77ed3..3adb1058f82 100644 --- a/src/mito2/src/region/opener.rs +++ b/src/mito2/src/region/opener.rs @@ -975,7 +975,8 @@ where region_write_ctx.push_bulk( OptionOutputTx::none(), part, - Some(bulk_sequence_from_wal) + Some(bulk_sequence_from_wal), + false ), RegionCorruptedSnafu { region_id, diff --git a/src/mito2/src/region_write_ctx.rs b/src/mito2/src/region_write_ctx.rs index f98d311a807..4bc0164cc01 100644 --- a/src/mito2/src/region_write_ctx.rs +++ b/src/mito2/src/region_write_ctx.rs @@ -318,24 +318,26 @@ impl RegionWriteCtx { sender: OptionOutputTx, mut bulk: BulkPart, sequence: Option, + skip_wal: bool, ) -> bool { if let Some(sequence) = sequence { self.next_sequence = sequence; } bulk.sequence = self.next_sequence; - let entry = match BulkWalEntry::try_from(&bulk) { - Ok(entry) => entry, - Err(e) => { - sender.send(Err(e)); - return false; - } - }; + if !skip_wal { + let entry = match BulkWalEntry::try_from(&bulk) { + Ok(entry) => entry, + Err(e) => { + sender.send(Err(e)); + return false; + } + }; + self.wal_entry.bulk_entries.push(entry); + } self.bulk_notifiers .push(WriteNotify::new(sender, bulk.num_rows())); - // Add bulk wal entry - self.wal_entry.bulk_entries.push(entry); self.next_sequence += bulk.num_rows() as u64; self.bulk_parts.push(bulk); true @@ -555,11 +557,16 @@ mod tests { #[test] fn test_request_skip_wal_preserves_sequences_and_other_writes() { // Ablate only the request flag: the workload and sequence allocation stay identical. - check_request_skip_wal_preserves_sequences_and_other_writes(false); - check_request_skip_wal_preserves_sequences_and_other_writes(true); + check_request_skip_wal_preserves_sequences_and_other_writes(false, false); + check_request_skip_wal_preserves_sequences_and_other_writes(false, true); + check_request_skip_wal_preserves_sequences_and_other_writes(true, false); + check_request_skip_wal_preserves_sequences_and_other_writes(true, true); } - fn check_request_skip_wal_preserves_sequences_and_other_writes(skip_wal: bool) { + fn check_request_skip_wal_preserves_sequences_and_other_writes( + skip_wal: bool, + bulk_skip_wal: bool, + ) { let builder = VersionControlBuilder::new(); let region_id = builder.region_id(); let version_control = Arc::new(builder.build()); @@ -600,10 +607,13 @@ mod tests { { assert_eq!(mutation.rows.as_ref().unwrap().rows.len(), notify.num_rows); } - assert!(ctx.push_bulk(OptionOutputTx::none(), new_bulk_part(), None)); + assert!(ctx.push_bulk(OptionOutputTx::none(), new_bulk_part(), None, bulk_skip_wal)); assert!(!ctx.skip_wal()); assert_eq!(ctx.next_sequence, 9); - assert_eq!(ctx.wal_entry.bulk_entries.len(), 1); + assert_eq!( + ctx.wal_entry.bulk_entries.len(), + usize::from(!bulk_skip_wal) + ); assert_eq!(ctx.bulk_parts[0].sequence, 7); let sequences: Vec<_> = ctx.wal_entry.mutations.iter().map(|m| m.sequence).collect(); assert_eq!(sequences, if skip_wal { vec![3, 6] } else { vec![1, 3, 6] }); @@ -709,8 +719,54 @@ mod tests { assert!(rx.blocking_recv().unwrap().is_err()); } + #[test] + fn test_bulk_skip_wal_preserves_sequences() { + check_bulk_skip_wal_preserves_sequences(false); + check_bulk_skip_wal_preserves_sequences(true); + } + + fn check_bulk_skip_wal_preserves_sequences(skip_wal: bool) { + let builder = VersionControlBuilder::new(); + let region_id = builder.region_id(); + let version_control = Arc::new(builder.build()); + let mut ctx = RegionWriteCtx::new( + region_id, + &version_control, + Provider::raft_engine_provider(region_id.as_u64()), + None, + ); + // Alternate policies in one context, retaining all parts for the memtable. + for skip in [skip_wal, false, skip_wal] { + assert!(ctx.push_bulk(OptionOutputTx::none(), new_bulk_part(), None, skip)); + } + assert_eq!(ctx.next_sequence, 7); + assert_eq!(ctx.bulk_notifiers.len(), 3); + assert_eq!( + ctx.bulk_parts + .iter() + .map(|p| p.sequence) + .collect::>(), + vec![1, 3, 5] + ); + let encoded = crate::wal::encoder::WalEntryEncoder::new().encode_to_vec(&ctx.wal_entry); + let decoded = WalEntry::decode(encoded.as_slice()).unwrap(); + assert_eq!( + decoded + .bulk_entries + .iter() + .map(|e| e.sequence) + .collect::>(), + if skip_wal { vec![3] } else { vec![1, 3, 5] } + ); + } + #[test] fn test_set_error_marks_bulk_notifiers_failed() { + check_set_error_marks_bulk_notifiers_failed(false); + check_set_error_marks_bulk_notifiers_failed(true); + } + + fn check_set_error_marks_bulk_notifiers_failed(skip_wal: bool) { let builder = VersionControlBuilder::new(); let region_id = builder.region_id(); let version_control = Arc::new(builder.build()); @@ -718,7 +774,9 @@ mod tests { RegionWriteCtx::new(region_id, &version_control, Provider::noop_provider(), None); let (tx, rx) = oneshot::channel(); - assert!(ctx.push_bulk(OptionOutputTx::from(tx), new_bulk_part(), None)); + assert!(ctx.push_bulk(OptionOutputTx::from(tx), new_bulk_part(), None, skip_wal)); + assert_eq!(ctx.wal_entry.bulk_entries.len(), usize::from(!skip_wal)); + assert_eq!(ctx.bulk_parts.len(), 1); ctx.set_error(Arc::new( UnexpectedSnafu { reason: "wal failed".to_string(), diff --git a/src/mito2/src/request.rs b/src/mito2/src/request.rs index 0d918e542bf..16b7eb4a6be 100644 --- a/src/mito2/src/request.rs +++ b/src/mito2/src/request.rs @@ -560,6 +560,7 @@ pub(crate) struct SenderWriteRequest { } pub(crate) struct SenderBulkRequest { + pub(crate) skip_wal: bool, pub(crate) sender: OptionOutputTx, pub(crate) region_id: RegionId, pub(crate) request: BulkPart, diff --git a/src/mito2/src/worker/handle_bulk_insert.rs b/src/mito2/src/worker/handle_bulk_insert.rs index b4a7ae4a182..c094c1fa6dc 100644 --- a/src/mito2/src/worker/handle_bulk_insert.rs +++ b/src/mito2/src/worker/handle_bulk_insert.rs @@ -96,6 +96,7 @@ impl RegionWorkerLoop { None }; pending_bulk_request.push(SenderBulkRequest { + skip_wal: request.skip_wal, sender, request: part, region_id: request.region_id, diff --git a/src/mito2/src/worker/handle_write.rs b/src/mito2/src/worker/handle_write.rs index 856fb75978d..a42f27eaf1e 100644 --- a/src/mito2/src/worker/handle_write.rs +++ b/src/mito2/src/worker/handle_write.rs @@ -565,7 +565,7 @@ impl RegionWorkerLoop { } // Collect requests by region. - if !region_ctx.push_bulk(bulk_req.sender, bulk_req.request, None) { + if !region_ctx.push_bulk(bulk_req.sender, bulk_req.request, None, bulk_req.skip_wal) { return; } } @@ -930,17 +930,32 @@ mod tests { async fn test_request_skip_wal_does_not_append_empty_batch() { // Only change the request flag. A failing log store demonstrates that // the all-skipped path never invokes append_batch, including empty appends. - check_request_skip_wal_does_not_append_empty_batch(false).await; - check_request_skip_wal_does_not_append_empty_batch(true).await; + check_request_skip_wal_does_not_append_empty_batch(false, false).await; + check_request_skip_wal_does_not_append_empty_batch(false, true).await; + check_request_skip_wal_does_not_append_empty_batch(true, false).await; + check_request_skip_wal_does_not_append_empty_batch(true, true).await; } - async fn check_request_skip_wal_does_not_append_empty_batch(skip_wal: bool) { + async fn check_request_skip_wal_does_not_append_empty_batch(skip_wal: bool, bulk: bool) { let region_id = RegionId::new(1, 1); let wal = Wal::new(Arc::new(MockLogStore { fail_append: true, ..Default::default() })); - let (ctx, rx) = new_region_ctx(region_id, skip_wal); + let (ctx, rx) = if bulk { + let version_control = Arc::new(VersionControlBuilder::new().build()); + let mut ctx = RegionWriteCtx::new( + region_id, + &version_control, + Provider::raft_engine_provider(region_id.as_u64()), + None, + ); + let (tx, rx) = oneshot::channel(); + assert!(ctx.push_bulk(OptionOutputTx::from(tx), new_bulk_part(1), None, skip_wal)); + (ctx, rx) + } else { + new_region_ctx(region_id, skip_wal) + }; let version_control = ctx.version_control().clone(); let mut contexts = HashMap::from([(region_id, ctx)]); assert_eq!(write_wal(&wal, &mut contexts).await, skip_wal); @@ -948,6 +963,7 @@ mod tests { let ctx = contexts.get_mut(®ion_id).unwrap(); assert_eq!(ctx.next_entry_id(), 1); ctx.write_memtable().await; + ctx.write_bulk().await; ctx.publish_sequence_and_entry_id(); assert_eq!(version_control.committed_sequence(), 1); assert_eq!(version_control.current().last_entry_id, 0); @@ -1007,6 +1023,11 @@ mod tests { #[tokio::test] async fn test_bulk_write_sequence_not_committed_before_install_worker_level() { + check_bulk_write_sequence_not_committed_before_install_worker_level(false).await; + check_bulk_write_sequence_not_committed_before_install_worker_level(true).await; + } + + async fn check_bulk_write_sequence_not_committed_before_install_worker_level(skip_wal: bool) { let region_id = RegionId::new(1, 1); let version_control = Arc::new(VersionControlBuilder::new().build()); @@ -1018,7 +1039,7 @@ mod tests { None, ); let (tx, rx) = oneshot::channel(); - assert!(ctx.push_bulk(OptionOutputTx::from(tx), new_bulk_part(3), None)); + assert!(ctx.push_bulk(OptionOutputTx::from(tx), new_bulk_part(3), None, skip_wal)); region_ctxs.insert(region_id, ctx); let wal = Wal::new(Arc::new(MockLogStore::default())); diff --git a/src/operator/src/bulk_insert.rs b/src/operator/src/bulk_insert.rs index dc570bf8761..7abfaa69ee1 100644 --- a/src/operator/src/bulk_insert.rs +++ b/src/operator/src/bulk_insert.rs @@ -48,6 +48,7 @@ impl Inserter { raw_flight_data: FlightData, record_batch: RecordBatch, schema_bytes: Bytes, + skip_wal: bool, ) -> error::Result { let table_info = table.table_info(); let table_id = table_info.table_id(); @@ -101,6 +102,7 @@ impl Inserter { ..Default::default() }), body: Some(region_request::Body::BulkInsert(BulkInsertRequest { + skip_wal, region_id: region_id.as_u64(), partition_expr_version: partition_expr_version .map(|value| PartitionExprVersion { value }), @@ -222,6 +224,7 @@ impl Inserter { ..Default::default() }), body: Some(region_request::Body::BulkInsert(BulkInsertRequest { + skip_wal, region_id: region_id.as_u64(), partition_expr_version: partition_expr_version .map(|value| PartitionExprVersion { value }), diff --git a/src/operator/src/req_convert/common/partitioner.rs b/src/operator/src/req_convert/common/partitioner.rs index d83cf1d207b..0b33782bd9d 100644 --- a/src/operator/src/req_convert/common/partitioner.rs +++ b/src/operator/src/req_convert/common/partitioner.rs @@ -181,7 +181,7 @@ impl TablePartitioner { .rule .split_record_batch(&batch) .context(SplitInsertSnafu)?; - self.build_bulk_insert_requests(batch, masks, original_ipc) + self.build_bulk_insert_requests(batch, masks, original_ipc, false) } fn build_bulk_insert_requests( @@ -189,6 +189,7 @@ impl TablePartitioner { batch: RecordBatch, masks: HashMap, mut original_ipc: Option, + skip_wal: bool, ) -> Result> { let mut requests = Vec::with_capacity(masks.len()); for (region, mask) in masks { @@ -219,6 +220,7 @@ impl TablePartitioner { } }; requests.push(BulkInsertRequest { + skip_wal, region_id: RegionId::new(self.table_id, region).as_u64(), partition_expr_version: self.version(region), // The destination datanode validates against its current schema. @@ -296,7 +298,7 @@ impl TablePartitioner { .split_record_batch(routing_batch.as_ref().unwrap_or(&batch)) .context(SplitInsertSnafu)?; Ok(self - .build_bulk_insert_requests(batch, masks, Some(ipc))? + .build_bulk_insert_requests(batch, masks, Some(ipc), bulk.skip_wal)? .into_iter() .map(Body::BulkInsert) .collect()) @@ -937,7 +939,12 @@ mod tests { } } - fn build_bulk_request(region: u32, version: Option, values: Vec) -> RegionRequest { + fn build_bulk_request( + region: u32, + version: Option, + values: Vec, + skip_wal: bool, + ) -> RegionRequest { let batch = RecordBatch::try_new( Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])), vec![Arc::new(Int32Array::from(values))], @@ -945,6 +952,7 @@ mod tests { .unwrap(); let (schema, data_header, payload) = flight::record_batch_to_ipc(batch).unwrap(); build_request(Body::BulkInsert(BulkInsertRequest { + skip_wal, region_id: RegionId::new(1, region).as_u64(), partition_expr_version: version.map(|value| PartitionExprVersion { value }), aligned_schema_version: Some(AlignedSchemaVersion { @@ -960,13 +968,18 @@ mod tests { #[tokio::test] async fn test_rebuild_bulk_reencodes_only_changed_partitions() { + check_rebuild_bulk_reencodes_only_changed_partitions(false).await; + check_rebuild_bulk_reencodes_only_changed_partitions(true).await; + } + + async fn check_rebuild_bulk_reencodes_only_changed_partitions(skip_wal: bool) { let (manager, cache) = create_fixture().await; let table_info = cache.get(1).await.unwrap().unwrap(); let (_, versions) = manager .find_table_partition_rule(&table_info) .await .unwrap(); - let request = build_bulk_request(99, Some(0), vec![1, 20, 70]); + let request = build_bulk_request(99, Some(0), vec![1, 20, 70], skip_wal); let header = request.header.clone(); let rerouted = rebuild_with_cached_table(&manager, &cache, request) .await @@ -980,6 +993,7 @@ mod tests { }; let region = RegionId::from_u64(bulk.region_id); assert_eq!(region.table_id(), 1); + assert_eq!(bulk.skip_wal, skip_wal); assert_eq!(bulk.aligned_schema_version, None); assert_eq!( bulk.partition_expr_version.map(|v| v.value), @@ -1004,7 +1018,7 @@ mod tests { assert_eq!(actual, [(1, 70), (2, 20), (3, 1)]); // A leader-only change retains the original IPC buffers and schema hint. - let request = build_bulk_request(3, versions[&3], vec![1]); + let request = build_bulk_request(3, versions[&3], vec![1], skip_wal); let Some(Body::BulkInsert(bulk)) = request.body.as_ref() else { unreachable!() }; @@ -1023,6 +1037,7 @@ mod tests { let Some(Body::BulkInsert(bulk)) = rerouted[0].body.as_ref() else { unreachable!() }; + assert_eq!(bulk.skip_wal, skip_wal); assert_eq!( bulk.aligned_schema_version, Some(AlignedSchemaVersion { diff --git a/src/operator/src/write_dispatcher/tests.rs b/src/operator/src/write_dispatcher/tests.rs index d3345bbbf7a..d9a273d0c30 100644 --- a/src/operator/src/write_dispatcher/tests.rs +++ b/src/operator/src/write_dispatcher/tests.rs @@ -857,6 +857,7 @@ fn build_partitioned_request( .unwrap(); let (schema, data_header, payload) = flight::record_batch_to_ipc(batch).unwrap(); Body::BulkInsert(BulkInsertRequest { + skip_wal: false, region_id, partition_expr_version, aligned_schema_version: Some(AlignedSchemaVersion { @@ -1383,6 +1384,7 @@ async fn test_dispatch_preserves_write_payloads_and_envelopes() { ..Default::default() }), body: Some(Body::BulkInsert(BulkInsertRequest { + skip_wal: false, region_id: region.as_u64(), partition_expr_version: version, aligned_schema_version: None, diff --git a/src/servers/src/pending_rows_batcher.rs b/src/servers/src/pending_rows_batcher.rs index 7063f9773a1..59d58c7a385 100644 --- a/src/servers/src/pending_rows_batcher.rs +++ b/src/servers/src/pending_rows_batcher.rs @@ -214,6 +214,7 @@ struct BatchKey { catalog: String, schema: String, physical_table: String, + skip_wal: bool, } /// An aligned logical record batch and its timestamp column index. @@ -318,8 +319,7 @@ enum WorkerCommand { Ack { ack_tx: oneshot::Sender<()> }, } -// Batch key is derived from QueryContext; it assumes catalog/schema/physical_table fully -// define the write target and must remain consistent across the batch. +// Requests can share a batch only when their write target and WAL policy match. fn batch_key_from_ctx(ctx: &QueryContextRef) -> BatchKey { let physical_table = ctx .extension(PHYSICAL_TABLE_KEY) @@ -329,6 +329,7 @@ fn batch_key_from_ctx(ctx: &QueryContextRef) -> BatchKey { catalog: ctx.current_catalog().to_string(), schema: ctx.current_schema(), physical_table, + skip_wal: ctx.skip_wal(), } } @@ -1538,7 +1539,8 @@ pub async fn flush_batch_physical( let planned_batches = plan_region_batches(combined_batch, physical_table_id, partition_rule.as_ref())?; - let mut region_writes = encode_region_write_requests(planned_batches, &partition_versions)?; + let mut region_writes = + encode_region_write_requests(planned_batches, &partition_versions, ctx.skip_wal())?; for request in &mut region_writes { if let Some(header) = &mut request.header { header.dbname = ctx.get_db_string(); @@ -1690,6 +1692,7 @@ fn plan_region_batches( fn encode_region_write_requests( planned_batches: Vec, partition_versions: &HashMap>, + skip_wal: bool, ) -> Result> { let mut region_writes = Vec::with_capacity(planned_batches.len()); for planned in planned_batches { @@ -1707,6 +1710,7 @@ fn encode_region_write_requests( ..Default::default() }), body: Some(region_request::Body::BulkInsert(BulkInsertRequest { + skip_wal, region_id: region_id.as_u64(), partition_expr_version: partition_versions .get(®ion_id.region_number()) @@ -1807,8 +1811,8 @@ mod tests { BatchKey, Error, FlushBatch, FlushWaiter, PendingBatch, PendingRowsBatcher, PendingWorker, PhysicalFlushCatalogProvider, PhysicalFlushPartitionProvider, PhysicalFlushWriteDispatcher, PhysicalTableMetadata, PlannedRegionBatch, RecordBatchWithTsIdx, TableBatch, WorkerCommand, - columns_taxonomy, drain_batch, encode_region_write_requests, extract_timestamps, - flush_batch, flush_batch_physical, greptime_timestamp, + batch_key_from_ctx, columns_taxonomy, drain_batch, encode_region_write_requests, + extract_timestamps, flush_batch, flush_batch_physical, greptime_timestamp, notify_flow_dirty_windows_after_flush, plan_region_batches, remove_worker_if_same_channel, should_close_worker_on_idle_timeout, start_flow_notification_worker, start_worker, transform_logical_batches_to_physical, try_enqueue_flow_notification, @@ -2674,6 +2678,31 @@ mod tests { ); } + #[test] + fn test_batch_key_groups_by_skip_wal() { + let wal_ctx = session::context::QueryContext::arc(); + let skip_wal_ctx = session::context::QueryContext::arc(); + skip_wal_ctx.set_skip_wal(true); + let another_skip_wal_ctx = session::context::QueryContext::arc(); + another_skip_wal_ctx.set_skip_wal(true); + + let mut batches = HashMap::new(); + *batches.entry(batch_key_from_ctx(&wal_ctx)).or_insert(0) += 1; + *batches + .entry(batch_key_from_ctx(&skip_wal_ctx)) + .or_insert(0) += 1; + *batches + .entry(batch_key_from_ctx(&another_skip_wal_ctx)) + .or_insert(0) += 1; + *batches + .entry(batch_key_from_ctx(&session::context::QueryContext::arc())) + .or_insert(0) += 1; + + assert_eq!(batches.len(), 2); + assert_eq!(batches[&batch_key_from_ctx(&wal_ctx)], 2); + assert_eq!(batches[&batch_key_from_ctx(&skip_wal_ctx)], 2); + } + #[test] fn test_remove_worker_if_same_channel_removes_matching_entry() { let workers = DashMap::new(); @@ -2681,6 +2710,7 @@ mod tests { catalog: "greptime".to_string(), schema: "public".to_string(), physical_table: "phy".to_string(), + skip_wal: false, }; let (tx, _rx) = mpsc::channel::(1); @@ -2697,6 +2727,7 @@ mod tests { catalog: "greptime".to_string(), schema: "public".to_string(), physical_table: "phy".to_string(), + skip_wal: false, }; let (stale_tx, _stale_rx) = mpsc::channel::(1); @@ -2788,6 +2819,7 @@ mod tests { catalog: "greptime".to_string(), schema: "public".to_string(), physical_table: "phy".to_string(), + skip_wal: false, }; let workers = Arc::new(DashMap::new()); let (worker_tx, worker_rx) = mpsc::channel(1); @@ -3409,6 +3441,7 @@ mod tests { let writes = encode_region_write_requests( planned_batches, &HashMap::from([(1, Some(7)), (2, Some(8))]), + false, ) .unwrap(); for (index, write) in writes.into_iter().enumerate() { @@ -3443,7 +3476,8 @@ mod tests { #[test] fn test_encode_region_write_requests_builds_bulk_insert_requests() { - check_encode_region_write_requests(); + check_encode_region_write_requests(false); + check_encode_region_write_requests(true); } struct TwoRegionPartitionRule { @@ -3482,7 +3516,7 @@ mod tests { } } - fn check_encode_region_write_requests() { + fn check_encode_region_write_requests(skip_wal: bool) { let planned_batch = PlannedRegionBatch { region_id: RegionId::new(1024, 1), batch: RecordBatch::try_new( @@ -3503,15 +3537,19 @@ mod tests { ) .unwrap(), }; - let writes = - encode_region_write_requests(vec![planned_batch], &HashMap::from([(1, Some(7))])) - .unwrap(); + let writes = encode_region_write_requests( + vec![planned_batch], + &HashMap::from([(1, Some(7))]), + skip_wal, + ) + .unwrap(); assert_eq!(1, writes.len()); let Some(region_request::Body::BulkInsert(request)) = &writes[0].body else { panic!("expected bulk insert request"); }; assert_eq!(RegionId::new(1024, 1).as_u64(), request.region_id); + assert_eq!(skip_wal, request.skip_wal); assert_eq!( Some(PartitionExprVersion { value: 7 }), request.partition_expr_version diff --git a/src/store-api/src/region_request.rs b/src/store-api/src/region_request.rs index 01cf6a4acb1..6a40f3dbc6b 100644 --- a/src/store-api/src/region_request.rs +++ b/src/store-api/src/region_request.rs @@ -472,6 +472,7 @@ fn make_region_truncate(truncate: TruncateRequest) -> Result Result> { let region_id = request.region_id.into(); + let skip_wal = request.skip_wal; let partition_expr_version = request.partition_expr_version.map(|v| v.value); let aligned_schema_version = request.aligned_schema_version.map(|v| v.schema_version); let Some(Body::ArrowIpc(request)) = request.body else { @@ -493,6 +494,7 @@ fn make_region_bulk_inserts(request: BulkInsertRequest) -> Result= Duration::from_secs(1)); assert!(matches!(output.data, OutputData::Stream(_))); } + + #[rstest] + #[case::standalone_single_region(false, false)] + #[case::standalone_partitioned(false, true)] + #[case::distributed_single_region(true, false)] + #[case::distributed_partitioned(true, true)] + #[tokio::test(flavor = "multi_thread")] + async fn test_flight_bulk_skip_wal(#[case] distributed: bool, #[case] partitioned: bool) { + let mut db = + MockInstanceImpl::new(&format!("flight_bulk_skip_wal_{partitioned}"), distributed) + .await; + let runtime = common_runtime::global_runtime().clone(); + let handler = GreptimeRequestHandler::new( + db.frontend(), + None, + Some(runtime.clone()), + FlightCompression::default(), + ); + let mut server = GrpcServerBuilder::new(GrpcServerConfig::default(), runtime) + .flight_handler(Arc::new(handler)) + .build(); + server.start("127.0.0.1:0".parse().unwrap()).await.unwrap(); + let client = Database::new( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + Client::with_urls(vec![server.bind_addr().unwrap().to_string()]), + ); + let partition = if partitioned { + " PARTITION ON COLUMNS (a) (a < 0, a >= 0)" + } else { + "" + }; + client.sql(&format!( + "CREATE TABLE foo (ts TIMESTAMP TIME INDEX, a INT NOT NULL, \"B\" STRING, PRIMARY KEY (a)){partition}" + )).await.unwrap(); + let mut before = db.flush_and_snapshot_wal().await; + assert_eq!(before.len(), if partitioned { 2 } else { 1 }); + // The same stream spans both partitions and includes multiple data messages. + // Repeated rows still exercise real writes; each round flushes before the next. + for hint in [None, Some("true"), Some("false"), None] { + let hints = hint.map(|value| [(INSERT_SKIP_WAL_HINT, value)]); + test_put_record_batches_with_hints( + &client, + create_record_batches(-4), + hints.as_ref().map_or(&[], |hints| hints.as_slice()), + ) + .await; + let after = db.flush_and_snapshot_wal().await; + assert_wal_delta(&before, &after, hint == Some("true")); + // Both regions must receive rows, not merely exist in the catalog. + for (id, &(flushed_sequence, _)) in &after { + assert!( + flushed_sequence > before[id].0, + "region {id:?} received no rows" + ); + } + before = after; + } + server.shutdown().await.unwrap(); + db.shutdown().await; + } + #[tokio::test(flavor = "multi_thread")] async fn test_standalone_flight_do_put() { common_telemetry::init_default_ut_logging(); @@ -735,6 +799,14 @@ mod test { } async fn test_put_record_batches(client: &Database, record_batches: Vec) { + test_put_record_batches_with_hints(client, record_batches, &[]).await; + } + + async fn test_put_record_batches_with_hints( + client: &Database, + record_batches: Vec, + hints: &[(&str, &str)], + ) { let requests_count = record_batches.len(); let schema = record_batches[0].schema.arrow_schema().clone(); @@ -767,7 +839,7 @@ mod test { ) .boxed(); - let response_stream = client.do_put(stream).await.unwrap(); + let response_stream = client.do_put_with_hints(stream, hints).await.unwrap(); let responses = response_stream.collect::>().await; let responses_count = responses.len(); diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index 0704b184e89..18e1d144024 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -62,6 +62,8 @@ use standalone::{StandaloneDatanodeManager, StandaloneRepartitionProcedureFactor use crate::test_util::{self, StorageType, TestGuard, create_tmp_dir_and_datanode_opts}; pub struct GreptimeDbStandalone { + /// Storage engine for assertions across protocol and storage boundaries. + pub mito_engine: mito2::engine::MitoEngine, pub frontend: Arc, pub opts: StandaloneOptions, pub guard: TestGuard, @@ -342,6 +344,7 @@ impl GreptimeDbStandaloneBuilder { }; GreptimeDbStandalone { + mito_engine: datanode.region_server().mito_engine().unwrap(), frontend: Arc::new(frontend), opts, guard, diff --git a/tests-integration/src/test_util.rs b/tests-integration/src/test_util.rs index 7f36c74cb7b..d903023a857 100644 --- a/tests-integration/src/test_util.rs +++ b/tests-integration/src/test_util.rs @@ -12,6 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::BTreeMap; +#[cfg(test)] +use std::collections::HashMap; use std::env; use std::fmt::Display; use std::net::SocketAddr; @@ -26,7 +29,10 @@ use axum::Router; use catalog::kvbackend::KvBackendCatalogManager; use client::{Client, Database}; use common_base::Plugins; +use common_catalog::consts::MIN_USER_TABLE_ID; use common_config::Configurable; +#[cfg(test)] +use common_meta::DatanodeId; use common_meta::key::TableMetadataManager; use common_meta::key::catalog_name::CatalogNameKey; use common_meta::key::schema_name::SchemaNameKey; @@ -38,8 +44,13 @@ use common_test_util::ports; use common_test_util::temp_dir::{TempDir, create_temp_dir}; use common_wal::config::DatanodeWalConfig; use datanode::config::{DatanodeOptions, StorageConfig}; +#[cfg(test)] +use datanode::datanode::Datanode; use frontend::instance::Instance; use frontend::service_config::{MysqlOptions, PostgresOptions}; +#[cfg(test)] +use meta_srv::metasrv::Metasrv; +use mito2::engine::MitoEngine; use mito2::gc::GcConfig; use object_store::config::{ AzblobConfig, FileConfig, GcsConfig, ObjectStoreConfig, OssConfig, S3Config, @@ -62,9 +73,178 @@ use servers::request_memory_limiter::ServerMemoryLimiter; use servers::server::Server; use servers::tls::ReloadableTlsServerConfig; use session::context::QueryContext; +use store_api::metric_engine_consts::METRIC_METADATA_REGION_GROUP; +use store_api::region_engine::RegionEngine; +use store_api::region_request::{RegionFlushRequest, RegionRequest}; +use store_api::storage::RegionId; +use crate::cluster::{GreptimeDbCluster, GreptimeDbClusterBuilder}; use crate::standalone::{GreptimeDbStandalone, GreptimeDbStandaloneBuilder}; +/// Maps `(node_id, region_id)` to `(flushed_sequence, flushed_entry_id)`. +pub type WalSnapshot = BTreeMap<(u64, RegionId), (u64, u64)>; + +/// Flush user data regions before observing their persisted WAL watermarks. +/// Metadata regions write WAL independently and are excluded from this check. +async fn flush_and_snapshot_region_wal(engine: &MitoEngine) -> WalSnapshot { + let mut snapshot = BTreeMap::new(); + for region in engine.regions() { + let id = region.region_id(); + if id.table_id() < MIN_USER_TABLE_ID || id.region_group() == METRIC_METADATA_REGION_GROUP { + continue; + } + engine + .handle_request(id, RegionRequest::Flush(RegionFlushRequest::default())) + .await + .unwrap(); + let statistic = engine.region_statistic(id).unwrap(); + snapshot.insert( + (0, id), + ( + region.flushed_sequence(), + statistic.manifest.data_flushed_entry_id(), + ), + ); + } + snapshot +} + +/// Require a real write, then check WAL policy independently of row deduplication. +pub fn assert_wal_delta(before: &WalSnapshot, after: &WalSnapshot, skip_wal: bool) { + assert_eq!( + before.keys().collect::>(), + after.keys().collect::>(), + "warm up table creation before taking the snapshot" + ); + let mut written = 0; + for (id, &(flushed_sequence, flushed_entry_id)) in after { + let (previous_flushed_sequence, previous_flushed_entry_id) = before[id]; + if flushed_sequence > previous_flushed_sequence { + written += 1; + if skip_wal { + assert_eq!( + flushed_entry_id, previous_flushed_entry_id, + "region {id:?} wrote WAL despite the request policy" + ); + } else { + assert!( + flushed_entry_id > previous_flushed_entry_id, + "region {id:?} did not write WAL" + ); + } + } + } + assert!( + written > 0, + "request must reach a data region, not merely return success" + ); +} + +/// A test instance backed by embedded storage or a distributed cluster. +pub enum MockInstanceImpl { + Standalone(GreptimeDbStandalone), + Distributed(GreptimeDbCluster), +} + +impl MockInstanceImpl { + /// Returns the metasrv of a distributed instance. + /// + /// # Panics + /// Panics if this is a standalone instance. + #[cfg(test)] + pub(crate) fn metasrv(&self) -> &Arc { + match self { + Self::Standalone(_) => unreachable!(), + Self::Distributed(instance) => &instance.metasrv, + } + } + + /// Returns the datanodes of a distributed instance. + /// + /// # Panics + /// Panics if this is a standalone instance. + #[cfg(test)] + pub(crate) fn datanodes(&self) -> &HashMap { + match self { + Self::Standalone(_) => unreachable!(), + Self::Distributed(instance) => &instance.datanode_instances, + } + } + + /// Creates a standalone instance or a three-datanode cluster using local storage. + pub async fn new(name: &str, distributed: bool) -> Self { + let name = format!( + "{name}_{}", + if distributed { + "distributed" + } else { + "standalone" + } + ); + if distributed { + // The repository cluster harness uses real tonic/protobuf services + // over duplex transports between FE and datanodes, not direct calls. + Self::Distributed( + GreptimeDbClusterBuilder::new(&name) + .await + .with_datanodes(3) + .build(false) + .await, + ) + } else { + Self::Standalone(GreptimeDbStandaloneBuilder::new(&name).build().await) + } + } + + /// Returns the frontend instance. + pub fn frontend(&self) -> Arc { + match self { + Self::Standalone(instance) => instance.fe_instance().clone(), + Self::Distributed(cluster) => cluster.fe_instance().clone(), + } + } + + /// Flushes user data regions and returns their write and WAL watermarks. + pub async fn flush_and_snapshot_wal(&self) -> WalSnapshot { + match self { + Self::Standalone(instance) => { + flush_and_snapshot_region_wal(&instance.mito_engine).await + } + Self::Distributed(cluster) => { + let mut result = BTreeMap::new(); + for (node_id, datanode) in &cluster.datanode_instances { + let engine = datanode.region_server().mito_engine().unwrap(); + result.extend( + flush_and_snapshot_region_wal(&engine) + .await + .into_iter() + .map(|((_, region_id), watermarks)| { + ((*node_id, region_id), watermarks) + }), + ); + } + result + } + } + } + + /// Shuts down cluster services and removes the test storage. + pub async fn shutdown(&mut self) { + match self { + Self::Standalone(instance) => instance.guard.remove_all().await, + Self::Distributed(cluster) => { + cluster.metasrv.shutdown().await.unwrap(); + for datanode in cluster.datanode_instances.values_mut() { + datanode.shutdown().await.unwrap(); + } + for guard in &mut cluster.guards { + guard.remove_all().await; + } + } + } + } +} + pub const PEER_PLACEHOLDER_ADDR: &str = "127.0.0.1:3001"; #[derive(Debug, Clone, Copy, Eq, PartialEq)] @@ -663,11 +843,27 @@ async fn setup_test_prom_app_with_frontend_inner( let sql = "INSERT INTO mito(host, val, ts) VALUES (1, 1.1, 0)"; run_sql(sql, &instance).await; + let http_server = build_test_prom_server( + instance.fe_instance().clone(), + enable_batcher, + experimental_enable_prometheus_native_histogram, + ) + .with_greptime_config_options(instance.opts.datanode_options().to_toml().unwrap()) + .build(); + let app = http_server.build(http_server.make_app()).unwrap(); + (app, instance.guard) +} + +/// Builds Prometheus HTTP routes for either a standalone or distributed frontend. +pub fn build_test_prom_server( + frontend_ref: Arc, + enable_batcher: bool, + experimental_enable_prometheus_native_histogram: bool, +) -> HttpServerBuilder { let http_opts = HttpOptions { addr: format!("127.0.0.1:{}", ports::get_port()), ..Default::default() }; - let frontend_ref = instance.fe_instance().clone(); // Mirror the production wiring at `frontend::server`: build the batcher from the // instance's managers. A short flush interval keeps the test responsive. let pending_rows_batcher = if enable_batcher { @@ -689,9 +885,10 @@ async fn setup_test_prom_app_with_frontend_inner( } else { None }; - let http_server = HttpServerBuilder::new(http_opts) + assert_eq!(pending_rows_batcher.is_some(), enable_batcher); + HttpServerBuilder::new(http_opts) .with_sql_handler(frontend_ref.clone()) - .with_logs_handler(instance.fe_instance().clone()) + .with_logs_handler(frontend_ref.clone()) .with_prom_handler( frontend_ref.clone(), Some(frontend_ref.clone()), @@ -701,10 +898,6 @@ async fn setup_test_prom_app_with_frontend_inner( pending_rows_batcher, ) .with_prometheus_handler(frontend_ref) - .with_greptime_config_options(instance.opts.datanode_options().to_toml().unwrap()) - .build(); - let app = http_server.build(http_server.make_app()).unwrap(); - (app, instance.guard) } pub async fn setup_grpc_server( diff --git a/tests-integration/src/tests/instance_test.rs b/tests-integration/src/tests/instance_test.rs index a716284c758..3cd625ba204 100644 --- a/tests-integration/src/tests/instance_test.rs +++ b/tests-integration/src/tests/instance_test.rs @@ -1174,6 +1174,55 @@ async fn test_execute_query_external_table_csv(instance: Arc) check_output_stream(output, expect).await; } +#[tokio::test(flavor = "multi_thread")] +async fn test_copy_from_inherits_skip_wal() { + use crate::test_util::{MockInstanceImpl, assert_wal_delta}; + + // Local COPY access is intentionally restricted to the standalone harness. + let mut env = MockInstanceImpl::new("copy_from_skip_wal", false).await; + let instance = env.frontend(); + let directory = create_local_file_test_dir("copy_from_skip_wal"); + let csv_path = directory.path().join("copy_skip_wal.csv"); + std::fs::copy( + find_testing_resource("/tests/data/csv/headerless.csv"), + &csv_path, + ) + .unwrap(); + execute_sql( + &instance, + "CREATE TABLE copy_skip_wal(host_id INT, host_name STRING, reading_value DOUBLE, ts TIMESTAMP TIME INDEX);", + ) + .await; + + let commands = [ + format!( + "COPY copy_skip_wal FROM '{}' WITH (FORMAT='csv', HEADERS='false');", + prepare_path(&csv_path.display().to_string()) + ), + format!( + "COPY DATABASE public FROM '{}/' WITH (FORMAT='csv', HEADERS='false');", + prepare_path(&directory.path().display().to_string()) + ), + ]; + for command in commands { + for skip_wal in [Some(true), Some(false), None] { + let ctx = QueryContext::arc(); + if let Some(skip_wal) = skip_wal { + ctx.set_skip_wal(skip_wal); + } + let before = env.flush_and_snapshot_wal().await; + let output = execute_sql_with(&instance, &command, ctx).await; + assert!(matches!(output.data, OutputData::AffectedRows(2))); + assert_wal_delta( + &before, + &env.flush_and_snapshot_wal().await, + skip_wal == Some(true), + ); + } + } + env.shutdown().await; +} + #[apply(standalone_instance_case)] async fn test_execute_copy_from_headerless_csv(instance: Arc) { let instance = instance.frontend(); diff --git a/tests-integration/src/tests/test_util.rs b/tests-integration/src/tests/test_util.rs index ab860367509..5ddcd0748a9 100644 --- a/tests-integration/src/tests/test_util.rs +++ b/tests-integration/src/tests/test_util.rs @@ -42,6 +42,7 @@ use session::context::{QueryContext, QueryContextRef}; use crate::cluster::{GreptimeDbCluster, GreptimeDbClusterBuilder}; use crate::standalone::{GreptimeDbStandalone, GreptimeDbStandaloneBuilder}; +pub(crate) use crate::test_util::MockInstanceImpl; use crate::test_util::StorageType; use crate::tests::{MockDistributedInstance, create_distributed_instance}; @@ -85,33 +86,9 @@ pub(crate) enum MockInstanceBuilder { Distributed(GreptimeDbClusterBuilder), } -pub(crate) enum MockInstanceImpl { - Standalone(GreptimeDbStandalone), - Distributed(GreptimeDbCluster), -} - -impl MockInstanceImpl { - pub(crate) fn metasrv(&self) -> &Arc { - match self { - MockInstanceImpl::Standalone(_) => unreachable!(), - MockInstanceImpl::Distributed(instance) => &instance.metasrv, - } - } - - pub(crate) fn datanodes(&self) -> &HashMap { - match self { - MockInstanceImpl::Standalone(_) => unreachable!(), - MockInstanceImpl::Distributed(instance) => &instance.datanode_instances, - } - } -} - impl MockInstance for MockInstanceImpl { fn frontend(&self) -> Arc { - match self { - MockInstanceImpl::Standalone(instance) => instance.frontend(), - MockInstanceImpl::Distributed(instance) => instance.fe_instance().clone(), - } + MockInstanceImpl::frontend(self) } fn is_distributed_mode(&self) -> bool { diff --git a/tests-integration/tests/grpc.rs b/tests-integration/tests/grpc.rs index 1d9b123d1b7..9607d611627 100644 --- a/tests-integration/tests/grpc.rs +++ b/tests-integration/tests/grpc.rs @@ -15,13 +15,16 @@ use std::sync::Arc; use api::v1::alter_table_expr::Kind; +use api::v1::greptime_database_client::GreptimeDatabaseClient; +use api::v1::greptime_request::Request as RequestBody; +use api::v1::greptime_response::Response as ResponseBody; use api::v1::promql_request::Promql; use api::v1::value::ValueData; use api::v1::{ - AddColumn, AddColumns, AlterTableExpr, Basic, Column, ColumnDataType, ColumnDef, - CreateTableExpr, InsertRequest, InsertRequests, PromInstantQuery, PromRangeQuery, - PromqlRequest, RequestHeader, Row, RowInsertRequest, RowInsertRequests, SemanticType, Value, - column, + AddColumn, AddColumns, AlterTableExpr, Basic, Column, ColumnDataType, ColumnDef, ColumnSchema, + CreateTableExpr, GreptimeRequest, InsertRequest, InsertRequests, PromInstantQuery, + PromRangeQuery, PromqlRequest, RequestHeader, Row, RowInsertRequest, RowInsertRequests, Rows, + SemanticType, Value, column, }; use auth::user_provider_from_option; use base64::prelude::{BASE64_STANDARD, Engine as _}; @@ -52,6 +55,7 @@ use otel_arrow_rust::proto::opentelemetry::arrow::v1::{ }; use otel_arrow_rust::proto::opentelemetry::metrics::v1::AggregationTemporality; use otel_arrow_rust::schema::consts as arrow_consts; +use rstest_reuse::apply; use servers::grpc::GrpcServerConfig; use servers::grpc::builder::GrpcServerBuilder; use servers::http::prometheus::{ @@ -61,13 +65,17 @@ use servers::http::prometheus::{ use servers::request_memory_limiter::ServerMemoryLimiter; use servers::server::Server; use servers::tls::{TlsMode, TlsOption}; +use session::hints::{HINTS_KEY, INSERT_SKIP_WAL_HINT}; use tests_integration::test_util::{ - StorageType, setup_grpc_server, setup_grpc_server_with, + MockInstanceImpl, StorageType, assert_wal_delta, setup_grpc_server, + setup_grpc_server_for_frontend_instance, setup_grpc_server_with, setup_grpc_server_with_auto_create_table_disabled, setup_grpc_server_with_user_provider, }; use tonic::Request; use tonic::metadata::MetadataValue; +use crate::both_deployment_cases; + #[macro_export] macro_rules! grpc_test { ($service:ident, $($(#[$meta:meta])* $test:ident),*,) => { @@ -776,6 +784,44 @@ fn gauge_arrow_batch(batch_id: i64, reserved_attr: bool) -> BatchArrowRecords { } } +#[apply(both_deployment_cases)] +async fn test_skip_wal_otel_arrow_metrics(distributed: bool) { + // OTEL Arrow metrics decode into ordinary metric inserts, unlike Flight DoPut bulk inserts. + let mut env = MockInstanceImpl::new("skip_wal_otel_arrow", distributed).await; + let server = setup_grpc_server_for_frontend_instance(env.frontend(), None).await; + let mut client = + ArrowMetricsServiceClient::connect(format!("http://{}", server.bind_addr().unwrap())) + .await + .unwrap(); + // Warm up auto-created logical/physical tables and metadata before comparing data-region WAL. + for (batch_id, hint) in [None, Some("true"), Some("false"), None] + .into_iter() + .enumerate() + { + let before = env.flush_and_snapshot_wal().await; + let batch = gauge_arrow_batch(batch_id as i64, false); + let request = with_skip_wal_hint(futures::stream::iter([batch]), hint); + let mut response = client.arrow_metrics(request).await.unwrap().into_inner(); + let status = response.message().await.unwrap().unwrap(); + assert_eq!( + status.status_code, + ArrowStatusCode::Ok as i32, + "{}", + status.status_message + ); + assert!(response.message().await.unwrap().is_none()); + if batch_id != 0 { + assert_wal_delta( + &before, + &env.flush_and_snapshot_wal().await, + hint == Some("true"), + ); + } + } + server.shutdown().await.unwrap(); + env.shutdown().await; +} + pub async fn test_otel_arrow_delta_histogram(store_type: StorageType) { let (_instance, server) = setup_grpc_server(store_type, "test_otel_arrow_delta_histogram").await; @@ -1152,6 +1198,139 @@ fn expect_data() -> (Column, Column, Column, Column) { ) } +fn with_skip_wal_hint(body: T, hint: Option<&str>) -> Request { + let mut request = Request::new(body); + if let Some(hint) = hint { + request.metadata_mut().insert( + HINTS_KEY, + format!("{INSERT_SKIP_WAL_HINT}={hint}").parse().unwrap(), + ); + } + request +} + +fn skip_wal_insert_body(columnar: bool) -> RequestBody { + if columnar { + RequestBody::Inserts(InsertRequests { + inserts: vec![InsertRequest { + table_name: "skip_wal_grpc".to_string(), + row_count: 1, + columns: vec![Column { + column_name: "ts".to_string(), + semantic_type: SemanticType::Timestamp as i32, + datatype: ColumnDataType::TimestampMillisecond as i32, + values: Some(column::Values { + timestamp_millisecond_values: vec![1000], + ..Default::default() + }), + ..Default::default() + }], + }], + }) + } else { + RequestBody::RowInserts(RowInsertRequests { + inserts: vec![RowInsertRequest { + table_name: "skip_wal_grpc".to_string(), + rows: Some(Rows { + schema: vec![ColumnSchema { + column_name: "ts".to_string(), + semantic_type: SemanticType::Timestamp as i32, + datatype: ColumnDataType::TimestampMillisecond as i32, + ..Default::default() + }], + rows: vec![Row { + values: vec![Value { + value_data: Some(ValueData::TimestampMillisecondValue(1000)), + }], + }], + }), + }], + }) + } +} + +#[apply(both_deployment_cases)] +async fn test_skip_wal_grpc_unary_stream_and_flight_sql(distributed: bool) { + let mut env = MockInstanceImpl::new("skip_wal_grpc_protocols", distributed).await; + let server = setup_grpc_server_for_frontend_instance(env.frontend(), None).await; + let addr = server.bind_addr().unwrap().to_string(); + let mut grpc = GreptimeDatabaseClient::connect(format!("http://{addr}")) + .await + .unwrap(); + let database = Database::new_with_dbname("greptime-public", Client::with_urls(vec![addr])); + database + .sql("CREATE TABLE skip_wal_grpc (ts TIMESTAMP TIME INDEX)") + .await + .unwrap(); + + for streaming in [false, true] { + for columnar in [false, true] { + for hint in [Some("true"), Some("false"), None] { + let before = env.flush_and_snapshot_wal().await; + let request = GreptimeRequest { + header: Some(RequestHeader { + catalog: "greptime".to_string(), + schema: "public".to_string(), + ..Default::default() + }), + request: Some(skip_wal_insert_body(columnar)), + }; + let response = if streaming { + grpc.handle_requests(with_skip_wal_hint( + futures::stream::iter([request.clone(), request]), + hint, + )) + .await + .unwrap() + } else { + grpc.handle(with_skip_wal_hint(request, hint)) + .await + .unwrap() + } + .into_inner(); + let expected_rows = if streaming { 2 } else { 1 }; + assert!( + matches!(response.response, Some(ResponseBody::AffectedRows(rows)) if rows.value == expected_rows) + ); + assert_wal_delta( + &before, + &env.flush_and_snapshot_wal().await, + hint == Some("true"), + ); + } + } + } + for hint in [Some("true"), Some("false"), None] { + let before = env.flush_and_snapshot_wal().await; + let hints = hint + .map(|value| vec![(INSERT_SKIP_WAL_HINT, value)]) + .unwrap_or_default(); + database + .sql_with_hint("INSERT INTO skip_wal_grpc VALUES (1000)", &hints) + .await + .unwrap(); + assert_wal_delta( + &before, + &env.flush_and_snapshot_wal().await, + hint == Some("true"), + ); + } + // Strict validation must reject the request before any data write. + let before = env.flush_and_snapshot_wal().await; + assert!( + database + .sql_with_hint( + "INSERT INTO skip_wal_grpc VALUES (1000)", + &[(INSERT_SKIP_WAL_HINT, "yes")] + ) + .await + .is_err() + ); + assert_eq!(before, env.flush_and_snapshot_wal().await); + server.shutdown().await.unwrap(); + env.shutdown().await; +} + pub async fn test_insert_and_select(store_type: StorageType) { common_telemetry::init_default_ut_logging(); let (_db, fe_grpc_server) = setup_grpc_server(store_type, "test_insert_and_select").await; diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index c696bd40ac4..c8f18be5d66 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -58,8 +58,8 @@ use opentelemetry_proto::tonic::collector::trace::v1::{ }; use pipeline::GREPTIME_INTERNAL_TRACE_PIPELINE_V1_NAME; use prost::Message; +use rstest_reuse::apply; use serde_json::{Value, json}; -use servers::http::GreptimeQueryOutput; use servers::http::handler::HealthResponse; use servers::http::header::constants::{ GREPTIME_LOG_EXTRACT_KEYS_HEADER_NAME, GREPTIME_LOG_TABLE_NAME_HEADER_NAME, @@ -72,20 +72,23 @@ use servers::http::result::error_result::ErrorResponse; use servers::http::result::greptime_result_v1::GreptimedbV1Response; use servers::http::result::influxdb_result_v1::{InfluxdbOutput, InfluxdbV1Response}; use servers::http::test_helpers::{TestClient, TestResponse}; +use servers::http::{GreptimeQueryOutput, HttpOptions, HttpServerBuilder}; use servers::prom_remote_write::v2::test_util as remote_write_v2; +use servers::prom_remote_write::validation::PromValidationMode; use servers::prom_store::{self, mock_timeseries_new_label}; use servers::request_memory_limiter::ServerMemoryLimiter; use standalone::options::StandaloneOptions; use table::table_name::TableName; use tests_integration::test_util::{ - StorageType, setup_test_http_app, setup_test_http_app_with_frontend, - setup_test_http_app_with_frontend_and_slow_query_threshold, + MockInstanceImpl, StorageType, assert_wal_delta, build_test_prom_server, setup_test_http_app, + setup_test_http_app_with_frontend, setup_test_http_app_with_frontend_and_slow_query_threshold, setup_test_http_app_with_frontend_and_user_provider, setup_test_prom_app_with_frontend, setup_test_prom_app_with_frontend_batched, setup_test_prom_app_with_frontend_native_histogram, }; use urlencoding::encode; use yaml_rust::YamlLoader; +use crate::both_deployment_cases; use crate::event_recorder_test_util::assert_procedure_actor_by_table; #[macro_export] @@ -444,6 +447,33 @@ pub async fn test_cors() { guard.remove_all().await; } +#[apply(both_deployment_cases)] +async fn test_sql_skip_wal(distributed: bool) { + let mut cases = vec![HttpWalCase::new( + "SQL", + "/v1/sql", + "application/x-www-form-urlencoded", + "sql=INSERT+INTO+wal_sql+VALUES+(1%2C-1)%2C(2%2C1)", + )]; + let mut sql_get = HttpWalCase::new( + "SQL GET", + "/v1/sql?sql=INSERT%20INTO%20wal_sql%20VALUES%20(1%2C-1)%2C(2%2C1)", + "application/x-www-form-urlencoded", + Vec::new(), + ); + sql_get.get = true; + cases.push(sql_get); + for case in &mut cases { + case.create_table = Some(if distributed { + "CREATE TABLE IF NOT EXISTS wal_sql (ts TIMESTAMP TIME INDEX, v INT) PARTITION ON COLUMNS (v) (v < 0, v >= 0)" + } else { + "CREATE TABLE IF NOT EXISTS wal_sql (ts TIMESTAMP TIME INDEX, v INT)" + }); + case.expected_written_nodes = distributed.then_some(2); + } + check_http_skip_wal("sql", &cases, distributed).await; +} + pub async fn test_sql_api(store_type: StorageType) { let (app, mut guard) = setup_test_http_app_with_frontend(store_type, "sql_api").await; let client = TestClient::new(app).await; @@ -1628,6 +1658,25 @@ pub async fn test_splunk_health_is_public(store_type: StorageType) { assert_eq!(StatusCode::OK, res.status()); } +#[apply(both_deployment_cases)] +async fn test_splunk_skip_wal(distributed: bool) { + let cases = vec![ + HttpWalCase::new( + "Splunk event", + "/v1/splunk/services/collector/event", + "application/json", + r#"{"event":"wal test","time":1700000000,"index":"wal_splunk"}"#, + ), + HttpWalCase::new( + "Splunk raw", + "/v1/splunk/services/collector/raw?index=wal_splunk_raw&time=1700000000", + "text/plain", + "wal test", + ), + ]; + check_http_skip_wal("splunk", &cases, distributed).await; +} + pub async fn test_splunk_logs(store_type: StorageType) { common_telemetry::init_default_ut_logging(); @@ -2649,6 +2698,43 @@ pub async fn test_dashboard_api(store_type: StorageType) { #[cfg(not(feature = "dashboard"))] pub async fn test_dashboard_api(_: StorageType) {} +#[apply(both_deployment_cases)] +async fn test_prometheus_skip_wal(distributed: bool) { + let mut cases = Vec::new(); + let prom = api::prom_store::remote::WriteRequest { + timeseries: prom_store::mock_timeseries(), + ..Default::default() + }; + let mut prom_case = HttpWalCase::new( + "Prometheus v1", + "/v1/prometheus/write", + "application/x-protobuf", + prom_store::snappy_compress(&prom.encode_to_vec()).unwrap(), + ); + prom_case.headers.push(("content-encoding", "snappy")); + cases.push(prom_case); + let prom_v2 = servers::prom_remote_write::v2::test_util::request_with_labels_and_samples( + vec![ + (prom_store::METRIC_NAME_LABEL, "wal_prom_v2"), + ("host", "a"), + ], + vec![api::greptime_proto::io::prometheus::write::v2::Sample { + value: 1.0, + timestamp: 1700000000000, + start_timestamp: 0, + }], + ); + let mut prom_v2_case = HttpWalCase::new( + "Prometheus v2", + "/v1/prometheus/write", + "application/x-protobuf;proto=io.prometheus.write.v2.Request", + prom_store::snappy_compress(&prom_v2.encode_to_vec()).unwrap(), + ); + prom_v2_case.headers.push(("content-encoding", "snappy")); + cases.push(prom_v2_case); + check_http_skip_wal("prometheus", &cases, distributed).await; +} + pub async fn test_prometheus_remote_write(store_type: StorageType) { common_telemetry::init_default_ut_logging(); let (app, mut guard) = @@ -3224,6 +3310,94 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag guard.remove_all().await; } +#[apply(both_deployment_cases)] +async fn test_prometheus_remote_write_batched_skip_wal(distributed: bool) { + check_prometheus_remote_write_batched_skip_wal(distributed, false).await; +} + +#[apply(both_deployment_cases)] +async fn test_prometheus_remote_write_v2_batched_skip_wal(distributed: bool) { + check_prometheus_remote_write_batched_skip_wal(distributed, true).await; +} + +async fn check_prometheus_remote_write_batched_skip_wal(distributed: bool, v2: bool) { + common_telemetry::init_default_ut_logging(); + let mut instance = + MockInstanceImpl::new(&format!("prom_bulk_skip_wal_v2_{v2}"), distributed).await; + let server = build_test_prom_server(instance.frontend(), true, false).build(); + let client = TestClient::new(server.build(server.make_app()).unwrap()).await; + + write_prometheus_skip_wal_sample(&client, v2, 1000, None).await; + wait_for_data(&client, "SELECT count(*) FROM wal_prom_bulk", "[[1]]").await; + let mut before = instance.flush_and_snapshot_wal().await; + // Reuse the same batcher across policy transitions to detect leaked policy. + for (index, hint) in [None, Some("true"), Some("false"), None] + .into_iter() + .enumerate() + { + write_prometheus_skip_wal_sample(&client, v2, (index as i64 + 2) * 1000, hint).await; + wait_for_data( + &client, + "SELECT count(*) FROM wal_prom_bulk", + &format!("[[{}]]", index + 2), + ) + .await; + let after = instance.flush_and_snapshot_wal().await; + assert_wal_delta(&before, &after, hint == Some("true")); + before = after; + } + instance.shutdown().await; +} + +async fn write_prometheus_skip_wal_sample( + client: &TestClient, + v2: bool, + timestamp: i64, + hint: Option<&str>, +) { + let payload = if v2 { + remote_write_v2::request_with_labels_and_samples( + vec![(prom_store::METRIC_NAME_LABEL, "wal_prom_bulk")], + vec![RemoteWriteV2Sample { + value: 1.0, + timestamp, + start_timestamp: 0, + }], + ) + .encode_to_vec() + } else { + WriteRequest { + timeseries: vec![TimeSeries { + labels: vec![Label { + name: prom_store::METRIC_NAME_LABEL.to_string(), + value: "wal_prom_bulk".to_string(), + }], + samples: vec![Sample { + value: 1.0, + timestamp, + }], + ..Default::default() + }], + ..Default::default() + } + .encode_to_vec() + }; + let mut request = client + .post("/v1/prometheus/write") + .header("Content-Encoding", "snappy") + .body(prom_store::snappy_compress(&payload).unwrap()); + if v2 { + request = request.header( + "Content-Type", + "application/x-protobuf;proto=io.prometheus.write.v2.Request", + ); + } + if let Some(hint) = hint { + request = request.header("x-greptime-insert-skip-wal", hint); + } + assert_eq!(request.send().await.status(), StatusCode::NO_CONTENT); +} + /// Covers the batched (pending-rows-batcher) Prometheus remote write path, which /// bypasses `PromStoreProtocolHandler::write`. Verifies the metric table is created /// asynchronously and still carries the Prometheus semantic identity stamped on the @@ -4135,6 +4309,25 @@ transform: guard.remove_all().await; } +#[apply(both_deployment_cases)] +async fn test_json_logs_skip_wal(distributed: bool) { + let cases = vec![ + HttpWalCase::new( + "JSON logs", + "/v1/ingest?table=wal_logs&pipeline_name=greptime_identity", + "application/json", + r#"[{"message":"wal test"}]"#, + ), + HttpWalCase::new( + "Events logs", + "/v1/events/logs?table=wal_events&pipeline_name=greptime_identity", + "application/json", + r#"[{"message":"wal test"}]"#, + ), + ]; + check_http_skip_wal("json_logs", &cases, distributed).await; +} + pub async fn test_identity_pipeline(store_type: StorageType) { common_telemetry::init_default_ut_logging(); let (app, mut guard) = @@ -5343,6 +5536,25 @@ transform: guard.remove_all().await; } +#[apply(both_deployment_cases)] +async fn test_influxdb_skip_wal(distributed: bool) { + let cases = vec![ + HttpWalCase::new( + "InfluxDB v1", + "/v1/influxdb/write?db=public", + "text/plain", + "wal_influx,host=a value=1 1700000000000000000", + ), + HttpWalCase::new( + "InfluxDB v2", + "/v1/influxdb/api/v2/write?bucket=public", + "text/plain", + "wal_influx_v2,host=a value=1 1700000000000000000", + ), + ]; + check_http_skip_wal("influxdb", &cases, distributed).await; +} + pub async fn test_influxdb_write_with_hints(storage_type: StorageType) { common_telemetry::init_default_ut_logging(); let (app, mut guard) = @@ -6523,6 +6735,21 @@ pub async fn test_pipeline_auto_transform_with_select(store_type: StorageType) { guard.remove_all().await; } +#[apply(both_deployment_cases)] +async fn test_otlp_metrics_skip_wal(distributed: bool) { + let mut cases = Vec::new(); + let metrics: ExportMetricsServiceRequest = serde_json::from_value(json!({ + "resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"wal_otlp", "gauge":{"dataPoints":[{"timeUnixNano":"1700000000000000000","asDouble":1.0}]}}]}]}] + })).unwrap(); + cases.push(HttpWalCase::new( + "OTLP metrics", + "/v1/otlp/v1/metrics", + "application/x-protobuf", + metrics.encode_to_vec(), + )); + check_http_skip_wal("otlp_metrics", &cases, distributed).await; +} + pub async fn test_otlp_metrics_new(store_type: StorageType) { // init common_telemetry::init_default_ut_logging(); @@ -7318,6 +7545,42 @@ pub async fn test_otlp_metrics_resource_info_conflicts(store_type: StorageType) guard.remove_all().await; } +#[apply(both_deployment_cases)] +async fn test_otlp_traces_skip_wal(distributed: bool) { + let mut cases = Vec::new(); + let traces = make_trace_v1_request( + "wal-test", + vec![json!({ + "traceId":"0102030405060708090a0b0c0d0e0f10", + "spanId":"0102030405060708", "name":"wal test", + "startTimeUnixNano":"1700000000000000000", + "endTimeUnixNano":"1700000001000000000" + })], + ); + let mut traces_case = HttpWalCase::new( + "OTLP traces", + "/v1/otlp/v1/traces", + "application/x-protobuf", + traces.encode_to_vec(), + ); + traces_case + .headers + .push(("x-greptime-pipeline-name", "greptime_trace_v0")); + cases.push(traces_case); + let mut traces_v1 = HttpWalCase::new( + "OTLP traces v1", + "/v1/otlp/v1/traces", + "application/x-protobuf", + traces.encode_to_vec(), + ); + traces_v1.headers = vec![ + ("x-greptime-pipeline-name", "greptime_trace_v1"), + ("x-greptime-trace-table-name", "wal_traces_v1"), + ]; + cases.push(traces_v1); + check_http_skip_wal("otlp_traces", &cases, distributed).await; +} + pub async fn test_otlp_traces_v0(store_type: StorageType) { // init common_telemetry::init_default_ut_logging(); @@ -8388,6 +8651,22 @@ pub async fn test_otlp_traces_v1(store_type: StorageType) { guard.remove_all().await; } +#[apply(both_deployment_cases)] +async fn test_otlp_logs_skip_wal(distributed: bool) { + let mut cases = Vec::new(); + let logs = make_log_request(vec![json!({ + "timeUnixNano": "1700000000000000000", + "body": {"stringValue": "wal test"} + })]); + cases.push(HttpWalCase::new( + "OTLP logs", + "/v1/otlp/v1/logs", + "application/x-protobuf", + logs.encode_to_vec(), + )); + check_http_skip_wal("otlp_logs", &cases, distributed).await; +} + pub async fn test_otlp_logs(store_type: StorageType) { common_telemetry::init_default_ut_logging(); let (app, mut guard) = setup_test_http_app_with_frontend(store_type, "test_otlp_logs").await; @@ -8678,6 +8957,37 @@ pub async fn test_otlp_logs(store_type: StorageType) { guard.remove_all().await; } +#[apply(both_deployment_cases)] +async fn test_loki_skip_wal(distributed: bool) { + let mut cases = vec![HttpWalCase::new( + "Loki JSON", + "/v1/loki/api/v1/push", + "application/json", + r#"{"streams":[{"stream":{"host":"a"},"values":[["1700000000000000000","wal test"]]}]}"#, + )]; + let loki = loki_proto::logproto::PushRequest { + streams: vec![loki_proto::logproto::StreamAdapter { + labels: r#"{host="a"}"#.to_string(), + entries: vec![loki_proto::logproto::EntryAdapter { + timestamp: Some(loki_proto::prost_types::Timestamp { + seconds: 1700000000, + nanos: 0, + }), + line: "wal test".to_string(), + ..Default::default() + }], + ..Default::default() + }], + }; + cases.push(HttpWalCase::new( + "Loki protobuf", + "/v1/loki/api/v1/push", + "application/x-protobuf", + prom_store::snappy_compress(&loki.encode_to_vec()).unwrap(), + )); + check_http_skip_wal("loki", &cases, distributed).await; +} + pub async fn test_loki_pb_logs(store_type: StorageType) { common_telemetry::init_default_ut_logging(); let (app, mut guard) = setup_test_http_app_with_frontend(store_type, "test_loki_pb_logs").await; @@ -9086,6 +9396,17 @@ processors: guard.remove_all().await; } +#[apply(both_deployment_cases)] +async fn test_elasticsearch_skip_wal(distributed: bool) { + let cases = vec![HttpWalCase::new( + "Elasticsearch", + "/v1/elasticsearch/_bulk", + "application/json", + "{\"create\":{\"_index\":\"wal_elastic\"}}\n{\"message\":\"wal test\"}\n", + )]; + check_http_skip_wal("elasticsearch", &cases, distributed).await; +} + pub async fn test_elasticsearch_logs(store_type: StorageType) { common_telemetry::init_default_ut_logging(); let (app, mut guard) = @@ -10789,6 +11110,17 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { guard.remove_all().await; } +#[apply(both_deployment_cases)] +async fn test_opentsdb_skip_wal(distributed: bool) { + let cases = vec![HttpWalCase::new( + "OpenTSDB", + "/v1/opentsdb/api/put", + "application/json", + r#"{"metric":"wal_tsdb","timestamp":1700000000,"value":1,"tags":{"host":"a"}}"#, + )]; + check_http_skip_wal("opentsdb", &cases, distributed).await; +} + pub async fn test_influxdb_write(store_type: StorageType) { common_telemetry::init_default_ut_logging(); let (app, mut guard) = @@ -10897,6 +11229,123 @@ async fn execute_sql(client: &TestClient, sql: &str) -> TestResponse { .await } +struct HttpWalCase { + name: &'static str, + get: bool, + create_table: Option<&'static str>, + expected_written_nodes: Option, + path: &'static str, + content_type: &'static str, + body: Vec, + headers: Vec<(&'static str, &'static str)>, +} + +impl HttpWalCase { + fn new( + name: &'static str, + path: &'static str, + content_type: &'static str, + body: impl Into>, + ) -> Self { + Self { + name, + get: false, + create_table: None, + expected_written_nodes: None, + path, + content_type, + body: body.into(), + headers: Vec::new(), + } + } +} + +async fn check_http_skip_wal(name: &str, cases: &[HttpWalCase], distributed: bool) { + common_telemetry::init_default_ut_logging(); + let mut instance = MockInstanceImpl::new(&format!("http_skip_wal_{name}"), distributed).await; + let fe = instance.frontend(); + let server = HttpServerBuilder::new(HttpOptions::default()) + .with_sql_handler(fe.clone()) + .with_influxdb_handler(fe.clone()) + .with_opentsdb_handler(fe.clone()) + .with_log_ingest_handler(fe.clone(), None, None) + .with_otlp_handler(fe.clone(), true, false) + // The pending batcher uses BulkInsert, deliberately outside this PR. + .with_prom_handler( + fe.clone(), + Some(fe), + true, + PromValidationMode::Strict, + false, + None, + ) + .build(); + let client = TestClient::new(server.build(server.make_app()).unwrap()).await; + for case in cases { + if let Some(create_table) = case.create_table { + let response = execute_sql(&client, create_table).await; + assert!(response.status().is_success(), "{}", response.text().await); + } + // Warm up schema-on-write, then change only the request hint. Reusing + // the same payload also verifies policy is not persisted on the table. + for (round, hint) in [None, Some("true"), Some("false"), None] + .into_iter() + .enumerate() + { + common_telemetry::info!("Protocol WAL case: {}, hint: {:?}", case.name, hint); + let before = instance.flush_and_snapshot_wal().await; + let mut headers = vec![( + HeaderName::from_static("content-type"), + HeaderValue::from_static(case.content_type), + )]; + headers.extend(case.headers.iter().map(|(key, value)| { + ( + HeaderName::from_static(key), + HeaderValue::from_static(value), + ) + })); + if let Some(value) = hint { + headers.push(( + HeaderName::from_static("x-greptime-insert-skip-wal"), + HeaderValue::from_static(value), + )); + } + let response = if case.get { + let mut request = client.get(case.path).body(case.body.clone()); + for (key, value) in headers { + request = request.header(key, value); + } + request.send().await + } else { + send_req(&client, headers, case.path, case.body.clone(), false).await + }; + let status = response.status(); + let body = response.text().await; + assert!(status.is_success(), "{}: {status}: {body}", case.name); + let after = instance.flush_and_snapshot_wal().await; + if before.keys().eq(after.keys()) { + assert_wal_delta(&before, &after, hint == Some("true")); + if let Some(expected_written_nodes) = case.expected_written_nodes { + let written_nodes = after + .iter() + .filter(|(key, (flushed_sequence, _))| *flushed_sequence > before[*key].0) + .map(|((node_id, _), _)| *node_id) + .collect::>(); + assert_eq!( + written_nodes.len(), + expected_written_nodes, + "{}: unexpected number of written datanodes", + case.name + ); + } + } else { + assert!(round == 0, "only warmup may create regions: {}", case.name); + } + } + } + instance.shutdown().await; +} + async fn send_req( client: &TestClient, headers: Vec<(HeaderName, HeaderValue)>, diff --git a/tests-integration/tests/main.rs b/tests-integration/tests/main.rs index 56108d4d3ff..28a81155724 100644 --- a/tests-integration/tests/main.rs +++ b/tests-integration/tests/main.rs @@ -41,6 +41,13 @@ mod reconciliation_event; mod view_ddl_event; mod wal_prune_event; +#[rstest_reuse::template] +#[rstest::rstest] +#[case::standalone(false)] +#[case::distributed(true)] +#[tokio::test(flavor = "multi_thread")] +fn both_deployment_cases(#[case] distributed: bool) {} + grpc_tests!(File, S3, S3WithCache, Oss, Azblob, Gcs); http_tests!(File, S3, S3WithCache, Oss, Azblob, Gcs); diff --git a/tests/cases/standalone/common/session_skip_wal.result b/tests/cases/standalone/common/session_skip_wal.result new file mode 100644 index 00000000000..4343e3cd334 --- /dev/null +++ b/tests/cases/standalone/common/session_skip_wal.result @@ -0,0 +1,224 @@ +CREATE TABLE session_skip_wal(host STRING, ts TIMESTAMP TIME INDEX); + +Affected Rows: 0 + +-- MYSQL: SET persists across statements on the same connection. +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = true; + +affected_rows: 0 + +-- Invalid SET must leave the enabled policy unchanged. +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = 'invalid'; + +Failed to execute query, err: MySqlError { ERROR 1235 (42000): (Unsupported): Not supported: Invalid skip_wal value "invalid": expected true or false } + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = 1; + +Failed to execute query, err: MySqlError { ERROR 1235 (42000): (Unsupported): Not supported: SET skip_wal requires true or false } + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = NULL; + +Failed to execute query, err: MySqlError { ERROR 1235 (42000): (Unsupported): Not supported: SET skip_wal requires true or false } + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = false, true; + +Failed to execute query, err: MySqlError { ERROR 1235 (42000): (Unsupported): Not supported: SET skip_wal requires exactly one boolean value } + +-- SQLNESS PROTOCOL MYSQL +INSERT INTO session_skip_wal VALUES ('skipped', 1000); + +affected_rows: 1 + +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM session_skip_wal ORDER BY ts; + ++---------+---------------------+ +| host | ts | ++---------+---------------------+ +| skipped | 1970-01-01 00:00:01 | ++---------+---------------------+ + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM session_skip_wal ORDER BY ts; + +affected_rows: 0 + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = false; + +affected_rows: 0 + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = 'true'; + +affected_rows: 0 + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = 'false'; + +affected_rows: 0 + +-- SQLNESS PROTOCOL MYSQL +INSERT INTO session_skip_wal VALUES ('persisted', 2000); + +affected_rows: 1 + +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM session_skip_wal ORDER BY ts; + ++-----------+---------------------+ +| host | ts | ++-----------+---------------------+ +| persisted | 1970-01-01 00:00:02 | ++-----------+---------------------+ + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM session_skip_wal ORDER BY ts; + ++-----------+---------------------+ +| host | ts | ++-----------+---------------------+ +| persisted | 1970-01-01 00:00:02 | ++-----------+---------------------+ + +TRUNCATE TABLE session_skip_wal; + +Affected Rows: 0 + +-- POSTGRES: SET persists across statements on the same connection. +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = true; + +Affected Rows: 0 + +-- Invalid SET must leave the enabled policy unchanged. +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = 'invalid'; + +Failed to execute query, encountered: Error { kind: Db, cause: Some(DbError { severity: "ERROR", parsed_severity: None, code: SqlState(E0A000), message: "Not supported: Invalid skip_wal value \"invalid\": expected true or false", detail: None, hint: None, position: None, where_: None, schema: None, table: None, column: None, datatype: None, constraint: None, file: None, line: None, routine: None }) } + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = 1; + +Failed to execute query, encountered: Error { kind: Db, cause: Some(DbError { severity: "ERROR", parsed_severity: None, code: SqlState(E0A000), message: "Not supported: SET skip_wal requires true or false", detail: None, hint: None, position: None, where_: None, schema: None, table: None, column: None, datatype: None, constraint: None, file: None, line: None, routine: None }) } + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = NULL; + +Failed to execute query, encountered: Error { kind: Db, cause: Some(DbError { severity: "ERROR", parsed_severity: None, code: SqlState(E0A000), message: "Not supported: SET skip_wal requires true or false", detail: None, hint: None, position: None, where_: None, schema: None, table: None, column: None, datatype: None, constraint: None, file: None, line: None, routine: None }) } + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = false, true; + +Failed to execute query, encountered: Error { kind: Db, cause: Some(DbError { severity: "ERROR", parsed_severity: None, code: SqlState(E0A000), message: "Not supported: SET skip_wal requires exactly one boolean value", detail: None, hint: None, position: None, where_: None, schema: None, table: None, column: None, datatype: None, constraint: None, file: None, line: None, routine: None }) } + +-- SQLNESS PROTOCOL POSTGRES +INSERT INTO session_skip_wal VALUES ('skipped', 1000); + +Affected Rows: 1 + +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM session_skip_wal ORDER BY ts; + ++---------+----------------------------+ +| host | ts | ++---------+----------------------------+ +| skipped | 1970-01-01 00:00:01.000000 | ++---------+----------------------------+ + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM session_skip_wal ORDER BY ts; + ++------+----+ +| host | ts | ++------+----+ ++------+----+ + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = false; + +Affected Rows: 0 + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = 'true'; + +Affected Rows: 0 + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = 'false'; + +Affected Rows: 0 + +-- SQLNESS PROTOCOL POSTGRES +INSERT INTO session_skip_wal VALUES ('persisted', 2000); + +Affected Rows: 1 + +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM session_skip_wal ORDER BY ts; + ++-----------+----------------------------+ +| host | ts | ++-----------+----------------------------+ +| persisted | 1970-01-01 00:00:02.000000 | ++-----------+----------------------------+ + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM session_skip_wal ORDER BY ts; + ++-----------+----------------------------+ +| host | ts | ++-----------+----------------------------+ +| persisted | 1970-01-01 00:00:02.000000 | ++-----------+----------------------------+ + +TRUNCATE TABLE session_skip_wal; + +Affected Rows: 0 + +DROP TABLE session_skip_wal; + +Affected Rows: 0 + +-- The default gRPC protocol has request-scoped contexts, not a persistent SQL session. +SET skip_wal = true; + +Affected Rows: 0 + +SET skip_wal = false; + +Affected Rows: 0 + +SET skip_wal = 'true'; + +Affected Rows: 0 + +SET skip_wal = 'false'; + +Affected Rows: 0 + +SET skip_wal = 'invalid'; + +Error: 1001(Unsupported), Not supported: Invalid skip_wal value "invalid": expected true or false + +SET skip_wal = 1; + +Error: 1001(Unsupported), Not supported: SET skip_wal requires true or false + +SET skip_wal = NULL; + +Error: 1001(Unsupported), Not supported: SET skip_wal requires true or false + +SET skip_wal = false, true; + +Error: 1001(Unsupported), Not supported: SET skip_wal requires exactly one boolean value + diff --git a/tests/cases/standalone/common/session_skip_wal.sql b/tests/cases/standalone/common/session_skip_wal.sql new file mode 100644 index 00000000000..8be063253bc --- /dev/null +++ b/tests/cases/standalone/common/session_skip_wal.sql @@ -0,0 +1,118 @@ +CREATE TABLE session_skip_wal(host STRING, ts TIMESTAMP TIME INDEX); + +-- MYSQL: SET persists across statements on the same connection. +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = true; + +-- Invalid SET must leave the enabled policy unchanged. + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = 'invalid'; + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = 1; + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = NULL; + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = false, true; + +-- SQLNESS PROTOCOL MYSQL +INSERT INTO session_skip_wal VALUES ('skipped', 1000); + +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM session_skip_wal ORDER BY ts; + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM session_skip_wal ORDER BY ts; + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = false; + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = 'true'; + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = 'false'; + +-- SQLNESS PROTOCOL MYSQL +INSERT INTO session_skip_wal VALUES ('persisted', 2000); + +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM session_skip_wal ORDER BY ts; + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM session_skip_wal ORDER BY ts; + +TRUNCATE TABLE session_skip_wal; + +-- POSTGRES: SET persists across statements on the same connection. +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = true; + +-- Invalid SET must leave the enabled policy unchanged. + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = 'invalid'; + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = 1; + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = NULL; + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = false, true; + +-- SQLNESS PROTOCOL POSTGRES +INSERT INTO session_skip_wal VALUES ('skipped', 1000); + +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM session_skip_wal ORDER BY ts; + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM session_skip_wal ORDER BY ts; + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = false; + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = 'true'; + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = 'false'; + +-- SQLNESS PROTOCOL POSTGRES +INSERT INTO session_skip_wal VALUES ('persisted', 2000); + +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM session_skip_wal ORDER BY ts; + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM session_skip_wal ORDER BY ts; + +TRUNCATE TABLE session_skip_wal; + +DROP TABLE session_skip_wal; + +-- The default gRPC protocol has request-scoped contexts, not a persistent SQL session. +SET skip_wal = true; + +SET skip_wal = false; + +SET skip_wal = 'true'; + +SET skip_wal = 'false'; + +SET skip_wal = 'invalid'; + +SET skip_wal = 1; + +SET skip_wal = NULL; + +SET skip_wal = false, true; diff --git a/tests/cases/standalone/copy/copy_skip_wal.result b/tests/cases/standalone/copy/copy_skip_wal.result new file mode 100644 index 00000000000..9f7a2b71cea --- /dev/null +++ b/tests/cases/standalone/copy/copy_skip_wal.result @@ -0,0 +1,274 @@ +CREATE TABLE copy_skip_wal(host STRING, ts TIMESTAMP TIME INDEX); + +Affected Rows: 0 + +INSERT INTO copy_skip_wal VALUES ('host1', 1000), ('host2', 2000); + +Affected Rows: 2 + +COPY copy_skip_wal TO '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv'); + +Affected Rows: 2 + +TRUNCATE TABLE copy_skip_wal; + +Affected Rows: 0 + +-- MYSQL: COPY TABLE inherits the connection WAL policy. +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = true; + +affected_rows: 0 + +-- SQLNESS PROTOCOL MYSQL +COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv'); + +affected_rows: 2 + +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + ++-------+---------------------+ +| host | ts | ++-------+---------------------+ +| host1 | 1970-01-01 00:00:01 | +| host2 | 1970-01-01 00:00:02 | ++-------+---------------------+ + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + +affected_rows: 0 + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = false; + +affected_rows: 0 + +-- SQLNESS PROTOCOL MYSQL +COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv'); + +affected_rows: 2 + +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + ++-------+---------------------+ +| host | ts | ++-------+---------------------+ +| host1 | 1970-01-01 00:00:01 | +| host2 | 1970-01-01 00:00:02 | ++-------+---------------------+ + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + ++-------+---------------------+ +| host | ts | ++-------+---------------------+ +| host1 | 1970-01-01 00:00:01 | +| host2 | 1970-01-01 00:00:02 | ++-------+---------------------+ + +TRUNCATE TABLE copy_skip_wal; + +Affected Rows: 0 + +-- MYSQL: COPY DATABASE inherits the connection WAL policy. +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = true; + +affected_rows: 0 + +-- SQLNESS PROTOCOL MYSQL +COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv'); + +affected_rows: 2 + +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + ++-------+---------------------+ +| host | ts | ++-------+---------------------+ +| host1 | 1970-01-01 00:00:01 | +| host2 | 1970-01-01 00:00:02 | ++-------+---------------------+ + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + +affected_rows: 0 + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = false; + +affected_rows: 0 + +-- SQLNESS PROTOCOL MYSQL +COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv'); + +affected_rows: 2 + +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + ++-------+---------------------+ +| host | ts | ++-------+---------------------+ +| host1 | 1970-01-01 00:00:01 | +| host2 | 1970-01-01 00:00:02 | ++-------+---------------------+ + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + ++-------+---------------------+ +| host | ts | ++-------+---------------------+ +| host1 | 1970-01-01 00:00:01 | +| host2 | 1970-01-01 00:00:02 | ++-------+---------------------+ + +TRUNCATE TABLE copy_skip_wal; + +Affected Rows: 0 + +-- POSTGRES: COPY TABLE inherits the connection WAL policy. +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = true; + +Affected Rows: 0 + +-- SQLNESS PROTOCOL POSTGRES +COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv'); + +Affected Rows: 2 + +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + ++-------+----------------------------+ +| host | ts | ++-------+----------------------------+ +| host1 | 1970-01-01 00:00:01.000000 | +| host2 | 1970-01-01 00:00:02.000000 | ++-------+----------------------------+ + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + ++------+----+ +| host | ts | ++------+----+ ++------+----+ + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = false; + +Affected Rows: 0 + +-- SQLNESS PROTOCOL POSTGRES +COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv'); + +Affected Rows: 2 + +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + ++-------+----------------------------+ +| host | ts | ++-------+----------------------------+ +| host1 | 1970-01-01 00:00:01.000000 | +| host2 | 1970-01-01 00:00:02.000000 | ++-------+----------------------------+ + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + ++-------+----------------------------+ +| host | ts | ++-------+----------------------------+ +| host1 | 1970-01-01 00:00:01.000000 | +| host2 | 1970-01-01 00:00:02.000000 | ++-------+----------------------------+ + +TRUNCATE TABLE copy_skip_wal; + +Affected Rows: 0 + +-- POSTGRES: COPY DATABASE inherits the connection WAL policy. +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = true; + +Affected Rows: 0 + +-- SQLNESS PROTOCOL POSTGRES +COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv'); + +Affected Rows: 2 + +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + ++-------+----------------------------+ +| host | ts | ++-------+----------------------------+ +| host1 | 1970-01-01 00:00:01.000000 | +| host2 | 1970-01-01 00:00:02.000000 | ++-------+----------------------------+ + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + ++------+----+ +| host | ts | ++------+----+ ++------+----+ + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = false; + +Affected Rows: 0 + +-- SQLNESS PROTOCOL POSTGRES +COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv'); + +Affected Rows: 2 + +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + ++-------+----------------------------+ +| host | ts | ++-------+----------------------------+ +| host1 | 1970-01-01 00:00:01.000000 | +| host2 | 1970-01-01 00:00:02.000000 | ++-------+----------------------------+ + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + ++-------+----------------------------+ +| host | ts | ++-------+----------------------------+ +| host1 | 1970-01-01 00:00:01.000000 | +| host2 | 1970-01-01 00:00:02.000000 | ++-------+----------------------------+ + +TRUNCATE TABLE copy_skip_wal; + +Affected Rows: 0 + +DROP TABLE copy_skip_wal; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/copy/copy_skip_wal.sql b/tests/cases/standalone/copy/copy_skip_wal.sql new file mode 100644 index 00000000000..7ead655de58 --- /dev/null +++ b/tests/cases/standalone/copy/copy_skip_wal.sql @@ -0,0 +1,125 @@ +CREATE TABLE copy_skip_wal(host STRING, ts TIMESTAMP TIME INDEX); + +INSERT INTO copy_skip_wal VALUES ('host1', 1000), ('host2', 2000); + +COPY copy_skip_wal TO '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv'); + +TRUNCATE TABLE copy_skip_wal; + +-- MYSQL: COPY TABLE inherits the connection WAL policy. +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = true; + +-- SQLNESS PROTOCOL MYSQL +COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv'); + +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = false; + +-- SQLNESS PROTOCOL MYSQL +COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv'); + +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + +TRUNCATE TABLE copy_skip_wal; + +-- MYSQL: COPY DATABASE inherits the connection WAL policy. +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = true; + +-- SQLNESS PROTOCOL MYSQL +COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv'); + +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + +-- SQLNESS PROTOCOL MYSQL +SET skip_wal = false; + +-- SQLNESS PROTOCOL MYSQL +COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv'); + +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL MYSQL +SELECT * FROM copy_skip_wal ORDER BY ts; + +TRUNCATE TABLE copy_skip_wal; + +-- POSTGRES: COPY TABLE inherits the connection WAL policy. +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = true; + +-- SQLNESS PROTOCOL POSTGRES +COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv'); + +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = false; + +-- SQLNESS PROTOCOL POSTGRES +COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv'); + +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + +TRUNCATE TABLE copy_skip_wal; + +-- POSTGRES: COPY DATABASE inherits the connection WAL policy. +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = true; + +-- SQLNESS PROTOCOL POSTGRES +COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv'); + +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + +-- SQLNESS PROTOCOL POSTGRES +SET skip_wal = false; + +-- SQLNESS PROTOCOL POSTGRES +COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv'); + +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + +-- SQLNESS ARG restart=true +-- SQLNESS PROTOCOL POSTGRES +SELECT * FROM copy_skip_wal ORDER BY ts; + +TRUNCATE TABLE copy_skip_wal; + +DROP TABLE copy_skip_wal;