mz_storage_types/

dyncfgs.rs

// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

//! Dyncfgs used by the storage layer. Despite their name, these can be used
//! "statically" during rendering, or dynamically within timely operators.

use mz_dyncfg::{Config, ConfigSet, ParameterScope};
use std::time::Duration;

/// When dataflows observe an invariant violation it is either due to a bug or due to the cluster
/// being shut down. This configuration defines the amount of time to wait before panicking the
/// process, which will register the invariant violation.
pub const CLUSTER_SHUTDOWN_GRACE_PERIOD: Config<Duration> = Config::new(
    "storage_cluster_shutdown_grace_period",
    Duration::from_secs(10 * 60),
    "When dataflows observe an invariant violation it is either due to a bug or due to \
        the cluster being shut down. This configuration defines the amount of time to \
        wait before panicking the process, which will register the invariant violation.",
    ParameterScope::Replica,
);

// Flow control

/// Whether rendering should use `mz_join_core` rather than DD's `JoinCore::join_core`.
/// Configuration for basic hydration backpressure.
pub const DELAY_SOURCES_PAST_REHYDRATION: Config<bool> = Config::new(
    "storage_dataflow_delay_sources_past_rehydration",
    // This was original `false`, but it is not enabled everywhere.
    true,
    "Whether or not to delay sources producing values in some scenarios \
        (namely, upsert) till after rehydration is finished",
    ParameterScope::Environment,
);

/// Whether storage dataflows should suspend execution while downstream operators are still
/// processing data.
pub const SUSPENDABLE_SOURCES: Config<bool> = Config::new(
    "storage_dataflow_suspendable_sources",
    true,
    "Whether storage dataflows should suspend execution while downstream operators are still \
        processing data.",
    ParameterScope::Environment,
);

// Controller

/// When enabled, force-downgrade the controller's since handle on the shard
/// during shard finalization.
pub const STORAGE_DOWNGRADE_SINCE_DURING_FINALIZATION: Config<bool> = Config::new(
    "storage_downgrade_since_during_finalization",
    // This was original `false`, but it is not enabled everywhere.
    true,
    "When enabled, force-downgrade the controller's since handle on the shard\
    during shard finalization",
    ParameterScope::Environment,
);

/// The interval of time to keep when truncating the replica metrics history.
pub const REPLICA_METRICS_HISTORY_RETENTION_INTERVAL: Config<Duration> = Config::new(
    "replica_metrics_history_retention_interval",
    Duration::from_secs(60 * 60 * 24 * 30), // 30 days
    "The interval of time to keep when truncating the replica metrics history.",
    ParameterScope::Environment,
);

/// The interval of time to keep when truncating the wallclock lag history.
pub const WALLCLOCK_LAG_HISTORY_RETENTION_INTERVAL: Config<Duration> = Config::new(
    "wallclock_lag_history_retention_interval",
    Duration::from_secs(60 * 60 * 24 * 30), // 30 days
    "The interval of time to keep when truncating the wallclock lag history.",
    ParameterScope::Environment,
);

/// The interval of time to keep when truncating the wallclock lag histogram.
pub const WALLCLOCK_GLOBAL_LAG_HISTOGRAM_RETENTION_INTERVAL: Config<Duration> = Config::new(
    "wallclock_global_lag_histogram_retention_interval",
    Duration::from_secs(60 * 60 * 24 * 30), // 30 days
    "The interval of time to keep when truncating the wallclock lag histogram.",
    ParameterScope::Environment,
);

// Kafka

/// Rules for enriching the `client.id` property of Kafka clients with
/// additional data.
///
/// The configuration value must be a JSON array of objects containing keys
/// named `pattern` and `payload`, both of type string. Rules are checked in the
/// order they are defined. The rule's pattern must be a regular expression
/// understood by the Rust `regex` crate. If the rule's pattern matches the
/// address of any broker in the connection, then the payload is appended to the
/// client ID. A rule's payload is always prefixed with `-`, to separate it from
/// the preceding data in the client ID.
pub const KAFKA_CLIENT_ID_ENRICHMENT_RULES: Config<fn() -> serde_json::Value> = Config::new(
    "kafka_client_id_enrichment_rules",
    || serde_json::json!([]),
    "Rules for enriching the `client.id` property of Kafka clients with additional data.",
    ParameterScope::Environment,
);

