feat(config): support overriding region engine options from env (#9700)

* feat(config): support overriding region engine options from env

`region_engine` is a `Vec` of externally tagged enums, so plain dotted
environment keys cannot address it: `config` walks the path, replaces the
default sequence with a table and loading fails with `invalid type: map,
expected a sequence`.

Address the entries by tag name instead
(`<PREFIX>__REGION_ENGINE__<TAG>__<FIELD>`) and merge the overrides onto the
default entries in a dedicated source, so the engines and fields that are not
mentioned keep their configured value. An entry the default sequence does not
have is appended; a numeric index, an override of the whole sequence, a field
below a non-table default and two variables that set a path and a field below
it are reported with a dedicated error instead of a serde type error.

A configuration file that configures the sequence as a whole still replaces
the merged value, which the example configs document.

Closes #8620

Signed-off-by: NightG1owZ <316247938+NightG1owZ@users.noreply.github.com>

* docs(config): clarify that the metric engine is always enabled

`region_engine` defaults to [mito, file], but the datanode builds and
registers the metric engine unconditionally. Reword the example configs so
`mito` and `file` read as the default configuration entries and naming
`metric` only supplies its options, then regenerate config.md.

Signed-off-by: NightG1owZ <316247938+NightG1owZ@users.noreply.github.com>

* docs(config): drop the env-var explanation from the config reference

Environment variables are documented in the GreptimeTeam/docs repository, so the example configs and the generated config reference only describe the config file. Keep the note that the metric engine is always registered.

Signed-off-by: NightG1owZ <316247938+NightG1owZ@users.noreply.github.com>

* test(config): cover tagged sequence environment collection

Add unit tests for TaggedSeqEnvSource::collect: merging by tag, appending unknown tags, empty and null defaults, flat key removal, the error paths and the take_entries and is_index_like helpers. The variables are injected through Environment::source, so the tests do not touch the process environment.

Signed-off-by: NightG1owZ <316247938+NightG1owZ@users.noreply.github.com>

---------

Signed-off-by: NightG1owZ <316247938+NightG1owZ@users.noreply.github.com>
Co-authored-by: NightG1owZ <316247938+NightG1owZ@users.noreply.github.com>

config/config.md

@@ -174,7 +174,7 @@
 | `storage.http_client.timeout` | String | `30s` | The total request timeout, applied from when the request starts connecting until the response body has finished.<br/>Also considered a total deadline. |
 | `storage.http_client.pool_idle_timeout` | String | `90s` | The timeout for idle sockets being kept-alive. |
 | `storage.http_client.skip_ssl_validation` | Bool | `false` | To skip the ssl verification<br/>**Security Notice**: Setting `skip_ssl_validation = true` disables certificate verification, making connections vulnerable to man-in-the-middle attacks. Only use this in development or trusted private networks. |
-| `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines.<br/>Each engine type (mito, file, metric) may appear only once; duplicates cause startup to fail. |
+| `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines.<br/>Each engine type (mito, file, metric) may appear only once; duplicates cause startup to fail.<br/>The default configuration lists `mito` and `file`; `metric` is always registered, so naming<br/>it only supplies its options rather than enabling the engine. |
 | `region_engine.mito` | -- | -- | The Mito engine options. |
 | `region_engine.mito.num_workers` | Integer | Auto | Number of region workers. Defaults to half of the available CPU cores, rounded up. |
 | `region_engine.mito.worker_channel_size` | Integer | `128` | Request channel size of each worker. |
@@ -650,7 +650,7 @@
 | `storage.http_client.timeout` | String | `30s` | The total request timeout, applied from when the request starts connecting until the response body has finished.<br/>Also considered a total deadline. |
 | `storage.http_client.pool_idle_timeout` | String | `90s` | The timeout for idle sockets being kept-alive. |
 | `storage.http_client.skip_ssl_validation` | Bool | `false` | To skip the ssl verification<br/>**Security Notice**: Setting `skip_ssl_validation = true` disables certificate verification, making connections vulnerable to man-in-the-middle attacks. Only use this in development or trusted private networks. |
-| `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines.<br/>Each engine type (mito, file, metric) may appear only once; duplicates cause startup to fail. |
+| `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines.<br/>Each engine type (mito, file, metric) may appear only once; duplicates cause startup to fail.<br/>The default configuration lists `mito` and `file`; `metric` is always registered, so naming<br/>it only supplies its options rather than enabling the engine. |
 | `region_engine.mito` | -- | -- | The Mito engine options. |
 | `region_engine.mito.num_workers` | Integer | Auto | Number of region workers. Defaults to half of the available CPU cores, rounded up. |
 | `region_engine.mito.worker_channel_size` | Integer | `128` | Request channel size of each worker. |

config/datanode.example.toml

@@ -509,6 +509,8 @@ skip_ssl_validation = false
 
 ## The region engine options. You can configure multiple region engines.
 ## Each engine type (mito, file, metric) may appear only once; duplicates cause startup to fail.
+## The default configuration lists `mito` and `file`; `metric` is always registered, so naming
+## it only supplies its options rather than enabling the engine.
 [[region_engine]]
 
 ## The Mito engine options.

config/standalone.example.toml

@@ -729,6 +729,8 @@ skip_ssl_validation = false
 
 ## The region engine options. You can configure multiple region engines.
 ## Each engine type (mito, file, metric) may appear only once; duplicates cause startup to fail.
+## The default configuration lists `mito` and `file`; `metric` is always registered, so naming
+## it only supplies its options rather than enabling the engine.
 [[region_engine]]
 
 ## The Mito engine options.

src/cmd/src/options.rs

@@ -65,6 +65,10 @@ where
     fn env_list_keys() -> Option<&'static [&'static str]> {
         T::env_list_keys()
     }
+
+    fn env_tagged_seq_keys() -> Option<&'static [&'static str]> {
+        T::env_tagged_seq_keys()
+    }
 }
 
 /// A serde `deserialize_with` helper that parses the `plugins` field leniently.

src/cmd/tests/load_config_test.rs

@@ -902,3 +902,216 @@ fn test_load_metric_config_with_removed_sparse_primary_key_encoding() {
         Duration::from_secs(30)
     );
 }
