feat: Allow ALTER TABLE to modify the skip_wal option dynamically

scripts/fetch-dashboard-assets.sh

@@ -23,7 +23,7 @@ function retry_fetch() {
     local url=$1
     local filename=$2
 
-    curl --connect-timeout 10 --retry 3 -fsSL $url --output $filename || {
+    curl --connect-timeout 30 --max-time 60 --retry 3 -fsSL $url --output $filename || {
         echo "Failed to download $url"
         echo "You may try to set http_proxy and https_proxy environment variables."
         if [[ -z "$GITHUB_PROXY_URL" ]]; then

src/common/meta/src/ddl/alter_logical_tables/executor.rs

@@ -137,6 +137,7 @@ impl<'a> AlterLogicalTablesExecutor<'a> {
                 current_table_info_value,
                 Some(region_distribution),
                 new_raw_table_info,
+                None,
             )
             .await?;
 

src/common/meta/src/ddl/alter_table.rs

@@ -283,13 +283,16 @@ impl AlterTableProcedure {
         };
 
         // Safety: region distribution is set in `submit_alter_region_requests`.
+        // Note: We don't reallocate WAL options when skip_wal changes.
+        // The region_wal_options in DatanodeTableValue are preserved.
         self.executor
             .on_alter_metadata(
                 &self.context.table_metadata_manager,
                 table_info_value,
                 self.data.region_distribution.as_ref(),
                 new_info.into(),
                 &self.data.column_metadatas,
+                None, // Don't update region_wal_options
             )
             .await?;
 

src/common/meta/src/ddl/alter_table/executor.rs

@@ -121,6 +121,9 @@ impl AlterTableExecutor {
         region_distribution: Option<&RegionDistribution>,
         mut raw_table_info: RawTableInfo,
         column_metadatas: &[ColumnMetadata],
+        new_region_wal_options: Option<
+            std::collections::HashMap<store_api::storage::RegionNumber, String>,
+        >,
     ) -> Result<()> {
         let table_ref = self.table.table_ref();
         let table_id = self.table_id;
@@ -155,6 +158,7 @@ impl AlterTableExecutor {
                     current_table_info_value,
                     region_distribution.cloned(),
                     raw_table_info,
+                    new_region_wal_options,
                 )
                 .await?;
         }

src/common/meta/src/ddl/create_logical_tables/update_metadata.rs

@@ -63,7 +63,7 @@ impl CreateLogicalTablesProcedure {
         // Update physical table's metadata and we don't need to touch per-region settings.
         self.context
             .table_metadata_manager
-            .update_table_info(&physical_table_info, None, new_table_info)
+            .update_table_info(&physical_table_info, None, new_table_info, None)
             .await?;
 
         // Invalid physical table cache

src/common/meta/src/key.rs

@@ -1119,6 +1119,9 @@ impl TableMetadataManager {
         current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
         region_distribution: Option<RegionDistribution>,
         new_table_info: RawTableInfo,
+        new_region_wal_options: Option<
+            std::collections::HashMap<store_api::storage::RegionNumber, String>,
+        >,
     ) -> Result<()> {
         let table_id = current_table_info_value.table_info.ident.table_id;
         let new_table_info_value = current_table_info_value.update(new_table_info);
@@ -1133,7 +1136,12 @@ impl TableMetadataManager {
             let new_region_options = new_table_info_value.table_info.to_region_options();
             let update_datanode_table_options_txn = self
                 .datanode_table_manager
-                .build_update_table_options_txn(table_id, region_distribution, new_region_options)
+                .build_update_table_options_txn(
+                    table_id,
+                    region_distribution,
+                    new_region_options,
+                    new_region_wal_options,
+                )
                 .await?;
             Txn::merge_all([update_table_info_txn, update_datanode_table_options_txn])
         } else {
@@ -2002,12 +2010,22 @@ mod tests {
             DeserializedValueWithBytes::from_inner(TableInfoValue::new(table_info.clone()));
         // should be ok.
         table_metadata_manager
-            .update_table_info(&current_table_info_value, None, new_table_info.clone())
+            .update_table_info(
+                &current_table_info_value,
+                None,
+                new_table_info.clone(),
+                None,
+            )
             .await
             .unwrap();
         // if table info was updated, it should be ok.
         table_metadata_manager
-            .update_table_info(&current_table_info_value, None, new_table_info.clone())
+            .update_table_info(
+                &current_table_info_value,
+                None,
+                new_table_info.clone(),
+                None,
+            )
             .await
             .unwrap();
 
@@ -2030,7 +2048,7 @@ mod tests {
         // The ABA problem.
         assert!(
             table_metadata_manager
-                .update_table_info(&wrong_table_info_value, None, new_table_info)
+                .update_table_info(&wrong_table_info_value, None, new_table_info, None)
                 .await
                 .is_err()
         )

src/common/meta/src/key/datanode_table.rs

@@ -267,6 +267,7 @@ impl DatanodeTableManager {
         table_id: TableId,
         region_distribution: RegionDistribution,
         new_region_options: HashMap<String, String>,
+        new_region_wal_options: Option<HashMap<RegionNumber, String>>,
     ) -> Result<Txn> {
         assert!(!region_distribution.is_empty());
         // safety: region_distribution must not be empty
@@ -284,12 +285,27 @@ impl DatanodeTableManager {
             .and_then(|r| DatanodeTableValue::try_from_raw_value(&r.value))?
             .region_info;
 
-        // If the region options are the same, we don't need to update it.
-        if region_info.region_options == new_region_options {
+        // If the region options are the same and WAL options are not being updated, we don't need to update it.
+        let need_update_options = region_info.region_options != new_region_options;
+        let need_update_wal_options = if let Some(ref new_wal_options) = new_region_wal_options {
+            region_info.region_wal_options != *new_wal_options
+        } else {
+            false
+        };
+
+        if !need_update_options && !need_update_wal_options {
             return Ok(Txn::new());
         }
-        // substitute region options only.
-        region_info.region_options = new_region_options;
+
+        // substitute region options.
+        if need_update_options {
+            region_info.region_options = new_region_options;
+        }
+
+        // substitute region WAL options if provided.
+        if let Some(new_wal_options) = new_region_wal_options {
+            region_info.region_wal_options = new_wal_options;
+        }
 
         let mut txns = Vec::with_capacity(region_distribution.len());
 

src/common/meta/src/reconciliation/reconcile_table/update_table_info.rs

@@ -96,6 +96,7 @@ impl State for UpdateTableInfo {
                 current_table_info_value,
                 Some(region_distribution),
                 new_table_info,
+                None,
             )
             .await?;
 

src/mito2/src/compaction/window.rs

@@ -252,6 +252,7 @@ mod tests {
                 memtable: None,
                 merge_mode: None,
                 sst_format: None,
+                original_wal_options: None,
             },
             compaction_time_window: None,
         }

src/mito2/src/region/options.rs

@@ -79,6 +79,10 @@ pub struct RegionOptions {
     pub merge_mode: Option<MergeMode>,
     /// SST format type.
     pub sst_format: Option<FormatType>,
+    /// Original WAL options saved when skip_wal is enabled.
+    /// Used to restore WAL options when skip_wal is disabled.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub original_wal_options: Option<WalOptions>,
 }
 
 impl RegionOptions {
@@ -166,6 +170,7 @@ impl TryFrom<&HashMap<String, String>> for RegionOptions {
             memtable,
             merge_mode: options.merge_mode,
             sst_format: options.sst_format,
+            original_wal_options: None,
         };
         opts.validate()?;
 
@@ -673,6 +678,7 @@ mod tests {
             })),
             merge_mode: Some(MergeMode::LastNonNull),
             sst_format: None,
+            original_wal_options: None,
         };
         assert_eq!(expect, options);
     }
@@ -708,6 +714,7 @@ mod tests {
             })),
             merge_mode: Some(MergeMode::LastNonNull),
             sst_format: None,
+            original_wal_options: None,
         };
         let region_options_json_str = serde_json::to_string(&options).unwrap();
         let got: RegionOptions = serde_json::from_str(&region_options_json_str).unwrap();
@@ -773,6 +780,7 @@ mod tests {
             })),
             merge_mode: Some(MergeMode::LastNonNull),
             sst_format: None,
+            original_wal_options: None,
         };
         assert_eq!(options, got);
     }

