fix: support to compat the old layered memtable options (apache#1568)

## Rationale
We introduce the explicit flag to control should we enable layered
memtable, but it has some compatibility problem when upgrading from old
version.
This pr add an option to support compating the old layered memtable
on/off control method.

## Detailed Changes
Add an option to support compating the old layered memtable on/off
control method.

## Test Plan
Manually.

src/analytic_engine/src/instance/open.rs

@@ -154,6 +154,7 @@ impl Instance {
             preflush_write_buffer_size_ratio: ctx.config.preflush_write_buffer_size_ratio,
             manifest_snapshot_every_n_updates: ctx.config.manifest.snapshot_every_n_updates,
             enable_primary_key_sampling: ctx.config.enable_primary_key_sampling,
+            try_compat_old_layered_memtable_opts: ctx.config.try_compat_old_layered_memtable_opts,
             metrics_opt: ctx.config.metrics.clone(),
         });
         let manifest = ManifestImpl::open(

src/analytic_engine/src/lib.rs

@@ -69,6 +69,12 @@ pub struct Config {
     /// Default options for table
     pub table_opts: TableOptions,
 
+    /// Should we try to compat the `LayeredMemtableOptions` in `TableOptions`
+    /// The old one use if `mutable_segment_switch_threshold` > 0 to control
+    /// the on/off of layered memtable(`0`:off, `>0`:on).
+    /// The new one use a explicit flag `enable` to do that.
+    pub try_compat_old_layered_memtable_opts: bool,
+
     pub compaction: SchedulerConfig,
 
     /// Offload the compaction task to remote nodes or not.
@@ -182,6 +188,7 @@ impl Default for Config {
             replay_batch_size: 500,
             max_replay_tables_per_batch: 64,
             table_opts: TableOptions::default(),
+            try_compat_old_layered_memtable_opts: false,
             compaction: SchedulerConfig::default(),
             compaction_offload: false,
             sst_meta_cache_cap: Some(1000),

src/analytic_engine/src/manifest/details.rs

@@ -776,6 +776,7 @@ mod tests {
                     manifest_snapshot_every_n_updates: NonZeroUsize::new(usize::MAX).unwrap(),
                     metrics_opt: MetricsOptions::default(),
                     enable_primary_key_sampling: false,
+                    try_compat_old_layered_memtable_opts: false,
                 },
                 &purger,
                 mem_size_options,

src/analytic_engine/src/table/data.rs

@@ -68,6 +68,7 @@ use crate::{
         sst_util,
         version::{MemTableForWrite, MemTableState, SamplingMemTable, TableVersion},
     },
+    table_options::UpdateMode,
     MetricsOptions, TableOptions,
 };
 
@@ -155,6 +156,7 @@ pub struct TableConfig {
     pub manifest_snapshot_every_n_updates: NonZeroUsize,
     pub metrics_opt: MetricsOptions,
     pub enable_primary_key_sampling: bool,
+    pub try_compat_old_layered_memtable_opts: bool,
 }
 
 #[derive(Debug, Clone)]
@@ -314,11 +316,13 @@ impl TableData {
             name,
             schema,
         } = desc;
+
         let TableConfig {
             preflush_write_buffer_size_ratio,
             manifest_snapshot_every_n_updates,
             metrics_opt,
             enable_primary_key_sampling,
+            ..
         } = config;
 
         let memtable_factory: MemTableFactoryRef = match opts.memtable_type {
@@ -406,6 +410,7 @@ impl TableData {
             manifest_snapshot_every_n_updates,
             metrics_opt,
             enable_primary_key_sampling,
+            try_compat_old_layered_memtable_opts,
         } = config;
 
         let memtable_factory: MemTableFactoryRef = match add_meta.opts.memtable_type {
@@ -421,17 +426,31 @@ impl TableData {
                 .mutable_segment_switch_threshold
                 .0 as usize;
 
-            ensure!(
-                mutable_segment_switch_threshold > 0,
-                InvalidTableOpts {
+            if mutable_segment_switch_threshold > 0 {
+                ensure!(
+                    add_meta.opts.update_mode != UpdateMode::Overwrite,
+                    InvalidTableOpts {
+                        msg: "layered memtable is enabled but update mode is Overwrite",
+                    }
+                );
+
+                Arc::new(LayeredMemtableFactory::new(
+                    memtable_factory,
+                    mutable_segment_switch_threshold,
+                )) as _
+            } else if try_compat_old_layered_memtable_opts {
+                // Maybe some old layered memtable opts controlling the on/off of this feature
+                // by checking `mutable_segment_switch_threshold`(`0`:disable, `>0`:enable)
+                // were persisted.
+                // If `try_compat_old_layered_memtable_opts` is true, we will try to follow the
+                // old behavior.
+                memtable_factory as _
+            } else {
+                return InvalidTableOpts {
                     msg: "layered memtable is enabled but mutable_switch_threshold is 0",
                 }
-            );
-
-            Arc::new(LayeredMemtableFactory::new(
-                memtable_factory,
-                mutable_segment_switch_threshold,
-            )) as _
+                .fail();
+            }
         } else {
             memtable_factory as _
         };
@@ -1028,6 +1047,7 @@ pub mod tests {
                     manifest_snapshot_every_n_updates: self.manifest_snapshot_every_n_updates,
                     metrics_opt: MetricsOptions::default(),
                     enable_primary_key_sampling: false,
+                    try_compat_old_layered_memtable_opts: false,
                 },
                 &purger,
                 mem_size_options,

src/analytic_engine/src/table_meta_set_impl.rs

@@ -54,6 +54,7 @@ pub(crate) struct TableMetaSetImpl {
     pub(crate) preflush_write_buffer_size_ratio: f32,
     pub(crate) manifest_snapshot_every_n_updates: NonZeroUsize,
     pub(crate) enable_primary_key_sampling: bool,
+    pub(crate) try_compat_old_layered_memtable_opts: bool,
     pub(crate) metrics_opt: MetricsOptions,
 }
 
@@ -140,6 +141,8 @@ impl TableMetaSetImpl {
                                 .manifest_snapshot_every_n_updates,
                             metrics_opt: self.metrics_opt.clone(),
                             enable_primary_key_sampling: self.enable_primary_key_sampling,
+                            try_compat_old_layered_memtable_opts: self
+                                .try_compat_old_layered_memtable_opts,
                         },
                         &self.file_purger,
                         mem_size_options,
@@ -271,6 +274,7 @@ impl TableMetaSetImpl {
                     manifest_snapshot_every_n_updates: self.manifest_snapshot_every_n_updates,
                     metrics_opt: self.metrics_opt.clone(),
                     enable_primary_key_sampling: self.enable_primary_key_sampling,
+                    try_compat_old_layered_memtable_opts: self.try_compat_old_layered_memtable_opts,
                 },
                 mem_size_options,
                 allocator,