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

test: cover request-level insert WAL skipping end to end (#9093)

Browse Source 
* test: cover request-level WAL skipping end to end

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

* test: cover session WAL policy and COPY recovery in sqlness

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

* fix(test): isolate Mito test feature in dev dependencies

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

* test: parameterize WAL protocol cases and make setup explicit

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

* test: cover skip-WAL hints across streaming messages

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

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
Weny Xu
2026-09-11 07:48:57 +00:00
committed by GitHub
parent 9619085310
commit a673e084b2
12 changed files with 1531 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
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/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
+180
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 `(written_bytes, 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),
(
statistic.written_bytes,
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, &(written_bytes, flushed_entry_id)) in after {
let (previous_written_bytes, previous_flushed_entry_id) = before[id];
if written_bytes > previous_written_bytes {
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)]
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
+364 -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, 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) =
@@ -4135,6 +4221,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 +5448,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 +6647,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 +7457,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 +8563,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 +8869,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 +9308,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 +11022,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 +11141,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, (written_bytes, _))| *written_bytes > 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;