rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-16 08:52:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: backport bulk skip-WAL support and integration coverage (#9627)

Browse Source 
* test: backport request-level skip-WAL integration coverage

Backport #9093 as the shared protocol and storage test harness prerequisite for bulk request WAL skipping.

Signed-off-by: WenyXu <wenymedia@gmail.com>

(cherry picked from commit a673e084b2)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* feat: backport request-level bulk WAL skipping

Backport #9110 with its protocol update and Flight/Prometheus coverage. Keep the release branch's write dispatcher and partition-column retention, including the skip-WAL flag when retry routing rebuilds bulk requests.

Signed-off-by: WenyXu <wenymedia@gmail.com>

(cherry picked from commit 737025760e)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

---------

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
Co-authored-by: Weny Xu <wenymedia@gmail.com>
This commit is contained in:
Lei, HUANG
2026-10-16 11:12:07 +08:00
committed by GitHub
co-authored by Weny Xu
parent bd01d2a138
commit 12b7ebe7d5
33 changed files with 2124 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1 -1
View File
@@ -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",
Cargo.toml
+1 -1
View File
@@ -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"
src/client/src/database.rs
+10
View File
@@ -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<DoPutResponseStream> {
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<DoPutResponseStream> {
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 })),
src/frontend/src/instance/grpc.rs
+1
View File
@@ -453,6 +453,7 @@ impl Instance {
request.flight_data,
request.record_batch,
request.schema_bytes,
ctx.skip_wal(),
)
.await
.context(TableOperationSnafu)?;
src/metric-engine/src/engine/bulk_insert.rs
+79 -18
View File
@@ -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::<usize>(), 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::<usize>(),
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)
src/mito2/src/engine/alter_test.rs
+1
View File
@@ -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,
src/mito2/src/engine/edit_region_test.rs
+1
View File
@@ -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 {
src/mito2/src/engine/scan_test.rs
+134
View File
@@ -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::<usize>(),
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::<usize>(),
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]
src/mito2/src/engine/staging_test.rs
+1
View File
@@ -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 {
src/mito2/src/flush.rs
+1
View File
@@ -1728,6 +1728,7 @@ mod tests {
(
SenderBulkRequest {
skip_wal: false,
sender: OptionOutputTx::from(sender),
region_id,
request: converter.convert().unwrap(),
src/mito2/src/region/opener.rs
+2 -1
View File
@@ -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,
src/mito2/src/region_write_ctx.rs
+73 -15
View File
@@ -318,24 +318,26 @@ impl RegionWriteCtx {
sender: OptionOutputTx,
mut bulk: BulkPart,
sequence: Option<SequenceNumber>,
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<_>>(),
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::<Vec<_>>(),
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(),
src/mito2/src/request.rs
+1
View File
@@ -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,
src/mito2/src/worker/handle_bulk_insert.rs
+1
View File
@@ -96,6 +96,7 @@ impl<S: LogStore> RegionWorkerLoop<S> {
None
};
pending_bulk_request.push(SenderBulkRequest {
skip_wal: request.skip_wal,
sender,
request: part,
region_id: request.region_id,
src/mito2/src/worker/handle_write.rs
+27 -6
View File
@@ -565,7 +565,7 @@ impl<S> RegionWorkerLoop<S> {
}
// 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(&region_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()));
src/operator/src/bulk_insert.rs
+3
View File
@@ -48,6 +48,7 @@ impl Inserter {
raw_flight_data: FlightData,
record_batch: RecordBatch,
schema_bytes: Bytes,
skip_wal: bool,
) -> error::Result<AffectedRows> {
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 }),
src/operator/src/req_convert/common/partitioner.rs
+20 -5
View File
@@ -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<RegionNumber, RegionMask>,
mut original_ipc: Option<ArrowIpc>,
skip_wal: bool,
) -> Result<Vec<BulkInsertRequest>> {
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<u64>, values: Vec<i32>) -> RegionRequest {
fn build_bulk_request(
region: u32,
version: Option<u64>,
values: Vec<i32>,
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 {
src/operator/src/write_dispatcher/tests.rs
+2
View File
@@ -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,
src/servers/src/pending_rows_batcher.rs
+48 -10
View File
@@ -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<PlannedRegionBatch>,
partition_versions: &HashMap<u32, Option<u64>>,
skip_wal: bool,
) -> Result<Vec<RegionRequest>> {
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(&region_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::<WorkerCommand>(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::<WorkerCommand>(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
src/store-api/src/region_request.rs
+4
View File
@@ -472,6 +472,7 @@ fn make_region_truncate(truncate: TruncateRequest) -> Result<Vec<(RegionId, Regi
/// Convert [BulkInsertRequest] to [RegionRequest] and group by [RegionId].
fn make_region_bulk_inserts(request: BulkInsertRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
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<Vec<(RegionId,
region_id,
payload,
raw_data: request,
skip_wal,
partition_expr_version,
aligned_schema_version,
}),
@@ -1755,6 +1757,8 @@ pub struct RegionCatchupRequest {
#[derive(Debug, Clone)]
pub struct RegionBulkInsertsRequest {
/// Whether this request should skip WAL.
pub skip_wal: bool,
pub region_id: RegionId,
pub payload: DfRecordBatch,
pub raw_data: ArrowIpc,
tests-integration/Cargo.toml
+1
View File
@@ -106,6 +106,7 @@ hex.workspace = true
http.workspace = true
itertools.workspace = true
jsonb.workspace = true
mito2 = { workspace = true, features = ["test"] }
mysql_async = { version = "0.37", default-features = false, features = [
"time",
"default-rustls-ring",
tests-integration/src/grpc/flight.rs
+74 -2
View File
@@ -49,6 +49,7 @@ mod test {
use futures_util::{Stream, StreamExt};
use hyper_util::rt::TokioIo;
use itertools::Itertools;
use rstest::rstest;
use servers::grpc::builder::GrpcServerBuilder;
use servers::grpc::flight::{
FlightCraft, FlightCraftWrapper, FlightRecordBatchSource, FlightRecordBatchStream,
@@ -59,6 +60,7 @@ mod test {
use servers::query_handler::grpc::GrpcQueryHandler;
use servers::server::Server;
use session::context::QueryContextRef;
use session::hints::INSERT_SKIP_WAL_HINT;
use tokio::net::TcpListener;
use tokio_stream::wrappers::TcpListenerStream;
use tonic::transport::Server as TonicServer;
@@ -67,7 +69,7 @@ mod test {
use crate::cluster::GreptimeDbClusterBuilder;
use crate::grpc::query_and_expect;
use crate::test_util::{StorageType, setup_grpc_server};
use crate::test_util::{MockInstanceImpl, StorageType, assert_wal_delta, setup_grpc_server};
use crate::tests::test_util::MockInstance;
struct SlowFlightCraft;
@@ -399,6 +401,68 @@ mod test {
assert!(start.elapsed() >= 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<RecordBatch>) {
test_put_record_batches_with_hints(client, record_batches, &[]).await;
}
async fn test_put_record_batches_with_hints(
client: &Database,
record_batches: Vec<RecordBatch>,
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::<Vec<_>>().await;
let responses_count = responses.len();
tests-integration/src/standalone.rs
+3
View File
@@ -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<Frontend>,
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,
tests-integration/src/test_util.rs
+200 -7
View File
@@ -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::<Vec<_>>(),
after.keys().collect::<Vec<_>>(),
"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<Metasrv> {
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<DatanodeId, Datanode> {
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<Instance> {
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<Instance>,
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(
tests-integration/src/tests/instance_test.rs
+49
View File
@@ -1174,6 +1174,55 @@ async fn test_execute_query_external_table_csv(instance: Arc<dyn MockInstance>)
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<dyn MockInstance>) {
let instance = instance.frontend();
tests-integration/src/tests/test_util.rs
+2 -25
View File
@@ -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<Metasrv> {
match self {
MockInstanceImpl::Standalone(_) => unreachable!(),
MockInstanceImpl::Distributed(instance) => &instance.metasrv,
}
}
pub(crate) fn datanodes(&self) -> &HashMap<DatanodeId, Datanode> {
match self {
MockInstanceImpl::Standalone(_) => unreachable!(),
MockInstanceImpl::Distributed(instance) => &instance.datanode_instances,
}
}
}
impl MockInstance for MockInstanceImpl {
fn frontend(&self) -> Arc<Instance> {
match self {
MockInstanceImpl::Standalone(instance) => instance.frontend(),
MockInstanceImpl::Distributed(instance) => instance.fe_instance().clone(),
}
MockInstanceImpl::frontend(self)
}
fn is_distributed_mode(&self) -> bool {
tests-integration/tests/grpc.rs
+184 -5
View File
@@ -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<T>(body: T, hint: Option<&str>) -> Request<T> {
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;
tests-integration/tests/http.rs
+452 -3
View File
@@ -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<usize>,
path: &'static str,
content_type: &'static str,
body: Vec<u8>,
headers: Vec<(&'static str, &'static str)>,
}
impl HttpWalCase {
fn new(
name: &'static str,
path: &'static str,
content_type: &'static str,
body: impl Into<Vec<u8>>,
) -> 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::<std::collections::BTreeSet<_>>();
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)>,
tests-integration/tests/main.rs
+7
View File
@@ -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);
tests/cases/standalone/common/session_skip_wal.result
+224
View File
@@ -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
tests/cases/standalone/common/session_skip_wal.sql
+118
View File
@@ -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;
tests/cases/standalone/copy/copy_skip_wal.result
+274
View File
@@ -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
tests/cases/standalone/copy/copy_skip_wal.sql
+125
View File
@@ -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;