/// The maximum time we will wait before re-polling rdkafka to see if new partitions/data are
/// available.
pub const KAFKA_POLL_MAX_WAIT: Config<Duration> = Config::new(
    "kafka_poll_max_wait",
    Duration::from_secs(1),
    "The maximum time we will wait before re-polling rdkafka to see if new partitions/data are \
    available.",
    ParameterScope::Replica,
);

/// How often Kafka sources recommit offsets that have not changed, so that brokers do not expire
/// them. Must stay well below the broker's `offsets.retention.minutes`. Zero disables it.
pub const KAFKA_OFFSET_COMMIT_REFRESH_INTERVAL: Config<Duration> = Config::new(
    "kafka_offset_commit_refresh_interval",
    Duration::from_secs(10 * 60),
    "How often Kafka sources recommit offsets that have not changed, which must stay well below \
    the broker's offsets.retention.minutes. Zero disables it.",
    ParameterScope::Replica,
);

/// Whether to check the low watermark for Kafka sources and error if the start offset/resume
/// upper has been compacted away.
/// Environment-scoped because it decides whether a definite error is emitted.
/// Replicas of one cluster disagreeing would write different collection
/// contents, so the value has to be coherent across them.
pub const KAFKA_LOW_WATERMARK_CHECK: Config<bool> = Config::new(
    "kafka_low_watermark_check",
    true,
    "Whether to check the low watermark for Kafka sources and error if the start \
    offset/resume upper has been compacted away.",
    ParameterScope::Environment,
);

pub const KAFKA_DEFAULT_AWS_PRIVATELINK_ENDPOINT_IDENTIFICATION_ALGORITHM: Config<&'static str> =
    Config::new(
        "kafka_default_aws_privatelink_endpoint_identification_algorithm",
        // Default to no hostname verification, which is the default in versions of `librdkafka <1.9.2`.
        "none",
        "The value we set for the 'ssl.endpoint.identification.algorithm' option in the Kafka \
    Connection config. default: 'none'",
        ParameterScope::Environment,
    );

pub const KAFKA_BUFFERED_EVENT_RESIZE_THRESHOLD_ELEMENTS: Config<usize> = Config::new(
    "kafka_buffered_event_resize_threshold_elements",
    1000,
    "In the Kafka sink operator we might need to buffer messages before emitting them. As a \
        performance optimization we reuse the buffer allocations, but shrink it to retain at \
        most this number of elements.",
    ParameterScope::Replica,
);

/// Sets retry.backoff.ms in librdkafka for sources and sinks.
/// See <https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html>
pub const KAFKA_RETRY_BACKOFF: Config<Duration> = Config::new(
    "kafka_retry_backoff",
    Duration::from_millis(100),
    "Sets retry.backoff.ms in librdkafka for sources and sinks.",
    ParameterScope::Replica,
);

/// Sets retry.backoff.max.ms in librdkafka for sources and sinks.
/// See <https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html>
pub const KAFKA_RETRY_BACKOFF_MAX: Config<Duration> = Config::new(
    "kafka_retry_backoff_max",
    Duration::from_secs(1),
    "Sets retry.backoff.max.ms in librdkafka for sources and sinks.",
    ParameterScope::Replica,
);

/// Sets reconnect.backoff.ms in librdkafka for sources and sinks.
/// See <https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html>
pub const KAFKA_RECONNECT_BACKOFF: Config<Duration> = Config::new(
    "kafka_reconnect_backoff",
    Duration::from_millis(100),
    "Sets reconnect.backoff.ms in librdkafka for sources and sinks.",
    ParameterScope::Replica,
);

