diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml
index 5e2a63015b..a2d6c70e39 100644
--- a/tests-integration/Cargo.toml
+++ b/tests-integration/Cargo.toml
@@ -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",
diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs
index 0704b184e8..18e1d14402 100644
--- a/tests-integration/src/standalone.rs
+++ b/tests-integration/src/standalone.rs
@@ -62,6 +62,8 @@ use standalone::{StandaloneDatanodeManager, StandaloneRepartitionProcedureFactor
 use crate::test_util::{self, StorageType, TestGuard, create_tmp_dir_and_datanode_opts};
 
 pub struct GreptimeDbStandalone {
+    /// Storage engine for assertions across protocol and storage boundaries.
+    pub mito_engine: mito2::engine::MitoEngine,
     pub frontend: Arc<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,
diff --git a/tests-integration/src/test_util.rs b/tests-integration/src/test_util.rs
index 9d12848ce6..4d44935477 100644
--- a/tests-integration/src/test_util.rs
+++ b/tests-integration/src/test_util.rs
@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::BTreeMap;
+#[cfg(test)]
+use std::collections::HashMap;
 use std::env;
 use std::fmt::Display;
 use std::net::SocketAddr;
@@ -26,7 +29,10 @@ use axum::Router;
 use catalog::kvbackend::KvBackendCatalogManager;
 use client::{Client, Database};
 use common_base::Plugins;
+use common_catalog::consts::MIN_USER_TABLE_ID;
 use common_config::Configurable;
+#[cfg(test)]
+use common_meta::DatanodeId;
 use common_meta::key::TableMetadataManager;
 use common_meta::key::catalog_name::CatalogNameKey;
 use common_meta::key::schema_name::SchemaNameKey;
@@ -38,8 +44,13 @@ use common_test_util::ports;
 use common_test_util::temp_dir::{TempDir, create_temp_dir};
 use common_wal::config::DatanodeWalConfig;
 use datanode::config::{DatanodeOptions, StorageConfig};
+#[cfg(test)]
+use datanode::datanode::Datanode;
 use frontend::instance::Instance;
 use frontend::service_config::{MysqlOptions, PostgresOptions};
+#[cfg(test)]
+use meta_srv::metasrv::Metasrv;
+use mito2::engine::MitoEngine;
 use mito2::gc::GcConfig;
 use object_store::config::{
     AzblobConfig, FileConfig, GcsConfig, ObjectStoreConfig, OssConfig, S3Config,
@@ -62,9 +73,178 @@ use servers::request_memory_limiter::ServerMemoryLimiter;
 use servers::server::Server;
 use servers::tls::ReloadableTlsServerConfig;
 use session::context::QueryContext;
+use store_api::metric_engine_consts::METRIC_METADATA_REGION_GROUP;
+use store_api::region_engine::RegionEngine;
+use store_api::region_request::{RegionFlushRequest, RegionRequest};
+use store_api::storage::RegionId;
 
+use crate::cluster::{GreptimeDbCluster, GreptimeDbClusterBuilder};
 use crate::standalone::{GreptimeDbStandalone, GreptimeDbStandaloneBuilder};
 
+/// Maps `(node_id, region_id)` to `(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)]
diff --git a/tests-integration/src/tests/instance_test.rs b/tests-integration/src/tests/instance_test.rs
index a716284c75..3cd625ba20 100644
--- a/tests-integration/src/tests/instance_test.rs
+++ b/tests-integration/src/tests/instance_test.rs
@@ -1174,6 +1174,55 @@ async fn test_execute_query_external_table_csv(instance: Arc<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();
diff --git a/tests-integration/src/tests/test_util.rs b/tests-integration/src/tests/test_util.rs
index ab86036750..5ddcd0748a 100644
--- a/tests-integration/src/tests/test_util.rs
+++ b/tests-integration/src/tests/test_util.rs
@@ -42,6 +42,7 @@ use session::context::{QueryContext, QueryContextRef};
 
 use crate::cluster::{GreptimeDbCluster, GreptimeDbClusterBuilder};
 use crate::standalone::{GreptimeDbStandalone, GreptimeDbStandaloneBuilder};
+pub(crate) use crate::test_util::MockInstanceImpl;
 use crate::test_util::StorageType;
 use crate::tests::{MockDistributedInstance, create_distributed_instance};
 
@@ -85,33 +86,9 @@ pub(crate) enum MockInstanceBuilder {
     Distributed(GreptimeDbClusterBuilder),
 }
 
-pub(crate) enum MockInstanceImpl {
-    Standalone(GreptimeDbStandalone),
-    Distributed(GreptimeDbCluster),
-}
-
-impl MockInstanceImpl {
-    pub(crate) fn metasrv(&self) -> &Arc<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 {
diff --git a/tests-integration/tests/grpc.rs b/tests-integration/tests/grpc.rs
index 1d9b123d1b..9607d61162 100644
--- a/tests-integration/tests/grpc.rs
+++ b/tests-integration/tests/grpc.rs
@@ -15,13 +15,16 @@
 use std::sync::Arc;
 
 use api::v1::alter_table_expr::Kind;
+use api::v1::greptime_database_client::GreptimeDatabaseClient;
+use api::v1::greptime_request::Request as RequestBody;
+use api::v1::greptime_response::Response as ResponseBody;
 use api::v1::promql_request::Promql;
 use api::v1::value::ValueData;
 use api::v1::{
-    AddColumn, AddColumns, AlterTableExpr, Basic, Column, ColumnDataType, ColumnDef,
-    CreateTableExpr, InsertRequest, InsertRequests, PromInstantQuery, PromRangeQuery,
-    PromqlRequest, RequestHeader, Row, RowInsertRequest, RowInsertRequests, SemanticType, Value,
-    column,
+    AddColumn, AddColumns, AlterTableExpr, Basic, Column, ColumnDataType, ColumnDef, ColumnSchema,
+    CreateTableExpr, GreptimeRequest, InsertRequest, InsertRequests, PromInstantQuery,
+    PromRangeQuery, PromqlRequest, RequestHeader, Row, RowInsertRequest, RowInsertRequests, Rows,
+    SemanticType, Value, column,
 };
 use auth::user_provider_from_option;
 use base64::prelude::{BASE64_STANDARD, Engine as _};
@@ -52,6 +55,7 @@ use otel_arrow_rust::proto::opentelemetry::arrow::v1::{
 };
 use otel_arrow_rust::proto::opentelemetry::metrics::v1::AggregationTemporality;
 use otel_arrow_rust::schema::consts as arrow_consts;
+use rstest_reuse::apply;
 use servers::grpc::GrpcServerConfig;
 use servers::grpc::builder::GrpcServerBuilder;
 use servers::http::prometheus::{
@@ -61,13 +65,17 @@ use servers::http::prometheus::{
 use servers::request_memory_limiter::ServerMemoryLimiter;
 use servers::server::Server;
 use servers::tls::{TlsMode, TlsOption};
+use session::hints::{HINTS_KEY, INSERT_SKIP_WAL_HINT};
 use tests_integration::test_util::{
-    StorageType, setup_grpc_server, setup_grpc_server_with,
+    MockInstanceImpl, StorageType, assert_wal_delta, setup_grpc_server,
+    setup_grpc_server_for_frontend_instance, setup_grpc_server_with,
     setup_grpc_server_with_auto_create_table_disabled, setup_grpc_server_with_user_provider,
 };
 use tonic::Request;
 use tonic::metadata::MetadataValue;
 
+use crate::both_deployment_cases;
+
 #[macro_export]
 macro_rules! grpc_test {
     ($service:ident, $($(#[$meta:meta])* $test:ident),*,) => {
@@ -776,6 +784,44 @@ fn gauge_arrow_batch(batch_id: i64, reserved_attr: bool) -> BatchArrowRecords {
     }
 }
 
+#[apply(both_deployment_cases)]
+async fn test_skip_wal_otel_arrow_metrics(distributed: bool) {
+    // OTEL Arrow metrics decode into ordinary metric inserts, unlike Flight DoPut bulk inserts.
+    let mut env = MockInstanceImpl::new("skip_wal_otel_arrow", distributed).await;
+    let server = setup_grpc_server_for_frontend_instance(env.frontend(), None).await;
+    let mut client =
+        ArrowMetricsServiceClient::connect(format!("http://{}", server.bind_addr().unwrap()))
+            .await
+            .unwrap();
+    // Warm up auto-created logical/physical tables and metadata before comparing data-region WAL.
+    for (batch_id, hint) in [None, Some("true"), Some("false"), None]
+        .into_iter()
+        .enumerate()
+    {
+        let before = env.flush_and_snapshot_wal().await;
+        let batch = gauge_arrow_batch(batch_id as i64, false);
+        let request = with_skip_wal_hint(futures::stream::iter([batch]), hint);
+        let mut response = client.arrow_metrics(request).await.unwrap().into_inner();
+        let status = response.message().await.unwrap().unwrap();
+        assert_eq!(
+            status.status_code,
+            ArrowStatusCode::Ok as i32,
+            "{}",
+            status.status_message
+        );
+        assert!(response.message().await.unwrap().is_none());
+        if batch_id != 0 {
+            assert_wal_delta(
+                &before,
+                &env.flush_and_snapshot_wal().await,
+                hint == Some("true"),
+            );
+        }
+    }
+    server.shutdown().await.unwrap();
+    env.shutdown().await;
+}
+
 pub async fn test_otel_arrow_delta_histogram(store_type: StorageType) {
     let (_instance, server) =
         setup_grpc_server(store_type, "test_otel_arrow_delta_histogram").await;
@@ -1152,6 +1198,139 @@ fn expect_data() -> (Column, Column, Column, Column) {
     )
 }
 
+fn with_skip_wal_hint<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;
diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs
index c696bd40ac..84d1f7d957 100644
--- a/tests-integration/tests/http.rs
+++ b/tests-integration/tests/http.rs
@@ -58,8 +58,8 @@ use opentelemetry_proto::tonic::collector::trace::v1::{
 };
 use pipeline::GREPTIME_INTERNAL_TRACE_PIPELINE_V1_NAME;
 use prost::Message;
+use rstest_reuse::apply;
 use serde_json::{Value, json};
-use servers::http::GreptimeQueryOutput;
 use servers::http::handler::HealthResponse;
 use servers::http::header::constants::{
     GREPTIME_LOG_EXTRACT_KEYS_HEADER_NAME, GREPTIME_LOG_TABLE_NAME_HEADER_NAME,
@@ -72,20 +72,23 @@ use servers::http::result::error_result::ErrorResponse;
 use servers::http::result::greptime_result_v1::GreptimedbV1Response;
 use servers::http::result::influxdb_result_v1::{InfluxdbOutput, InfluxdbV1Response};
 use servers::http::test_helpers::{TestClient, TestResponse};
+use servers::http::{GreptimeQueryOutput, HttpOptions, HttpServerBuilder};
 use servers::prom_remote_write::v2::test_util as remote_write_v2;
+use servers::prom_remote_write::validation::PromValidationMode;
 use servers::prom_store::{self, mock_timeseries_new_label};
 use servers::request_memory_limiter::ServerMemoryLimiter;
 use standalone::options::StandaloneOptions;
 use table::table_name::TableName;
 use tests_integration::test_util::{
-    StorageType, setup_test_http_app, setup_test_http_app_with_frontend,
-    setup_test_http_app_with_frontend_and_slow_query_threshold,
+    MockInstanceImpl, StorageType, assert_wal_delta, 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)>,
diff --git a/tests-integration/tests/main.rs b/tests-integration/tests/main.rs
index 56108d4d3f..28a8115572 100644
--- a/tests-integration/tests/main.rs
+++ b/tests-integration/tests/main.rs
@@ -41,6 +41,13 @@ mod reconciliation_event;
 mod view_ddl_event;
 mod wal_prune_event;
 
+#[rstest_reuse::template]
+#[rstest::rstest]
+#[case::standalone(false)]
+#[case::distributed(true)]
+#[tokio::test(flavor = "multi_thread")]
+fn both_deployment_cases(#[case] distributed: bool) {}
+
 grpc_tests!(File, S3, S3WithCache, Oss, Azblob, Gcs);
 
 http_tests!(File, S3, S3WithCache, Oss, Azblob, Gcs);
diff --git a/tests/cases/standalone/common/session_skip_wal.result b/tests/cases/standalone/common/session_skip_wal.result
new file mode 100644
index 0000000000..4343e3cd33
--- /dev/null
+++ b/tests/cases/standalone/common/session_skip_wal.result
@@ -0,0 +1,224 @@
+CREATE TABLE session_skip_wal(host STRING, ts TIMESTAMP TIME INDEX);
+
+Affected Rows: 0
+
+-- MYSQL: SET persists across statements on the same connection.
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = true;
+
+affected_rows: 0
+
+-- Invalid SET must leave the enabled policy unchanged.
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = 'invalid';
+
+Failed to execute query, err: MySqlError { ERROR 1235 (42000): (Unsupported): Not supported: Invalid skip_wal value "invalid": expected true or false }
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = 1;
+
+Failed to execute query, err: MySqlError { ERROR 1235 (42000): (Unsupported): Not supported: SET skip_wal requires true or false }
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = NULL;
+
+Failed to execute query, err: MySqlError { ERROR 1235 (42000): (Unsupported): Not supported: SET skip_wal requires true or false }
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = false, true;
+
+Failed to execute query, err: MySqlError { ERROR 1235 (42000): (Unsupported): Not supported: SET skip_wal requires exactly one boolean value }
+
+-- SQLNESS PROTOCOL MYSQL
+INSERT INTO session_skip_wal VALUES ('skipped', 1000);
+
+affected_rows: 1
+
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM session_skip_wal ORDER BY ts;
+
++---------+---------------------+
+| host    | ts                  |
++---------+---------------------+
+| skipped | 1970-01-01 00:00:01 |
++---------+---------------------+
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM session_skip_wal ORDER BY ts;
+
+affected_rows: 0
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = false;
+
+affected_rows: 0
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = 'true';
+
+affected_rows: 0
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = 'false';
+
+affected_rows: 0
+
+-- SQLNESS PROTOCOL MYSQL
+INSERT INTO session_skip_wal VALUES ('persisted', 2000);
+
+affected_rows: 1
+
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM session_skip_wal ORDER BY ts;
+
++-----------+---------------------+
+| host      | ts                  |
++-----------+---------------------+
+| persisted | 1970-01-01 00:00:02 |
++-----------+---------------------+
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM session_skip_wal ORDER BY ts;
+
++-----------+---------------------+
+| host      | ts                  |
++-----------+---------------------+
+| persisted | 1970-01-01 00:00:02 |
++-----------+---------------------+
+
+TRUNCATE TABLE session_skip_wal;
+
+Affected Rows: 0
+
+-- POSTGRES: SET persists across statements on the same connection.
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = true;
+
+Affected Rows: 0
+
+-- Invalid SET must leave the enabled policy unchanged.
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = 'invalid';
+
+Failed to execute query, encountered: Error { kind: Db, cause: Some(DbError { severity: "ERROR", parsed_severity: None, code: SqlState(E0A000), message: "Not supported: Invalid skip_wal value \"invalid\": expected true or false", detail: None, hint: None, position: None, where_: None, schema: None, table: None, column: None, datatype: None, constraint: None, file: None, line: None, routine: None }) }
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = 1;
+
+Failed to execute query, encountered: Error { kind: Db, cause: Some(DbError { severity: "ERROR", parsed_severity: None, code: SqlState(E0A000), message: "Not supported: SET skip_wal requires true or false", detail: None, hint: None, position: None, where_: None, schema: None, table: None, column: None, datatype: None, constraint: None, file: None, line: None, routine: None }) }
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = NULL;
+
+Failed to execute query, encountered: Error { kind: Db, cause: Some(DbError { severity: "ERROR", parsed_severity: None, code: SqlState(E0A000), message: "Not supported: SET skip_wal requires true or false", detail: None, hint: None, position: None, where_: None, schema: None, table: None, column: None, datatype: None, constraint: None, file: None, line: None, routine: None }) }
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = false, true;
+
+Failed to execute query, encountered: Error { kind: Db, cause: Some(DbError { severity: "ERROR", parsed_severity: None, code: SqlState(E0A000), message: "Not supported: SET skip_wal requires exactly one boolean value", detail: None, hint: None, position: None, where_: None, schema: None, table: None, column: None, datatype: None, constraint: None, file: None, line: None, routine: None }) }
+
+-- SQLNESS PROTOCOL POSTGRES
+INSERT INTO session_skip_wal VALUES ('skipped', 1000);
+
+Affected Rows: 1
+
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM session_skip_wal ORDER BY ts;
+
++---------+----------------------------+
+| host    | ts                         |
++---------+----------------------------+
+| skipped | 1970-01-01 00:00:01.000000 |
++---------+----------------------------+
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM session_skip_wal ORDER BY ts;
+
++------+----+
+| host | ts |
++------+----+
++------+----+
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = false;
+
+Affected Rows: 0
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = 'true';
+
+Affected Rows: 0
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = 'false';
+
+Affected Rows: 0
+
+-- SQLNESS PROTOCOL POSTGRES
+INSERT INTO session_skip_wal VALUES ('persisted', 2000);
+
+Affected Rows: 1
+
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM session_skip_wal ORDER BY ts;
+
++-----------+----------------------------+
+| host      | ts                         |
++-----------+----------------------------+
+| persisted | 1970-01-01 00:00:02.000000 |
++-----------+----------------------------+
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM session_skip_wal ORDER BY ts;
+
++-----------+----------------------------+
+| host      | ts                         |
++-----------+----------------------------+
+| persisted | 1970-01-01 00:00:02.000000 |
++-----------+----------------------------+
+
+TRUNCATE TABLE session_skip_wal;
+
+Affected Rows: 0
+
+DROP TABLE session_skip_wal;
+
+Affected Rows: 0
+
+-- The default gRPC protocol has request-scoped contexts, not a persistent SQL session.
+SET skip_wal = true;
+
+Affected Rows: 0
+
+SET skip_wal = false;
+
+Affected Rows: 0
+
+SET skip_wal = 'true';
+
+Affected Rows: 0
+
+SET skip_wal = 'false';
+
+Affected Rows: 0
+
+SET skip_wal = 'invalid';
+
+Error: 1001(Unsupported), Not supported: Invalid skip_wal value "invalid": expected true or false
+
+SET skip_wal = 1;
+
+Error: 1001(Unsupported), Not supported: SET skip_wal requires true or false
+
+SET skip_wal = NULL;
+
+Error: 1001(Unsupported), Not supported: SET skip_wal requires true or false
+
+SET skip_wal = false, true;
+
+Error: 1001(Unsupported), Not supported: SET skip_wal requires exactly one boolean value
+
diff --git a/tests/cases/standalone/common/session_skip_wal.sql b/tests/cases/standalone/common/session_skip_wal.sql
new file mode 100644
index 0000000000..8be063253b
--- /dev/null
+++ b/tests/cases/standalone/common/session_skip_wal.sql
@@ -0,0 +1,118 @@
+CREATE TABLE session_skip_wal(host STRING, ts TIMESTAMP TIME INDEX);
+
+-- MYSQL: SET persists across statements on the same connection.
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = true;
+
+-- Invalid SET must leave the enabled policy unchanged.
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = 'invalid';
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = 1;
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = NULL;
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = false, true;
+
+-- SQLNESS PROTOCOL MYSQL
+INSERT INTO session_skip_wal VALUES ('skipped', 1000);
+
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM session_skip_wal ORDER BY ts;
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM session_skip_wal ORDER BY ts;
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = false;
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = 'true';
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = 'false';
+
+-- SQLNESS PROTOCOL MYSQL
+INSERT INTO session_skip_wal VALUES ('persisted', 2000);
+
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM session_skip_wal ORDER BY ts;
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM session_skip_wal ORDER BY ts;
+
+TRUNCATE TABLE session_skip_wal;
+
+-- POSTGRES: SET persists across statements on the same connection.
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = true;
+
+-- Invalid SET must leave the enabled policy unchanged.
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = 'invalid';
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = 1;
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = NULL;
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = false, true;
+
+-- SQLNESS PROTOCOL POSTGRES
+INSERT INTO session_skip_wal VALUES ('skipped', 1000);
+
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM session_skip_wal ORDER BY ts;
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM session_skip_wal ORDER BY ts;
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = false;
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = 'true';
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = 'false';
+
+-- SQLNESS PROTOCOL POSTGRES
+INSERT INTO session_skip_wal VALUES ('persisted', 2000);
+
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM session_skip_wal ORDER BY ts;
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM session_skip_wal ORDER BY ts;
+
+TRUNCATE TABLE session_skip_wal;
+
+DROP TABLE session_skip_wal;
+
+-- The default gRPC protocol has request-scoped contexts, not a persistent SQL session.
+SET skip_wal = true;
+
+SET skip_wal = false;
+
+SET skip_wal = 'true';
+
+SET skip_wal = 'false';
+
+SET skip_wal = 'invalid';
+
+SET skip_wal = 1;
+
+SET skip_wal = NULL;
+
+SET skip_wal = false, true;
diff --git a/tests/cases/standalone/copy/copy_skip_wal.result b/tests/cases/standalone/copy/copy_skip_wal.result
new file mode 100644
index 0000000000..9f7a2b71ce
--- /dev/null
+++ b/tests/cases/standalone/copy/copy_skip_wal.result
@@ -0,0 +1,274 @@
+CREATE TABLE copy_skip_wal(host STRING, ts TIMESTAMP TIME INDEX);
+
+Affected Rows: 0
+
+INSERT INTO copy_skip_wal VALUES ('host1', 1000), ('host2', 2000);
+
+Affected Rows: 2
+
+COPY copy_skip_wal TO '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv');
+
+Affected Rows: 2
+
+TRUNCATE TABLE copy_skip_wal;
+
+Affected Rows: 0
+
+-- MYSQL: COPY TABLE inherits the connection WAL policy.
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = true;
+
+affected_rows: 0
+
+-- SQLNESS PROTOCOL MYSQL
+COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv');
+
+affected_rows: 2
+
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++-------+---------------------+
+| host  | ts                  |
++-------+---------------------+
+| host1 | 1970-01-01 00:00:01 |
+| host2 | 1970-01-01 00:00:02 |
++-------+---------------------+
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+affected_rows: 0
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = false;
+
+affected_rows: 0
+
+-- SQLNESS PROTOCOL MYSQL
+COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv');
+
+affected_rows: 2
+
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++-------+---------------------+
+| host  | ts                  |
++-------+---------------------+
+| host1 | 1970-01-01 00:00:01 |
+| host2 | 1970-01-01 00:00:02 |
++-------+---------------------+
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++-------+---------------------+
+| host  | ts                  |
++-------+---------------------+
+| host1 | 1970-01-01 00:00:01 |
+| host2 | 1970-01-01 00:00:02 |
++-------+---------------------+
+
+TRUNCATE TABLE copy_skip_wal;
+
+Affected Rows: 0
+
+-- MYSQL: COPY DATABASE inherits the connection WAL policy.
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = true;
+
+affected_rows: 0
+
+-- SQLNESS PROTOCOL MYSQL
+COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv');
+
+affected_rows: 2
+
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++-------+---------------------+
+| host  | ts                  |
++-------+---------------------+
+| host1 | 1970-01-01 00:00:01 |
+| host2 | 1970-01-01 00:00:02 |
++-------+---------------------+
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+affected_rows: 0
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = false;
+
+affected_rows: 0
+
+-- SQLNESS PROTOCOL MYSQL
+COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv');
+
+affected_rows: 2
+
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++-------+---------------------+
+| host  | ts                  |
++-------+---------------------+
+| host1 | 1970-01-01 00:00:01 |
+| host2 | 1970-01-01 00:00:02 |
++-------+---------------------+
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++-------+---------------------+
+| host  | ts                  |
++-------+---------------------+
+| host1 | 1970-01-01 00:00:01 |
+| host2 | 1970-01-01 00:00:02 |
++-------+---------------------+
+
+TRUNCATE TABLE copy_skip_wal;
+
+Affected Rows: 0
+
+-- POSTGRES: COPY TABLE inherits the connection WAL policy.
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = true;
+
+Affected Rows: 0
+
+-- SQLNESS PROTOCOL POSTGRES
+COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv');
+
+Affected Rows: 2
+
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++-------+----------------------------+
+| host  | ts                         |
++-------+----------------------------+
+| host1 | 1970-01-01 00:00:01.000000 |
+| host2 | 1970-01-01 00:00:02.000000 |
++-------+----------------------------+
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++------+----+
+| host | ts |
++------+----+
++------+----+
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = false;
+
+Affected Rows: 0
+
+-- SQLNESS PROTOCOL POSTGRES
+COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv');
+
+Affected Rows: 2
+
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++-------+----------------------------+
+| host  | ts                         |
++-------+----------------------------+
+| host1 | 1970-01-01 00:00:01.000000 |
+| host2 | 1970-01-01 00:00:02.000000 |
++-------+----------------------------+
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++-------+----------------------------+
+| host  | ts                         |
++-------+----------------------------+
+| host1 | 1970-01-01 00:00:01.000000 |
+| host2 | 1970-01-01 00:00:02.000000 |
++-------+----------------------------+
+
+TRUNCATE TABLE copy_skip_wal;
+
+Affected Rows: 0
+
+-- POSTGRES: COPY DATABASE inherits the connection WAL policy.
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = true;
+
+Affected Rows: 0
+
+-- SQLNESS PROTOCOL POSTGRES
+COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv');
+
+Affected Rows: 2
+
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++-------+----------------------------+
+| host  | ts                         |
++-------+----------------------------+
+| host1 | 1970-01-01 00:00:01.000000 |
+| host2 | 1970-01-01 00:00:02.000000 |
++-------+----------------------------+
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++------+----+
+| host | ts |
++------+----+
++------+----+
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = false;
+
+Affected Rows: 0
+
+-- SQLNESS PROTOCOL POSTGRES
+COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv');
+
+Affected Rows: 2
+
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++-------+----------------------------+
+| host  | ts                         |
++-------+----------------------------+
+| host1 | 1970-01-01 00:00:01.000000 |
+| host2 | 1970-01-01 00:00:02.000000 |
++-------+----------------------------+
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
++-------+----------------------------+
+| host  | ts                         |
++-------+----------------------------+
+| host1 | 1970-01-01 00:00:01.000000 |
+| host2 | 1970-01-01 00:00:02.000000 |
++-------+----------------------------+
+
+TRUNCATE TABLE copy_skip_wal;
+
+Affected Rows: 0
+
+DROP TABLE copy_skip_wal;
+
+Affected Rows: 0
+
diff --git a/tests/cases/standalone/copy/copy_skip_wal.sql b/tests/cases/standalone/copy/copy_skip_wal.sql
new file mode 100644
index 0000000000..7ead655de5
--- /dev/null
+++ b/tests/cases/standalone/copy/copy_skip_wal.sql
@@ -0,0 +1,125 @@
+CREATE TABLE copy_skip_wal(host STRING, ts TIMESTAMP TIME INDEX);
+
+INSERT INTO copy_skip_wal VALUES ('host1', 1000), ('host2', 2000);
+
+COPY copy_skip_wal TO '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv');
+
+TRUNCATE TABLE copy_skip_wal;
+
+-- MYSQL: COPY TABLE inherits the connection WAL policy.
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = true;
+
+-- SQLNESS PROTOCOL MYSQL
+COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv');
+
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = false;
+
+-- SQLNESS PROTOCOL MYSQL
+COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv');
+
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+TRUNCATE TABLE copy_skip_wal;
+
+-- MYSQL: COPY DATABASE inherits the connection WAL policy.
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = true;
+
+-- SQLNESS PROTOCOL MYSQL
+COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv');
+
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+-- SQLNESS PROTOCOL MYSQL
+SET skip_wal = false;
+
+-- SQLNESS PROTOCOL MYSQL
+COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv');
+
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL MYSQL
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+TRUNCATE TABLE copy_skip_wal;
+
+-- POSTGRES: COPY TABLE inherits the connection WAL policy.
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = true;
+
+-- SQLNESS PROTOCOL POSTGRES
+COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv');
+
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = false;
+
+-- SQLNESS PROTOCOL POSTGRES
+COPY copy_skip_wal FROM '${SQLNESS_HOME}/copy_skip_wal/copy_skip_wal.csv' WITH (FORMAT='csv');
+
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+TRUNCATE TABLE copy_skip_wal;
+
+-- POSTGRES: COPY DATABASE inherits the connection WAL policy.
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = true;
+
+-- SQLNESS PROTOCOL POSTGRES
+COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv');
+
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+-- SQLNESS PROTOCOL POSTGRES
+SET skip_wal = false;
+
+-- SQLNESS PROTOCOL POSTGRES
+COPY DATABASE public FROM '${SQLNESS_HOME}/copy_skip_wal/' WITH (FORMAT='csv');
+
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+-- SQLNESS ARG restart=true
+-- SQLNESS PROTOCOL POSTGRES
+SELECT * FROM copy_skip_wal ORDER BY ts;
+
+TRUNCATE TABLE copy_skip_wal;
+
+DROP TABLE copy_skip_wal;