+
+/// Returns the mito engine config of a loaded `region_engine` sequence.
+fn find_mito(region_engine: &[RegionEngineConfig]) -> &MitoConfig {
+    region_engine
+        .iter()
+        .find_map(|engine| match engine {
+            RegionEngineConfig::Mito(config) => Some(config),
+            _ => None,
+        })
+        .expect("the mito engine config")
+}
+
+#[test]
+fn test_load_region_engine_mito_field_from_env() {
+    let env_prefix = "REGION_ENGINE_MITO_FIELD_UT";
+    let env_key = [
+        env_prefix,
+        "REGION_ENGINE",
+        "MITO",
+        "GLOBAL_WRITE_BUFFER_REJECT_SIZE",
+    ]
+    .join(ENV_VAR_SEP);
+
+    temp_env::with_var(env_key, Some("4GB"), || {
+        let options =
+            GreptimeOptions::<DatanodeOptions>::load_layered_options(None, env_prefix).unwrap();
+
+        // The addressed field wins, the fields next to it keep their default value.
+        let mito = find_mito(&options.component.region_engine);
+        assert_eq!(mito.global_write_buffer_reject_size, ReadableSize::gb(4));
+        assert_eq!(
+            mito.global_write_buffer_size,
+            MitoConfig::default().global_write_buffer_size
+        );
+        assert_eq!(mito.index, MitoConfig::default().index);
+
+        // The other entries of the default sequence are intact.
+        assert!(
+            options
+                .component
+                .region_engine
+                .iter()
+                .any(|engine| matches!(engine, RegionEngineConfig::File(_))),
+            "the file engine config should survive an env override of the mito engine"
+        );
+    });
+}
+
+#[test]
+fn test_load_region_engine_mito_field_from_env_standalone() {
+    let env_prefix = "REGION_ENGINE_MITO_FIELD_STANDALONE_UT";
+    let env_key = [
+        env_prefix,
+        "REGION_ENGINE",
+        "MITO",
+        "GLOBAL_WRITE_BUFFER_REJECT_SIZE",
+    ]
+    .join(ENV_VAR_SEP);
+
+    temp_env::with_var(env_key, Some("4GB"), || {
+        let options =
+            GreptimeOptions::<StandaloneOptions>::load_layered_options(None, env_prefix).unwrap();
+
+        let mito = find_mito(&options.component.region_engine);
+        assert_eq!(mito.global_write_buffer_reject_size, ReadableSize::gb(4));
+        assert_eq!(
+            mito.global_write_buffer_size,
+            MitoConfig::default().global_write_buffer_size
+        );
+        assert!(
+            options
+                .component
+                .region_engine
+                .iter()
+                .any(|engine| matches!(engine, RegionEngineConfig::File(_))),
+            "the file engine config should survive an env override of the mito engine"
+        );
+    });
+}
+
+#[test]
+fn test_load_region_engine_metric_entry_from_env() {
+    let env_prefix = "REGION_ENGINE_METRIC_FIELD_UT";
+    let env_key = [
+        env_prefix,
+        "REGION_ENGINE",
+        "METRIC",
+        "FLUSH_METADATA_REGION_INTERVAL",
+    ]
+    .join(ENV_VAR_SEP);
+
+    temp_env::with_var(env_key, Some("1m"), || {
+        let options =
+            GreptimeOptions::<DatanodeOptions>::load_layered_options(None, env_prefix).unwrap();
+
+        // `metric` is not in the default sequence, so it is appended instead of replacing
+        // the whole sequence.
+        assert_eq!(
+            options.component.region_engine.len(),
+            DatanodeOptions::default().region_engine.len() + 1
+        );
+        let metric = options
+            .component
+            .region_engine
+            .iter()
+            .find_map(|engine| match engine {
+                RegionEngineConfig::Metric(config) => Some(config),
+                _ => None,
+            })
+            .expect("the metric engine config");
+        assert_eq!(
+            metric.flush_metadata_region_interval,
+            Duration::from_secs(60)
+        );
+
+        // The engines of the default sequence keep their default options.
+        let mito = find_mito(&options.component.region_engine);
+        assert_eq!(
+            mito.global_write_buffer_reject_size,
+            MitoConfig::default().global_write_buffer_reject_size
+        );
+        assert!(
+            options
+                .component
+                .region_engine
+                .iter()
+                .any(|engine| matches!(engine, RegionEngineConfig::File(_)))
+        );
+    });
+}
+
+#[test]
+fn test_load_region_engine_env_patch_is_replaced_by_config_file() {
+    let config = tempfile::NamedTempFile::new().unwrap();
+    std::fs::write(
+        config.path(),
+        "[[region_engine]]\n[region_engine.metric]\nflush_metadata_region_interval = \"2m\"\n",
+    )
+    .unwrap();
+
+    let env_prefix = "REGION_ENGINE_FILE_WINS_UT";
+    let env_key = [
+        env_prefix,
+        "REGION_ENGINE",
+        "MITO",
+        "GLOBAL_WRITE_BUFFER_REJECT_SIZE",
+    ]
+    .join(ENV_VAR_SEP);
+
+    temp_env::with_var(env_key, Some("4GB"), || {
+        let options = GreptimeOptions::<DatanodeOptions>::load_layered_options(
+            config.path().to_str(),
+            env_prefix,
+        )
+        .unwrap();
+
+        // A sequence is a single value: the config file replaces the whole value produced by
+        // the environment, so the environment patch (which is merged onto the default
+        // entries) is dropped. That keeps the precedence file > env > default of every other
+        // option, and means a config file has to spell out every engine it wants.
+        assert_eq!(
+            options.component.region_engine,
+            vec![RegionEngineConfig::Metric(MetricEngineConfig {
+                flush_metadata_region_interval: Duration::from_secs(120),
+            })]
+        );
+    });
+}
+
+#[test]
+fn test_load_region_engine_env_patch_with_config_file_lacking_region_engine() {
+    // The usual deployment shape: a config file that sets other options, plus an environment
+    // variable patching one region engine. The file says nothing about `region_engine`, so
+    // there is nothing that could replace the patched sequence.
+    let config = tempfile::NamedTempFile::new().unwrap();
+    std::fs::write(config.path(), "node_id = 7\n").unwrap();
+
+    let env_prefix = "REGION_ENGINE_FILE_NO_ENGINE_UT";
+    let env_key = [
+        env_prefix,
+        "REGION_ENGINE",
+        "MITO",
+        "GLOBAL_WRITE_BUFFER_REJECT_SIZE",
+    ]
+    .join(ENV_VAR_SEP);
+
+    temp_env::with_var(env_key, Some("4GB"), || {
+        let options = GreptimeOptions::<DatanodeOptions>::load_layered_options(
+            config.path().to_str(),
+            env_prefix,
+        )
+        .unwrap();
+
+        // The option of the file is applied...
+        assert_eq!(options.component.node_id, Some(7));
+        // ...the environment patch applies on top of the default engine entries...
+        let mito = find_mito(&options.component.region_engine);
+        assert_eq!(mito.global_write_buffer_reject_size, ReadableSize::gb(4));
+        assert_eq!(
+            mito.global_write_buffer_size,
+            MitoConfig::default().global_write_buffer_size
+        );
+        // ...and the rest of the default sequence is still configured.
+        assert!(
+            options
+                .component
+                .region_engine
+                .iter()
+                .any(|engine| matches!(engine, RegionEngineConfig::File(_))),
+            "the file engine entry should survive"
+        );
+    });
+}

src/common/config/src/config.rs

@@ -17,7 +17,10 @@ use serde::Serialize;
 use serde::de::DeserializeOwned;
 use snafu::ResultExt;
 
-use crate::error::{LoadLayeredConfigSnafu, Result, SerdeJsonSnafu, TomlFormatSnafu};
+use crate::error::{
+    LoadLayeredConfigSnafu, Result, SerdeJsonSnafu, TaggedSeqNotTopLevelSnafu, TomlFormatSnafu,
+};
+use crate::tagged_seq::TaggedSeqEnvSource;
 
 /// Separator for environment variables. For example, `DATANODE__STORAGE__MANIFEST__CHECKPOINT_MARGIN`.
 pub const ENV_VAR_SEP: &str = "__";
