adapter: use occ for read-then write, with incremental retries

doc/developer/design/20260210_incremental_occ_read_then_write.md

@@ -128,7 +128,7 @@ Session Task                         Coordinator
   |                                      |
   |-- acquire OCC semaphore              |
   |                                      |
-  |-- CreateReadThenWriteSubscribe ----> |
+  |-- CreateInternalSubscribe ---------> |
   | <------------ subscribe channel -----|
   |                                      |
   |   +-- OCC Loop ------------------+   |
@@ -141,7 +141,7 @@ Session Task                         Coordinator
   |   |   if Success: break          |   |
   |   +------------------------------+   |
   |                                      |
-  |-- DropReadThenWriteSubscribe ------> |
+  |-- DropInternalSubscribe -----------> |
   |                                      |
 ```
 
@@ -193,7 +193,7 @@ subscribe.
 The subscribes created for read-then-write are internal: they do not appear in
 `mz_subscriptions` or other introspection tables, and they don't increment the
 active subscribes metric. They are created and dropped via dedicated `Command`
-variants (`CreateReadThenWriteSubscribe`, `DropReadThenWriteSubscribe`).
+variants (`CreateInternalSubscribe`, `DropInternalSubscribe`).
 
 ## Correctness
 

misc/python/materialize/mzcompose/__init__.py

@@ -185,6 +185,11 @@ def get_variable_system_parameters(
             "true",
             ["true", "false"],
         ),
+        VariableSystemParameter(
+            "enable_adapter_frontend_occ_read_then_write",
+            "true",
+            ["true", "false"],
+        ),
         VariableSystemParameter(
             "enable_cast_elimination",
             "false",

misc/python/materialize/parallel_workload/action.py

@@ -1520,6 +1520,9 @@ def __init__(
             "false",
         ]
         self.flags_with_values["enable_cast_elimination"] = BOOLEAN_FLAG_VALUES
+        self.flags_with_values["enable_adapter_frontend_occ_read_then_write"] = (
+            BOOLEAN_FLAG_VALUES
+        )
 
         # If you are adding a new config flag in Materialize, consider using it
         # here instead of just marking it as uninteresting to silence the

src/adapter-types/src/dyncfgs.rs

@@ -155,6 +155,13 @@ pub const ENABLE_S3_TABLES_REGION_CHECK: Config<bool> = Config::new(
     "Whether to enforce that S3 Tables connections are in the same region as the environment.",
 );
 
+pub const FRONTEND_READ_THEN_WRITE: Config<bool> = Config::new(
+    "enable_adapter_frontend_occ_read_then_write",
+    // WIP: true for testing in ci, Should be false before merging.
+    true,
+    "Use frontend sequencing for DELETE/UPDATE operations.",
+);
+
 /// Adds the full set of all adapter `Config`s.
 pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
     configs
@@ -178,4 +185,5 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&CONSTRAINT_BASED_TIMESTAMP_SELECTION)
         .add(&PERSIST_FAST_PATH_ORDER)
         .add(&ENABLE_S3_TABLES_REGION_CHECK)
+        .add(&FRONTEND_READ_THEN_WRITE)
 }

src/adapter/src/active_compute_sink.rs

@@ -116,6 +116,9 @@ pub struct ActiveSubscribe {
     pub start_time: EpochMillis,
     /// How to present the subscribe's output.
     pub output: SubscribeOutput,
+    /// If true, this is an internal subscribe that should not appear in
+    /// introspection tables like mz_subscriptions.
+    pub internal: bool,
 }
 
 impl ActiveSubscribe {