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,44 @@
         };
 
         // Safety: region distribution is set in `submit_alter_region_requests`.
+        // Check if skip_wal changed and update WAL options if needed
+        let current_skip_wal = table_info_value.table_info.meta.options.skip_wal;
+        let new_skip_wal = new_info.meta.options.skip_wal;
+        let new_region_wal_options = if current_skip_wal != new_skip_wal
+            && self.data.region_distribution.is_some()
+        {
+            // Get region numbers from region_distribution
+            let region_distribution = self.data.region_distribution.as_ref().unwrap();
+            let region_numbers: Vec<_> = region_distribution
+                .values()
+                .flat_map(|region_role_set| {
+                    region_role_set
+                        .leader_regions
+                        .iter()
+                        .chain(region_role_set.follower_regions.iter())
+                        .copied()
+                })
+                .collect();
+            // Allocate new WAL options based on skip_wal
+            Some(
+                self.context
+                    .table_metadata_allocator
+                    .wal_options_allocator()
+                    .allocate(&region_numbers, new_skip_wal)
+                    .await?,
+            )
+        } else {
+            None
+        };
+
         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,
+                new_region_wal_options,
             )
             .await?;
 

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

@@ -118,9 +118,10 @@
         &self,
         table_metadata_manager: &TableMetadataManagerRef,
         current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
         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 +156,7 @@
                     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/ddl/table_meta.rs

@@ -164,4 +164,8 @@ impl TableMetadataAllocator {
     pub fn table_id_allocator(&self) -> ResourceIdAllocatorRef {
         self.table_id_allocator.clone()
     }
+
+    pub fn wal_options_allocator(&self) -> &WalOptionsAllocatorRef {
+        &self.wal_options_allocator
+    }
 }

src/common/meta/src/key.rs

@@ -1116,9 +1116,10 @@
     /// if region_distribution is present.
     pub async fn update_table_info(
         &self,
         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 +1134,12 @@
             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 {
@@ -1990,15 +1996,15 @@
         let mut new_table_info = table_info.clone();
         new_table_info.name = "hi".to_string();
         let current_table_info_value =
             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();
 
@@ -2021,7 +2027,7 @@
         // 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/worker/handle_alter.rs

@@ -17,11 +17,12 @@
 use std::str::FromStr;
 use std::sync::Arc;
 
 use common_base::readable_size::ReadableSize;
 use common_telemetry::info;
 use common_telemetry::tracing::warn;
 use humantime_serde::re::humantime;
 use snafu::{ResultExt, ensure};
+use common_wal::options::WalOptions;
 use store_api::logstore::LogStore;
 use store_api::metadata::{
     InvalidSetRegionOptionRequestSnafu, MetadataError, RegionMetadata, RegionMetadataBuilder,
@@ -229,6 +230,24 @@
                         );
                     }
                 }
+                SetRegionOption::SkipWal(skip_wal) => {
+                    info!(
+                        "Update region skip_wal: {}, previous: {:?} new: {}",
+                        region.region_id, current_options.wal_options, skip_wal
+                    );
+                    if skip_wal {
+                        // Disable WAL by setting to Noop
+                        current_options.wal_options = WalOptions::Noop;
+                    } else {
+                        // Enable WAL: restore to default (RaftEngine)
+                        // TODO: In distributed mode, this should be allocated by metasrv,
+                        // but for simplicity, we use RaftEngine as default here.
+                        // The actual WAL options will be persisted in metasrv.
+                        // We should read the correct WAL options from DatanodeTableValue
+                        // or pass them through the AlterRegionRequest.
+                        current_options.wal_options = WalOptions::RaftEngine;
+                    }
+                }
             }
         }
         region.version_control.alter_options(current_options);
@@ -264,6 +283,13 @@
 
                 current_options.sst_format = Some(new_format);
             }
