diff --git a/ci/nightly/pipeline.template.yml b/ci/nightly/pipeline.template.yml index f25c31d5ef852..3e3fec1668155 100644 --- a/ci/nightly/pipeline.template.yml +++ b/ci/nightly/pipeline.template.yml @@ -3026,6 +3026,16 @@ steps: - group: Language key: language-tests steps: + - id: adbc + label: Arrow ADBC driver + depends_on: build-aarch64 + timeout_in_minutes: 30 + plugins: + - ./ci/plugins/mzcompose: + composition: adbc + agents: + queue: hetzner-aarch64-4cpu-8gb + - id: lang-csharp label: ":csharp:" depends_on: build-aarch64 diff --git a/console/types/materialize.d.ts b/console/types/materialize.d.ts index c47728f078583..25fc3e22c6e41 100644 --- a/console/types/materialize.d.ts +++ b/console/types/materialize.d.ts @@ -4019,6 +4019,7 @@ export interface MzTypePgMetadata { id: Generated; typinput: Generated; typreceive: Generated; + typsend: Generated; } export interface MzTypes { @@ -4260,6 +4261,7 @@ export interface PgTypeAllDatabases { typowner: number; typreceive: number; typrelid: number; + typsend: string; typtype: string; typtypmod: number; } diff --git a/doc/user/content/manage/terraform/get-started.md b/doc/user/content/manage/terraform/get-started.md index 0af61d91bd090..f62de7578f245 100644 --- a/doc/user/content/manage/terraform/get-started.md +++ b/doc/user/content/manage/terraform/get-started.md @@ -152,9 +152,21 @@ export MZ_PASSWORD= export MZ_HOST= ``` +{{< warning >}} +`MZ_HOST` is intended for use with Self-Managed Materialize. If `MZ_HOST` is set, +the provider connects directly to that host, using `MZ_USERNAME` and +`MZ_PASSWORD`, instead of connecting through Materialize Cloud. + +If you leave `MZ_HOST` exported and then run Terraform against Materialize +Cloud, the provider uses your self-managed settings instead, which usually +results in an `invalid password` error. + +Unset `MZ_HOST` before working with Materialize Cloud. +{{< /warning >}} + ```hcl provider "materialize" { - # Configuration will be read from MZ_HOST, MZ_PORT, MZ_USER, + # Configuration will be read from MZ_HOST, MZ_PORT, MZ_USERNAME, # MZ_DATABASE, MZ_PASSWORD, MZ_SSLMODE environment variables } ``` @@ -189,7 +201,7 @@ provider "materialize" { |-----------|-------------|---------------------|---------| | `host` | Materialize host address | `MZ_HOST` | - | | `port` | Materialize port | `MZ_PORT` | `6875` | -| `username` | Database username | `MZ_USER` | `materialize` | +| `username` | Database username | `MZ_USERNAME` | `materialize` | | `database` | Database name | `MZ_DATABASE` | `materialize` | | `password` | Database password | `MZ_PASSWORD` | - | | `sslmode` | SSL mode (`disable`, `require`, `verify-ca`, `verify-full`) | `MZ_SSLMODE` | `require` | diff --git a/doc/user/content/support.md b/doc/user/content/support.md index fc3a5a5ac243d..c0853ed50c367 100644 --- a/doc/user/content/support.md +++ b/doc/user/content/support.md @@ -11,24 +11,24 @@ aliases: Whether you're building, scaling, or debugging, we're here to help. -{{< note >}} -We're retiring access to Slack and Microsoft Teams workspaces. We've introduced the messenger app to streamline and scale how users engage with support — starting with AI-first assistance and seamless escalation paths when needed. -{{< /note >}} - ## Materialize Support -We offer multiple support pathways depending on your platform and plan: +We offer multiple support pathways depending on your platform and plan. + +### Support chat -### Messenger App +Our support chat is available directly on: -You can reach us directly from: - **materialize.com** - **Materialize Console** (for Cloud customers) -Our messenger now features Matty, our AI-powered support agent. Matty can help +The support chat features Matty, our AI-powered support agent. Matty can help answer questions, suggest documentation, and assist with troubleshooting. You -can also submit a request to **Talk to a human**. Just look for the messenger -widget on the bottom right. +can also submit a request to **Talk to a human**. Just look for the support chat +icon {{% icons/support-chat %}} in the bottom-right corner. + +**Don’t see the chat icon?** Make sure cookies are enabled for +**materialize.com**, then refresh the page. {{< tip >}} If you are an SLA Supported Customer who is experiencing an [**urgent** issue](#urgent-issues-for-sla-supported-customers-only), @@ -37,7 +37,7 @@ you can use the **Report Production Outage** option for fastest response. ### Email -You can also email us your questions at `support@materialize.com`. For +You can also email us your questions at `support@materialize.com`. If you are a Materialize Cloud SLA Supported Customer, email from your Materialize login email address. @@ -47,67 +47,38 @@ Connect with other users, share ideas, and get informal help from the [Materiali ## Urgent Issues (For SLA Supported Customers Only) -If you are an SLA Supported Customer who is experiencing an **urgent** issue, +If you are an SLA Supported Customer who is experiencing an **urgent** issue: -- Use the [messenger's](#messenger-app) **Report Production Outage** option for - fastest response. +- Use the **Report Production Outage** option in the [support + chat](#support-chat) for the fastest response. - Alternatively, you can email `support@materialize.com`. To ensure priority handling, include **urgent** in the subject line. - - For Materialize Cloud SLA Supported Customer, email from your Materialize - login email address. + - If you are a Materialize Cloud SLA Supported Customer, email from your + Materialize login email address. Use the following definitions to determine what priority to assign to your support ticket: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
PriorityResponse timeDefinition
UrgentRefer to your support contract - Critical severity error: an Error in the Platform that: (a) renders the Platform completely inoperative, or (b) makes Customer's use of material features of the Platform impossible, with no alternative available. -
HighRefer to your support contract - High severity error: an Error in the Platform that (a) has a high impact to key portions of the Platform, or (b) seriously impairs Customer's use of material features of the Platform and Customer cannot reasonably circumvent or avoid the Error on a temporary basis without the expenditure of significant time or effort. -
NormalRefer to your support contract - Medium severity error: an Error that has a medium-to-low impact on the Platform, but Customer can still access and use some functionality of the Platform. -
LowRefer to your support contract - Low severity error: an Error that has low-to-no impact on Customer's access to and use of the Platform. -
+| Priority | Response time | Definition | +|----------|---------------|------------| +| Urgent | Refer to your support contract | **Critical severity error:** an Error in the Platform that: (a) renders the Platform completely inoperative, or (b) makes Customer's use of material features of the Platform impossible, with no alternative available. | +| High | Refer to your support contract | **High severity error:** an Error in the Platform that (a) has a high impact to key portions of the Platform, or (b) seriously impairs Customer's use of material features of the Platform and Customer cannot reasonably circumvent or avoid the Error on a temporary basis without the expenditure of significant time or effort. | +| Normal | Refer to your support contract | **Medium severity error:** an Error that has a medium-to-low impact on the Platform, but Customer can still access and use some functionality of the Platform. | +| Low | Refer to your support contract | **Low severity error:** an Error that has low-to-no impact on Customer's access to and use of the Platform. | ## Share your feedback -Have ideas to improve Materialize? Use the messenger to start a conversation — you'll be guided to submit feature requests or report bugs. Your input goes straight to our Product and Engineering teams. +Have ideas to improve Materialize? Use the [support chat](#support-chat) to open +a ticket with support. Your input goes straight to our Product and Engineering +teams. ## Questions about your support plan? -We tailor support experiences based on your workspace and plan. To learn more about what's available to your team, reach out via the messenger or email us. +We tailor support experiences based on your workspace and plan. To learn more +about what's available to your team, reach out via the [support +chat](#support-chat) or email us. ## Checking Materialize Cloud service status diff --git a/doc/user/data/self_managed/self_managed_editions_table.yml b/doc/user/data/self_managed/self_managed_editions_table.yml index fb42574d93de9..b6c4b368ed6a4 100644 --- a/doc/user/data/self_managed/self_managed_editions_table.yml +++ b/doc/user/data/self_managed/self_managed_editions_table.yml @@ -13,7 +13,7 @@ rows: "Enterprise Edition": | [Per terms of your license](https://materialize.com/self-managed/enterprise-license/) "Community Edition": | - [Community slack or messenger app](/support/) + [Community slack or support chat](/support/) - "Feature": "**License**" "Enterprise Edition": | [Enterprise License - Contact Us](https://materialize.com/self-managed/enterprise-license/) diff --git a/doc/user/layouts/shortcodes/icons/support-chat.html b/doc/user/layouts/shortcodes/icons/support-chat.html new file mode 100644 index 0000000000000..9d3c9031d521c --- /dev/null +++ b/doc/user/layouts/shortcodes/icons/support-chat.html @@ -0,0 +1,4 @@ +{{- /* Resolve the src like render-image.html does, so the image works under + any baseURL (production /docs, PR previews, self-managed versions). */ -}} +{{- $parentPath := partial "relative-link.html" . -}} +Materialize support chat icon diff --git a/doc/user/layouts/shortcodes/self-managed/self-managed-editions-table.html b/doc/user/layouts/shortcodes/self-managed/self-managed-editions-table.html index 208ab0e64c106..b8478cadfc1fc 100644 --- a/doc/user/layouts/shortcodes/self-managed/self-managed-editions-table.html +++ b/doc/user/layouts/shortcodes/self-managed/self-managed-editions-table.html @@ -5,7 +5,7 @@ | --- | --- | --- | | **Maximum Usage Limits (Memory)** | None | 24 GiB | | **Maximum Usage Limits (Disk)** | None | 48 GiB | -| **[Support](/support/)** | [Per terms of your license] | [Community slack or messenger app](/support/) | +| **[Support](/support/)** | [Per terms of your license] | [Community slack or support chat](/support/) | | **License** | [Paid Enterprise License - Contact Us] | [BSL/Privacy Policy] | [Per terms of your license]: https://materialize.com/self-managed/enterprise-license/ diff --git a/doc/user/static/images/messenger-widget.png b/doc/user/static/images/messenger-widget.png new file mode 100644 index 0000000000000..58f6dbb7aa6ab Binary files /dev/null and b/doc/user/static/images/messenger-widget.png differ diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index a3e7aad96db01..1be7ba3c7ec07 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -338,6 +338,19 @@ def get_variable_system_parameters( "true", ["true", "false"], ), + VariableSystemParameter( + "enable_upsert_paged_spill", + "true", + ["true", "false"], + ), + # On by default so CI exercises the chunked stash flavor, which is + # off in production while it earns trust. Only meaningful when + # enable_upsert_v2 is true. + VariableSystemParameter( + "enable_upsert_chunked_stash", + "true", + ["true", "false"], + ), VariableSystemParameter( "enable_upsert_v2", "false", @@ -648,7 +661,6 @@ def get_default_system_parameters( "column_paged_batcher_spill_worker_count", "column_paged_batcher_eager_backing", "column_paged_batcher_pool_rss_target_fraction", - "enable_upsert_paged_spill", "enable_lgalloc_eager_reclamation", "lgalloc_background_interval", "lgalloc_file_growth_dampener", diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 65dda1529e115..194d7b2f31516 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -3138,6 +3138,7 @@ def __init__( "0.02", ] self.flags_with_values["enable_upsert_paged_spill"] = BOOLEAN_FLAG_VALUES + self.flags_with_values["enable_upsert_chunked_stash"] = BOOLEAN_FLAG_VALUES self.flags_with_values["column_chunk_compress_min_depth"] = [ "0", # compress every spilled body "1", # the default: fresh chunks store uncompressed diff --git a/src/adapter/src/catalog.rs b/src/adapter/src/catalog.rs index f21271b1d89f3..b29d44a2ad12f 100644 --- a/src/adapter/src/catalog.rs +++ b/src/adapter/src/catalog.rs @@ -3059,6 +3059,7 @@ mod tests { array: u32, input: u32, receive: u32, + send: u32, } struct PgOper { @@ -3098,7 +3099,7 @@ mod tests { let pg_type: BTreeMap<_, _> = query( &client, sql!( - "SELECT oid, typname, typtype::text, typelem, typarray, typinput::oid, typreceive::oid as typreceive FROM pg_type" + "SELECT oid, typname, typtype::text, typelem, typarray, typinput::oid, typreceive::oid as typreceive, typsend::oid as typsend FROM pg_type" ), &[], ) @@ -3114,6 +3115,7 @@ mod tests { array: row.get("typarray"), input: row.get("typinput"), receive: row.get("typreceive"), + send: row.get("typsend"), }; (oid, pg_type) }) @@ -3205,10 +3207,15 @@ mod tests { ty.oid, pg_ty.name, ty.name, ); - let (typinput_oid, typreceive_oid) = match &ty.details.pg_metadata { - None => (0, 0), - Some(pgmeta) => (pgmeta.typinput_oid, pgmeta.typreceive_oid), - }; + let (typinput_oid, typreceive_oid, typsend_oid) = + match &ty.details.pg_metadata { + None => (0, 0, 0), + Some(pgmeta) => ( + pgmeta.typinput_oid, + pgmeta.typreceive_oid, + pgmeta.typsend_oid, + ), + }; assert_eq!( typinput_oid, pg_ty.input, "type {} has typinput OID {:?} in mz but {:?} in pg", @@ -3219,6 +3226,15 @@ mod tests { "type {} has typreceive OID {:?} in mz but {:?} in pg", ty.name, typreceive_oid, pg_ty.receive, ); + // Unlike typinput and typreceive below, typsend is not also + // checked against `func_oids`. Nothing resolves a typsend OID + // to a name, so the corresponding `*send` functions are + // deliberately not registered as builtins. + assert_eq!( + typsend_oid, pg_ty.send, + "type {} has typsend OID {:?} in mz but {:?} in pg", + ty.name, typsend_oid, pg_ty.send, + ); if typinput_oid != 0 { assert!( func_oids.contains(&typinput_oid), diff --git a/src/adapter/src/catalog/builtin_table_updates.rs b/src/adapter/src/catalog/builtin_table_updates.rs index bb10328aabb56..b71258e3898f0 100644 --- a/src/adapter/src/catalog/builtin_table_updates.rs +++ b/src/adapter/src/catalog/builtin_table_updates.rs @@ -575,6 +575,7 @@ impl CatalogState { Datum::String(&id.to_string()), Datum::UInt32(pg_metadata.typinput_oid), Datum::UInt32(pg_metadata.typreceive_oid), + Datum::UInt32(pg_metadata.typsend_oid), ]), diff, )); diff --git a/src/adapter/src/catalog/open/builtin_schema_migration.rs b/src/adapter/src/catalog/open/builtin_schema_migration.rs index a96caf5281975..5ddc4839943e5 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration.rs @@ -324,6 +324,15 @@ static MIGRATIONS: LazyLock> = LazyLock::new(|| { MZ_INTERNAL_SCHEMA, "mz_kafka_source_tables", ), + // `mz_type_pg_metadata` gained a trailing `typsend` column. See the NOTE + // above: this version must stay at the workspace's current dev version + // until the change ships. + MigrationStep::replacement( + "26.40.0-dev.0", + CatalogItemType::Table, + MZ_INTERNAL_SCHEMA, + "mz_type_pg_metadata", + ), // Converting the connection-detail builtin tables to materialized views // changes their catalog fingerprint, so each needs an explicit // replacement step. diff --git a/src/catalog/src/builtin.rs b/src/catalog/src/builtin.rs index a3dbda548f476..6fb54b7867f51 100644 --- a/src/catalog/src/builtin.rs +++ b/src/catalog/src/builtin.rs @@ -1655,6 +1655,112 @@ mod tests { use super::*; + /// Recomputes `mz_pgrepr::regproc::NAMES` from the builtin function registry + /// and fails when the checked-in copy has drifted. It is checked in as data + /// because `mz-pgrepr` sits below this crate in the dependency graph and so + /// cannot read the registry itself. + /// + /// Run with `REWRITE=1` to splice the recomputed table back into + /// `src/pgrepr-consts/src/regproc.rs`. + #[mz_ore::test] + #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux` + fn test_regproc_names_match_builtin_functions() { + // `effective_search_path` unconditionally prepends these two, so a + // uniquely named function in either resolves from its bare name. + const IMPLICITLY_SEARCHED: &[&str] = &[MZ_CATALOG_SCHEMA, PG_CATALOG_SCHEMA]; + + // A bare name only identifies one OID when exactly one impl anywhere in + // the registry carries it, so count impls across schemas. + let mut impls_per_name: BTreeMap<&str, usize> = BTreeMap::new(); + for func in BUILTINS::funcs() { + *impls_per_name.entry(func.name).or_default() += func.inner.func_impls().len(); + } + + let mut expected: BTreeMap = BTreeMap::new(); + for func in BUILTINS::funcs() { + // Mirrors PostgreSQL's `regprocout`, which qualifies a name that + // would not resolve back to this OID on its own. + let rendered = + if impls_per_name[func.name] == 1 && IMPLICITLY_SEARCHED.contains(&func.schema) { + func.name.to_string() + } else { + format!("{}.{}", func.schema, func.name) + }; + for details in func.inner.func_impls() { + let previous = expected.insert(details.oid, rendered.clone()); + assert_eq!( + previous, None, + "two builtin functions share OID {}", + details.oid + ); + } + } + + let table: String = expected + .iter() + .map(|(oid, name)| format!(" ({}, \"{}\"),\n", oid, name)) + .collect(); + + if std::env::var_os("REWRITE").is_some() { + rewrite_regproc_names(&table); + return; + } + + let actual: BTreeMap = mz_pgrepr::regproc::NAMES + .iter() + .map(|(oid, name)| (*oid, name.to_string())) + .collect(); + + if actual != expected { + panic!( + "mz_pgrepr::regproc::NAMES has drifted from the builtin function \ + registry. Regenerate it with:\n\n \ + REWRITE=1 cargo test -p mz-catalog \ + test_regproc_names_match_builtin_functions\n" + ); + } + } + + /// Replaces the generated region of `mz_pgrepr::regproc::NAMES` with + /// `table`, leaving every other byte of the file alone. + /// + /// The path is relative to this crate's directory, which is the working + /// directory `cargo test` runs in. + /// + /// A splice anchored anywhere but the table would clobber the lookup + /// functions below it, so both markers have to appear exactly once. + fn rewrite_regproc_names(table: &str) { + const PATH: &str = "../pgrepr-consts/src/regproc.rs"; + const BEGIN: &str = " // BEGIN GENERATED\n"; + const END: &str = " // END GENERATED\n"; + + let contents = + std::fs::read_to_string(PATH).unwrap_or_else(|e| panic!("reading '{PATH}': {e}")); + for marker in [BEGIN, END] { + let count = contents.matches(marker).count(); + assert_eq!( + count, + 1, + "'{}' appears {} times in '{}', expected exactly once", + marker.trim(), + count, + PATH + ); + } + let begin = contents.find(BEGIN).expect("checked above") + BEGIN.len(); + let end = contents.find(END).expect("checked above"); + assert!( + begin <= end, + "'{}' precedes '{}' in '{}'", + END.trim(), + BEGIN.trim(), + PATH + ); + + let rewritten = format!("{}{}{}", &contents[..begin], table, &contents[end..]); + std::fs::write(PATH, rewritten).unwrap_or_else(|e| panic!("writing '{PATH}': {e}")); + } + #[mz_ore::test] #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux` fn test_builtin_type_schema() { diff --git a/src/catalog/src/builtin/mz_catalog.rs b/src/catalog/src/builtin/mz_catalog.rs index 2179e7a62af82..3874ac8d7d177 100644 --- a/src/catalog/src/builtin/mz_catalog.rs +++ b/src/catalog/src/builtin/mz_catalog.rs @@ -194,6 +194,7 @@ pub const TYPE_MZ_ACL_ITEM_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; diff --git a/src/catalog/src/builtin/mz_internal.rs b/src/catalog/src/builtin/mz_internal.rs index 463df5d4168cd..123c870f2fce3 100644 --- a/src/catalog/src/builtin/mz_internal.rs +++ b/src/catalog/src/builtin/mz_internal.rs @@ -705,6 +705,10 @@ pub static MZ_TYPE_PG_METADATA: LazyLock = LazyLock::new(|| Builti .with_column("id", SqlScalarType::String.nullable(false)) .with_column("typinput", SqlScalarType::Oid.nullable(false)) .with_column("typreceive", SqlScalarType::Oid.nullable(false)) + // NOTE: `pg_type_all_databases` still needs `COALESCE` on this column, + // because its `LEFT JOIN` against this table yields NULLs for types with + // no PostgreSQL metadata. + .with_column("typsend", SqlScalarType::Oid.nullable(false)) .finish(), column_comments: BTreeMap::new(), is_retained_metrics_object: false, @@ -4753,6 +4757,7 @@ pub static PG_TYPE_ALL_DATABASES: LazyLock = LazyLock::new(|| { .with_column("typcollation", SqlScalarType::Oid.nullable(false)) .with_column("typdefault", SqlScalarType::String.nullable(true)) .with_column("database_name", SqlScalarType::String.nullable(true)) + .with_column("typsend", SqlScalarType::RegProc.nullable(false)) .finish(), column_comments: BTreeMap::new(), sql: " @@ -4821,7 +4826,8 @@ SELECT -- MZ doesn't support COLLATE so typcollation is filled with 0 0::pg_catalog.oid AS typcollation, NULL::pg_catalog.text AS typdefault, - d.name as database_name + d.name as database_name, + COALESCE(mz_internal.mz_type_pg_metadata.typsend, 0)::pg_catalog.regproc AS typsend FROM mz_catalog.mz_types LEFT JOIN mz_internal.mz_type_pg_metadata ON mz_catalog.mz_types.id = mz_internal.mz_type_pg_metadata.id diff --git a/src/catalog/src/builtin/pg_catalog.rs b/src/catalog/src/builtin/pg_catalog.rs index ffc6f2484cf50..78d914d54a86e 100644 --- a/src/catalog/src/builtin/pg_catalog.rs +++ b/src/catalog/src/builtin/pg_catalog.rs @@ -52,6 +52,7 @@ pub const TYPE_BOOL: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 1242, typreceive_oid: 2436, + typsend_oid: 2437, }), }, }; @@ -66,6 +67,7 @@ pub const TYPE_BYTEA: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 1244, typreceive_oid: 2412, + typsend_oid: 2413, }), }, }; @@ -80,6 +82,7 @@ pub const TYPE_INT8: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 460, typreceive_oid: 2408, + typsend_oid: 2409, }), }, }; @@ -94,6 +97,7 @@ pub const TYPE_INT4: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 42, typreceive_oid: 2406, + typsend_oid: 2407, }), }, }; @@ -108,6 +112,7 @@ pub const TYPE_TEXT: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 46, typreceive_oid: 2414, + typsend_oid: 2415, }), }, }; @@ -122,6 +127,7 @@ pub const TYPE_OID: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 1798, typreceive_oid: 2418, + typsend_oid: 2419, }), }, }; @@ -136,6 +142,7 @@ pub const TYPE_FLOAT4: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 200, typreceive_oid: 2424, + typsend_oid: 2425, }), }, }; @@ -150,6 +157,7 @@ pub const TYPE_FLOAT8: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 214, typreceive_oid: 2426, + typsend_oid: 2427, }), }, }; @@ -166,6 +174,7 @@ pub const TYPE_BOOL_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -182,6 +191,7 @@ pub const TYPE_BYTEA_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -198,6 +208,7 @@ pub const TYPE_INT4_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -214,6 +225,7 @@ pub const TYPE_TEXT_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -230,6 +242,7 @@ pub const TYPE_INT8_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -246,6 +259,7 @@ pub const TYPE_FLOAT4_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -262,6 +276,7 @@ pub const TYPE_FLOAT8_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -278,6 +293,7 @@ pub const TYPE_OID_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -292,6 +308,7 @@ pub const TYPE_DATE: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 1084, typreceive_oid: 2468, + typsend_oid: 2469, }), }, }; @@ -306,6 +323,7 @@ pub const TYPE_TIME: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 1143, typreceive_oid: 2470, + typsend_oid: 2471, }), }, }; @@ -320,6 +338,7 @@ pub const TYPE_TIMESTAMP: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 1312, typreceive_oid: 2474, + typsend_oid: 2475, }), }, }; @@ -336,6 +355,7 @@ pub const TYPE_TIMESTAMP_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -352,6 +372,7 @@ pub const TYPE_DATE_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -368,6 +389,7 @@ pub const TYPE_TIME_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -382,6 +404,7 @@ pub const TYPE_TIMESTAMPTZ: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 1150, typreceive_oid: 2476, + typsend_oid: 2477, }), }, }; @@ -398,6 +421,7 @@ pub const TYPE_TIMESTAMPTZ_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -412,6 +436,7 @@ pub const TYPE_INTERVAL: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 1160, typreceive_oid: 2478, + typsend_oid: 2479, }), }, }; @@ -428,6 +453,7 @@ pub const TYPE_INTERVAL_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -442,6 +468,7 @@ pub const TYPE_NAME: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 34, typreceive_oid: 2422, + typsend_oid: 2423, }), }, }; @@ -458,6 +485,7 @@ pub const TYPE_NAME_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -472,6 +500,7 @@ pub const TYPE_NUMERIC: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 1701, typreceive_oid: 2460, + typsend_oid: 2461, }), }, }; @@ -488,6 +517,7 @@ pub const TYPE_NUMERIC_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -502,6 +532,7 @@ pub const TYPE_RECORD: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 2290, typreceive_oid: 2402, + typsend_oid: 2403, }), }, }; @@ -518,6 +549,7 @@ pub const TYPE_RECORD_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -532,6 +564,7 @@ pub const TYPE_UUID: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 2952, typreceive_oid: 2961, + typsend_oid: 2962, }), }, }; @@ -548,6 +581,7 @@ pub const TYPE_UUID_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -562,6 +596,7 @@ pub const TYPE_JSONB: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 3806, typreceive_oid: 3805, + typsend_oid: 3803, }), }, }; @@ -578,6 +613,7 @@ pub const TYPE_JSONB_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -592,6 +628,7 @@ pub const TYPE_ANY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 2294, typreceive_oid: 0, + typsend_oid: 0, }), }, }; @@ -606,6 +643,7 @@ pub const TYPE_ANYARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 2296, typreceive_oid: 2502, + typsend_oid: 2503, }), }, }; @@ -620,6 +658,7 @@ pub const TYPE_ANYELEMENT: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 2312, typreceive_oid: 0, + typsend_oid: 0, }), }, }; @@ -634,6 +673,7 @@ pub const TYPE_ANYNONARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 2777, typreceive_oid: 0, + typsend_oid: 0, }), }, }; @@ -648,6 +688,7 @@ pub const TYPE_ANYRANGE: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 3832, typreceive_oid: 0, + typsend_oid: 0, }), }, }; @@ -662,6 +703,7 @@ pub const TYPE_CHAR: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 1245, typreceive_oid: 2434, + typsend_oid: 2435, }), }, }; @@ -676,6 +718,7 @@ pub const TYPE_VARCHAR: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 1046, typreceive_oid: 2432, + typsend_oid: 2433, }), }, }; @@ -690,6 +733,7 @@ pub const TYPE_INT2: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 38, typreceive_oid: 2404, + typsend_oid: 2405, }), }, }; @@ -706,6 +750,7 @@ pub const TYPE_INT2_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -720,6 +765,7 @@ pub const TYPE_BPCHAR: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 1044, typreceive_oid: 2430, + typsend_oid: 2431, }), }, }; @@ -736,6 +782,7 @@ pub const TYPE_CHAR_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -752,6 +799,7 @@ pub const TYPE_VARCHAR_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -768,6 +816,7 @@ pub const TYPE_BPCHAR_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -782,6 +831,7 @@ pub const TYPE_REGPROC: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 44, typreceive_oid: 2444, + typsend_oid: 2445, }), }, }; @@ -798,6 +848,7 @@ pub const TYPE_REGPROC_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -812,6 +863,7 @@ pub const TYPE_REGTYPE: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 2220, typreceive_oid: 2454, + typsend_oid: 2455, }), }, }; @@ -828,6 +880,7 @@ pub const TYPE_REGTYPE_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -842,6 +895,7 @@ pub const TYPE_REGCLASS: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 2218, typreceive_oid: 2452, + typsend_oid: 2453, }), }, }; @@ -858,6 +912,7 @@ pub const TYPE_REGCLASS_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -872,6 +927,7 @@ pub const TYPE_INT2_VECTOR: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 40, typreceive_oid: 2410, + typsend_oid: 2411, }), }, }; @@ -888,6 +944,7 @@ pub const TYPE_INT2_VECTOR_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -902,6 +959,7 @@ pub const TYPE_ANYCOMPATIBLE: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 5086, typreceive_oid: 0, + typsend_oid: 0, }), }, }; @@ -916,6 +974,7 @@ pub const TYPE_ANYCOMPATIBLEARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 5088, typreceive_oid: 5090, + typsend_oid: 5091, }), }, }; @@ -930,6 +989,7 @@ pub const TYPE_ANYCOMPATIBLENONARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 5092, typreceive_oid: 0, + typsend_oid: 0, }), }, }; @@ -944,6 +1004,7 @@ pub const TYPE_ANYCOMPATIBLERANGE: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 5094, typreceive_oid: 0, + typsend_oid: 0, }), }, }; @@ -960,6 +1021,7 @@ pub const TYPE_INT4_RANGE: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 3834, typreceive_oid: 3836, + typsend_oid: 3837, }), }, }; @@ -976,6 +1038,7 @@ pub const TYPE_INT4_RANGE_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -992,6 +1055,7 @@ pub const TYPE_INT8_RANGE: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 3834, typreceive_oid: 3836, + typsend_oid: 3837, }), }, }; @@ -1008,6 +1072,7 @@ pub const TYPE_INT8_RANGE_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -1024,6 +1089,7 @@ pub const TYPE_DATE_RANGE: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 3834, typreceive_oid: 3836, + typsend_oid: 3837, }), }, }; @@ -1040,6 +1106,7 @@ pub const TYPE_DATE_RANGE_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -1056,6 +1123,7 @@ pub const TYPE_NUM_RANGE: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 3834, typreceive_oid: 3836, + typsend_oid: 3837, }), }, }; @@ -1072,6 +1140,7 @@ pub const TYPE_NUM_RANGE_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -1088,6 +1157,7 @@ pub const TYPE_TS_RANGE: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 3834, typreceive_oid: 3836, + typsend_oid: 3837, }), }, }; @@ -1104,6 +1174,7 @@ pub const TYPE_TS_RANGE_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -1120,6 +1191,7 @@ pub const TYPE_TSTZ_RANGE: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 3834, typreceive_oid: 3836, + typsend_oid: 3837, }), }, }; @@ -1136,6 +1208,7 @@ pub const TYPE_TSTZ_RANGE_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -1150,6 +1223,7 @@ pub const TYPE_ACL_ITEM: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 1031, typreceive_oid: 0, + typsend_oid: 0, }), }, }; @@ -1166,6 +1240,7 @@ pub const TYPE_ACL_ITEM_ARRAY: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 750, typreceive_oid: 2400, + typsend_oid: 2401, }), }, }; @@ -1180,6 +1255,7 @@ pub const TYPE_INTERNAL: BuiltinType = BuiltinType { pg_metadata: Some(CatalogTypePgMetadata { typinput_oid: 2304, typreceive_oid: 0, + typsend_oid: 0, }), }, }; @@ -1492,17 +1568,22 @@ pub static PG_TYPE: LazyLock = LazyLock::new(|| BuiltinView { .with_column("typelem", SqlScalarType::Oid.nullable(false)) .with_column("typarray", SqlScalarType::Oid.nullable(false)) .with_column("typinput", SqlScalarType::RegProc.nullable(true)) - .with_column("typreceive", SqlScalarType::Oid.nullable(false)) + // The underlying `mz_internal.pg_type_all_databases` keeps `typreceive` + // as `oid` because it backs a builtin index, and resolving a `regproc` to + // a name reads `current_database()`, which is unmaterializable. + .with_column("typreceive", SqlScalarType::RegProc.nullable(false)) .with_column("typnotnull", SqlScalarType::Bool.nullable(false)) .with_column("typbasetype", SqlScalarType::Oid.nullable(false)) .with_column("typtypmod", SqlScalarType::Int32.nullable(false)) .with_column("typcollation", SqlScalarType::Oid.nullable(false)) .with_column("typdefault", SqlScalarType::String.nullable(true)) + .with_column("typsend", SqlScalarType::RegProc.nullable(false)) .finish(), column_comments: BTreeMap::new(), sql: "SELECT oid, typname, typnamespace, typowner, typlen, typtype, typcategory, typdelim, typrelid, typelem, - typarray, typinput, typreceive, typnotnull, typbasetype, typtypmod, typcollation, typdefault + typarray, typinput, typreceive::pg_catalog.regproc AS typreceive, typnotnull, typbasetype, + typtypmod, typcollation, typdefault, typsend FROM mz_internal.pg_type_all_databases WHERE database_name IS NULL OR database_name = pg_catalog.current_database();", access: vec![PUBLIC_SELECT], diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index 41f927b4eb29b..69da191e92f4f 100644 --- a/src/compute/src/compute_state.rs +++ b/src/compute/src/compute_state.rs @@ -456,6 +456,12 @@ impl ComputeState { let compute_spill = ENABLE_COLUMN_PAGED_BATCHER_SPILL.get(config); let storage_spill = mz_storage_types::dyncfgs::ENABLE_UPSERT_PAGED_SPILL.get(config); + // Set compute's leg of the process-wide chunk spill gate. The + // gate ORs this leg with storage's, so chunks spill while either + // subsystem's flag is set. Storage's config application writes + // only its own leg, keeping the two flags from clobbering each + // other. + mz_timely_util::columnar::chunk::set_compute_spill_enabled(compute_spill); if !(compute_spill || storage_spill) { debug!("chunk spill: gates off, leaving the buffer pool uninstalled"); } else { diff --git a/src/interchange/src/json.rs b/src/interchange/src/json.rs index 34a9c1affd438..36470fc7b02dc 100644 --- a/src/interchange/src/json.rs +++ b/src/interchange/src/json.rs @@ -117,6 +117,10 @@ impl ToJson for TypedDatum<'_> { SqlScalarType::Int32 => json!(datum.unwrap_int32()), SqlScalarType::Int64 => json!(datum.unwrap_int64()), SqlScalarType::UInt16 => json!(datum.unwrap_uint16()), + // NOTE: `regproc` stays a number here even though the pgwire text + // encoding resolves it to a function name. `build_row_schema_json` + // below declares these columns as 4-byte unsigned integers, so a + // name would contradict the schema this module publishes for them. SqlScalarType::UInt32 | SqlScalarType::Oid | SqlScalarType::RegClass diff --git a/src/pgrepr-consts/src/lib.rs b/src/pgrepr-consts/src/lib.rs index a9fa3f684635c..b69d78d247031 100644 --- a/src/pgrepr-consts/src/lib.rs +++ b/src/pgrepr-consts/src/lib.rs @@ -10,3 +10,4 @@ //! Constants used in the representation of and serialization for PostgreSQL datums. pub mod oid; +pub mod regproc; diff --git a/src/pgrepr-consts/src/regproc.rs b/src/pgrepr-consts/src/regproc.rs new file mode 100644 index 0000000000000..d3ad9c9c76674 --- /dev/null +++ b/src/pgrepr-consts/src/regproc.rs @@ -0,0 +1,730 @@ +// 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. + +//! Text renderings of `regproc` values. +//! +//! PostgreSQL renders a `regproc` in text format through `regprocout`, which +//! resolves the OID to the function's name. Materialize has no +//! `CREATE FUNCTION`, so every function is a builtin and the OID to name +//! mapping is fixed when the binary is built. That is what makes a static +//! table viable: the text encoder has no catalog access and cannot gain any, +//! because the catalog crates depend on `mz-pgrepr` rather than the reverse, +//! and some encode paths run in `clusterd`, which has no catalog at all. +//! +//! [`NAMES`] is derived from Materialize's builtin function registry, reached +//! through `mz_catalog::builtin::BUILTINS::funcs()`, which is the authoritative +//! definition of every function OID. `mz_catalog::builtin`'s +//! `test_regproc_names_match_builtin_functions` recomputes the whole table from +//! that registry and fails when the two have drifted, so do not hand-edit +//! entries here. Add the function to the registry, then regenerate the table: +//! +//! ```shell +//! REWRITE=1 cargo test -p mz-catalog test_regproc_names_match_builtin_functions +//! ``` +//! +//! The OIDs are not invented by Materialize wherever PostgreSQL already assigns +//! one. `mz_sql::func`'s `PG_CATALOG_BUILTINS` records that its literal OIDs +//! were read out of a PostgreSQL 13 `pg_proc`, and points at +//! +//! for the same values. Everything else draws from [`crate::oid`], which carves +//! out `FIRST_UNPINNED_OID` upward for functions PostgreSQL ships without a +//! pinned OID, and `FIRST_MATERIALIZE_OID` upward for functions that exist only +//! in Materialize. + +/// Every builtin function OID paired with the text `regproc` renders it as, +/// sorted by OID. +/// +/// The name is schema-qualified whenever the bare name would not resolve back to +/// this OID, which is the rule `regprocout` applies. +/// +/// NOTE: The rendering assumes the default search path, which is all a static +/// table can do. A session with `mz_internal` or `mz_unsafe` on its +/// `search_path` makes those bare names resolvable, so the SQL cast from +/// `regproc` to `text` renders them bare where this table still qualifies them. +/// +/// NOTE: Reserved words are left unquoted here, so `left` renders as `left` +/// where PostgreSQL's `regprocout` renders `"left"`. This matches the SQL cast +/// from `regproc` to `text`, and the unquoted form still resolves because +/// `regprocin` accepts a reserved word either way. +pub const NAMES: &[(u32, &str)] = &[ + // BEGIN GENERATED + (34, "namein"), + (38, "int2in"), + (40, "int2vectorin"), + (42, "int4in"), + (44, "regprocin"), + (46, "textin"), + (89, "version"), + (200, "float4in"), + (214, "float8in"), + (376, "pg_catalog.string_to_array"), + (383, "array_cat"), + (384, "pg_catalog.array_to_string"), + (394, "pg_catalog.string_to_array"), + (395, "pg_catalog.array_to_string"), + (460, "int8in"), + (720, "pg_catalog.octet_length"), + (721, "get_byte"), + (723, "get_bit"), + (745, "current_user"), + (746, "session_user"), + (750, "array_in"), + (849, "position"), + (861, "current_database"), + (868, "strpos"), + (870, "pg_catalog.lower"), + (871, "pg_catalog.upper"), + (872, "initcap"), + (873, "pg_catalog.lpad"), + (875, "pg_catalog.ltrim"), + (876, "pg_catalog.rtrim"), + (877, "pg_catalog.substr"), + (878, "translate"), + (879, "pg_catalog.lpad"), + (881, "pg_catalog.ltrim"), + (882, "pg_catalog.rtrim"), + (883, "pg_catalog.substr"), + (884, "pg_catalog.btrim"), + (885, "pg_catalog.btrim"), + (936, "pg_catalog.substring"), + (937, "pg_catalog.substring"), + (938, "pg_catalog.generate_series"), + (939, "pg_catalog.generate_series"), + (940, "pg_catalog.mod"), + (941, "pg_catalog.mod"), + (947, "pg_catalog.mod"), + (1026, "pg_catalog.timezone"), + (1031, "aclitemin"), + (1044, "bpcharin"), + (1046, "varcharin"), + (1066, "pg_catalog.generate_series"), + (1067, "pg_catalog.generate_series"), + (1068, "pg_catalog.generate_series"), + (1069, "pg_catalog.generate_series"), + (1081, "format_type"), + (1084, "date_in"), + (1143, "time_in"), + (1150, "timestamptz_in"), + (1158, "to_timestamp"), + (1159, "pg_catalog.timezone"), + (1160, "interval_in"), + (1171, "pg_catalog.date_part"), + (1172, "pg_catalog.date_part"), + (1175, "justify_hours"), + (1178, "pg_catalog.date"), + (1192, "generate_subscripts"), + (1193, "pg_catalog.array_fill"), + (1194, "pg_catalog.log10"), + (1199, "pg_catalog.age"), + (1215, "obj_description"), + (1216, "col_description"), + (1217, "pg_catalog.date_trunc"), + (1218, "pg_catalog.date_trunc"), + (1242, "boolin"), + (1244, "byteain"), + (1245, "charin"), + (1268, "pg_catalog.parse_ident"), + (1269, "pg_column_size"), + (1282, "quote_ident"), + (1286, "pg_catalog.array_fill"), + (1295, "justify_days"), + (1299, "now"), + (1312, "timestamp_in"), + (1317, "pg_catalog.length"), + (1340, "pg_catalog.log"), + (1341, "pg_catalog.ln"), + (1342, "pg_catalog.round"), + (1343, "pg_catalog.trunc"), + (1344, "pg_catalog.sqrt"), + (1345, "cbrt"), + (1346, "pow"), + (1347, "pg_catalog.exp"), + (1365, "makeaclitem"), + (1368, "pg_catalog.power"), + (1374, "pg_catalog.octet_length"), + (1375, "pg_catalog.octet_length"), + (1381, "char_length"), + (1385, "pg_catalog.date_part"), + (1387, "pg_catalog.pg_get_constraintdef"), + (1394, "pg_catalog.abs"), + (1395, "pg_catalog.abs"), + (1396, "pg_catalog.abs"), + (1397, "pg_catalog.abs"), + (1398, "pg_catalog.abs"), + (1402, "current_schema"), + (1403, "current_schemas"), + (1481, "pg_catalog.log10"), + (1573, "pg_catalog.pg_get_ruledef"), + (1597, "pg_encoding_to_char"), + (1600, "asin"), + (1601, "acos"), + (1602, "atan"), + (1604, "sin"), + (1605, "cos"), + (1606, "tan"), + (1607, "cot"), + (1608, "degrees"), + (1609, "radians"), + (1619, "pg_typeof"), + (1620, "ascii"), + (1621, "chr"), + (1622, "repeat"), + (1637, "like_escape"), + (1640, "pg_catalog.pg_get_viewdef"), + (1641, "pg_catalog.pg_get_viewdef"), + (1642, "pg_get_userbyid"), + (1643, "pg_catalog.pg_get_indexdef"), + (1689, "aclexplode"), + (1701, "numeric_in"), + (1705, "pg_catalog.abs"), + (1707, "pg_catalog.round"), + (1708, "pg_catalog.round"), + (1710, "pg_catalog.trunc"), + (1711, "pg_catalog.ceil"), + (1712, "pg_catalog.floor"), + (1713, "pg_catalog.length"), + (1714, "convert_from"), + (1716, "pg_catalog.pg_get_expr"), + (1728, "pg_catalog.mod"), + (1730, "pg_catalog.sqrt"), + (1732, "pg_catalog.exp"), + (1734, "pg_catalog.ln"), + (1736, "pg_catalog.log"), + (1741, "pg_catalog.log"), + (1770, "pg_catalog.to_char"), + (1798, "oidin"), + (1810, "pg_catalog.bit_length"), + (1811, "pg_catalog.bit_length"), + (1922, "pg_catalog.has_table_privilege"), + (1923, "pg_catalog.has_table_privilege"), + (1924, "pg_catalog.has_table_privilege"), + (1925, "pg_catalog.has_table_privilege"), + (1926, "pg_catalog.has_table_privilege"), + (1927, "pg_catalog.has_table_privilege"), + (1928, "pg_stat_get_numscans"), + (1946, "encode"), + (1947, "decode"), + (2010, "pg_catalog.length"), + (2020, "pg_catalog.date_trunc"), + (2021, "pg_catalog.date_part"), + (2026, "pg_backend_pid"), + (2029, "pg_catalog.date"), + (2037, "pg_catalog.timezone"), + (2038, "pg_catalog.timezone"), + (2049, "pg_catalog.to_char"), + (2058, "pg_catalog.age"), + (2069, "pg_catalog.timezone"), + (2070, "pg_catalog.timezone"), + (2077, "pg_catalog.current_setting"), + (2079, "pg_table_is_visible"), + (2080, "pg_type_is_visible"), + (2081, "pg_function_is_visible"), + (2087, "replace"), + (2088, "split_part"), + (2091, "array_lower"), + (2092, "array_upper"), + (2100, "pg_catalog.avg"), + (2101, "pg_catalog.avg"), + (2102, "pg_catalog.avg"), + (2104, "pg_catalog.avg"), + (2105, "pg_catalog.avg"), + (2106, "pg_catalog.avg"), + (2107, "pg_catalog.sum"), + (2108, "pg_catalog.sum"), + (2109, "pg_catalog.sum"), + (2110, "pg_catalog.sum"), + (2111, "pg_catalog.sum"), + (2113, "pg_catalog.sum"), + (2114, "pg_catalog.sum"), + (2115, "pg_catalog.max"), + (2116, "pg_catalog.max"), + (2117, "pg_catalog.max"), + (2119, "pg_catalog.max"), + (2120, "pg_catalog.max"), + (2122, "pg_catalog.max"), + (2123, "pg_catalog.max"), + (2126, "pg_catalog.max"), + (2127, "pg_catalog.max"), + (2128, "pg_catalog.max"), + (2129, "pg_catalog.max"), + (2131, "pg_catalog.min"), + (2132, "pg_catalog.min"), + (2133, "pg_catalog.min"), + (2135, "pg_catalog.min"), + (2136, "pg_catalog.min"), + (2138, "pg_catalog.min"), + (2139, "pg_catalog.min"), + (2142, "pg_catalog.min"), + (2143, "pg_catalog.min"), + (2144, "pg_catalog.min"), + (2145, "pg_catalog.min"), + (2147, "pg_catalog.count"), + (2148, "pg_catalog.variance"), + (2149, "pg_catalog.variance"), + (2150, "pg_catalog.variance"), + (2151, "pg_catalog.variance"), + (2152, "pg_catalog.variance"), + (2154, "pg_catalog.stddev"), + (2155, "pg_catalog.stddev"), + (2156, "pg_catalog.stddev"), + (2157, "pg_catalog.stddev"), + (2158, "pg_catalog.stddev"), + (2167, "pg_catalog.ceiling"), + (2169, "pg_catalog.power"), + (2171, "pg_cancel_backend"), + (2176, "array_length"), + (2218, "regclassin"), + (2220, "regtypein"), + (2244, "pg_catalog.max"), + (2245, "pg_catalog.min"), + (2250, "pg_catalog.has_database_privilege"), + (2251, "pg_catalog.has_database_privilege"), + (2252, "pg_catalog.has_database_privilege"), + (2253, "pg_catalog.has_database_privilege"), + (2254, "pg_catalog.has_database_privilege"), + (2255, "pg_catalog.has_database_privilege"), + (2268, "pg_catalog.has_schema_privilege"), + (2269, "pg_catalog.has_schema_privilege"), + (2270, "pg_catalog.has_schema_privilege"), + (2271, "pg_catalog.has_schema_privilege"), + (2272, "pg_catalog.has_schema_privilege"), + (2273, "pg_catalog.has_schema_privilege"), + (2284, "pg_catalog.regexp_replace"), + (2285, "pg_catalog.regexp_replace"), + (2290, "record_in"), + (2294, "any_in"), + (2296, "anyarray_in"), + (2304, "internal_in"), + (2308, "pg_catalog.ceil"), + (2309, "pg_catalog.floor"), + (2311, "pg_catalog.md5"), + (2312, "anyelement_in"), + (2320, "pg_catalog.ceiling"), + (2321, "pg_catalog.md5"), + (2325, "pg_catalog.pg_relation_size"), + (2331, "mz_catalog.unnest"), + (2332, "pg_catalog.pg_relation_size"), + (2335, "pg_catalog.array_agg"), + (2400, "array_recv"), + (2402, "record_recv"), + (2404, "int2recv"), + (2406, "int4recv"), + (2408, "int8recv"), + (2410, "int2vectorrecv"), + (2412, "bytearecv"), + (2414, "textrecv"), + (2418, "oidrecv"), + (2422, "namerecv"), + (2424, "float4recv"), + (2426, "float8recv"), + (2430, "bpcharrecv"), + (2432, "varcharrecv"), + (2434, "charrecv"), + (2436, "boolrecv"), + (2444, "regprocrecv"), + (2452, "regclassrecv"), + (2454, "regtyperecv"), + (2460, "numeric_recv"), + (2462, "sinh"), + (2463, "cosh"), + (2464, "tanh"), + (2465, "asinh"), + (2466, "acosh"), + (2467, "atanh"), + (2468, "date_recv"), + (2470, "time_recv"), + (2474, "timestamp_recv"), + (2476, "timestamptz_recv"), + (2478, "interval_recv"), + (2502, "anyarray_recv"), + (2504, "pg_catalog.pg_get_ruledef"), + (2505, "pg_catalog.pg_get_viewdef"), + (2506, "pg_catalog.pg_get_viewdef"), + (2507, "pg_catalog.pg_get_indexdef"), + (2508, "pg_catalog.pg_get_constraintdef"), + (2509, "pg_catalog.pg_get_expr"), + (2517, "bool_and"), + (2518, "bool_or"), + (2560, "pg_postmaster_start_time"), + (2641, "pg_catalog.var_samp"), + (2642, "pg_catalog.var_samp"), + (2643, "pg_catalog.var_samp"), + (2644, "pg_catalog.var_samp"), + (2645, "pg_catalog.var_samp"), + (2705, "pg_catalog.pg_has_role"), + (2706, "pg_catalog.pg_has_role"), + (2707, "pg_catalog.pg_has_role"), + (2708, "pg_catalog.pg_has_role"), + (2709, "pg_catalog.pg_has_role"), + (2710, "pg_catalog.pg_has_role"), + (2711, "justify_interval"), + (2712, "pg_catalog.stddev_samp"), + (2713, "pg_catalog.stddev_samp"), + (2714, "pg_catalog.stddev_samp"), + (2715, "pg_catalog.stddev_samp"), + (2716, "pg_catalog.stddev_samp"), + (2718, "pg_catalog.var_pop"), + (2719, "pg_catalog.var_pop"), + (2720, "pg_catalog.var_pop"), + (2721, "pg_catalog.var_pop"), + (2722, "pg_catalog.var_pop"), + (2724, "pg_catalog.stddev_pop"), + (2725, "pg_catalog.stddev_pop"), + (2726, "pg_catalog.stddev_pop"), + (2727, "pg_catalog.stddev_pop"), + (2728, "pg_catalog.stddev_pop"), + (2763, "pg_catalog.regexp_matches"), + (2764, "pg_catalog.regexp_matches"), + (2765, "pg_catalog.regexp_split_to_table"), + (2766, "pg_catalog.regexp_split_to_table"), + (2767, "pg_catalog.regexp_split_to_array"), + (2768, "pg_catalog.regexp_split_to_array"), + (2777, "anynonarray_in"), + (2803, "pg_catalog.count"), + (2952, "uuid_in"), + (2961, "uuid_recv"), + (3058, "concat"), + (3059, "concat_ws"), + (3060, "left"), + (3061, "right"), + (3062, "reverse"), + (3100, "row_number"), + (3101, "rank"), + (3102, "dense_rank"), + (3106, "pg_catalog.lag"), + (3107, "pg_catalog.lag"), + (3108, "pg_catalog.lag"), + (3109, "pg_catalog.lead"), + (3110, "pg_catalog.lead"), + (3111, "pg_catalog.lead"), + (3112, "first_value"), + (3113, "last_value"), + (3138, "mz_catalog.has_type_privilege"), + (3139, "mz_catalog.has_type_privilege"), + (3140, "mz_catalog.has_type_privilege"), + (3141, "mz_catalog.has_type_privilege"), + (3142, "mz_catalog.has_type_privilege"), + (3143, "mz_catalog.has_type_privilege"), + (3159, "pg_catalog.pg_get_viewdef"), + (3166, "pg_size_pretty"), + (3167, "array_remove"), + (3207, "jsonb_array_length"), + (3208, "jsonb_each"), + (3210, "jsonb_typeof"), + (3219, "jsonb_array_elements"), + (3262, "jsonb_strip_nulls"), + (3267, "jsonb_agg"), + (3270, "jsonb_object_agg"), + (3271, "pg_catalog.jsonb_build_array"), + (3272, "pg_catalog.jsonb_build_array"), + (3273, "pg_catalog.jsonb_build_object"), + (3274, "pg_catalog.jsonb_build_object"), + (3277, "pg_catalog.array_position"), + (3278, "pg_catalog.array_position"), + (3294, "pg_catalog.current_setting"), + (3306, "jsonb_pretty"), + (3396, "pg_catalog.regexp_match"), + (3397, "pg_catalog.regexp_match"), + (3419, "sha224"), + (3420, "sha256"), + (3421, "sha384"), + (3422, "sha512"), + (3461, "make_timestamp"), + (3465, "jsonb_array_elements_text"), + (3538, "pg_catalog.string_agg"), + (3545, "pg_catalog.string_agg"), + (3696, "starts_with"), + (3778, "pg_tablespace_location"), + (3787, "to_jsonb"), + (3805, "jsonb_recv"), + (3806, "jsonb_in"), + (3810, "pg_is_in_recovery"), + (3832, "anyrange_in"), + (3834, "range_in"), + (3836, "range_recv"), + (3840, "pg_catalog.int4range"), + (3841, "pg_catalog.int4range"), + (3844, "pg_catalog.numrange"), + (3845, "pg_catalog.numrange"), + (3848, "pg_catalog.lower"), + (3849, "pg_catalog.upper"), + (3850, "isempty"), + (3851, "lower_inc"), + (3852, "upper_inc"), + (3853, "lower_inf"), + (3854, "upper_inf"), + (3931, "jsonb_object_keys"), + (3932, "jsonb_each_text"), + (3933, "pg_catalog.tsrange"), + (3934, "pg_catalog.tsrange"), + (3937, "pg_catalog.tstzrange"), + (3938, "pg_catalog.tstzrange"), + (3941, "pg_catalog.daterange"), + (3942, "pg_catalog.daterange"), + (3945, "pg_catalog.int8range"), + (3946, "pg_catalog.int8range"), + (4053, "pg_catalog.array_agg"), + (4350, "normalize"), + (5086, "anycompatible_in"), + (5088, "anycompatiblearray_in"), + (5090, "anycompatiblearray_recv"), + (5092, "anycompatiblenonarray_in"), + (5094, "anycompatiblerange_in"), + (6163, "bit_count"), + (6177, "pg_catalog.date_bin"), + (6178, "pg_catalog.date_bin"), + (6199, "pg_catalog.extract"), + (6200, "pg_catalog.extract"), + (6202, "pg_catalog.extract"), + (6203, "pg_catalog.extract"), + (6204, "pg_catalog.extract"), + (12000, "information_schema._pg_expandarray"), + (12001, "pg_catalog.digest"), + (12002, "pg_catalog.digest"), + (12003, "pg_catalog.hmac"), + (12004, "pg_catalog.hmac"), + (16386, "pg_catalog.ceil"), + (16387, "concat_agg"), + (16388, "csv_extract"), + (16389, "current_timestamp"), + (16390, "pg_catalog.floor"), + (16392, "list_append"), + (16393, "list_cat"), + (16394, "list_length_max"), + (16395, "list_length"), + (16396, "list_n_layers"), + (16397, "list_prepend"), + (16398, "pg_catalog.max"), + (16399, "pg_catalog.min"), + (16400, "mz_unsafe.mz_all"), + (16401, "mz_unsafe.mz_any"), + (16403, "mz_unsafe.mz_avg_promotion_internal_v1"), + (16404, "mz_unsafe.mz_avg_promotion_internal_v1"), + (16405, "mz_unsafe.mz_avg_promotion_internal_v1"), + (16407, "mz_environment_id"), + (16409, "mz_logical_timestamp"), + (16410, "mz_internal.mz_render_typmod"), + (16411, "mz_version"), + (16412, "regexp_extract"), + (16413, "repeat_row"), + (16414, "pg_catalog.round"), + (16416, "mz_catalog.unnest"), + (16434, "mz_unsafe.mz_sleep"), + (16435, "mz_internal.mz_session_id"), + (16436, "mz_uptime"), + (16440, "mz_row_size"), + (16441, "pg_catalog.max"), + (16442, "pg_catalog.min"), + (16443, "mz_unsafe.mz_avg_promotion_internal_v1"), + (16444, "list_agg"), + (16445, "mz_unsafe.mz_error_if_null"), + (16446, "pg_catalog.date_bin"), + (16447, "pg_catalog.date_bin"), + (16448, "list_remove"), + (16449, "pg_catalog.date_bin_hopping"), + (16450, "pg_catalog.date_bin_hopping"), + (16451, "pg_catalog.date_bin_hopping"), + (16452, "pg_catalog.date_bin_hopping"), + (16453, "mz_internal.mz_type_name"), + (16456, "map_length"), + (16457, "mz_unsafe.mz_panic"), + (16458, "mz_version_num"), + (16459, "pg_catalog.trunc"), + (16496, "pg_catalog.max"), + (16497, "pg_catalog.max"), + (16498, "pg_catalog.max"), + (16499, "pg_catalog.min"), + (16500, "pg_catalog.min"), + (16501, "pg_catalog.min"), + (16502, "pg_catalog.sum"), + (16503, "pg_catalog.sum"), + (16504, "pg_catalog.sum"), + (16505, "pg_catalog.avg"), + (16506, "pg_catalog.avg"), + (16507, "pg_catalog.avg"), + (16508, "pg_catalog.mod"), + (16509, "pg_catalog.mod"), + (16510, "pg_catalog.mod"), + (16511, "pg_catalog.stddev"), + (16512, "pg_catalog.stddev"), + (16513, "pg_catalog.stddev"), + (16514, "pg_catalog.stddev_pop"), + (16515, "pg_catalog.stddev_pop"), + (16516, "pg_catalog.stddev_pop"), + (16517, "pg_catalog.stddev_samp"), + (16518, "pg_catalog.stddev_samp"), + (16519, "pg_catalog.stddev_samp"), + (16520, "pg_catalog.variance"), + (16521, "pg_catalog.variance"), + (16522, "pg_catalog.variance"), + (16523, "pg_catalog.var_pop"), + (16524, "pg_catalog.var_pop"), + (16525, "pg_catalog.var_pop"), + (16526, "pg_catalog.var_samp"), + (16527, "pg_catalog.var_samp"), + (16528, "pg_catalog.var_samp"), + (16550, "mz_unsafe.mz_avg_promotion_internal_v1"), + (16551, "mz_unsafe.mz_avg_promotion_internal_v1"), + (16560, "mz_now"), + (16561, "pg_catalog.max"), + (16562, "pg_catalog.min"), + (16563, "pg_catalog.date"), + (16564, "pg_catalog.ceiling"), + (16565, "uuid_generate_v5"), + (16570, "mz_internal.make_mz_aclitem"), + (16571, "mz_internal.mz_aclitem_grantor"), + (16572, "mz_internal.mz_aclitem_grantee"), + (16573, "mz_internal.mz_aclitem_privileges"), + (16574, "mz_internal.is_rbac_enabled"), + (16575, "mz_internal.mz_acl_item_contains_privilege"), + (16576, "mz_internal.mz_validate_privileges"), + (16577, "mz_internal.mz_role_oid"), + (16578, "mz_internal.mz_minimal_name_qualification"), + (16581, "mz_internal.mz_resolve_object_name"), + (16582, "mz_internal.mz_global_id_to_name"), + (16584, "pg_catalog.parse_ident"), + (16585, "mz_internal.mz_database_oid"), + (16586, "mz_internal.mz_schema_oid"), + (16587, "mz_catalog.has_cluster_privilege"), + (16588, "mz_catalog.has_cluster_privilege"), + (16589, "mz_catalog.has_cluster_privilege"), + (16591, "mz_catalog.has_connection_privilege"), + (16592, "mz_catalog.has_connection_privilege"), + (16593, "mz_catalog.has_connection_privilege"), + (16594, "mz_catalog.has_connection_privilege"), + (16595, "mz_catalog.has_connection_privilege"), + (16596, "mz_catalog.has_connection_privilege"), + (16597, "mz_catalog.has_secret_privilege"), + (16598, "mz_catalog.has_secret_privilege"), + (16599, "mz_catalog.has_secret_privilege"), + (16600, "mz_catalog.has_secret_privilege"), + (16601, "mz_catalog.has_secret_privilege"), + (16602, "mz_catalog.has_secret_privilege"), + (16603, "mz_internal.mz_normalize_object_name"), + (16604, "mz_catalog.datediff"), + (16605, "mz_catalog.datediff"), + (16606, "mz_catalog.datediff"), + (16607, "mz_catalog.datediff"), + (16608, "mz_catalog.has_system_privilege"), + (16609, "mz_catalog.has_system_privilege"), + (16610, "mz_catalog.has_system_privilege"), + (16611, "try_parse_monotonic_iso8601_timestamp"), + (16612, "mz_internal.mz_role_oid_memberships"), + (16613, "mz_internal.mz_validate_role_privilege"), + (16614, "mz_is_superuser"), + (16615, "mz_internal.mz_format_privileges"), + (16616, "mz_internal.aclitem_grantor"), + (16617, "mz_internal.aclitem_grantee"), + (16618, "mz_internal.aclitem_privileges"), + (16619, "mz_internal.mz_aclexplode"), + (16620, "mz_catalog.has_role"), + (16621, "mz_catalog.has_role"), + (16622, "mz_catalog.has_role"), + (16623, "mz_catalog.has_role"), + (16624, "mz_catalog.has_role"), + (16625, "mz_catalog.has_role"), + (16626, "mz_unsafe.mz_avg_promotion"), + (16627, "mz_unsafe.mz_avg_promotion"), + (16628, "mz_unsafe.mz_avg_promotion"), + (16629, "mz_unsafe.mz_avg_promotion"), + (16630, "mz_unsafe.mz_avg_promotion"), + (16631, "mz_unsafe.mz_avg_promotion"), + (16632, "mz_unsafe.mz_avg_promotion"), + (16633, "mz_unsafe.mz_avg_promotion"), + (16634, "mz_unsafe.mz_avg_promotion"), + (16635, "mz_catalog.avg_internal_v1"), + (16636, "mz_catalog.avg_internal_v1"), + (16637, "mz_catalog.avg_internal_v1"), + (16638, "mz_catalog.avg_internal_v1"), + (16639, "mz_catalog.avg_internal_v1"), + (16640, "mz_catalog.avg_internal_v1"), + (16641, "mz_catalog.avg_internal_v1"), + (16642, "mz_catalog.avg_internal_v1"), + (16643, "mz_catalog.avg_internal_v1"), + (16644, "mz_catalog.constant_time_eq"), + (16645, "mz_catalog.constant_time_eq"), + (16646, "timezone_offset"), + (16647, "mz_catalog.pretty_sql"), + (16648, "mz_catalog.pretty_sql"), + (16649, "mz_internal.mz_name_rank"), + (16650, "mz_internal.mz_connection_oid"), + (16651, "mz_internal.mz_secret_oid"), + (16652, "map_build"), + (16653, "map_agg"), + (16654, "mz_catalog.unnest"), + (16655, "mz_internal.mz_normalize_schema_name"), + (16957, "current_catalog"), + (16958, "current_role"), + (16959, "user"), + (16987, "mz_catalog.crc32"), + (16988, "mz_catalog.crc32"), + (16989, "mz_catalog.kafka_murmur2"), + (16990, "mz_catalog.kafka_murmur2"), + (16991, "mz_catalog.seahash"), + (16992, "mz_catalog.seahash"), + (17069, "mz_internal.parse_catalog_id"), + (17070, "mz_internal.parse_catalog_privileges"), + (17073, "mz_internal.parse_catalog_create_sql"), + (17074, "mz_internal.redact_sql"), + (17075, "repeat_row_non_negative"), + (17086, "mz_internal.mz_session_role_memberships"), + (17090, "mz_internal.parse_catalog_acl_mode"), + (17093, "mz_unsafe.generate_series_unoptimized"), + (17094, "mz_unsafe.generate_series_unoptimized"), + (17100, "mz_internal.parse_catalog_audit_log_details"), + (17105, "mz_internal.parse_postgres_source_details"), + (17106, "mz_internal.parse_kafka_source_details"), + (17111, "mz_internal.parse_source_export_details"), + (17112, "mz_internal.parse_connection_details"), + (17113, "mz_aws_account_id"), + (17114, "mz_aws_external_id_prefix"), + (17115, "mz_aws_connection_role_arn"), + // END GENERATED +]; + +/// Returns the text that `regproc` OID `oid` renders as, or `None` when no +/// builtin function has that OID. +/// +/// PostgreSQL renders an unmatched OID in decimal and OID 0 as `-`. Neither is +/// in [`NAMES`], so callers handle both. +pub fn name(oid: u32) -> Option<&'static str> { + NAMES + .binary_search_by_key(&oid, |(entry_oid, _)| *entry_oid) + .ok() + .map(|i| NAMES[i].1) +} + +/// Returns the OID whose [`NAMES`] rendering is `name`. +/// +/// `Err` distinguishes no match from an ambiguous one. A rendering is ambiguous +/// when several overloads share it, in which case PostgreSQL's `regprocin` +/// likewise refuses to resolve it. The scan is linear because reads of this +/// direction only happen while decoding text-format input, which is rare +/// compared to encoding. +pub fn oid(name: &str) -> Result { + let mut found = None; + for (entry_oid, entry_name) in NAMES { + if *entry_name == name { + if found.is_some() { + return Err(NameLookupError::Ambiguous); + } + found = Some(*entry_oid); + } + } + found.ok_or(NameLookupError::NotFound) +} + +/// Why [`oid`] could not resolve a name. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum NameLookupError { + /// No builtin function renders as the given name. + NotFound, + /// Several builtin functions render as the given name. + Ambiguous, +} diff --git a/src/pgrepr/src/lib.rs b/src/pgrepr/src/lib.rs index 9e540f8ac3e1c..4d0c58a3d6c2f 100644 --- a/src/pgrepr/src/lib.rs +++ b/src/pgrepr/src/lib.rs @@ -23,6 +23,7 @@ mod types; mod value; pub mod oid; +pub mod regproc; pub use types::{ ANYCOMPATIBLELIST, ANYCOMPATIBLEMAP, LIST, MAP, Type, TypeConversionError, TypeFromOidError, diff --git a/src/pgrepr/src/regproc.rs b/src/pgrepr/src/regproc.rs new file mode 100644 index 0000000000000..438dd72ddaa6e --- /dev/null +++ b/src/pgrepr/src/regproc.rs @@ -0,0 +1,63 @@ +// 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. + +#![allow(missing_docs)] + +pub use mz_pgrepr_consts::regproc::*; + +#[cfg(test)] +mod tests { + use super::*; + + /// [`name`] binary searches [`NAMES`], so the table must be sorted by OID + /// and free of duplicate OIDs. + #[mz_ore::test] + fn names_is_sorted_by_oid() { + for window in NAMES.windows(2) { + assert!( + window[0].0 < window[1].0, + "NAMES is not sorted by OID: {} precedes {}", + window[0].0, + window[1].0, + ); + } + } + + #[mz_ore::test] + // `oid` scans the table, so covering every entry is quadratic and far too + // slow interpreted. + #[cfg_attr(miri, ignore)] + fn lookups_round_trip() { + for (entry_oid, entry_name) in NAMES { + assert_eq!(name(*entry_oid), Some(*entry_name)); + // Overloads sharing a rendering can only resolve back to one OID, + // so ambiguity is the expected outcome for those. + match oid(entry_name) { + Ok(resolved) => assert_eq!(resolved, *entry_oid), + Err(err) => assert_eq!(err, NameLookupError::Ambiguous), + } + } + assert_eq!(name(0), None); + assert_eq!(oid("no_such_function"), Err(NameLookupError::NotFound)); + } + + /// Spot-checks the `pg_type` I/O function renderings, so a regeneration that + /// dropped them fails loudly. + #[mz_ore::test] + fn type_io_functions_resolve() { + for (func_oid, expected) in [ + (1242, "boolin"), + (2436, "boolrecv"), + (2400, "array_recv"), + (2414, "textrecv"), + ] { + assert_eq!(name(func_oid), Some(expected), "regproc {func_oid}"); + } + } +} diff --git a/src/pgrepr/src/value.rs b/src/pgrepr/src/value.rs index aaf01b100dfe0..bbb20f23f8d0f 100644 --- a/src/pgrepr/src/value.rs +++ b/src/pgrepr/src/value.rs @@ -35,7 +35,7 @@ use uuid::Uuid; use crate::types::{NumericConstraints, UINT2, UINT4, UINT8}; use crate::value::error::{IntoDatumError, NulCharacterError}; -use crate::{Interval, Jsonb, Numeric, Type, UInt2, UInt4, UInt8}; +use crate::{Interval, Jsonb, Numeric, Type, UInt2, UInt4, UInt8, regproc}; pub mod error; pub mod interval; @@ -92,6 +92,12 @@ pub enum Value { Numeric(Numeric), /// An object identifier. Oid(u32), + /// An object identifier naming a function. + /// + /// Kept distinct from [`Value::Oid`] because the text encoding resolves the + /// OID to the function's name, matching PostgreSQL's `regprocout`. The + /// binary encoding is identical to an `oid`, as it is in PostgreSQL. + RegProc(u32), /// A sequence of heterogeneous values. Record(Vec>), /// A time. @@ -141,8 +147,12 @@ impl Value { (Datum::UInt8(c), SqlScalarType::PgLegacyChar) => Some(Value::Char(c)), (Datum::UInt16(u), SqlScalarType::UInt16) => Some(Value::UInt2(UInt2(u))), (Datum::UInt32(oid), SqlScalarType::Oid) => Some(Value::Oid(oid)), + (Datum::UInt32(oid), SqlScalarType::RegProc) => Some(Value::RegProc(oid)), + // NOTE: `regclass` and `regtype` collapse into `Value::Oid`, so they + // render as digits where PostgreSQL renders a name. Unlike `regproc` + // they can name objects a user created, which no static table can + // know. Resolution for those two lives in the SQL cast to `text`. (Datum::UInt32(oid), SqlScalarType::RegClass) => Some(Value::Oid(oid)), - (Datum::UInt32(oid), SqlScalarType::RegProc) => Some(Value::Oid(oid)), (Datum::UInt32(oid), SqlScalarType::RegType) => Some(Value::Oid(oid)), (Datum::UInt32(u), SqlScalarType::UInt32) => Some(Value::UInt4(UInt4(u))), (Datum::UInt64(u), SqlScalarType::UInt64) => Some(Value::UInt8(UInt8(u))), @@ -305,7 +315,7 @@ impl Value { }) })? } - Value::Oid(oid) => Datum::UInt32(oid), + Value::Oid(oid) | Value::RegProc(oid) => Datum::UInt32(oid), Value::Record(_) => { // This situation is handled gracefully by Value::decode; if we // wind up here it's a programming error. @@ -422,6 +432,14 @@ impl Value { }) .expect("provided closure never fails"), Value::Oid(oid) => strconv::format_uint32(buf, *oid), + // Mirrors PostgreSQL's `regprocout`. + Value::RegProc(oid) => match *oid { + 0 => strconv::format_string(buf, REGPROC_NULL), + oid => match regproc::name(oid) { + Some(name) => strconv::format_string(buf, name), + None => strconv::format_uint32(buf, oid), + }, + }, Value::Record(elems) => strconv::format_record(buf, elems, |buf, elem| match elem { None => Ok::<_, ()>(buf.write_null()), Some(elem) => Ok(elem.encode_text(buf.nonnull_buffer(), settings)), @@ -544,7 +562,9 @@ impl Value { Err("binary encoding of map types is not implemented".into()) } Value::Name(s) => s.to_sql(&PgType::NAME, buf), - Value::Oid(i) => i.to_sql(&PgType::OID, buf), + // PostgreSQL's `regprocsend` is `int4send`, so binary is identical to + // an `oid`. + Value::Oid(i) | Value::RegProc(i) => i.to_sql(&PgType::OID, buf), Value::Record(fields) => { let nfields = pg_len("record field length", fields.len())?; buf.put_i32(nfields); @@ -734,9 +754,8 @@ impl Value { strconv::parse_numeric(s)?, constraints.as_ref(), )?)), - Type::Oid | Type::RegClass | Type::RegProc | Type::RegType => { - Value::Oid(strconv::parse_oid(s)?) - } + Type::Oid | Type::RegClass | Type::RegType => Value::Oid(strconv::parse_oid(s)?), + Type::RegProc => Value::RegProc(parse_regproc(s)?), Type::Record(_) => { return Err("input of anonymous composite types is not implemented".into()); } @@ -844,9 +863,10 @@ impl Value { strconv::parse_numeric(s)?, constraints.as_ref(), )?)), - Type::Oid | Type::RegClass | Type::RegProc | Type::RegType => { + Type::Oid | Type::RegClass | Type::RegType => { packer.push(Datum::UInt32(strconv::parse_oid(s)?)) } + Type::RegProc => packer.push(Datum::UInt32(parse_regproc(s)?)), Type::Record(_) => { return Err("input of anonymous composite types is not implemented".into()); } @@ -933,9 +953,10 @@ impl Value { constraints.as_ref(), )?))) } - Type::Oid | Type::RegClass | Type::RegProc | Type::RegType => { + Type::Oid | Type::RegClass | Type::RegType => { u32::from_sql(ty.inner(), raw).map(Value::Oid) } + Type::RegProc => u32::from_sql(ty.inner(), raw).map(Value::RegProc), Type::Record(_) => Err("input of anonymous composite types is not implemented".into()), Type::Text => decode_binary_string(ty, raw).map(Value::Text), Type::BpChar { .. } => decode_binary_string(ty, raw).map(Value::BpChar), @@ -986,6 +1007,44 @@ impl Value { } } +/// The text PostgreSQL's `regprocout` emits for OID 0, and that `regprocin` +/// reads back as OID 0. +const REGPROC_NULL: &str = "-"; + +/// Parses the text format of a `regproc`, matching PostgreSQL's `regprocin`. +/// +/// Accepting names is what keeps a `COPY ... TO` rendering loadable by +/// `COPY ... FROM`, since the text encoding emits names. An overloaded name +/// renders schema-qualified and names every one of its impls, so it does not +/// round trip, which is true of PostgreSQL as well. +fn parse_regproc(s: &str) -> Result> { + if s == REGPROC_NULL { + return Ok(0); + } + match regproc::oid(s) { + Ok(oid) => Ok(oid), + // PostgreSQL refuses the same input rather than picking an overload. + Err(regproc::NameLookupError::Ambiguous) => { + Err(format!("more than one function named \"{}\"", s).into()) + } + Err(regproc::NameLookupError::NotFound) => match strconv::parse_oid(s) { + Ok(oid) => Ok(oid), + // Number-shaped input keeps its numeric diagnosis, which is what + // reports an out of range OID. Anything else was meant as a name, so + // report a missing function, matching PostgreSQL. + Err(err) if is_number_shaped(s) => Err(err.into()), + Err(_) => Err(format!("function \"{}\" does not exist", s).into()), + }, + } +} + +/// Whether `s` is shaped like a decimal number, ignoring whether it is in range. +fn is_number_shaped(s: &str) -> bool { + let digits = s.trim(); + let digits = digits.strip_prefix(['+', '-']).unwrap_or(digits); + !digits.is_empty() && digits.bytes().all(|b| b.is_ascii_digit()) +} + /// Session settings that affect how values are encoded as text. /// /// PostgreSQL's text output for some types depends on session state. Encoders @@ -1158,6 +1217,71 @@ mod tests { }); } + /// A `regproc` renders as the function's name in text format and as a bare + /// OID in binary, which is what PostgreSQL's `regprocout` and + /// `regprocsend` do. + #[mz_ore::test] + fn regproc_text_encoding_resolves_names() { + // 1242 is `boolin`, uniquely named. 1398 is one of six `abs` overloads, + // so PostgreSQL qualifies it. 99999 names no function. The qualified + // overload does not round trip, because its rendering names all six. + for (oid, expected, round_trips) in [ + (0, "-", true), + (1242, "boolin", true), + (1398, "pg_catalog.abs", false), + (99999, "99999", true), + (u32::MAX, "4294967295", true), + ] { + let mut buf = BytesMut::new(); + Value::from_datum(Datum::UInt32(oid), &SqlScalarType::RegProc) + .expect("a non-null datum encodes") + .encode_text(&mut buf, TextEncodeSettings::STABLE); + assert_eq!(str::from_utf8(&buf).unwrap(), expected, "regproc {oid}"); + + if round_trips { + let Value::RegProc(decoded) = Value::decode_text(&Type::RegProc, &buf).unwrap() + else { + panic!("decoding a regproc must yield Value::RegProc"); + }; + assert_eq!(decoded, oid, "text rendering {expected} did not round trip"); + } + + let mut binary = BytesMut::new(); + Value::RegProc(oid) + .encode_binary(&Type::RegProc, &mut binary) + .expect("regproc has a binary encoding"); + assert_eq!(&binary[..], &oid.to_be_bytes()[..], "regproc {oid} binary"); + } + } + + /// An overloaded name is an error rather than an arbitrary choice, which is + /// what PostgreSQL does too. + #[mz_ore::test] + fn regproc_text_decoding_rejects_ambiguous_names() { + assert_eq!( + Value::decode_text(&Type::RegProc, b"pg_catalog.abs") + .map(|_| ()) + .map_err(|e| e.to_string()) + .unwrap_err(), + "more than one function named \"pg_catalog.abs\"", + ); + assert_eq!( + Value::decode_text(&Type::RegProc, b"no_such_function") + .map(|_| ()) + .map_err(|e| e.to_string()) + .unwrap_err(), + "function \"no_such_function\" does not exist", + ); + let out_of_range = Value::decode_text(&Type::RegProc, b"99999999999999") + .map(|_| ()) + .map_err(|e| e.to_string()) + .unwrap_err(); + assert!( + out_of_range.starts_with("invalid input syntax for type oid"), + "unexpected error for an out of range OID: {out_of_range}", + ); + } + /// [`format_float_limited`] must match C's `%.*g`, which PostgreSQL's /// `float4out`/`float8out` use when `extra_float_digits` is zero or /// negative. diff --git a/src/sql/src/catalog.rs b/src/sql/src/catalog.rs index 18cc768faea9e..705a2f98816ec 100644 --- a/src/sql/src/catalog.rs +++ b/src/sql/src/catalog.rs @@ -1083,6 +1083,8 @@ pub struct CatalogTypePgMetadata { pub typinput_oid: u32, /// The OID of the `typreceive` function in PostgreSQL. pub typreceive_oid: u32, + /// The OID of the `typsend` function in PostgreSQL. + pub typsend_oid: u32, } /// Represents a reference to type in the catalog diff --git a/src/sql/src/plan/typeconv.rs b/src/sql/src/plan/typeconv.rs index 6b69af3b8f78d..6f09ee1c5cec7 100644 --- a/src/sql/src/plan/typeconv.rs +++ b/src/sql/src/plan/typeconv.rs @@ -123,6 +123,8 @@ impl From<[UnaryFunc; N]> for CastTemplate { /// A reg* type represents a specific type of object by oid. /// /// Casting from a string to a reg*: +/// - Accepts `-`, the absent reference, as OID 0 in all cases, matching +/// PostgreSQL's `parseDashOrOid`. /// - Accepts a string that looks like an OID and converts the value to the /// specified reg* type. This is available in all cases except explicitly /// casting text values to regclass (e.g. `SELECT '2'::text::regclass`) @@ -138,6 +140,9 @@ const STRING_REG_CAST_TEMPLATE: &str = " (SELECT CASE WHEN $1 IS NULL THEN NULL +-- Handle the absent reference, which the reg* text output spells `-`. PostgreSQL's +-- `parseDashOrOid` accepts it for every reg* type, including text to regclass. + WHEN $1 = '-' THEN 0::pg_catalog.oid::pg_catalog.{0} -- Handle OID-like input, if available via {2} WHEN {2} AND pg_catalog.substring($1, 1, 1) BETWEEN '0' AND '9' THEN $1::pg_catalog.oid::pg_catalog.{0} @@ -203,9 +208,16 @@ static STRING_TO_REGTYPE: LazyLock = LazyLock::new(|| { .to_string() }); +// OID 0 renders as `-` to match PostgreSQL's `regprocout`, `regclassout` and +// `regtypeout`, which all spell the absent reference that way. A nonzero OID +// that names nothing still renders as its digits, also as PostgreSQL does. const REG_STRING_CAST_TEMPLATE: &str = "( SELECT - COALESCE(mz_internal.mz_global_id_to_name(o.id), CAST($1 AS pg_catalog.oid)::pg_catalog.text) + CASE + WHEN CAST($1 AS pg_catalog.oid) = 0::pg_catalog.oid THEN '-' + ELSE + COALESCE(mz_internal.mz_global_id_to_name(o.id), CAST($1 AS pg_catalog.oid)::pg_catalog.text) + END AS text FROM ( diff --git a/src/storage-types/src/dyncfgs.rs b/src/storage-types/src/dyncfgs.rs index 3df7d647302e8..8fecf867089e6 100644 --- a/src/storage-types/src/dyncfgs.rs +++ b/src/storage-types/src/dyncfgs.rs @@ -393,21 +393,52 @@ pub const STORAGE_UPSERT_MAX_SNAPSHOT_BATCH_BUFFERING: Config> = C ParameterScope::Replica, ); -/// Allow the upsert-v2 source stash's chunk batcher to spill cold chains out -/// of RSS via the process buffer pool. The stash draws from the same shared -/// pool budget as the compute chunk batchers — there is one budget — but -/// this flag gates the stash's participation independently of the -/// compute-side `enable_column_paged_batcher_spill`. +/// 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 storage's leg of the process-wide chunk spill gate +/// (`mz_timely_util::columnar::chunk`). The gate is the OR of a compute +/// leg (`enable_column_paged_batcher_spill`) and this storage leg: chunks +/// spill while either is set, so this flag cannot veto spilling that the +/// compute flag has enabled. 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. /// -/// Off by default; the stash keeps every chunk resident until enabled. /// 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 = Config::new( "enable_upsert_paged_spill", false, - "Allow the upsert-v2 source stash to spill chunks to the shared buffer pool, gated \ - independently of the compute `enable_column_paged_batcher_spill`.", + "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 = 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, ); @@ -534,6 +565,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { .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) diff --git a/src/storage/src/render/sources.rs b/src/storage/src/render/sources.rs index 207efb62b899f..e64b4a2fa5b6a 100644 --- a/src/storage/src/render/sources.rs +++ b/src/storage/src/render/sources.rs @@ -339,6 +339,13 @@ where if dyncfgs::ENABLE_UPSERT_V2 .get(storage_state.storage_configuration.config_set()) { + // Resolved here, at operator construction, so the + // dataflow keeps one stash flavor for its whole + // life even if the flag flips underneath it. + let stash_flavor = + crate::upsert_continual_feedback_v2::UpsertStashFlavor::from_config( + storage_state.storage_configuration.config_set(), + ); crate::upsert::upsert_v2( upsert_input.enter(scope), upsert_envelope.clone(), @@ -347,6 +354,7 @@ where previous_token, export_config, backpressure_metrics, + stash_flavor, ) } else { crate::upsert::upsert( diff --git a/src/storage/src/storage_state.rs b/src/storage/src/storage_state.rs index fc1549174491e..44922849a9f45 100644 --- a/src/storage/src/storage_state.rs +++ b/src/storage/src/storage_state.rs @@ -846,16 +846,29 @@ impl<'w> Worker<'w> { STORAGE_SERVER_MAINTENANCE_INTERVAL .get(self.storage_state.storage_configuration.config_set()); - // Gate the upsert source-stash's use of the column pager. The - // pager's budget pool, backend, and codec are the shared ones - // configured by compute's `apply_worker_config` (compute and - // storage run in the same process); storage only decides whether - // its stash participates, via its own dyncfg. + // Apply storage's upsert spill flag to both stash flavors' + // mechanisms: the storage leg of the process-wide chunk + // spill gate (chunked flavor) and the storage-owned column + // pager (paged flavor). The buffer pool, the pager pool, and + // their budgets are the shared ones configured by compute's + // `apply_worker_config` (compute and storage run in the same + // process). The chunk gate ORs storage's leg with compute's, + // so chunks spill while either subsystem's flag is set. + // + // The flag is replica-scoped: the storage controller merges + // per-replica overrides into the `UpdateConfiguration` + // commands it sends, so reading this worker's `ConfigSet` + // here observes them. { use mz_storage_types::dyncfgs::ENABLE_UPSERT_PAGED_SPILL; let enabled = ENABLE_UPSERT_PAGED_SPILL .get(self.storage_state.storage_configuration.config_set()); + debug!( + worker = self.timely_worker.index(), + enabled, "upsert stash spill: applying gate", + ); + crate::upsert::upsert_stash_spill::set_enabled(enabled); crate::upsert::upsert_stash_pager::set_enabled(enabled); } } diff --git a/src/storage/src/upsert.rs b/src/storage/src/upsert.rs index 58000c8157410..6bdb4bcd6acd4 100644 --- a/src/storage/src/upsert.rs +++ b/src/storage/src/upsert.rs @@ -296,7 +296,27 @@ macro_rules! upsert_source_time_unit { } upsert_source_time_unit!(GtidPartition, Lsn); -/// Pager for the upsert-v2 source stash. +/// Storage's leg of the process-wide chunk spill gate, used by the chunked +/// upsert-v2 stash flavor. +/// +/// In that flavor the source stash and feedback arrangement spill through the +/// process buffer pool ([`mz_timely_util::columnar::chunk`]): committed chunk +/// bodies land in the pool once compute's config handler has installed and +/// budgeted it (storage and compute run in the same `clusterd` process). +/// +/// The gate is process-wide with one leg per subsystem, and chunks spill +/// while either leg is set. Storage sets its leg from +/// `enable_upsert_paged_spill`, so that flag alone cannot veto spilling +/// enabled by compute's leg. The gate is consulted at every chunk commit, so +/// flips apply to running dataflows. +pub mod upsert_stash_spill { + /// Enable or disable spilling of upsert chunk bodies to the buffer pool. + pub fn set_enabled(enabled: bool) { + mz_timely_util::columnar::chunk::set_storage_spill_enabled(enabled); + } +} + +/// Pager for the paged upsert-v2 stash flavor. /// /// This draws from the same process-wide [`TieredPolicy`] budget pool as the /// compute column-paged batcher — there is one budget and one underlying @@ -306,6 +326,9 @@ upsert_source_time_unit!(GtidPartition, Lsn); /// codec are configured by compute's `apply_tiered_config` (storage and compute /// run in the same `clusterd` process). /// +/// Flipping the flag takes effect on dataflows created after the change: the +/// paged flavor captures the pager once at operator construction. +/// /// [`TieredPolicy`]: mz_timely_util::column_pager::policy::TieredPolicy pub mod upsert_stash_pager { use std::sync::{LazyLock, RwLock}; @@ -619,6 +642,7 @@ pub(crate) fn upsert_v2<'scope, T, FromTime>( previous_token: Option>, source_config: crate::source::SourceExportCreationConfig, backpressure_metrics: Option, + stash_flavor: upsert_continual_feedback_v2::UpsertStashFlavor, ) -> ( VecCollection<'scope, T, Result, Diff>, StreamVec<'scope, T, (Option, HealthStatusUpdate)>, @@ -645,10 +669,12 @@ where tracing::info!( worker_id = %source_config.worker_id, source_id = %source_config.id, + ?stash_flavor, "rendering upsert source (btreemap backend)" ); upsert_continual_feedback_v2::upsert_inner( + stash_flavor, thin_input, upsert_envelope.key_indices, resume_upper, diff --git a/src/storage/src/upsert_continual_feedback_v2.rs b/src/storage/src/upsert_continual_feedback_v2.rs index 1885e17f80cf5..5b4b699ac2048 100644 --- a/src/storage/src/upsert_continual_feedback_v2.rs +++ b/src/storage/src/upsert_continual_feedback_v2.rs @@ -31,60 +31,81 @@ //! 1. **Ingest source data.** Read upsert commands from the source input, //! wrap each in an `UpsertDiff` (carrying a columnar order key projected //! from `FromTime` via [`UpsertSourceTime`] for dedup), and push into the -//! source-stash batcher. The batcher is a paged columnar merge batcher: it -//! consolidates entries for the same `(key, time)` via the `UpsertDiff` -//! Semigroup — keeping the update with the highest order key (latest source -//! offset) — through amortized geometric merging as data is pushed in, and -//! pages cold chains out of RSS through the pager. This bounds resident -//! memory to O(unique key-time pairs) even during large source snapshots. +//! source-stash batcher. The batcher consolidates entries for the same +//! `(key, time)` via the `UpsertDiff` Semigroup, keeping the update with +//! the highest order key (latest source offset), through amortized +//! geometric merging as data is pushed in, and cold stash state leaves RSS +//! through the flavor's spill path (see below). This bounds resident +//! memory even during large source snapshots. //! //! 2. **Read persist frontier.** Check the probe on the persist arrangement //! to learn which times have been committed. When the persist frontier //! reaches the resume upper, rehydration is complete. //! //! 3. **Seal & drain.** Call `batcher.seal(input_upper)` to extract all -//! source-finalized entries as sorted, consolidated `Column` chunks. Each -//! entry is classified: +//! source-finalized entries as sorted, consolidated chunks. Each entry is +//! classified: //! - **Eligible** (at the persist frontier): the persist trace has the -//! correct "before" state for this time. Look up the old value via a -//! cursor, emit a retraction if present, and emit the new value. +//! correct "before" state for this time. Look up the old value in the +//! feedback arrangement, emit a retraction if present, and emit the new +//! value. //! - **Ineligible** (between persist and input frontiers): persist hasn't //! caught up yet. Push back into the batcher for the next iteration. //! - **Already persisted** (below the persist frontier): some writer has -//! already advanced the shard past this time, so it is dropped. See -//! `drain_sealed_input` for why re-stashing it would strand the data -//! and pin the output frontier below the shard upper. +//! already advanced the shard past this time, so it is dropped. See the +//! drain functions for why re-stashing it would strand the data and pin +//! the output frontier below the shard upper. //! //! 4. **Capability management.** Downgrade the output capability to the //! minimum time of any remaining buffered data (in the batcher or pushed //! back as ineligible). Drop the capability entirely when the batcher is //! empty. //! +//! ## Stash flavors +//! +//! [`UpsertStashFlavor`], resolved from `enable_upsert_chunked_stash` at +//! operator construction, selects between two instantiations of the same +//! loop: +//! +//! * **Chunked**: the stash is differential's chunk merge batcher over +//! `ColumnChunk`s and the feedback arrangement is a spine of chunk batches. +//! Committed chunk bodies spill to the process buffer pool, the drain +//! loads sealed chunks one at a time, and prior state comes back through +//! bulk probes of the trace's batches. +//! * **Paged**: the stash is the paged columnar merge batcher and the +//! feedback arrangement is a `ValRowSpine`. Cold chains page out of RSS +//! through the storage-owned column pager, and prior state comes back +//! through a trace cursor. +//! +//! Both flavors' spill paths are gated by `enable_upsert_paged_spill`. +//! //! ## Eligibility condition (total order) //! //! For a total-order timestamp with `input_upper = {i}` and //! `persist_upper = {p}`, an entry at time `ts` is eligible when //! `ts == p < i` — the source has finalized it and persist is exactly at -//! that time, so the trace cursor returns the correct prior state. An entry +//! that time, so the feedback trace holds the correct prior state. An entry //! with `p < ts` is ineligible (persist hasn't caught up), and one with //! `ts < p` is already persisted and dropped. use std::fmt::Debug; -use columnar::Index as _; use differential_dataflow::difference::{IsZero, Semigroup}; use differential_dataflow::hashable::Hashable; use differential_dataflow::lattice::Lattice; use differential_dataflow::logging::Logger; use differential_dataflow::operators::arrange::agent::TraceAgent; -use differential_dataflow::operators::arrange::arrangement::arrange_core; +use differential_dataflow::operators::arrange::arrangement::{Arranged, arrange_core}; +use differential_dataflow::trace::chunk::{ChunkBatcher, ChunkBuilder, ChunkSpine}; use differential_dataflow::trace::{Batcher, Cursor, Description, TraceReader}; use differential_dataflow::{AsCollection, VecCollection}; +use mz_dyncfg::ConfigSet; use mz_repr::{Datum, Diff, GlobalId, Row}; -use mz_row_spine::{ValRowColPagedBuilder, ValRowSpine}; // Only the fuzzing-gated `datum_seq_to_upsert_value` takes a `DatumSeq`. #[cfg(feature = "fuzzing")] use mz_row_spine::DatumSeq; +use mz_row_spine::{ValRowColPagedBuilder, ValRowSpine}; +use mz_storage_types::dyncfgs::ENABLE_UPSERT_CHUNKED_STASH; use mz_storage_types::errors::{DataflowError, EnvelopeError, UpsertError}; use mz_timely_util::builder_async::{ AsyncOutputHandle, Event as AsyncEvent, OperatorBuilder as AsyncOperatorBuilder, @@ -92,15 +113,17 @@ use mz_timely_util::builder_async::{ }; use mz_timely_util::columnar::batcher::ColumnChunker; use mz_timely_util::columnar::builder::ColumnBuilder; +use mz_timely_util::columnar::chunk::{ChunkChunker, ColumnChunk}; use mz_timely_util::columnar::merge_batcher::ColumnMergeBatcher; +use mz_timely_util::columnar::unload::UnloadBatch; use mz_timely_util::columnar::{Col2ValPagedBatcher, Column}; use mz_timely_util::containers::stack::FueledBuilder; use std::convert::Infallible; use timely::container::{CapacityContainerBuilder, PushInto}; -use timely::dataflow::StreamVec; use timely::dataflow::channels::pact::{Exchange, Pipeline}; use timely::dataflow::operators::generic::Operator; use timely::dataflow::operators::{Capability, CapabilitySet, Exchange as _}; +use timely::dataflow::{Stream, StreamVec}; use timely::order::{PartialOrder, TotalOrder}; use timely::progress::frontier::AntichainRef; use timely::progress::timestamp::Refines; @@ -108,20 +131,50 @@ use timely::progress::{Antichain, Timestamp}; use crate::healthcheck::HealthStatusUpdate; use crate::metrics::upsert::UpsertMetrics; +use crate::statistics::SourceStatistics; use crate::upsert::UpsertKey; use crate::upsert::UpsertSourceTime; use crate::upsert::UpsertValue; -/// The persist-feedback arrangement's batcher, wrapping [`Col2ValPagedBatcher`] -/// only to capture the storage upsert-stash pager at construction. +/// Which stash and feedback-arrangement representation the upsert-v2 +/// operator instantiates. The two flavors run the same operator loop; they +/// differ in the batcher, the feedback trace, and how the drain reads prior +/// state. See the module docs for the comparison. +#[derive(Clone, Copy, Debug)] +pub enum UpsertStashFlavor { + /// Paged columnar merge batcher stash, `ValRowSpine` feedback + /// arrangement, cursor-based drain. Spills through the storage-owned + /// column pager. + Paged, + /// Chunk merge batcher stash, chunk-spine feedback arrangement, + /// bulk-probe drain. Spills through the process buffer pool. + Chunked, +} + +impl UpsertStashFlavor { + /// Resolve the flavor from the replica's config set. Call this once per + /// source at operator construction time, so a dataflow keeps one flavor + /// for its whole life even if the flag flips underneath it. + pub fn from_config(config: &ConfigSet) -> Self { + if ENABLE_UPSERT_CHUNKED_STASH.get(config) { + Self::Chunked + } else { + Self::Paged + } + } +} + +/// The paged flavor's persist-feedback batcher, wrapping +/// [`Col2ValPagedBatcher`] only to capture the storage upsert-stash pager at +/// construction. /// /// `arrange_core` builds its batcher via [`Batcher::new`], which has no pager /// hook, so a plain `Col2ValPagedBatcher` falls back to the process-global -/// (compute) pager — meaning the feedback arrangement's spill would be gated by -/// compute's `enable_column_paged_batcher_spill` rather than storage's -/// `enable_upsert_paged_spill`. Injecting `upsert_stash_pager::pager()` in `new` -/// puts the feedback arrangement under the same flag as the source stash. Every -/// other method delegates to the inner batcher unchanged. +/// (compute) pager, meaning the feedback arrangement's spill would be gated +/// by compute's `enable_column_paged_batcher_spill` rather than storage's +/// `enable_upsert_paged_spill`. Injecting `upsert_stash_pager::pager()` in +/// `new` puts the feedback arrangement under the same flag as the source +/// stash. Every other method delegates to the inner batcher unchanged. struct UpsertFeedbackBatcher(Col2ValPagedBatcher); impl Batcher for UpsertFeedbackBatcher @@ -158,6 +211,21 @@ where } } +/// One persist-feedback update: a key, its current value row, the time, and +/// an additive count. +type FeedbackUpdate = ((UpsertKey, Row), T, Diff); + +/// One feedback-arrangement chunk: a sorted, consolidated run of updates, +/// resident or spilled to the buffer pool. Both batcher chains and sealed +/// spine batches are sequences of these, so the arrangement's state pages out +/// of RSS under the pool's budget, and the drain reads it back through the +/// bulk [`UnloadChunk`](mz_timely_util::columnar::unload::UnloadChunk) +/// surface: copy-out probes, no cursor borrows. +type FeedbackChunk = ColumnChunk<(UpsertKey, Row), T, Diff>; + +/// The feedback arrangement's trace: a spine of `Rc`-shared chunk batches. +type FeedbackSpine = ChunkSpine>; + // The source stash carries the upsert payload in a custom diff type so the // merge batcher consolidates by (key, time), keeping the update with the // highest `FromTime` (latest source offset) per group. The diff is `Columnar` @@ -215,50 +283,32 @@ where } } -/// Consolidate `updates` through `chunker` into `Column` chunks and push them -/// into `batcher`, emptying `updates` (keeping its capacity). The chunker -/// readies a fully-consolidated chunk per `push_into`, so the `extract` loop -/// drains everything it produced. -fn flush_to_batcher( - updates: &mut Vec>, - chunker: &mut UpsertChunker, - batcher: &mut UpsertBatcher, -) where - T: columnar::Columnar + Default + Clone + PartialOrder, - for<'a> columnar::Ref<'a, T>: Copy + Ord, - O: columnar::Columnar + Default + Ord + Clone, - for<'a> columnar::Ref<'a, O>: Ord, -{ - use timely::container::{ContainerBuilder as _, PushInto as _}; - if updates.is_empty() { - return; - } - let mut raw: Column> = Default::default(); - for update in updates.drain(..) { - raw.push_into(&update); - } - chunker.push_into(&mut raw); - while let Some(chunk) = chunker.extract() { - batcher.push_into(std::mem::take(chunk)); - } -} - -// The source stash uses the paged columnar merge batcher. Data is pushed in -// unsorted; the batcher maintains geometrically-sized sorted chains and -// consolidates via the UpsertDiff Semigroup automatically. Unlike DD's -// in-memory `VecMerger`, this batcher stores each chain entry as a `Column` -// routed through the process-global pager, so the not-yet-eligible backlog -// (the snapshot / persist-lag window) pages out of RSS instead of growing it. - /// One source-stash update: a key, its dataflow time, and the payload diff. /// `O` is the columnar order key projected from the source `FromTime` (see /// [`UpsertSourceTime`]). type UpsertUpdate = (UpsertKey, T, UpsertDiff); -type UpsertBatcher = ColumnMergeBatcher>; +/// One stash chunk: a sorted, consolidated run of updates, resident or +/// spilled to the buffer pool. +type UpsertChunk = ColumnChunk>; + +/// The chunked flavor's stash: differential's chunk merge batcher over +/// `ColumnChunk`s. Data is pushed in unsorted. The batcher maintains +/// geometrically-sized sorted chains and consolidates via the UpsertDiff +/// Semigroup automatically. Committed chunks spill their bodies to the +/// process buffer pool (see `mz_timely_util::columnar::chunk`), so the +/// not-yet-eligible backlog (the snapshot / persist-lag window) pages out of +/// RSS instead of growing it. +type UpsertChunkBatcher = ChunkBatcher>; + +/// The paged flavor's stash: the paged columnar merge batcher, consolidating +/// like [`UpsertChunkBatcher`] but storing each chain entry as a `Column` +/// routed through the storage-owned pager, which pages cold chains out of +/// RSS. +type UpsertPagedBatcher = ColumnMergeBatcher>; /// The chunker that sorts and consolidates raw input into the `Column` chunks -/// [`UpsertBatcher`] consumes. +/// both stash batchers consume. type UpsertChunker = ColumnChunker>; /// The operator's data-output handle. A fueled `Vec` builder so the drain can @@ -267,12 +317,10 @@ type UpsertChunker = ColumnChunker>; type UpsertOutputHandle = AsyncOutputHandle>>>; -// The persist-feedback arrangement uses a `ValRowSpine`: keys -// land in a columnation arena (`UpsertKey` is `[u8; 32]` + `Copy`, so it uses -// `CopyRegion`), and values are stored as packed `Row` bytes in a -// `DatumContainer`. `UpsertValue` is `Result>`, so we -// still need to fold both arms into a single `Row` with a leading tag column -// so they share the value container. +// The persist-feedback arrangement stores `(UpsertKey, Row)` pairs in columnar +// chunk batches ([`FeedbackSpine`]). `UpsertValue` is +// `Result>`, so we fold both arms into a single `Row` +// with a leading tag column so they share the value column. /// Encode an [`UpsertValue`] as a `Row` with a leading tag column so both `Ok` /// and `Err` payloads round-trip through `Row` byte storage. @@ -320,8 +368,8 @@ pub fn datum_seq_to_upsert_value(seq: DatumSeq<'_>) -> UpsertValue { decode_upsert_value(seq) } -/// Decode an [`UpsertValue`] produced by [`upsert_value_to_row`] from any datum -/// iterator — a `ValRowSpine` cursor's `DatumSeq` or a stashed `Row`'s `iter`. +/// Decode an [`UpsertValue`] produced by [`upsert_value_to_row`] from any +/// datum iterator, whether over a columnar `Row` reference or an owned `Row`. fn decode_upsert_value<'a>(mut iter: impl Iterator>) -> UpsertValue { let tag = match iter.next() { Some(Datum::UInt8(tag)) => tag, @@ -356,9 +404,13 @@ fn decode_upsert_value<'a>(mut iter: impl Iterator>) -> UpsertV /// Has two inputs: /// 1. **Source input** — upsert commands from the external source. /// 2. **Persist input** — feedback of the operator's own output, read back -/// from persist. Arranged into a trace for cursor-based lookups. +/// from persist. Arranged into a trace for prior-value lookups. +/// +/// `flavor` selects the stash and feedback-arrangement representation; see +/// the module docs. #[allow(clippy::disallowed_methods)] pub fn upsert_inner<'scope, T, FromTime>( + flavor: UpsertStashFlavor, input: VecCollection<'scope, T, (UpsertKey, Option, FromTime), Diff>, key_indices: Vec, resume_upper: Antichain, @@ -381,9 +433,80 @@ where FromTime: Debug + timely::ExchangeData + Clone + Ord + Sync, FromTime: UpsertSourceTime, { - // Arrange persist feedback. - // Extract (UpsertKey, UpsertValue) from the persist feedback collection - // and arrange it. DD manages the spine, batching, and compaction. + // The feedback keying and encoding are flavor-independent; the + // arrangement they feed is not, so each arm names its own arrange types + // and hands the arrangement to the shared operator loop. + let encoded = encode_feedback( + persist_input, + key_indices, + source_config.source_statistics.clone(), + ); + match flavor { + UpsertStashFlavor::Chunked => { + // Chains and sealed batches alike are `FeedbackChunk`s whose + // bodies spill to the buffer pool, behind the same process spill + // gate as the source stash. + let persist_arranged = arrange_core::< + _, + _, + ChunkChunker<(UpsertKey, Row), T, Diff>, + ChunkBatcher>, + ChunkBuilder>, + FeedbackSpine, + >(encoded, Pipeline, "Persist feedback"); + build_upsert_operator::( + input, + resume_upper, + persist_arranged, + persist_token, + upsert_metrics, + source_config, + ) + } + UpsertStashFlavor::Paged => { + // The batcher routes its spine input through the storage-owned + // pager, paging cold feedback chains out of RSS, while + // `ValRowSpine` keeps keys in a columnation arena (`UpsertKey` + // is fixed-size `[u8; 32]`) and values as packed `Row` bytes in + // a `DatumContainer`. + let persist_arranged = arrange_core::< + _, + _, + ColumnChunker<((UpsertKey, Row), T, Diff)>, + UpsertFeedbackBatcher, + ValRowColPagedBuilder, + ValRowSpine, + >(encoded, Pipeline, "Persist feedback"); + build_upsert_operator::( + input, + resume_upper, + persist_arranged, + persist_token, + upsert_metrics, + source_config, + ) + } + } +} + +/// Key the persist feedback by [`UpsertKey`], record source statistics, and +/// encode `(UpsertKey, UpsertValue)` as `(UpsertKey, Row)` `Column` chunks, +/// the input both flavors' feedback arrangements consume. Built with +/// `Pipeline` downstream of an `UpsertKey::hashed` exchange, so the +/// arrangement keeps that locality. +fn encode_feedback<'scope, T>( + persist_input: VecCollection<'scope, T, Result, Diff>, + key_indices: Vec, + source_statistics: SourceStatistics, +) -> Stream<'scope, T, Column<((UpsertKey, Row), T, Diff)>> +where + T: Timestamp + TotalOrder + Sync, + T: Refines + differential_dataflow::lattice::Lattice, + T: columnation::Columnation, + T: columnar::Columnar + Default, + for<'a> columnar::Ref<'a, T>: Copy + Ord, +{ + // Extract (UpsertKey, UpsertValue) from the persist feedback collection. let persist_keyed = persist_input.flat_map(move |result| { let value = match result { Ok(ok) => Ok(ok), @@ -406,21 +529,12 @@ where .exchange(move |((key, _), _, _)| UpsertKey::hashed(key)) .as_collection() .inspect(move |((_, row), _, diff)| { - source_config - .source_statistics - .update_records_indexed_by(diff.into_inner()); - source_config.source_statistics.update_bytes_indexed_by( + source_statistics.update_records_indexed_by(diff.into_inner()); + source_statistics.update_bytes_indexed_by( row.as_ref().map_or(0, |r| r.byte_len().try_into().unwrap()) * diff.into_inner(), ); }); - // Encode (UpsertKey, UpsertValue) → (UpsertKey, Row) into `Column` - // containers so the feedback arrangement uses the paged columnar path: the - // batcher routes its spine input through the process-global pager, paging - // cold feedback chains out of RSS, while `ValRowSpine` keeps keys in a - // columnation arena (UpsertKey is fixed-size [u8; 32]) and values as packed - // `Row` bytes in a `DatumContainer`. Built with `Pipeline` so we keep the - // locality established by the `UpsertKey::hashed` exchange above. - let encoded = persist_keyed + persist_keyed .inner .unary::, _, _, _>( Pipeline, @@ -436,15 +550,40 @@ where }); } }, - ); - let persist_arranged = arrange_core::< - _, - _, - ColumnChunker<((UpsertKey, Row), T, Diff)>, - UpsertFeedbackBatcher, - ValRowColPagedBuilder, - ValRowSpine, - >(encoded, Pipeline, "Persist feedback"); + ) +} + +/// The flavor-independent upsert-v2 operator loop, generic over an +/// [`UpsertStashArm`]. +/// +/// Consumes the source input and the arranged persist feedback (constructed +/// per flavor by [`upsert_inner`]) and drives the ingest / seal / drain / +/// capability loop described in the module docs, calling through the arm at +/// the few points where the flavors diverge. +#[allow(clippy::disallowed_methods)] +fn build_upsert_operator<'scope, A, T, FromTime>( + input: VecCollection<'scope, T, (UpsertKey, Option, FromTime), Diff>, + resume_upper: Antichain, + persist_arranged: Arranged<'scope, TraceAgent>, + persist_token: Option>, + upsert_metrics: UpsertMetrics, + source_config: crate::source::SourceExportCreationConfig, +) -> ( + VecCollection<'scope, T, Result, Diff>, + StreamVec<'scope, T, (Option, HealthStatusUpdate)>, + StreamVec<'scope, T, Infallible>, + PressOnDropButton, +) +where + A: UpsertStashArm, + T: Timestamp + TotalOrder + Sync, + T: Refines + differential_dataflow::lattice::Lattice, + T: columnation::Columnation, + T: columnar::Columnar + Default, + for<'a> columnar::Ref<'a, T>: Copy + Ord, + FromTime: Debug + timely::ExchangeData + Clone + Ord + Sync, + FromTime: UpsertSourceTime, +{ let mut persist_trace = persist_arranged.trace.clone(); // Probe the persist arrangement's stream for frontier tracking. @@ -485,20 +624,12 @@ where let mut hydrating = true; - // Source stash backed by the paged columnar merge batcher. The batcher - // maintains geometrically-sized sorted chains and consolidates via the - // UpsertDiff Semigroup as data is pushed in, bounding memory to - // O(unique key-time pairs) even during large initial snapshots, and - // pages cold chains out of RSS through the pager. - // - // The pager is storage-owned (configured by `UpdateConfiguration` from - // storage's own dyncfgs), distinct from the compute column-paged - // batcher's process-global pager. Captured once here: backend / budget / - // codec tunes take effect live (the policy is reconfigured in place), - // but flipping the enable flag takes effect on dataflows created after - // the change. While disabled, the pager keeps every chunk resident. - let mut batcher: UpsertBatcher = Batcher::new(None, 0); - batcher.set_pager(crate::upsert::upsert_stash_pager::pager()); + // Source stash. The batcher maintains geometrically-sized sorted + // chains and consolidates via the UpsertDiff Semigroup as data is + // pushed in, bounding memory to O(unique key-time pairs) even during + // large initial snapshots. How cold stash state leaves RSS is + // flavor-specific; see the arm impls. + let mut batcher = A::new_batcher(); // The chunker sorts and consolidates raw input into the `Column` chunks // the batcher consumes. let mut chunker: UpsertChunker = Default::default(); @@ -575,14 +706,14 @@ where // Flush buffered events through the chunker into the batcher. This // triggers the chunker + geometric chain merging, which consolidates // entries for the same (key, time) via the UpsertDiff Semigroup. - flush_to_batcher(&mut push_buffer, &mut chunker, &mut batcher); + A::flush(&mut push_buffer, &mut chunker, &mut batcher); // Step 2: Read persist frontier. // The persist probe tells us which output times have been // committed back through the feedback loop. This determines: // - Whether rehydration is complete (persist >= resume_upper). // - Which source entries are eligible for processing (their - // time must equal persist_upper so the trace cursor returns + // time must equal persist_upper so the feedback trace holds // the correct prior state). // - How far to compact the persist trace. let persist_upper = persist_probe.with_frontier(|f| f.to_owned()); @@ -623,10 +754,10 @@ where // stay in the batcher. // // Extracted entries are partitioned into: - // - Eligible (ts == persist_upper): processed now via cursor - // lookup on the persist trace. + // - Eligible (ts == persist_upper): processed now via a + // prior-value lookup in the persist trace. // - Ineligible (persist_upper < ts < input_upper): persist - // hasn't caught up yet; pushed back into the batcher. + // hasn't caught up yet, so pushed back into the batcher. // // We skip the seal entirely unless an eligible entry is at all // possible. `seal` performs an O(N) merge of all chains @@ -663,18 +794,18 @@ where let remaining_frontier = batcher.frontier().to_owned(); let mut ineligible = Vec::new(); - // `drain_sealed_input` emits eligible output directly through - // `output_handle` (fueled), so there is no intermediate output - // buffer to drain afterward. - let drain_stats = drain_sealed_input( + // The drain emits eligible output directly through + // `output_handle` (fueled), so there is no intermediate + // output buffer to drain afterward. + let drain_stats = A::drain( sealed, &mut ineligible, &output_handle, &*cap, &persist_upper, &mut persist_trace, - &source_config.worker_id, - &source_config.id, + source_config.worker_id, + source_config.id, ) .await; @@ -699,24 +830,21 @@ where // remaining data: either entries still in the batcher (above // input_upper) or ineligible entries being pushed back. let min_ineligible_ts = ineligible.iter().map(|(_, ts, _)| ts).min().cloned(); - flush_to_batcher(&mut ineligible, &mut chunker, &mut batcher); - - let has_remaining = !remaining_frontier.is_empty() || min_ineligible_ts.is_some(); - if has_remaining { - let min_ts = match ( - remaining_frontier.elements().first(), - min_ineligible_ts.as_ref(), - ) { - (Some(a), Some(b)) => std::cmp::min(a, b).clone(), - (Some(a), None) => a.clone(), - (None, Some(b)) => b.clone(), - (None, None) => unreachable!(), - }; - cap.downgrade(&min_ts); - } else { - // Batcher is completely empty — drop the capability so + A::flush(&mut ineligible, &mut chunker, &mut batcher); + + // `Option::min` alone would be wrong here, `None` sorts low. + // Chain the candidates and take the min over present ones. + let min_ts = remaining_frontier + .elements() + .first() + .into_iter() + .chain(min_ineligible_ts.as_ref()) + .min(); + match min_ts { + Some(min_ts) => cap.downgrade(min_ts), + // Batcher is completely empty. Drop the capability so // downstream operators can make progress. - stash_cap = None; + None => stash_cap = None, } } @@ -739,11 +867,206 @@ where ) } -/// Counts from a single call to [`drain_sealed_input`], used to update metrics. +/// The flavor-specific pieces of the upsert-v2 operator: the stash batcher, +/// the feedback arrangement's spine, and how the drain reads prior state. +/// [`build_upsert_operator`] holds the flavor-independent operator loop and +/// calls through this trait at the few points where the flavors diverge. +trait UpsertStashArm +where + T: Timestamp + Lattice + columnar::Columnar + Default, + for<'a> columnar::Ref<'a, T>: Copy + Ord, + O: columnar::Columnar + Default + Ord + Clone + Send + Sync + 'static, + for<'a> columnar::Ref<'a, O>: Ord + Copy, +{ + /// The feedback arrangement's spine. `'static` because the operator + /// future owns a trace agent for it. + type Spine: TraceReader