src/mito2/src/worker/handle_alter.rs

@@ -20,6 +20,7 @@ use std::sync::Arc;
 use common_base::readable_size::ReadableSize;
 use common_telemetry::info;
 use common_telemetry::tracing::warn;
+use common_wal::options::WalOptions;
 use humantime_serde::re::humantime;
 use snafu::{ResultExt, ensure};
 use store_api::logstore::LogStore;
@@ -229,6 +230,54 @@ impl<S: LogStore> RegionWorkerLoop<S> {
                         );
                     }
                 }
+                SetRegionOption::SkipWal {
+                    skip_wal,
+                    wal_options,
+                } => {
+                    info!(
+                        "Update region skip_wal: {}, previous: {:?}, new: {}, original: {:?}",
+                        region.region_id,
+                        current_options.wal_options,
+                        skip_wal,
+                        current_options.original_wal_options
+                    );
+                    if skip_wal {
+                        // Save original wal_options before disabling WAL
+                        if !matches!(current_options.wal_options, WalOptions::Noop) {
+                            current_options.original_wal_options =
+                                Some(current_options.wal_options.clone());
+                        }
+                        current_options.wal_options = WalOptions::Noop;
+                    } else {
+                        // Restore WAL options: priority order
+                        // 1. Provided wal_options from request
+                        // 2. Saved original_wal_options
+                        // 3. Fallback to RaftEngine
+                        if let Some(opts) = wal_options {
+                            match serde_json::from_str(&opts) {
+                                Ok(restored) => current_options.wal_options = restored,
+                                Err(e) => {
+                                    warn!(
+                                        "Failed to parse wal_options '{}': {}, trying original",
+                                        opts, e
+                                    );
+                                    current_options.wal_options = current_options
+                                        .original_wal_options
+                                        .take()
+                                        .unwrap_or(WalOptions::RaftEngine);
+                                }
+                            }
+                        } else if let Some(original) = current_options.original_wal_options.take() {
+                            current_options.wal_options = original;
+                        } else {
+                            warn!(
+                                "No wal_options to restore for region {}, using RaftEngine",
+                                region.region_id
+                            );
+                            current_options.wal_options = WalOptions::RaftEngine;
+                        }
+                    }
+                }
             }
         }
         region.version_control.alter_options(current_options);
