From 6aa62870093306f66af5fcb3d3716d9df41d0b19 Mon Sep 17 00:00:00 2001 From: materializekatrina <118202509+materializekatrina@users.noreply.github.com> Date: Thu, 27 Aug 2026 13:16:05 -0400 Subject: [PATCH 1/4] Docs: update unclear wording on support doc (#38513) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Updates the support docs to clarify wording, remove outdated information about retiring Slack/Teams access, and add a troubleshooting tip for users who can’t see Matty. --------- Co-authored-by: kay-kim --- doc/user/content/support.md | 85 ++++++------------ .../self_managed_editions_table.yml | 2 +- .../shortcodes/icons/support-chat.html | 4 + .../self-managed-editions-table.html | 2 +- doc/user/static/images/messenger-widget.png | Bin 0 -> 3458 bytes 5 files changed, 34 insertions(+), 59 deletions(-) create mode 100644 doc/user/layouts/shortcodes/icons/support-chat.html create mode 100644 doc/user/static/images/messenger-widget.png 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 0000000000000000000000000000000000000000..58f6dbb7aa6ab761175ab1a3963d87521d22d454 GIT binary patch literal 3458 zcmV-|4Sn*7P)>H>lwYNEJC zGl{zq1$Sa_6pcG-5XF6A7aBQpzpuNs7aF?nd)@sS-mSV-i2C~7^XaqPbI-lk&CLbc z^w8BkeETfx738fR>gT>Joc$a*!ZRw_-!nSb&m%hDw@=hAbo`9pjqr?4_H~bn^z(=c z5Au#3%6{%OUA3l%o6>A=%jR$C8qiPOb7bEb)d-L16ej3G*2`J1XPvC-?qRP= z=tBR#F)8f%sz9&Uo>2?7c2y+aO8Yu&UH1SlwHE`RW&$l{y~=9gw+}<;42Eg1V!hbk zGe#ZQKh`TEEUSB`hLEvxUzgE?5_&W9vV;k;lywa^gf>7}fMMD-td}x4OU4XJ?46mk zze}e9F>CVrZvK7MzJ8uDdwt!bt68V_y^jsu-Ma4Q9%E~-9^t9pJF;)A@9glLZXFiH z%&@F(Y>=b;JYtHNpfuJBl(%z$FzuQV9_k{z^USd9jsqqw_P<>M`^ODo-?{<@L&G|u z`+7v%rR$$BVb|c@F^2TGpIzIOjUBe)Q#P19HmC+m1C}ue)82rwU`#FtV(!%3ZUJ7g z)7T&?Y#q3a09fi30bX&_=1$$<8fI*qPtqs;p6VUU{=x=R2kjb+5o47#jL`!Veb~$1 z?!X{S`)!O_b|99{%m0W)BN6Nct2+@4eXqvYAp=>!nBbrCaX`PgRQ6&`oe+k;H$euF zg;QaS8j$c8W=pqpVh~0hvVcsS0z>7K^j8)J?omLPbS4wX#-T8R*|$J}(KeJ|M`_9g z|J08aGi9Cmf~@QuO>O%~=fD-AThSpa$joj)bfImNB5doUFUZWElfG(2$hOZ@B%eHv z^GW`slg+>-pV32xlE9cZEw7uuXY39|sA8Rb><-9MG>|~A_-P$HQRuLQBxETHM(D7W zpRxy4D8iJiS3ss*Ft9qt)GSXCqU5G4k89;IsDJzr_J9UkKhT=?r1A|VEZRe(1|-N< za4cM<>7kjlC;vR1KY7by=XiDlWNTYx$fO)N@&cZLSkqcsTIkQa&q)34ZVK{_m*cmX zrhdMb@;98KGle>1oK*m-BVKdB8ypG)31}OLMJf-Mwb`tx>$O^`k5WqM|XNzvO z34<31WD6Nv1!K;X4bYAv-T~Rt0OjuOC-Q;fsiuiR>9jfP1f4BxABG4Z0AB=ITcaJm zV!~Won;uHkf*nqW0>2SCb1M}byh6pNZrc@zPD^~-gprkUsEdE!*u7!~#nbdq7SGv6 zflh=1Ad6<_Q^6sJ03pa4GPejMWblgKV)IpvZ#7K9d{FM(d`#c5Z>YndFcTHv6^E>> zIgr!#2Lf3`=51wjmxvwYmYgE0dR@n@DR*vE3Q!zjzEnQR^zFZXps=7-Sif)v5Pas6 z)@yGkt?$nML8;ge80VWzc^L)VP~N@MLWzuEn&K5_HyGm>jI`K&^y8LuG?hhF&H@6N zL-ytqLGJ`Fsd83;@p~O-q00dXWDf?0d8?l6msL`Qj%HAD z0fpauT`R4igJyY1;W*4Y){g;qMNPGyyXN03NW;SN>N{bEf23go3DShANYN&Y3oogfatC4i?akUaZ$k2}BR& z@x#}YCO}zkrYSo&pW)V&X3aY)z4#~14&PwrQ(&gBu(X>^_}WXK%M1t@^l}USEODrm zp|0^bB5gagJ#Z`fT1(J&Sjso%l zN)=_L9HG%BP$mYZ3j(UIYvh8W(KOS=;=6SG&{a0~a6t`WEZ{?SITDDWPg$3I#LTBm z8nue{|0o8^J0V*DFfQmrb~y!zA(YJI!xS=jg^^FuLMfrT+PB>E6l4q9gCa{Hjy74Q zP@X@n7JN#`kQL^dvZq9p`GTgDoWJW@Cd-jd-4c`+&n!bZAO=bkpuDgMMV?OGL18Zs z6utR6^wULZXlUYkb`yhAVF8LP!(OQ;hKQgrgSMTD7RovneT57*o~MM5Swn|^zC;c6 zP28ID;Qk9?t|I%1q134pRJGqPdn%+%c%ZSHu!pgEMI`!4fCY19Z%N_0?UnOu}Q{jP_C0%}yZi-lqu z62pA`y8p5$O?mWy%~P09k!AW_>IJIzQ2NIUQ0@VW2%nO(>R0-F$Vww95a98{SLiLL zqen=LWr0fEFKQ`=K^Z&DSW~cOew;yR zu;EOo=0g4Urh0rEm= zYHSwFA4-LsTLu)TvP+>H+H(eF?TUj&DRw}CPXUw`UMR2XghK966DZThW>DdgKLp|O zw1)adIvY}m0_N-AZswPDjkF`VRKGtSa^BB8vj6^h@C zLt+LcG*c*|Zsdhh*GN|`mRq&i)2_wV+U*GOl59sXZGZyvWhgv^#7pNgC^sK-n=b*% z`XaOF3nf4(rsno3Y@ULevbM6s(r-tI_p;lDoz@!W6DV-71clep!+b$gLWd=pnJ?7w zHi`&|QB!6O6uWKMk!nku#redG8Z#(x^e|tyZ&ZqkzP`^YB-QZNq2L=Y4oVrFbH_ifRbc?;_vgH52!29$@~no`{c6gX)JR}>UxzRuZUTQp?u z^n(LlKC2O`P7I?jKv^^^U)V9l4W+7HYrcGvji7urK1(PqZRUp3`~gal0}l>Bn;o9( zkfSxADX76Ug@WC;VxYi$VQ-_Uru=h4wy-ThBL>QYms;~B(a{=^wQLTb(fgF_w4-L` z3$x|xS1S0SFh`%Cd(tG$qJ@Huo?@V&aKdnr%kGEIXzzB8AE?&wRSTcg)Q;XR%SGM3E&X01WN<{rbniAb{`(RYt`kxLg6 z!_gZa)d46G6LUmqAfS}reI{s|Lk=+0U*@G8@n@eS{vccB5KLD;1k+%xL^&o_vSVW7 zrinhc7q^iq2PsQ)kg{YqyQzGVl%vK)9yP9#`bIb9Fn$>w#;;sgA;aq`&=$|xs$A$| zxX5g~CzS}g-qRV}C@C)u^cc|ok!M$NZQkNy}h^hMu zj7_pcp(7h||GGu4h7JIt0b>!pp5g3<)iEqt>elErE^3|%Fb3|GJ~=il>Z8&^S7M}b z5fDXqXPXNWT&c~PoC7evx(Z1-hRM})vdF%%%lTdP}iy}@Gf<-*5I>C@h*5L zyqjBRZV_SG-LXePw`r(Ys_k{{Z6m{H;WP2scn3G7nYX9~+qz=E9X3Jgb_W;g_R)(u z+@N0fj6ysU&&FrqvlNMQv@cW{INYFD@Mj4_b!W3g;^Z70XQVst3x}}n!q39*v?m-z k;@{%`;Q!(ot}xdB17Z;Q^zfG{Q2+n{07*qoM6N<$g7*ZDQUCw| literal 0 HcmV?d00001 From 844b947128194cafd04b56870d4b9f240cf6b736 Mon Sep 17 00:00:00 2001 From: Dov Alperin Date: Thu, 27 Aug 2026 13:37:18 -0400 Subject: [PATCH 2/4] storage/upsert: v2 stash on ChunkBatcher with pool-backed spill (#38255) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### Motivation The upsert v2 (continual feedback) stash moves from its previous representation onto `ChunkBatcher`/`ChunkSpine`: stashed updates live in columnar chunks that participate in the buffer pool, and the feedback arrangement drains through `UnloadChunk` one chunk at a time instead of materializing whole batches. With `enable_upsert_paged_spill` on, stash and arrangement state past the residency budget spills as compressed, budget-accounted extents rather than raw kernel-swapped heap. Benchmarked extensively on AWS (self-managed EKS, MSK source, swap-enabled nodes), August 2026 campaign: * Hydration under pressure, 25cc replica: at 4.5x state-to-memory, v2-spill hydrates in 707 s vs 1187 s for v1 (RocksDB mem-env). At 9x: 1607 s vs 4607 s (v1 goes superlinear, v2 stays near-linear). At 13.5x, v1 and v2-resident crash-loop at the pod swap ceiling and never complete; v2-spill finishes in 2386 s with matching counts. * Swap traffic: v2-spill moves roughly half the swap bytes of v1 at every depth measured. * Steady-state ingest freshness under the same pressure: statistically identical across v1, v2-resident, and v2-spill. * Unpressured regimes: v2-resident tracks v1 within ~10% everywhere; spill costs appear only at a 50cc saturation edge and vanish one size up (or under the two chunk patches below this PR in the stack). The new representation is flagged: `enable_upsert_chunked_stash` (off in production, on and randomized in CI) selects between the `ChunkBatcher` representation above and the previous paged-columnar-merge-batcher + `ValRowSpine` representation, which stays the production default while the chunked flavor earns trust. The operator loop is shared; a small `UpsertStashArm` trait carries the flavor-specific pieces (stash batcher, feedback spine, flush, drain), and the arms are resolved from the config set once at operator construction, so a dataflow keeps its flavor for life. Spilling in either flavor remains gated by `enable_upsert_paged_spill`. ### Tips for reviewer * Stacked on #38253 and #38254 (the two chunk patches the campaign produced); the interesting upsert logic is `upsert_continual_feedback_v2.rs`. * `enable_upsert_paged_spill` is replica-scoped and composes with compute's spill gate as an OR on the process-wide pool. * The unit-test harness runs every scenario under both stash flavors and asserts they produce identical output, so the paged arm is exercised by the same tests as the chunked one. ### Checklist - [ ] This PR has adequate test coverage / QA involvement has been duly considered. ([trigger-ci for additional test/nightly runs](https://trigger-ci.dev.materialize.com/)) - [ ] This PR has an associated up-to-date [design doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md), is a design doc ([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)), or is sufficiently small to not require a design. - [ ] If this PR evolves [an existing `$T ⇔ Proto$T` mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md) (possibly in a backwards-incompatible way), then it is tagged with a `T-proto` label. - [ ] If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label ([example](https://github.com/MaterializeInc/cloud/pull/5021)). - [x] If this PR includes major [user-facing behavior changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note), I have pinged the relevant PM to schedule a changelog post. --- misc/python/materialize/mzcompose/__init__.py | 14 +- .../materialize/parallel_workload/action.py | 1 + src/compute/src/compute_state.rs | 6 + src/storage-types/src/dyncfgs.rs | 48 +- src/storage/src/render/sources.rs | 8 + src/storage/src/storage_state.rs | 23 +- src/storage/src/upsert.rs | 28 +- .../src/upsert_continual_feedback_v2.rs | 1144 +++++++++++++---- .../mzcompose.py | 1 + 9 files changed, 1011 insertions(+), 262 deletions(-) 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/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/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