/// Sets reconnect.backoff.max.ms in librdkafka for sources and sinks.
/// We default to 30s instead of 10s to avoid constant reconnection attempts in the event of
/// auth changes or unavailability.
/// See <https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html>
pub const KAFKA_RECONNECT_BACKOFF_MAX: Config<Duration> = Config::new(
    "kafka_reconnect_backoff_max",
    Duration::from_secs(30),
    "Sets reconnect.backoff.max.ms in librdkafka for sources and sinks.",
    ParameterScope::Replica,
);

/// Sets message.max.bytes in librdkafka for Kafka sink producers.
/// Maximum Kafka protocol request message size. Producer-side, this controls
/// the maximum size of a single message (including framing) that the client
/// will allow. Defaults to the librdkafka default of 1,000,000 bytes.
/// See <https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html>
pub const KAFKA_SINK_MESSAGE_MAX_BYTES: Config<usize> = Config::new(
    "kafka_sink_message_max_bytes",
    1_000_000,
    "Sets message.max.bytes in librdkafka for Kafka sink producers.",
    ParameterScope::Environment,
);

/// Sets batch.size in librdkafka for Kafka sink producers.
/// Maximum size (in bytes) of all messages batched in one MessageSet, including
/// protocol framing overhead. Defaults to the librdkafka default of 1,000,000
/// bytes.
/// See <https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html>
pub const KAFKA_SINK_BATCH_SIZE: Config<usize> = Config::new(
    "kafka_sink_batch_size",
    1_000_000,
    "Sets batch.size in librdkafka for Kafka sink producers.",
    ParameterScope::Environment,
);

/// Sets batch.num.messages in librdkafka for Kafka sink producers.
/// Maximum number of messages batched in one MessageSet. Defaults to the
/// librdkafka default of 10,000 messages.
/// See <https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html>
pub const KAFKA_SINK_BATCH_NUM_MESSAGES: Config<usize> = Config::new(
    "kafka_sink_batch_num_messages",
    10_000,
    "Sets batch.num.messages in librdkafka for Kafka sink producers.",
    ParameterScope::Environment,
);

/// Whether Kafka sinks attach a `materialize-sink-id` header, holding the
/// sink's `GlobalId`, to every message they produce.
///
/// Environment-scoped because it changes the messages the sink produces.
pub const KAFKA_SINK_EMIT_SINK_ID_HEADER: Config<bool> = Config::new(
    "kafka_sink_emit_sink_id_header",
    true,
    "Whether Kafka sinks attach a `materialize-sink-id` header to every message they produce.",
    ParameterScope::Environment,
);

// MySQL

/// Replication heartbeat interval requested from the MySQL server.
pub const MYSQL_REPLICATION_HEARTBEAT_INTERVAL: Config<Duration> = Config::new(
    "mysql_replication_heartbeat_interval",
    Duration::from_secs(30),
    "Replication heartbeat interval requested from the MySQL server.",
    ParameterScope::Replica,
);

/// Whether to split snapshot reads of tables with a supported single-column
/// primary key into per-worker PK ranges. When disabled, each table is read
/// whole by a single worker.
pub static MYSQL_SOURCE_SNAPSHOT_PARALLELISM: Config<bool> = Config::new(
    "mysql_source_snapshot_parallelism",
    true,
    "Whether to split MySQL snapshot reads across workers by primary-key ranges.",
    ParameterScope::Replica,
);

/// Smallest estimated row count the MySQL snapshot partitioner attempts to subdivide.
pub static MYSQL_SOURCE_SNAPSHOT_PARTITION_MIN_ROWS: Config<usize> = Config::new(
    "mysql_source_snapshot_partition_min_rows",
    50_000,
    "Minimum estimated rows the MySQL snapshot partitioner attempts to split.",
    ParameterScope::Replica,
);