+            SetRegionOption::SkipWal(skip_wal) => {
+                if *skip_wal {
+                    current_options.wal_options = WalOptions::Noop;
+                } else {
+                    current_options.wal_options = WalOptions::RaftEngine;
+                }
+            }
         }
     }
     Some(current_options)

src/query/src/sql/show_create_table.rs

@@ -32,10 +32,10 @@
 use sql::parser::ParserContext;
 use sql::statements::create::{Column, ColumnExtensions, CreateTable, TableConstraint};
 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::{
@@ -63,6 +63,10 @@
         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()

src/store-api/src/region_request.rs

@@ -49,9 +49,10 @@
     InvalidSetRegionOptionRequestSnafu, InvalidUnsetRegionOptionRequestSnafu, MetadataError,
     RegionMetadata, Result, UnexpectedSnafu,
 };
 use crate::metric_engine_consts::PHYSICAL_TABLE_METADATA_KEY;
 use crate::metrics;
 use crate::mito_engine_options::{
+    SKIP_WAL_KEY,
     SST_FORMAT_KEY, TTL_KEY, TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW, TWCS_TRIGGER_FILE_NUM,
 };
 use crate::path_utils::table_dir;
@@ -1297,6 +1298,8 @@
     Twsc(String, String),
     // Modifying the SST format.
     Format(String),
+    // Modifying skip_wal option.
+    SkipWal(bool),
 }
 
 impl TryFrom<&PbOption> for SetRegionOption {
@@ -1313,8 +1316,13 @@
             }
             TWCS_TRIGGER_FILE_NUM | TWCS_MAX_OUTPUT_FILE_SIZE | TWCS_TIME_WINDOW => {
                 Ok(Self::Twsc(key.clone(), value.clone()))
             }
             SST_FORMAT_KEY => Ok(Self::Format(value.clone())),
+            SKIP_WAL_KEY => {
+                let skip_wal = value.parse::<bool>()
+                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
+                Ok(Self::SkipWal(skip_wal))
+            }
             _ => InvalidSetRegionOptionRequestSnafu { key, value }.fail(),
         }
     }
@@ -1333,6 +1341,7 @@
                 SetRegionOption::Twsc(unset_option.to_string(), String::new())
             }
             UnsetRegionOption::Ttl => SetRegionOption::Ttl(Default::default()),
+            UnsetRegionOption::SkipWal => SetRegionOption::SkipWal(false),
         }
     }
 }
@@ -1346,6 +1355,7 @@
             TWCS_TRIGGER_FILE_NUM => Ok(Self::TwcsTriggerFileNum),
             TWCS_MAX_OUTPUT_FILE_SIZE => Ok(Self::TwcsMaxOutputFileSize),
             TWCS_TIME_WINDOW => Ok(Self::TwcsTimeWindow),
+            SKIP_WAL_KEY => Ok(Self::SkipWal),
             _ => InvalidUnsetRegionOptionRequestSnafu { key }.fail(),
         }
     }
@@ -1357,6 +1367,7 @@
     TwcsMaxOutputFileSize,
     TwcsTimeWindow,
     Ttl,
+    SkipWal,
 }
 
 impl UnsetRegionOption {
@@ -1366,6 +1377,7 @@
             Self::TwcsTriggerFileNum => TWCS_TRIGGER_FILE_NUM,
             Self::TwcsMaxOutputFileSize => TWCS_MAX_OUTPUT_FILE_SIZE,
             Self::TwcsTimeWindow => TWCS_TIME_WINDOW,
+            Self::SkipWal => SKIP_WAL_KEY,
         }
     }
 }

src/table/src/metadata.rs

@@ -299,6 +299,9 @@ impl TableMeta {
                         .extra_options
                         .insert(SST_FORMAT_KEY.to_string(), value.clone());
                 }
+                SetRegionOption::SkipWal(skip_wal) => {
+                    new_options.skip_wal = *skip_wal;
+                }
             }
         }
         let mut builder = self.new_meta_builder();