@@ -64,9 +67,27 @@ pub trait Configurable: Serialize + DeserializeOwned + Default + Sized {
         let default_config = File::from_str(&json_str, FileFormat::Json);
 
         // Add default values and environment variables as the sources of the configuration.
-        let mut layered_config = config::Config::builder()
-            .add_source(default_config)
-            .add_source(env_source);
+        let mut layered_config = config::Config::builder().add_source(default_config);
+
+        // A sequence of tagged tables cannot be addressed by plain dotted env keys: the path
+        // walk of `config` drops the default sequence. Such keys are merged onto the default
+        // entries by tag name instead.
+        layered_config = match Self::env_tagged_seq_keys() {
+            Some(keys) if !keys.is_empty() => {
+                // The default entries are looked up by literal key, so a dotted key would find
+                // none of them. Reject it instead of silently dropping the defaults.
+                if let Some(key) = keys.iter().find(|key| key.contains('.')) {
+                    return TaggedSeqNotTopLevelSnafu {
+                        key: (*key).to_string(),
+                    }
+                    .fail();
+                }
+
+                let defaults = serde_json::to_value(&default_opts).context(SerdeJsonSnafu)?;
+                layered_config.add_source(TaggedSeqEnvSource::new(env_source, keys, &defaults))
+            }
+            _ => layered_config.add_source(env_source),
+        };
 
         // Add config file as the source of the configuration if it is specified.
         if let Some(config_file) = config_file {
@@ -93,6 +114,22 @@ pub trait Configurable: Serialize + DeserializeOwned + Default + Sized {
         None
     }
 
+    /// Keys whose value is a sequence of single-key tagged tables (e.g. `region_engine`,
+    /// serialized from `Vec<Enum>` with externally tagged variants). Environment variables
+    /// address them by tag name (`KEY__<TAG>__<FIELD>`) instead of numeric index, and the
+    /// overrides are merged onto the corresponding default entries.
+    ///
+    /// Keys must name a **top-level** field of the options struct: the default entries to
+    /// merge onto are looked up by literal key in the serialized defaults, so a dotted key is
+    /// rejected instead of silently leaving the defaults out of the merged sequence.
+    ///
+    /// A configuration file that configures such a key as a whole replaces the merged value,
+    /// so the environment variables for it are ineffective then and the file has to spell out
+    /// every entry it wants.
+    fn env_tagged_seq_keys() -> Option<&'static [&'static str]> {
+        None
+    }
+
     /// Serialize the configuration to a TOML string.
     fn to_toml(&self) -> Result<String> {
         toml::to_string(&self).context(TomlFormatSnafu)
@@ -102,7 +139,9 @@ pub trait Configurable: Serialize + DeserializeOwned + Default + Sized {
 #[cfg(test)]
 mod tests {
     use std::io::Write;
+    use std::time::Duration;
 
+    use common_base::readable_size::ReadableSize;
     use common_telemetry::logging::LoggingOptions;
     use common_test_util::temp_dir::create_named_temp_file;
     use common_wal::config::DatanodeWalConfig;
@@ -264,4 +303,358 @@ mod tests {
             },
         );
     }
+
+    /// A sequence of single-key tagged tables, the shape of `Vec<Enum>` with externally
+    /// tagged variants.
+    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+    #[serde(default)]
+    struct TestIndexConfig {
+        aux_path: String,
+        cache_percent: u8,
+    }
+
+    impl Default for TestIndexConfig {
+        fn default() -> Self {
+            Self {
+                aux_path: "/default/aux".to_string(),
+                cache_percent: 30,
+            }
+        }
+    }
+
+    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+    #[serde(default)]
+    struct TestMitoConfig {
+        global_write_buffer_size: ReadableSize,
+        global_write_buffer_reject_size: ReadableSize,
+        index: TestIndexConfig,
+    }
+
+    impl Default for TestMitoConfig {
+        fn default() -> Self {
+            Self {
+                global_write_buffer_size: ReadableSize::mb(64),
+                global_write_buffer_reject_size: ReadableSize::gb(2),
+                index: TestIndexConfig::default(),
+            }
+        }
+    }
+
+    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+    #[serde(default)]
+    struct TestFileConfig {
+        data_dir: String,
+    }
+
+    impl Default for TestFileConfig {
+        fn default() -> Self {
+            Self {
+                data_dir: "/default/data".to_string(),
+            }
+        }
+    }
+
+    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+    #[serde(default)]
+    struct TestMetricConfig {
+        #[serde(with = "humantime_serde")]
+        flush_metadata_region_interval: Duration,
+    }
+
+    impl Default for TestMetricConfig {
+        fn default() -> Self {
+            Self {
+                flush_metadata_region_interval: Duration::from_secs(30),
+            }
+        }
+    }
+
+    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+    #[serde(rename_all = "snake_case")]
+    enum TestRegionEngine {
+        Mito(TestMitoConfig),
+        File(TestFileConfig),
+        Metric(TestMetricConfig),
+    }
+
+    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+    struct TestTaggedSeqConfig {
+        region_engine: Vec<TestRegionEngine>,
+    }
+
+    impl Default for TestTaggedSeqConfig {
+        fn default() -> Self {
+            Self {
+                region_engine: vec![
+                    TestRegionEngine::Mito(TestMitoConfig::default()),
+                    TestRegionEngine::File(TestFileConfig::default()),
+                ],
+            }
+        }
+    }
+
+    impl Configurable for TestTaggedSeqConfig {
+        fn env_tagged_seq_keys() -> Option<&'static [&'static str]> {
+            Some(&["region_engine"])
+        }
+    }
+
+    /// A config whose tagged sequence key is not a top-level field.
+    #[derive(Debug, Serialize, Deserialize, Default)]
+    struct TestDottedKeyConfig {
+        region_engine: Vec<TestRegionEngine>,
+    }
+
+    impl Configurable for TestDottedKeyConfig {
+        fn env_tagged_seq_keys() -> Option<&'static [&'static str]> {
+            Some(&["region_engine.mito"])
+        }
+    }
+
+    fn test_mito(config: &TestTaggedSeqConfig) -> &TestMitoConfig {
+        config
+            .region_engine
+            .iter()
+            .find_map(|engine| match engine {
+                TestRegionEngine::Mito(config) => Some(config),
+                _ => None,
+            })
+            .expect("the mito entry")
+    }
+
+    fn test_metric(config: &TestTaggedSeqConfig) -> &TestMetricConfig {
+        config
+            .region_engine
+            .iter()
+            .find_map(|engine| match engine {
+                TestRegionEngine::Metric(config) => Some(config),
+                _ => None,
+            })
+            .expect("the metric entry")
+    }
+
+    /// Loads with the given prefix and returns the message of the error, including the
+    /// source of the layered loading error.
+    fn load_error_text(env_prefix: &str) -> String {
+        let err = TestTaggedSeqConfig::load_layered_options(None, env_prefix).unwrap_err();
+        let mut text = err.to_string();
+        let mut source = std::error::Error::source(&err);
+        while let Some(error) = source {
+            text.push_str(&format!(": {error}"));
+            source = std::error::Error::source(error);
+        }
+        text
+    }
+
+    #[test]
+    fn test_tagged_seq_env_overrides_a_field() {
+        let env_prefix = "TAGGED_SEQ_FIELD_UT";
+        let env_key = [
+            env_prefix,
+            "REGION_ENGINE",
+            "MITO",
+            "GLOBAL_WRITE_BUFFER_REJECT_SIZE",
+        ]
+        .join(ENV_VAR_SEP);
+
+        temp_env::with_var(env_key, Some("4GB"), || {
+            let opts = TestTaggedSeqConfig::load_layered_options(None, env_prefix).unwrap();
+            let defaults = TestTaggedSeqConfig::default();
+
+            // The entries of the sequence that the environment does not mention are intact.
+            assert_eq!(opts.region_engine.len(), defaults.region_engine.len());
+            assert_eq!(opts.region_engine[1], defaults.region_engine[1]);
+
+            // The addressed field wins, the fields next to it keep their default value.
+            let mito = test_mito(&opts);
+            assert_eq!(mito.global_write_buffer_reject_size, ReadableSize::gb(4));
+            assert_eq!(
+                mito.global_write_buffer_size,
+                TestMitoConfig::default().global_write_buffer_size
+            );
+            assert_eq!(mito.index, TestIndexConfig::default());
+        });
+    }
+
+    #[test]
+    fn test_tagged_seq_env_appends_an_entry() {
+        let env_prefix = "TAGGED_SEQ_APPEND_UT";
+        let env_key = [
+            env_prefix,
+            "REGION_ENGINE",
+            "METRIC",
+            "FLUSH_METADATA_REGION_INTERVAL",
+        ]
+        .join(ENV_VAR_SEP);
+
+        temp_env::with_var(env_key, Some("1m"), || {
+            let opts = TestTaggedSeqConfig::load_layered_options(None, env_prefix).unwrap();
+            let defaults = TestTaggedSeqConfig::default();
+
+            // The entry is not in the default sequence, so it is appended.
+            assert_eq!(opts.region_engine.len(), defaults.region_engine.len() + 1);
+            assert_eq!(opts.region_engine[0], defaults.region_engine[0]);
+            assert_eq!(opts.region_engine[1], defaults.region_engine[1]);
+            assert_eq!(
+                test_metric(&opts).flush_metadata_region_interval,
+                Duration::from_secs(60)
+            );
+        });
+    }
+
+    #[test]
+    fn test_tagged_seq_env_overrides_a_nested_field() {
+        let env_prefix = "TAGGED_SEQ_NESTED_UT";
+        let env_key = [env_prefix, "REGION_ENGINE", "MITO", "INDEX", "AUX_PATH"].join(ENV_VAR_SEP);
+
+        temp_env::with_var(env_key, Some("/tmp/aux"), || {
+            let opts = TestTaggedSeqConfig::load_layered_options(None, env_prefix).unwrap();
+
+            let mito = test_mito(&opts);
+            assert_eq!(mito.index.aux_path, "/tmp/aux");
+            assert_eq!(
+                mito.index.cache_percent,
+                TestIndexConfig::default().cache_percent
+            );
+            assert_eq!(
+                mito.global_write_buffer_size,
+                TestMitoConfig::default().global_write_buffer_size
+            );
+        });
+    }
+
+    #[test]
+    fn test_tagged_seq_env_rejects_whole_key() {
+        let env_prefix = "TAGGED_SEQ_NO_TAG_UT";
+        let env_key = [env_prefix, "REGION_ENGINE"].join(ENV_VAR_SEP);
+
+        temp_env::with_var(env_key, Some("mito"), || {
+            let text = load_error_text(env_prefix);
+            assert!(text.contains("region_engine"), "{text}");
+            assert!(text.contains("region_engine__<TAG>__<FIELD>"), "{text}");
+        });
+    }
+
+    #[test]
+    fn test_tagged_seq_env_rejects_numeric_index() {
+        let env_prefix = "TAGGED_SEQ_NUMERIC_UT";
+        let env_key = [
+            env_prefix,
+            "REGION_ENGINE",
+            "0",
+            "MITO",
+            "GLOBAL_WRITE_BUFFER_SIZE",
+        ]
+        .join(ENV_VAR_SEP);
+
+        temp_env::with_var(env_key, Some("4GB"), || {
+            let text = load_error_text(env_prefix);
+            assert!(
+                text.contains("region_engine.0.mito.global_write_buffer_size"),
+                "{text}"
+            );
+            assert!(text.contains("addressed by tag name"), "{text}");
+        });
+    }
+
+    #[test]
+    fn test_tagged_seq_env_rejects_scalar_and_field_under_same_key() {
+        let env_prefix = "TAGGED_SEQ_NOT_A_TABLE_UT";
+        let whole_key = [env_prefix, "REGION_ENGINE", "MITO", "INDEX"].join(ENV_VAR_SEP);
+        let below_key =
+            [env_prefix, "REGION_ENGINE", "MITO", "INDEX", "AUX_PATH"].join(ENV_VAR_SEP);
+
+        temp_env::with_vars(
+            [
+                (whole_key, Some("not-a-table")),
+                (below_key, Some("/tmp/aux")),
+            ],
+            || {
+                // One variable sets `index` as a whole, another sets a field below it. The
+                // conflict is reported as such instead of failing later inside serde with a
+                // type error that does not name the variables involved.
+                let text = load_error_text(env_prefix);
+                assert!(
+                    text.contains("another environment variable sets 'region_engine.mito.index'"),
+                    "{text}"
+                );
+                assert!(text.contains("region_engine.mito.index.aux_path"), "{text}");
+            },
+        );
+    }
+
+    #[test]
+    fn test_tagged_seq_env_rejects_signed_and_float_indices() {
+        for index in ["-1", "+1", "1e3"] {
+            let env_prefix = "TAGGED_SEQ_INDEX_SHAPE_UT";
+            let env_key = [
+                env_prefix,
+                "REGION_ENGINE",
+                index,
+                "MITO",
+                "GLOBAL_WRITE_BUFFER_SIZE",
+            ]
+            .join(ENV_VAR_SEP);
+
+            temp_env::with_var(env_key, Some("4GB"), || {
+                let text = load_error_text(env_prefix);
+                assert!(text.contains("addressed by tag name"), "{index}: {text}");
+            });
+        }
+    }
+
+    #[test]
+    fn test_tagged_seq_env_rejects_tag_without_field() {
+        let env_prefix = "TAGGED_SEQ_NO_FIELD_UT";
+        let env_key = [env_prefix, "REGION_ENGINE", "MITO"].join(ENV_VAR_SEP);
+
+        temp_env::with_var(env_key, Some("mito"), || {
+            let text = load_error_text(env_prefix);
+            assert!(text.contains("region_engine.mito"), "{text}");
+            assert!(text.contains("region_engine__<TAG>__<FIELD>"), "{text}");
+        });
+    }
+
+    #[test]
+    fn test_tagged_seq_env_rejects_field_below_scalar_default() {
+        // The `data_dir` field of the file engine defaults to a string, so nothing can be set
+        // below it.
+        let env_prefix = "TAGGED_SEQ_SCALAR_DEFAULT_UT";
+        let env_key = [env_prefix, "REGION_ENGINE", "FILE", "DATA_DIR", "SUB"].join(ENV_VAR_SEP);
+
+        temp_env::with_var(env_key, Some("/tmp/sub"), || {
+            let text = load_error_text(env_prefix);
+            assert!(text.contains("region_engine.file.data_dir"), "{text}");
+            assert!(text.contains("default value is not a table"), "{text}");
+        });
+    }
+
+    #[test]
+    fn test_tagged_seq_env_type_error_names_the_environment() {
+        let env_prefix = "TAGGED_SEQ_TYPE_ERROR_UT";
+        let env_key = [
+            env_prefix,
+            "REGION_ENGINE",
+            "MITO",
+            "INDEX",
+            "CACHE_PERCENT",
+        ]
+        .join(ENV_VAR_SEP);
+
+        temp_env::with_var(env_key, Some("not-a-number"), || {
+            // The override replaces the default value as the environment collected it, so the
+            // type error still says where that value comes from.
+            let text = load_error_text(env_prefix);
+            assert!(text.contains("in the environment"), "{text}");
+        });
+    }
+
+    #[test]
+    fn test_tagged_seq_key_must_be_top_level() {
+        let err = TestDottedKeyConfig::load_layered_options(None, "TAGGED_SEQ_DOTTED_KEY_UT")
+            .unwrap_err();
+        let text = err.to_string();
+        assert!(text.contains("region_engine.mito"), "{text}");
+        assert!(text.contains("top-level"), "{text}");
+    }
 }