/// Cap on string primary key prefixes visited when attempting to partition a table for
/// parallel snapshotting in MySQL. This limits runtime on high-cardinality prefixes,
/// and will return correct but likely more skewed boundaries on budget exhaustion.
pub static MYSQL_SOURCE_SNAPSHOT_PARTITION_PROBED_PREFIXES_PER_BILLION_ROWS: Config<usize> =
    Config::new(
        "mysql_source_snapshot_partition_probed_prefixes_per_billion_rows",
        1_000,
        "Cap on MySQL snapshot PK-prefix partitioning probed prefixes per table, per billion \
     estimated rows; when exhausted, splitting stops early with coarser partition boundaries. \
     The per-table budget is additionally hard-capped at 5000.",
        ParameterScope::Replica,
    );

/// If the optimizer estimates the table has fewer rows than this, compute the exact row count
/// with `COUNT(*)`. Otherwise, report the `information_schema` estimate directly.
pub static MYSQL_SOURCE_SNAPSHOT_EXACT_COUNT_MAX_ROWS: Config<usize> = Config::new(
    "mysql_source_snapshot_exact_count_max_rows",
    1_000_000,
    "Maximum estimated table size for which MySQL snapshots compute an exact COUNT(*) \
     for the size gauge; larger tables report the information_schema estimate.",
    ParameterScope::Replica,
);

// Postgres

/// Interval to poll `confirmed_flush_lsn` to get a resumption lsn.
pub const PG_FETCH_SLOT_RESUME_LSN_INTERVAL: Config<Duration> = Config::new(
    "postgres_fetch_slot_resume_lsn_interval",
    Duration::from_millis(500),
    "Interval to poll `confirmed_flush_lsn` to get a resumption lsn.",
    ParameterScope::Replica,
);

/// Timeout for `CREATE_REPLICATION_SLOT` on the upstream server. Applied as the session's
/// `lock_timeout`, which bounds the wait for in-progress upstream transactions to end. This has
/// no effect when the upsteam is a physical standby.
pub const PG_SOURCE_REPLICATION_SLOT_CREATION_TIMEOUT: Config<Duration> = Config::new(
    "pg_source_replication_slot_creation_timeout",
    Duration::from_secs(60),
    "Timeout for creating a replication slot on the upstream PostgreSQL server. Applied as \
     `lock_timeout`, which bounds the wait for in-progress upstream transactions to end. \
     Has no effect with physical standbys.",
    ParameterScope::Replica,
);

/// Interval to re-validate the schemas of ingested tables.
pub const PG_SCHEMA_VALIDATION_INTERVAL: Config<Duration> = Config::new(
    "pg_schema_validation_interval",
    Duration::from_secs(15),
    "Interval to re-validate the schemas of ingested tables.",
    ParameterScope::Environment,
);

/// Controls behavior of PG Source when the upstream DB timeline changes. The default behavior
/// is to emit a definite error forcing source recreation. In cases of HA, the upstream DB may
/// provide guarantees of failover without loss of data (e.g. CloudSQL maintenance). Changing this
/// flag puts the onus on the customer to recreate the source if the upstream DB changes timeline
/// in a way that introduces data loss (e.g. manual failover, restore, etc.).
/// Environment-scoped because it decides whether a definite error is emitted.
/// Replicas of one cluster disagreeing would write different collection
/// contents, so the value has to be coherent across them.
pub static PG_SOURCE_VALIDATE_TIMELINE: Config<bool> = Config::new(
    "pg_source_validate_timeline",
    true,
    "Whether to treat a timeline switch as a definite error",
    ParameterScope::Environment,
);

/// Controls behavior of the SQL Server source when the upstream DB restore history changes. The
/// default behavior is to emit a definite error, forcing source recreation.  In cases of Always
/// On Availability Group (AOAG), the upstream DB may guarantee continuity without loss of data.
/// Changing this flag puts the onus on the customer to recreate the source if the upstream DB
/// changes in a way that introduces data loss.
/// Environment-scoped because it decides whether a definite error is emitted.
/// Replicas of one cluster disagreeing would write different collection
/// contents, so the value has to be coherent across them.
pub static SQL_SERVER_SOURCE_VALIDATE_RESTORE_HISTORY: Config<bool> = Config::new(
    "sql_server_source_validate_restore_history",
    true,
    "Whether to treat a restore history change as a definite error",
    ParameterScope::Environment,
);