@@ -264,6 +313,24 @@ fn new_region_options_on_empty_memtable(
 
                 current_options.sst_format = Some(new_format);
             }
+            SetRegionOption::SkipWal {
+                skip_wal,
+                wal_options,
+            } => {
+                if *skip_wal {
+                    if !matches!(current_options.wal_options, WalOptions::Noop) {
+                        current_options.original_wal_options =
+                            Some(current_options.wal_options.clone());
+                    }
+                    current_options.wal_options = WalOptions::Noop;
+                } else if let Some(opts) = wal_options {
+                    if let Ok(restored) = serde_json::from_str(opts) {
+                        current_options.wal_options = restored;
+                    }
+                } else if let Some(original) = current_options.original_wal_options.take() {
+                    current_options.wal_options = original;
+                }
+            }
         }
     }
     Some(current_options)

src/query/src/sql/show_create_table.rs

@@ -38,7 +38,8 @@ use sql::statements::{self, OptionMap};
 use store_api::metric_engine_consts::{is_metric_engine, is_metric_engine_internal_column};
 use table::metadata::{TableInfoRef, TableMeta};
 use table::requests::{
-    COMMENT_KEY as TABLE_COMMENT_KEY, FILE_TABLE_META_KEY, TTL_KEY, WRITE_BUFFER_SIZE_KEY,
+    COMMENT_KEY as TABLE_COMMENT_KEY, FILE_TABLE_META_KEY, SKIP_WAL_KEY, TTL_KEY,
+    WRITE_BUFFER_SIZE_KEY,
 };
 
 use crate::error::{
@@ -66,6 +67,10 @@ fn create_sql_options(table_meta: &TableMeta, schema_options: Option<SchemaOptio
         options.insert(TTL_KEY.to_string(), database_ttl);
     };
 
+    if table_opts.skip_wal {
+        options.insert(SKIP_WAL_KEY.to_string(), "true".to_string());
+    }
+
     for (k, v) in table_opts
         .extra_options
         .iter()