src/common/config/src/error.rs

@@ -65,6 +65,84 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display(
+        "Invalid environment variable key '{key}': '{target}' is a sequence that cannot be \
+         overridden as a whole; address one of its entries with '{target}__<TAG>__<FIELD>'"
+    ))]
+    TaggedSeqMissingTag {
+        key: String,
+        target: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display(
+        "Invalid environment variable key '{key}': '{tag}' is a numeric index, but '{target}' \
+         is addressed by tag name; use '{target}__<TAG>__<FIELD>' instead, e.g. \
+         '{target}__MITO__<FIELD>'"
+    ))]
+    TaggedSeqNumericTag {
+        key: String,
+        tag: String,
+        target: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display(
+        "Invalid environment variable key '{key}': the entry '{tag}' of '{target}' needs a \
+         field to override; use '{target}__<TAG>__<FIELD>' instead"
+    ))]
+    TaggedSeqMissingField {
+        key: String,
+        tag: String,
+        target: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display(
+        "Invalid environment variable key '{key}': another environment variable sets '{path}' \
+         as a whole, so this key cannot set a field below it; set either the whole '{path}' or \
+         its fields, not both"
+    ))]
+    TaggedSeqOverridesConflict {
+        key: String,
+        path: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display(
+        "'{path}' cannot be overridden from the environment: its default value is not a table, \
+         so a field below it cannot be set"
+    ))]
+    TaggedSeqNotATable {
+        path: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display(
+        "'{key}' is registered as a sequence of tagged tables, but it is not a top-level field \
+         of the options; the default entries to merge onto are looked up by literal key"
+    ))]
+    TaggedSeqNotTopLevel {
+        key: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display(
+        "The default value of '{target}' is not a sequence of single-key tables: {reason}"
+    ))]
+    TaggedSeqInvalidDefault {
+        target: String,
+        reason: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 impl ErrorExt for Error {
@@ -73,7 +151,14 @@ impl ErrorExt for Error {
             Error::TomlFormat { .. }
             | Error::LoadLayeredConfig { .. }
             | Error::FileWatch { .. }
-            | Error::InvalidPath { .. } => StatusCode::InvalidArguments,
+            | Error::InvalidPath { .. }
+            | Error::TaggedSeqMissingTag { .. }
+            | Error::TaggedSeqNumericTag { .. }
+            | Error::TaggedSeqMissingField { .. }
+            | Error::TaggedSeqOverridesConflict { .. }
+            | Error::TaggedSeqNotATable { .. }
+            | Error::TaggedSeqNotTopLevel { .. }
+            | Error::TaggedSeqInvalidDefault { .. } => StatusCode::InvalidArguments,
             Error::SerdeJson { .. } => StatusCode::Unexpected,
         }
     }

src/common/config/src/lib.rs

@@ -15,6 +15,7 @@
 pub mod config;
 pub mod error;
 pub mod file_watcher;
+mod tagged_seq;
 
 use std::time::Duration;
 

src/common/config/src/tagged_seq.rs

@@ -0,0 +1,711 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Environment overrides for configuration keys holding a sequence of tagged tables.
+//!
+//! Such a sequence is what a `Vec` of externally tagged enums serializes to, for
+//! example the `region_engine` option:
+//!
+//! ```text
+//! [{"mito": {...}}, {"file": {...}}]
+//! ```
+//!
+//! Handing the plain [`Environment`] source to `config` cannot work for these keys:
+//! `config` writes every flat dotted key by walking down its path, and the walk resets
+//! any intermediate value that is not a table. Walking `region_engine.mito.index.aux_path`
+//! therefore destroys the default sequence before anything is set.
+//!
+//! This source instead addresses the entries by their tag name and merges the
+//! overrides onto the matching default entries:
+//!
+//! ```text
+//! GREPTIMEDB_STANDALONE__REGION_ENGINE__MITO__INDEX__AUX_PATH=...
+//! ```
+//!
+//! Entries that only the environment mentions are appended, and every field that is
+//! not mentioned keeps its default value. A configuration file that configures the
+//! whole sequence replaces the merged one, which makes these variables ineffective;
+//! the file has to spell out every entry it wants.
+
+use config::{ConfigError, Environment, Map, Source, Value, ValueKind};
+use serde_json::Value as JsonValue;
+
+use crate::error::{
+    Error, Result, TaggedSeqInvalidDefaultSnafu, TaggedSeqMissingFieldSnafu,
+    TaggedSeqMissingTagSnafu, TaggedSeqNotATableSnafu, TaggedSeqNumericTagSnafu,
+    TaggedSeqOverridesConflictSnafu,
+};
+
+/// A [`Source`] that merges environment overrides into sequences of tagged tables.
+///
+/// See the [module documentation](self) for the addressed syntax.
+#[derive(Debug)]
+pub(crate) struct TaggedSeqEnvSource {
+    /// The wrapped environment source. Everything it collects is kept as is, apart
+    /// from the entries that address one of the `targets`.
+    inner: Environment,
+    /// The keys whose entries are addressed by tag name.
+    targets: Vec<Target>,
+}
+
+/// One sequence of tagged tables that the environment may override.
+#[derive(Debug, Clone)]
+struct Target {
+    /// The dotted key of the sequence, e.g. `region_engine`.
+    key: String,
+    /// The default value of `key`, whose entries the overrides are merged into.
+    default: Option<JsonValue>,
+}
+
+impl TaggedSeqEnvSource {
+    /// Wraps `inner` so that the sequence at every key of `keys` accepts tag name
+    /// overrides from the environment, merged onto the value that the same key has
+    /// in `defaults`.
+    pub(crate) fn new(
+        inner: Environment,
+        keys: &'static [&'static str],
+        defaults: &JsonValue,
+    ) -> Self {
+        let targets = keys
+            .iter()
+            .map(|key| Target {
+                key: (*key).to_string(),
+                default: defaults.get(key).cloned(),
+            })
+            .collect();
+
+        Self { inner, targets }
+    }
+}
+
+impl Source for TaggedSeqEnvSource {
+    fn clone_into_box(&self) -> Box<dyn Source + Send + Sync> {
+        Box::new(Self {
+            inner: self.inner.clone(),
+            targets: self.targets.clone(),
+        })
+    }
+
+    fn collect(&self) -> std::result::Result<Map<String, Value>, ConfigError> {
+        let mut collected = self.inner.collect()?;
+
+        for target in &self.targets {
+            // Take the entries out first: `config` walks the collected map in an
+            // unspecified order, so a leftover `region_engine.mito.x` entry could be
+            // written after the sequence and destroy it again.
+            let entries = take_entries(&mut collected, &target.key);
+            let merged = merge(target, entries).map_err(|e| ConfigError::Message(e.to_string()))?;
+            collected.insert(target.key.clone(), merged);
+        }
+
+        Ok(collected)
+    }
+}
+
+/// Removes and returns the entries that address `key` itself or one of its children.
+fn take_entries(collected: &mut Map<String, Value>, key: &str) -> Vec<(String, Value)> {
+    let child_prefix = format!("{key}.");
+    let mut keys = collected
+        .keys()
+        .filter(|existing| *existing == key || existing.starts_with(&child_prefix))
+        .cloned()
+        .collect::<Vec<_>>();
+    // `Map` is unordered, sort so that a reported error is deterministic.
+    keys.sort();
+
+    keys.into_iter()
+        .filter_map(|key| collected.remove(&key).map(|value| (key, value)))
+        .collect()
+}
+
+/// Rebuilds the sequence of `target` with the environment overrides `entries` applied.
+fn merge(target: &Target, entries: Vec<(String, Value)>) -> Result<Value> {
+    let mut overrides = group_entries(target, entries)?;
+
+    let mut merged = Vec::with_capacity(overrides.len() + 1);
+    for entry in default_entries(target)? {
+        let mut entry = json_to_value(entry);
+        let tag = single_tag(target, &entry)?;
+
+        if let Some(value) = overrides.remove(&tag) {
+            apply_override(target, &tag, &mut entry, value)?;
+        }
+
+        merged.push(entry);
+    }
+
+    // Entries that the default sequence does not have are appended.
+    let mut appended = overrides.into_iter().collect::<Vec<_>>();
+    appended.sort_by(|a, b| a.0.cmp(&b.0));
+    for (tag, value) in appended {
+        let mut table = Map::new();
+        table.insert(tag, value);
+        merged.push(Value::new(None, ValueKind::Table(table)));
+    }
+
+    Ok(Value::new(None, ValueKind::Array(merged)))
+}
+
+/// Merges the override of the entry `tag` into `entry`, whose only key is `tag`.
+fn apply_override(target: &Target, tag: &str, entry: &mut Value, value: Value) -> Result<()> {
+    let ValueKind::Table(body) = &mut entry.kind else {
+        return invalid_default(target, "expected an array of tables");
+    };
+
+    let Some(field) = body.get_mut(tag) else {
+        return invalid_default(target, "expected each entry to hold exactly one tag");
+    };
+
+    // Both sides are tables, this only fails on a field whose default value cannot hold
+    // the sub-table that the environment sets.
+    deep_merge(field, value).map_err(|path| not_a_table(&entry_path(target, tag, &path)))
+}
+
+/// Returns the default entries of `target`, each one a single-key table `{tag: table}`.
+fn default_entries(target: &Target) -> Result<&[JsonValue]> {
+    match &target.default {
+        None | Some(JsonValue::Null) => Ok(&[]),
+        Some(JsonValue::Array(entries)) => Ok(entries),
+        Some(_) => invalid_default(target, "expected an array of tables"),
+    }
+}
+
+/// Returns the tag of `entry`, which must map one tag to a table.
+fn single_tag(target: &Target, entry: &Value) -> Result<String> {
+    let ValueKind::Table(table) = &entry.kind else {
+        return invalid_default(target, "expected an array of tables");
+    };
+
+    let mut keys = table.keys();
+    let (Some(tag), None) = (keys.next(), keys.next()) else {
+        return invalid_default(target, "expected each entry to hold exactly one tag");
+    };
+    if !matches!(&table[tag].kind, ValueKind::Table(_)) {
+        return invalid_default(target, "expected each entry to map its tag to a table");
+    }
+
+    Ok(tag.clone())
+}
+
+/// Groups the overrides by tag, nesting every field path into a table.
+fn group_entries(target: &Target, entries: Vec<(String, Value)>) -> Result<Map<String, Value>> {
+    // Every entry but `key` itself is addressed as `<key>.<tag>.<field>...`.
+    let offset = target.key.len() + 1;
+    let mut groups: Map<String, Value> = Map::new();
+
+    for (key, value) in entries {
+        if key == target.key {
+            return TaggedSeqMissingTagSnafu {
+                key,
+                target: target.key.clone(),
+            }
+            .fail();
+        }
+
+        let rest = key[offset..].to_string();
+        let Some((tag, field)) = rest.split_once('.') else {
+            return TaggedSeqMissingFieldSnafu {
+                key,
+                tag: rest,
+                target: target.key.clone(),
+            }
+            .fail();
+        };
+
+        if is_index_like(tag) {
+            return TaggedSeqNumericTagSnafu {
+                key,
+                tag: tag.to_string(),
+                target: target.key.clone(),
+            }
+            .fail();
+        }
+
+        let field = nest(field, value);
+        match groups.get_mut(tag) {
+            Some(group) => deep_merge(group, field)
+                .map_err(|path| overrides_conflict(&key, &entry_path(target, tag, &path)))?,
+            None => {
+                groups.insert(tag.to_string(), field);
+            }
+        }
+    }
+
+    Ok(groups)
+}
+
+/// Whether `tag` looks like a position in the sequence rather than an engine name.
+///
+/// `config` also accepts negative and floating point looking path segments (`[-1]`,
+/// `[1e3]`), so anything that starts like a number is reported as a misplaced index;
+/// otherwise those variables would only fail later with the generic `unknown variant`
+/// error from serde.
+fn is_index_like(tag: &str) -> bool {
+    let mut chars = tag.chars();
+    match chars.next() {
+        Some(first) if first.is_ascii_digit() => true,
+        Some('-') | Some('+') => {
+            matches!(chars.next(), Some(next) if next.is_ascii_digit() || next == '.')
+        }
+        _ => false,
+    }
+}
+
+/// Builds a table whose key path is the dotted `path`, e.g. `a.b` and `v` give
+/// `{a: {b: v}}`.
+fn nest(path: &str, value: Value) -> Value {
+    let mut table = Map::new();
+    match path.split_once('.') {
+        Some((head, tail)) => {
+            table.insert(head.to_string(), nest(tail, value));
+        }
+        None => {
+            table.insert(path.to_string(), value);
+        }
+    }
+
+    Value::new(None, ValueKind::Table(table))
+}
+
+/// Merges `incoming` into `base`, recursively for tables. Returns the dotted path of
+/// the field that cannot be merged into, when `base` holds a non-table value where
+/// `incoming` needs a table.
+fn deep_merge(base: &mut Value, incoming: Value) -> std::result::Result<(), String> {
+    merge_at(base, incoming, "")
+}
+
+fn merge_at(base: &mut Value, incoming: Value, path: &str) -> std::result::Result<(), String> {
+    match incoming.kind {
+        ValueKind::Table(incoming) => {
+            let ValueKind::Table(base) = &mut base.kind else {
+                return Err(path.to_string());
+            };
+            for (field, value) in incoming {
+                let field_path = if path.is_empty() {
+                    field.clone()
+                } else {
+                    format!("{path}.{field}")
+                };
+                // A non-table override replaces whatever the key already holds. It is inserted
+                // as is, so the value keeps the origin it was collected with.
+                if matches!(value.kind, ValueKind::Table(_)) {
+                    match base.get_mut(&field) {
+                        Some(existing) => merge_at(existing, value, &field_path)?,
+                        None => {
+                            base.insert(field, value);
+                        }
+                    }
+                } else {
+                    base.insert(field, value);
+                }
+            }
+            Ok(())
+        }
+        kind => {
+            // A non-table override replaces whatever the key already holds.
+            base.kind = kind;
+            Ok(())
+        }
+    }
+}
+
+fn invalid_default<T>(target: &Target, reason: &str) -> Result<T> {
+    TaggedSeqInvalidDefaultSnafu {
+        target: target.key.clone(),
+        reason: reason.to_string(),
+    }
+    .fail()
+}
+
+/// Builds the dotted path of `path` below the entry `tag` of `target`.
+fn entry_path(target: &Target, tag: &str, path: &str) -> String {
+    if path.is_empty() {
+        format!("{}.{}", target.key, tag)
+    } else {
+        format!("{}.{}.{}", target.key, tag, path)
+    }
+}
+
+/// Error for a field whose default value is not a table, so nothing can be set below it.
+fn not_a_table(path: &str) -> Error {
+    TaggedSeqNotATableSnafu {
+        path: path.to_string(),
+    }
+    .build()
+}
+
+/// Error for a key that sets a field below a path another key sets as a whole.
+fn overrides_conflict(key: &str, path: &str) -> Error {
+    TaggedSeqOverridesConflictSnafu {
+        key: key.to_string(),
+        path: path.to_string(),
+    }
+    .build()
+}
+
+/// Converts a default value parsed from the JSON serialization of the options.
+fn json_to_value(json: &JsonValue) -> Value {
+    let kind = match json {
+        JsonValue::Null => ValueKind::Nil,
+        JsonValue::Bool(value) => ValueKind::Boolean(*value),
+        JsonValue::Number(value) => match (value.as_i64(), value.as_u64()) {
+            (Some(value), _) => ValueKind::I64(value),
+            (None, Some(value)) => ValueKind::U64(value),
+            (None, None) => ValueKind::Float(value.as_f64().unwrap_or_default()),
+        },
+        JsonValue::String(value) => ValueKind::String(value.clone()),
+        JsonValue::Array(values) => ValueKind::Array(values.iter().map(json_to_value).collect()),
+        JsonValue::Object(values) => ValueKind::Table(
+            values
+                .iter()
+                .map(|(key, value)| (key.clone(), json_to_value(value)))
+                .collect(),
+        ),
+    };
+
+    Value::new(None, kind)
+}
+
+#[cfg(test)]
+mod tests {
+    use serde_json::json;
+
+    use super::*;
+
+    const KEY: &str = "region_engine";
+    const KEYS: &[&str] = &[KEY];
+
+    /// The default sequence used by the tests: two exported entries, `mito` before `file`,
+    /// with one nested table to override.
+    fn defaults() -> JsonValue {
+        json!({
+            "region_engine": [
+                {
+                    "mito": {
+                        "global_write_buffer_size": "64MB",
+                        "index": {
+                            "aux_path": "/default/aux",
+                            "cache_percent": 30
+                        }
+                    }
+                },
+                {"file": {"data_dir": "/default/data"}}
+            ],
+            "data_home": "/default/data_home"
+        })
+    }
+
+    /// A fixed set of variables, so the tests never touch the process environment and can
+    /// run in parallel.
+    fn env(vars: &[(&str, &str)]) -> Environment {
+        let mut vars_map = Map::new();
+        for (key, value) in vars {
+            vars_map.insert((*key).to_string(), (*value).to_string());
+        }
+
+        Environment::default()
+            .separator("__")
+            .try_parsing(true)
+            .source(Some(vars_map))
+    }
+
+    fn collect_with(vars: &[(&str, &str)], defaults: &JsonValue) -> Map<String, Value> {
+        TaggedSeqEnvSource::new(env(vars), KEYS, defaults)
+            .collect()
+            .unwrap()
+    }
+
+    fn collect(vars: &[(&str, &str)]) -> Map<String, Value> {
+        collect_with(vars, &defaults())
+    }
+
+    fn to_json(collected: Map<String, Value>) -> JsonValue {
+        let mut map = serde_json::Map::new();
+        for (key, value) in collected {
+            map.insert(key, value.try_deserialize().unwrap());
+        }
+
+        JsonValue::Object(map)
+    }
+
+    fn collect_json(vars: &[(&str, &str)]) -> JsonValue {
+        to_json(collect(vars))
+    }
+
+    fn collect_json_with(vars: &[(&str, &str)], defaults: &JsonValue) -> JsonValue {
+        to_json(collect_with(vars, defaults))
+    }
+
+    fn collect_error_with(vars: &[(&str, &str)], defaults: &JsonValue) -> String {
+        match TaggedSeqEnvSource::new(env(vars), KEYS, defaults).collect() {
+            Ok(_) => panic!("expected an error, got {vars:?}"),
+            Err(ConfigError::Message(message)) => message,
+            Err(other) => panic!("expected a message error, got: {other}"),
+        }
+    }
+
+    fn collect_error(vars: &[(&str, &str)]) -> String {
+        collect_error_with(vars, &defaults())
+    }
+
+    #[test]
+    fn test_collect_overrides_a_scalar_field() {
+        let collected = collect_json(&[("REGION_ENGINE__MITO__GLOBAL_WRITE_BUFFER_SIZE", "128MB")]);
+
+        let engines = collected["region_engine"].as_array().unwrap();
+        assert_eq!(engines.len(), 2);
+        assert_eq!(engines[0]["mito"]["global_write_buffer_size"], "128MB");
+        // The fields next to the override keep their default value.
+        assert_eq!(engines[0]["mito"]["index"]["aux_path"], "/default/aux");
+        assert_eq!(engines[1]["file"]["data_dir"], "/default/data");
+    }
+
+    #[test]
+    fn test_collect_overrides_a_nested_field() {
+        let collected = collect_json(&[("REGION_ENGINE__MITO__INDEX__AUX_PATH", "/tmp/aux")]);
+
+        assert_eq!(
+            collected["region_engine"][0]["mito"]["index"]["aux_path"],
+            "/tmp/aux"
+        );
+        assert_eq!(
+            collected["region_engine"][0]["mito"]["index"]["cache_percent"],
+            30
+        );
+        assert_eq!(
+            collected["region_engine"][0]["mito"]["global_write_buffer_size"],
+            "64MB"
+        );
+    }
+
+    #[test]
+    fn test_collect_merges_several_variables_of_one_entry() {
+        let collected = collect_json(&[
+            ("REGION_ENGINE__MITO__INDEX__AUX_PATH", "/tmp/aux"),
+            ("REGION_ENGINE__MITO__INDEX__CACHE_PERCENT", "50"),
+        ]);
+
+        assert_eq!(
+            collected["region_engine"][0]["mito"]["index"]["aux_path"],
+            "/tmp/aux"
+        );
+        assert_eq!(
+            collected["region_engine"][0]["mito"]["index"]["cache_percent"],
+            50
+        );
+    }
+
+    #[test]
+    fn test_collect_appends_unknown_tags_in_tag_order() {
+        let collected = collect_json(&[
+            (
+                "REGION_ENGINE__METRIC__FLUSH_METADATA_REGION_INTERVAL",
+                "1m",
+            ),
+            ("REGION_ENGINE__AAA__DATA_DIR", "/tmp/aaa"),
+        ]);
+
+        let engines = collected["region_engine"].as_array().unwrap();
+        assert_eq!(engines.len(), 4);
+        // The default entries stay first and keep their order.
+        assert_eq!(engines[0]["mito"]["global_write_buffer_size"], "64MB");
+        assert_eq!(engines[1]["file"]["data_dir"], "/default/data");
+        // Entries that only the environment mentions are appended, sorted by tag.
+        assert_eq!(engines[2], json!({"aaa": {"data_dir": "/tmp/aaa"}}));
+        assert_eq!(
+            engines[3],
+            json!({"metric": {"flush_metadata_region_interval": "1m"}})
+        );
+    }
+
+    #[test]
+    fn test_collect_keeps_the_default_sequence_without_overrides() {
+        let collected = collect_json(&[("DATA_HOME", "/srv/data")]);
+
+        assert_eq!(collected["region_engine"], defaults()["region_engine"]);
+        assert_eq!(collected["data_home"], "/srv/data");
+    }
+
+    #[test]
+    fn test_collect_removes_the_flat_keys_of_the_sequence() {
+        let collected = collect(&[
+            ("REGION_ENGINE__MITO__INDEX__AUX_PATH", "/tmp/aux"),
+            ("REGION_ENGINE_EXTRA__MITO__AUX_PATH", "/tmp/other"),
+        ]);
+
+        // A leftover `region_engine.x` key would be written after the sequence and destroy
+        // it again, so all of them must be consumed by the merge.
+        assert!(
+            !collected
+                .keys()
+                .any(|key| key.starts_with("region_engine.")),
+            "flat keys left in the collected map: {:?}",
+            collected.keys().collect::<Vec<_>>()
+        );
+        // A key that only shares the prefix is not part of the sequence.
+        assert!(collected.contains_key("region_engine_extra.mito.aux_path"));
+    }
+
+    #[test]
+    fn test_collect_without_default_entries() {
+        let collected = collect_json_with(
+            &[
+                (
+                    "REGION_ENGINE__METRIC__FLUSH_METADATA_REGION_INTERVAL",
+                    "1m",
+                ),
+                ("REGION_ENGINE__FILE__DATA_DIR", "/tmp/data"),
+            ],
+            &json!({}),
+        );
+
+        assert_eq!(
+            collected["region_engine"],
+            json!([
+                {"file": {"data_dir": "/tmp/data"}},
+                {"metric": {"flush_metadata_region_interval": "1m"}}
+            ])
+        );
+    }
+
+    #[test]
+    fn test_collect_with_null_default_entries() {
+        let collected = collect_json_with(
+            &[("REGION_ENGINE__FILE__DATA_DIR", "/tmp/data")],
+            &json!({"region_engine": null}),
+        );
+
+        assert_eq!(
+            collected["region_engine"],
+            json!([{"file": {"data_dir": "/tmp/data"}}])
+        );
+    }
+
+    #[test]
+    fn test_collect_rejects_setting_the_whole_sequence() {
+        let message = collect_error(&[("REGION_ENGINE", "mito")]);
+
+        assert!(
+            message.contains("cannot be overridden as a whole"),
+            "{message}"
+        );
+        assert!(
+            message.contains("region_engine__<TAG>__<FIELD>"),
+            "{message}"
+        );
+    }
+
+    #[test]
+    fn test_collect_rejects_a_numeric_index() {
+        let message = collect_error(&[("REGION_ENGINE__0__MITO__INDEX__AUX_PATH", "/tmp/aux")]);
+
+        assert!(message.contains("numeric index"), "{message}");
+        assert!(message.contains("addressed by tag name"), "{message}");
+    }
+
+    #[test]
+    fn test_collect_rejects_a_tag_without_field() {
+        let message = collect_error(&[("REGION_ENGINE__MITO", "mito")]);
+
+        assert!(message.contains("needs a field"), "{message}");
+    }
+
+    #[test]
+    fn test_collect_rejects_conflicting_overrides() {
+        let message = collect_error(&[
+            ("REGION_ENGINE__MITO__INDEX", "not-a-table"),
+            ("REGION_ENGINE__MITO__INDEX__AUX_PATH", "/tmp/aux"),
+        ]);
+
+        assert!(
+            message.contains("another environment variable sets 'region_engine.mito.index'"),
+            "{message}"
+        );
+    }
+
+    #[test]
+    fn test_collect_rejects_a_field_below_a_scalar_default() {
+        let message = collect_error(&[("REGION_ENGINE__FILE__DATA_DIR__SUB", "/tmp/sub")]);
+
+        assert!(message.contains("region_engine.file.data_dir"), "{message}");
+        assert!(
+            message.contains("default value is not a table"),
+            "{message}"
+        );
+    }
+
+    #[test]
+    fn test_collect_rejects_a_scalar_default_sequence() {
+        let message = collect_error_with(
+            &[("REGION_ENGINE__MITO__INDEX__AUX_PATH", "/tmp/aux")],
+            &json!({"region_engine": "not-an-array"}),
+        );
+
+        assert!(
+            message.contains("not a sequence of single-key tables"),
+            "{message}"
+        );
+    }
+
+    #[test]
+    fn test_collect_rejects_a_default_entry_with_two_tags() {
+        let message = collect_error_with(
+            &[("REGION_ENGINE__MITO__INDEX__AUX_PATH", "/tmp/aux")],
+            &json!({"region_engine": [{"mito": {}, "file": {}}]}),
+        );
+
+        assert!(message.contains("exactly one tag"), "{message}");
+    }
+
+    #[test]
+    fn test_take_entries_keeps_unrelated_keys() {
+        let mut collected = Map::new();
+        for key in [
+            "region_engine",
+            "region_engine.mito.index",
+            "region_engine_extra.mito",
+            "other",
+        ] {
+            collected.insert(
+                (*key).to_string(),
+                Value::new(None, ValueKind::String("v".to_string())),
+            );
+        }
+
+        let entries = take_entries(&mut collected, "region_engine");
+
+        assert_eq!(
+            entries
+                .iter()
+                .map(|(key, _)| key.as_str())
+                .collect::<Vec<_>>(),
+            vec!["region_engine", "region_engine.mito.index"]
+        );
+        assert!(!collected.contains_key("region_engine"));
+        assert!(!collected.contains_key("region_engine.mito.index"));
+        assert!(collected.contains_key("region_engine_extra.mito"));
+        assert!(collected.contains_key("other"));
+    }
+
+    #[test]
+    fn test_is_index_like() {
+        for tag in ["0", "12", "-1", "+1", "1e3", "-1.5"] {
+            assert!(is_index_like(tag), "{tag} should look like an index");
+        }
+
+        for tag in ["mito", "file", "metric", "", "m1", "-", "+"] {
+            assert!(!is_index_like(tag), "{tag} should not look like an index");
+        }
+    }
+}

src/datanode/src/config.rs

@@ -173,6 +173,10 @@ impl Configurable for DatanodeOptions {
             "wal.broker_endpoints",
         ])
     }
+
+    fn env_tagged_seq_keys() -> Option<&'static [&'static str]> {
+        Some(&["region_engine"])
+    }
 }
 
 #[allow(clippy::large_enum_variant)]

src/standalone/src/options.rs

@@ -137,6 +137,10 @@ impl Configurable for StandaloneOptions {
             "pending_rows_batcher.logical_table.protocols",
         ])
     }
+
+    fn env_tagged_seq_keys() -> Option<&'static [&'static str]> {
+        Some(&["region_engine"])
+    }
 }
 
 /// The [`StandaloneOptions`] is only defined in `standalone` crate,