// AWS

/// The AWS SDK's connect timeout on the AssumeRole prefetcher's STS calls.
/// Raise it if credential fetches time out on slow connections.
///
/// The default matches the SDK's own 3.1 second default, so behavior only
/// changes when an operator raises it.
///
/// Read once when a connection is set up. Running sinks are unaffected by
/// changes until their dataflow restarts.
pub const AWS_PREFETCH_STS_CONNECT_TIMEOUT: Config<Duration> = Config::new(
    "aws_prefetch_sts_connect_timeout",
    Duration::from_millis(3100),
    "Connect timeout for the AWS AssumeRole credentials prefetcher's STS calls.",
    ParameterScope::Replica,
);

// Networking

/// Whether or not to enforce that external connection addresses are global
/// (not private or local) when resolving them.
///
/// Read on both `environmentd` (purification, `COPY` planning) and the replica.
/// Deliberately environment-scoped even so: this is a security control, and a
/// per-replica override would weaken it for part of the environment only.
pub const ENFORCE_EXTERNAL_ADDRESSES: Config<bool> = Config::new(
    "storage_enforce_external_addresses",
    false,
    "Whether or not to enforce that external connection addresses are global \
          (not private or local) when resolving them",
    ParameterScope::Environment,
);

// Upsert

/// Whether or not to prevent buffering the entire _upstream_ snapshot in
/// memory when processing it in memory. This is generally understood to reduce
/// memory consumption.
///
/// When false, in general the memory utilization while processing the snapshot is:
/// # of snapshot updates + (# of unique keys in snapshot * N), where N is some small
/// integer number of buffers
///
/// When true, in general the memory utilization while processing the snapshot is:
/// # of snapshot updates + (RocksDB buffers + # of keys in batch produced by upstream) * # of
/// workers.
///
/// Without hydration flow control, which is not yet implemented, there are workloads that may
/// cause the latter to use more memory, which is why we offer this configuration.
pub const STORAGE_UPSERT_PREVENT_SNAPSHOT_BUFFERING: Config<bool> = Config::new(
    "storage_upsert_prevent_snapshot_buffering",
    true,
    "Prevent snapshot buffering in upsert.",
    ParameterScope::Replica,
);

/// Whether to enable the merge operator in upsert for the RocksDB backend.
pub const STORAGE_ROCKSDB_USE_MERGE_OPERATOR: Config<bool> = Config::new(
    "storage_rocksdb_use_merge_operator",
    true,
    "Use the native rocksdb merge operator where possible.",
    ParameterScope::Environment,
);

/// If `storage_upsert_prevent_snapshot_buffering` is true, this prevents the upsert
/// operator from buffering too many events from the upstream snapshot. In the absence
/// of hydration flow control, this could prevent certain workloads from causing egregiously
/// large writes to RocksDB.
pub const STORAGE_UPSERT_MAX_SNAPSHOT_BATCH_BUFFERING: Config<Option<usize>> = Config::new(
    "storage_upsert_max_snapshot_batch_buffering",
    None,
    "Limit snapshot buffering in upsert.",
    ParameterScope::Replica,
);

/// Allow the upsert-v2 stash to spill out of RSS. Off by default; while off,
/// the stash keeps everything resident.
///
/// The spill mechanism depends on the stash flavor
/// ([`ENABLE_UPSERT_CHUNKED_STASH`]):
///
/// * Chunked: sets the storage chunk spill gate
///   (`mz_timely_util::columnar::chunk::StorageSpill`), which only upsert's
///   chunks read, independently of compute's
///   `enable_column_paged_batcher_spill`. Spilled chunks draw on the one
///   shared pool budget, and the gate is consulted at every chunk commit, so
///   flips apply to running dataflows.
/// * Paged: gates the storage-owned column pager the stash and feedback
///   arrangement route their chains through, independently of compute's
///   `enable_column_paged_batcher_spill`. Captured at operator
///   construction, so flips apply to dataflows created after the change.
///
/// Enabling it also installs the process buffer pool (via compute's config
/// handler, which reads this flag from the aggregate dyncfg set), so
/// storage-only spilling needs no compute-side gate.
pub const ENABLE_UPSERT_PAGED_SPILL: Config<bool> = Config::new(
    "enable_upsert_paged_spill",
    false,
    "Allow the upsert-v2 stash to spill out of RSS, through the buffer pool (chunked stash \
     flavor) or the column pager (paged stash flavor).",
    ParameterScope::Replica,
);

/// Use the chunked stash flavor for the upsert-v2 operator: differential's
/// chunk merge batcher for the source stash and a spine of chunk batches for
/// the feedback arrangement, with a bulk-probe drain. When `false` (the
/// default), the paged flavor is used: the paged columnar merge batcher and a
/// `ValRowSpine`, with a cursor-based drain. See
/// `mz_storage::upsert_continual_feedback_v2::UpsertStashFlavor` for the
/// comparison.
///
/// Read at operator construction time; flips take effect on dataflows created
/// after the change. Only meaningful when [`ENABLE_UPSERT_V2`] is `true`.
/// Spilling in either flavor is gated by [`ENABLE_UPSERT_PAGED_SPILL`].
pub const ENABLE_UPSERT_CHUNKED_STASH: Config<bool> = Config::new(
    "enable_upsert_chunked_stash",
    false,
    "Use the chunk batcher and chunk spine for the upsert-v2 stash and feedback arrangement, \
     instead of the paged columnar merge batcher and ValRowSpine. Only meaningful when \
     enable_upsert_v2 is true.",
    ParameterScope::Replica,
);

// RocksDB

/// How many times to try to cleanup old RocksDB DB's on disk before giving up.
pub const STORAGE_ROCKSDB_CLEANUP_TRIES: Config<usize> = Config::new(
    "storage_rocksdb_cleanup_tries",
    5,
    "How many times to try to cleanup old RocksDB DB's on disk before giving up.",
    ParameterScope::Replica,
);

/// Delay interval when reconnecting to a source / sink after halt.
pub const STORAGE_SUSPEND_AND_RESTART_DELAY: Config<Duration> = Config::new(
    "storage_suspend_and_restart_delay",
    Duration::from_secs(5),
    "Delay interval when reconnecting to a source / sink after halt.",
    ParameterScope::Replica,
);

/// Whether to use the new continual feedback upsert operator.
pub const STORAGE_USE_CONTINUAL_FEEDBACK_UPSERT: Config<bool> = Config::new(
    "storage_use_continual_feedback_upsert",
    true,
    "Whether to use the new continual feedback upsert operator.",
    ParameterScope::Environment,
);

/// Whether to use the v2 upsert operator.
pub const ENABLE_UPSERT_V2: Config<bool> = Config::new(
    "enable_upsert_v2",
    false,
    "Whether to use the v2 upsert operator.",
    ParameterScope::Environment,
);

/// The interval at which the storage server performs maintenance tasks.
pub const STORAGE_SERVER_MAINTENANCE_INTERVAL: Config<Duration> = Config::new(
    "storage_server_maintenance_interval",
    Duration::from_millis(10),
    "The interval at which the storage server performs maintenance tasks. Zero enables maintenance on every iteration.",
    ParameterScope::Replica,
);

/// If set, iteratively search the progress topic for a progress record with increasing lookback.
pub const SINK_PROGRESS_SEARCH: Config<bool> = Config::new(
    "storage_sink_progress_search",
    true,
    "If set, iteratively search the progress topic for a progress record with increasing lookback.",
    ParameterScope::Environment,
);

/// Configure how to behave when trying to create an existing topic with specified configs.
pub const SINK_ENSURE_TOPIC_CONFIG: Config<&'static str> = Config::new(
    "storage_sink_ensure_topic_config",
    "skip",
    "If `skip`, don't check the config of existing topics; if `check`, fetch the config and \
    warn if it does not match the expected configs; if `alter`, attempt to change the upstream to \
    match the expected configs.",
    ParameterScope::Environment,
);

/// How far ahead of the remap upper the source `persist_sink` commits a ceiling while a
/// snapshotting export holds its frontier pinned, so that it can group updates into one batch.
///
/// The persist sink mints descriptions based on the data frontier. During a snapshot, that frontier
/// does not progress. Providing a lookahead instructs the minter to commit to a ceiling based
/// on the remap upper, which is used by the batch writers to group updates into a batch.
/// The minter honors the ceiling by minting descriptions at or beyond it. The whole snapshot and
/// the CDC events that were captured concurrently, up to the ceiling, become one description,
/// appended once.
///
/// Because the reclock stamps every update below the remap upper, the ceiling is ahead of the data
/// by at least the lookahead even if the export has seen no data. It still has to reach the writers
/// ahead of the rows stamped under the newest binding, since a builder only takes updates at times
/// it was opened for, so this wants to be at least one `timestamp_interval`. An update that outruns
/// it writes a batch of its own instead, which costs a batch rather than correctness.
///
/// Only applies to an export of a Postgres source (for now) that is snapshotting in this dataflow
/// incarnation, and only until the frontier moves off the time its snapshot occupies. Zero disables
/// committing ahead, leaving descriptions derived from the frontier alone and every timestamp
/// writing its own batch.
///
/// A nonzero value also switches the Postgres snapshot operator to emitting its rewind requests
/// as soon as the snapshot bound is known instead of after the copy completes, so the replication
/// stream is read while the snapshot runs, and lets each export's frontier advance once it has no
/// snapshot data left to emit. Replication data received for an export that is still snapshotting
/// is staged in the dataflow until its snapshot completes, which is the many-timestamp shape the
/// lookahead groups. The two are one setting because reading the stream concurrently without the
/// grouping keeps one batch builder per timestamp resident until each reaches the blob target size.
pub const STORAGE_PERSIST_SINK_DESCRIPTION_LOOKAHEAD: Config<Duration> = Config::new(
    "storage_persist_sink_description_lookahead",
    Duration::ZERO,
    "Determines how far past the remap upper the source persist sink will commit to a ceiling \
    in order to group data into one batch and one description, and whether the Postgres \
    replication stream is read while a snapshot runs. Zero disables both, leaving every \
    timestamp writing its own batch. Other values below the tick interval are clamped to the \
    tick interval.",
    ParameterScope::Environment,
);

/// Configure mz-ore overflowing type behavior.
pub const ORE_OVERFLOWING_BEHAVIOR: Config<&'static str> = Config::new(
    "ore_overflowing_behavior",
    "soft_panic",
    "Overflow behavior for Overflowing types. One of 'ignore', 'panic', 'soft_panic'.",
    ParameterScope::Environment,
);

/// The time after which we delete per-replica statistics (for sources and
/// sinks) after there have been no updates.
///
/// This time is opportunistic, statistics are not guaranteed to be deleted
/// after the retention time runs out.
pub const STATISTICS_RETENTION_DURATION: Config<Duration> = Config::new(
    "storage_statistics_retention_duration",
    Duration::from_secs(86_400), /* one day */
    "The time after which we delete per replica statistics (for sources and sinks) after there have been no updates.",
    ParameterScope::Environment,
);

/// Adds the full set of all storage `Config`s.
pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
    configs
        .add(&AWS_PREFETCH_STS_CONNECT_TIMEOUT)
        .add(&CLUSTER_SHUTDOWN_GRACE_PERIOD)
        .add(&DELAY_SOURCES_PAST_REHYDRATION)
        .add(&ENFORCE_EXTERNAL_ADDRESSES)
        .add(&KAFKA_BUFFERED_EVENT_RESIZE_THRESHOLD_ELEMENTS)
        .add(&KAFKA_CLIENT_ID_ENRICHMENT_RULES)
        .add(&KAFKA_DEFAULT_AWS_PRIVATELINK_ENDPOINT_IDENTIFICATION_ALGORITHM)
        .add(&KAFKA_LOW_WATERMARK_CHECK)
        .add(&KAFKA_OFFSET_COMMIT_REFRESH_INTERVAL)
        .add(&KAFKA_POLL_MAX_WAIT)
        .add(&KAFKA_RETRY_BACKOFF)
        .add(&KAFKA_RETRY_BACKOFF_MAX)
        .add(&KAFKA_RECONNECT_BACKOFF)
        .add(&KAFKA_RECONNECT_BACKOFF_MAX)
        .add(&KAFKA_SINK_MESSAGE_MAX_BYTES)
        .add(&KAFKA_SINK_BATCH_SIZE)
        .add(&KAFKA_SINK_BATCH_NUM_MESSAGES)
        .add(&KAFKA_SINK_EMIT_SINK_ID_HEADER)
        .add(&MYSQL_REPLICATION_HEARTBEAT_INTERVAL)
        .add(&MYSQL_SOURCE_SNAPSHOT_EXACT_COUNT_MAX_ROWS)
        .add(&MYSQL_SOURCE_SNAPSHOT_PARALLELISM)
        .add(&MYSQL_SOURCE_SNAPSHOT_PARTITION_MIN_ROWS)
        .add(&MYSQL_SOURCE_SNAPSHOT_PARTITION_PROBED_PREFIXES_PER_BILLION_ROWS)
        .add(&ORE_OVERFLOWING_BEHAVIOR)
        .add(&PG_FETCH_SLOT_RESUME_LSN_INTERVAL)
        .add(&PG_SCHEMA_VALIDATION_INTERVAL)
        .add(&PG_SOURCE_REPLICATION_SLOT_CREATION_TIMEOUT)
        .add(&PG_SOURCE_VALIDATE_TIMELINE)
        .add(&REPLICA_METRICS_HISTORY_RETENTION_INTERVAL)
        .add(&SINK_ENSURE_TOPIC_CONFIG)
        .add(&SINK_PROGRESS_SEARCH)
        .add(&SQL_SERVER_SOURCE_VALIDATE_RESTORE_HISTORY)
        .add(&STORAGE_DOWNGRADE_SINCE_DURING_FINALIZATION)
        .add(&STORAGE_ROCKSDB_CLEANUP_TRIES)
        .add(&STORAGE_ROCKSDB_USE_MERGE_OPERATOR)
        .add(&STORAGE_PERSIST_SINK_DESCRIPTION_LOOKAHEAD)
        .add(&STORAGE_SERVER_MAINTENANCE_INTERVAL)
        .add(&STORAGE_SUSPEND_AND_RESTART_DELAY)
        .add(&STORAGE_UPSERT_MAX_SNAPSHOT_BATCH_BUFFERING)
        .add(&STORAGE_UPSERT_PREVENT_SNAPSHOT_BUFFERING)
        .add(&STORAGE_USE_CONTINUAL_FEEDBACK_UPSERT)
        .add(&ENABLE_UPSERT_V2)
        .add(&SUSPENDABLE_SOURCES)
        .add(&ENABLE_UPSERT_PAGED_SPILL)
        .add(&ENABLE_UPSERT_CHUNKED_STASH)
        .add(&WALLCLOCK_GLOBAL_LAG_HISTOGRAM_RETENTION_INTERVAL)
        .add(&WALLCLOCK_LAG_HISTORY_RETENTION_INTERVAL)
        .add(&crate::sources::sql_server::CDC_CLEANUP_CHANGE_TABLE)
        .add(&crate::sources::sql_server::CDC_CLEANUP_CHANGE_TABLE_MAX_DELETES)
        .add(&crate::sources::sql_server::MAX_LSN_WAIT)
        .add(&crate::sources::sql_server::SNAPSHOT_PROGRESS_REPORT_INTERVAL)
        .add(&STATISTICS_RETENTION_DURATION)
}