Repository: MaterializeInc/materialize (language: python)

Issue:
Disable the new instrospection sources
This PR disables the two introspection sources added in https://github.com/MaterializeInc/materialize/pull/29848, because Mz's 0dt mechanism is currently not prepared to handle new introspection sources.

"Disabling" here means removing the registration of the new `Log` builtins and the `View`s depending on them. The intention is to keep the diff as small as possible, to make this change less risky.

The plan is to revert this PR once https://github.com/MaterializeInc/database-issues/issues/8731 is resolved.

### Motivation

  * This PR fixes a recognized bug.

Mitigates https://github.com/MaterializeInc/database-issues/issues/8731 for the release.

### Tips for reviewer

<!--
Leave some tips for your reviewer, like:

    * The diff is much smaller if viewed with whitespace hidden.
    * [Some function/module/file] deserves extra attention.
    * [Some function/module/file] is pure code movement and only needs a skim.

Delete this section if no tips.
-->

### 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.
  <!-- Reference the design in the description. -->
- [ ] 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)).
  <!-- Ask in #team-cloud on Slack if you need help preparing the cloud PR. -->
- [ ] 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.



Discussion / hints:
We are still seeing the logging:
```
platform-checks-mz_2-1              | cluster-s3-replica-s3-gen-1: 2024-11-08T15:00:40.661245Z  WARN mz_compute::server: halting process: new instance configuration does not match existing instance configuration:
```
But I think that's actually not a sign of this not working, but instead we already have a v0.124.0 cut that has the problem. So ok from my side.

Fix patch:
diff --git a/doc/user/content/sql/system-catalog/mz_introspection.md b/doc/user/content/sql/system-catalog/mz_introspection.md
index ade7e4602a985..46836ff521e16 100644
--- a/doc/user/content/sql/system-catalog/mz_introspection.md
+++ b/doc/user/content/sql/system-catalog/mz_introspection.md
@@ -222,6 +222,19 @@ The `mz_dataflow_channel_operators` view associates [dataflow] channels with the
 
 <!-- RELATION_SPEC_UNDOCUMENTED mz_introspection.mz_dataflow_channel_operators_per_worker -->
 
+## `mz_dataflow_global_ids`
+
+The `mz_dataflow_global_ids` view associates [dataflow] ids with global ids (ids of the form `u8` or `t5`).
+
+<!-- RELATION_SPEC mz_introspection.mz_dataflow_global_ids -->
+
+| Field        | Type      | Meaning                                    |
+|------------- | -------   | --------                                   |
+| `id`         | [`uint8`] | The dataflow ID.                           |
+| `global_id`  | [`text`]  | A global ID associated with that dataflow. |
+
+<!-- RELATION_SPEC_UNDOCUMENTED mz_introspection.mz_compute_dataflow_global_ids_per_worker -->
+
 ## `mz_dataflow_operators`
 
 The `mz_dataflow_operators` view describes the [dataflow] operators in the system.
@@ -292,6 +305,29 @@ through a hierarchical scheme for either aggregation or Top K computations.
 | `savings`       | [`numeric`]          | A conservative estimate of the amount of memory in bytes to be saved by applying the hint.                |
 | `hint`          | [`double precision`] | The hint value that will eliminate `to_cut` levels from the region's hierarchy.                           |
 
+## `mz_lir_mapping`
+
+The `mz_lir_mapping` view describes the low-level internal representation (LIR) plan that corresponds to global ids.
+LIR is a higher-level representation than dataflows; this view is used for profiling and debugging indices and materialized views.
+Note that LIR is not a stable interface and may change at any time.
+In particular, you should not attempt to parse `operator` descriptions.
+LIR nodes are implemented by zero or more dataflow operators with sequential ids.
+We use the range `[operator_id_start, operator_id_end)` to record this information.
+If an LIR node was implemented without any dataflow operators, `operator_id_start` will be equal to `operator_id_end`.
+
+<!-- RELATION_SPEC mz_introspection.mz_lir_mapping -->
+| Field             | Type      | Meaning
+| ---------         | --------  | -----------
+| global_id         | [`text`]  | The global ID.
+| lir_id            | [`uint8`] | The LIR node ID.
+| operator          | [`text`]  | The LIR operator, in the format `OperatorName INPUTS [OPTIONS]`.
+| parent_lir_id     | [`uint8`] | The parent of this LIR node. May be `NULL`.
+| nesting           | [`uint2`] | The nesting level of this LIR node.
+| operator_id_start | [`uint8`] | The first dataflow operator ID implementing this LIR operator (inclusive).
+| operator_id_end   | [`uint8`] | The first dataflow operator ID _after_ this LIR operator (exclusive).
+
+<!-- RELATION_SPEC_UNDOCUMENTED mz_introspection.mz_compute_lir_mapping_per_worker -->
+
 ## `mz_message_counts`
 
 The `mz_message_counts` view describes the messages and message batches sent and received over the [dataflow] channels in the system.
diff --git a/src/catalog/src/builtin.rs b/src/catalog/src/builtin.rs
index 963897ec0c76b..a273a486724ac 100644
--- a/src/catalog/src/builtin.rs
+++ b/src/catalog/src/builtin.rs
@@ -9136,6 +9136,7 @@ pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = LazyLock::ne
         Builtin::Log(&MZ_DATAFLOW_ADDRESSES_PER_WORKER),
         Builtin::Log(&MZ_DATAFLOW_OPERATOR_REACHABILITY_RAW),
         Builtin::Log(&MZ_COMPUTE_EXPORTS_PER_WORKER),
+        Builtin::Log(&MZ_COMPUTE_DATAFLOW_GLOBAL_IDS_PER_WORKER),
         Builtin::Log(&MZ_MESSAGE_COUNTS_RECEIVED_RAW),
         Builtin::Log(&MZ_MESSAGE_COUNTS_SENT_RAW),
         Builtin::Log(&MZ_MESSAGE_BATCH_COUNTS_RECEIVED_RAW),
@@ -9231,6 +9232,7 @@ pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = LazyLock::ne
         Builtin::View(&MZ_DATAFLOW_ADDRESSES),
         Builtin::View(&MZ_DATAFLOW_CHANNELS),
         Builtin::View(&MZ_DATAFLOW_OPERATORS),
+        Builtin::View(&MZ_DATAFLOW_GLOBAL_IDS),
         Builtin::View(&MZ_DATAFLOW_OPERATOR_DATAFLOWS_PER_WORKER),
         Builtin::View(&MZ_DATAFLOW_OPERATOR_DATAFLOWS),
         Builtin::View(&MZ_OBJECT_TRANSITIVE_DEPENDENCIES),
@@ -9410,6 +9412,8 @@ pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = LazyLock::ne
         Builtin::View(&MZ_COMPUTE_ERROR_COUNTS),
         Builtin::Source(&MZ_COMPUTE_ERROR_COUNTS_RAW_UNIFIED),
         Builtin::Source(&MZ_COMPUTE_HYDRATION_TIMES),
+        Builtin::Log(&MZ_COMPUTE_LIR_MAPPING_PER_WORKER),
+        Builtin::View(&MZ_LIR_MAPPING),
         Builtin::View(&MZ_COMPUTE_OPERATOR_HYDRATION_STATUSES),
         Builtin::Source(&MZ_CLUSTER_REPLICA_FRONTIERS),
         Builtin::View(&MZ_COMPUTE_HYDRATION_STATUSES),


Test patch:
diff --git a/src/environmentd/tests/testdata/http/ws b/src/environmentd/tests/testdata/http/ws
index bee3e9252cb92..f8121b4e3b15a 100644
--- a/src/environmentd/tests/testdata/http/ws
+++ b/src/environmentd/tests/testdata/http/ws
@@ -402,7 +402,7 @@ ws-text
 ws-text
 {"query": "SELECT 1 FROM mz_sources LIMIT 1"}
 ----
-{"type":"Notice","payload":{"message":"{\n  \"plans\": {\n    \"raw\": {\n      \"text\": \"Finish limit=1 output=[#0]\\n  Project (#15)\\n    Map (1)\\n      Get mz_catalog.mz_sources\\n\\nTarget cluster: mz_catalog_server\\n\",\n      \"json\": {\n        \"Project\": {\n          \"input\": {\n            \"Map\": {\n              \"input\": {\n                \"Get\": {\n                  \"id\": {\n                    \"Global\": {\n                      \"System\": 460\n                    }\n                  },\n                  \"typ\": {\n                    \"column_types\": [\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"Oid\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": {\n                          \"Array\": \"MzAclItem\"\n                        },\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      }\n                    ],\n                    \"keys\": [\n                      [\n                        0\n                      ],\n                      [\n                        1\n                      ]\n                    ]\n                  }\n                }\n              },\n              \"scalars\": [\n                {\n                  \"Literal\": [\n                    {\n                      \"data\": [\n                        45,\n                        1\n                      ]\n                    },\n                    {\n                      \"scalar_type\": \"Int32\",\n                      \"nullable\": false\n                    }\n                  ]\n                }\n              ]\n            }\n          },\n          \"outputs\": [\n            15\n          ]\n        }\n      }\n    },\n    \"optimized\": {\n      \"global\": {\n        \"text\": \"t72:\\n  Finish limit=1 output=[#0]\\n    ArrangeBy keys=[[#0]]\\n      ReadGlobalFromSameDataflow t71\\n\\nt71:\\n  Project (#15)\\n    Map (1)\\n      ReadIndex on=mz_sources mz_sources_ind=[*** full scan ***]\\n\\nTarget cluster: mz_catalog_server\\n\",\n        \"json\": {\n          \"plans\": [\n            {\n              \"id\": \"t72\",\n              \"plan\": {\n                \"ArrangeBy\": {\n                  \"input\": {\n                    \"Get\": {\n                      \"id\": {\n                        \"Global\": {\n                          \"Transient\": 71\n                        }\n                      },\n                      \"typ\": {\n                        \"column_types\": [\n                          {\n                            \"scalar_type\": \"Int32\",\n                            \"nullable\": false\n                          }\n                        ],\n                        \"keys\": []\n                      },\n                      \"access_strategy\": \"SameDataflow\"\n                    }\n                  },\n                  \"keys\": [\n                    [\n                      {\n                        \"Column\": 0\n                      }\n                    ]\n                  ]\n                }\n              }\n            },\n            {\n              \"id\": \"t71\",\n              \"plan\": {\n                \"Project\": {\n                  \"input\": {\n                    \"Map\": {\n                      \"input\": {\n                        \"Get\": {\n                          \"id\": {\n                            \"Global\": {\n                              \"System\": 460\n                            }\n                          },\n                          \"typ\": {\n                            \"column_types\": [\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"Oid\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": {\n                                  \"Array\": \"MzAclItem\"\n                                },\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              }\n                            ],\n                            \"keys\": [\n                              [\n                                0\n                              ],\n                              [\n                                1\n                              ]\n                            ]\n                          },\n                          \"access_strategy\": {\n                            \"Index\": [\n                              [\n                                {\n                                  \"System\": 732\n                                },\n                                \"FullScan\"\n                              ]\n                            ]\n                          }\n                        }\n                      },\n                      \"scalars\": [\n                        {\n                          \"Literal\": [\n                            {\n                              \"Ok\": {\n                                \"data\": [\n                                  45,\n                                  1\n                                ]\n                              }\n                            },\n                            {\n                              \"scalar_type\": \"Int32\",\n                              \"nullable\": false\n                            }\n                          ]\n                        }\n                      ]\n                    }\n                  },\n                  \"outputs\": [\n                    15\n                  ]\n                }\n              }\n            }\n          ],\n          \"sources\": []\n        }\n      },\n      \"fast_path\": {\n        \"text\": \"Explained Query (fast path):\\n  Finish limit=1 output=[#0]\\n    Project (#15)\\n      Map (1)\\n        ReadIndex on=mz_catalog.mz_sources mz_sources_ind=[*** full scan ***]\\n\\nTarget cluster: mz_catalog_server\\n\",\n        \"json\": {\n          \"plans\": [\n            {\n              \"id\": \"Explained Query (fast path)\",\n              \"plan\": {\n                \"PeekExisting\": [\n                  {\n                    \"System\": 460\n                  },\n                  {\n                    \"System\": 732\n                  },\n                  null,\n                  {\n                    \"mfp\": {\n                      \"expressions\": [\n                        {\n                          \"Literal\": [\n                            {\n                              \"Ok\": {\n                                \"data\": [\n                                  45,\n                                  1\n                                ]\n                              }\n                            },\n                            {\n                              \"scalar_type\": \"Int32\",\n                              \"nullable\": false\n                            }\n                          ]\n                        }\n                      ],\n                      \"predicates\": [],\n                      \"projection\": [\n                        15\n                      ],\n                      \"input_arity\": 15\n                    }\n                  }\n                ]\n              }\n            }\n          ],\n          \"sources\": []\n        }\n      }\n    }\n  },\n  \"insights\": {\n    \"imports\": {\n      \"s732\": {\n        \"name\": {\n          \"schema\": \"mz_catalog\",\n          \"item\": \"mz_sources_ind\"\n        },\n        \"type\": \"compute\"\n      }\n    },\n    \"fast_path_clusters\": {},\n    \"fast_path_limit\": null,\n    \"persist_count\": []\n  },\n  \"cluster\": {\n    \"name\": \"mz_catalog_server\",\n    \"id\": {\n      \"System\": 2\n    }\n  },\n  \"redacted_sql\": \"SELECT '<REDACTED>' FROM [s460 AS mz_catalog.mz_sources] LIMIT '<REDACTED>'\"\n}","code":"MZ001","severity":"notice"}}
+{"type":"Notice","payload":{"message":"{\n  \"plans\": {\n    \"raw\": {\n      \"text\": \"Finish limit=1 output=[#0]\\n  Project (#15)\\n    Map (1)\\n      Get mz_catalog.mz_sources\\n\\nTarget cluster: mz_catalog_server\\n\",\n      \"json\": {\n        \"Project\": {\n          \"input\": {\n            \"Map\": {\n              \"input\": {\n                \"Get\": {\n                  \"id\": {\n                    \"Global\": {\n                      \"System\": 461\n                    }\n                  },\n                  \"typ\": {\n                    \"column_types\": [\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"Oid\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": {\n                          \"Array\": \"MzAclItem\"\n                        },\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      }\n                    ],\n                    \"keys\": [\n                      [\n                        0\n                      ],\n                      [\n                        1\n                      ]\n                    ]\n                  }\n                }\n              },\n              \"scalars\": [\n                {\n                  \"Literal\": [\n                    {\n                      \"data\": [\n                        45,\n                        1\n                      ]\n                    },\n                    {\n                      \"scalar_type\": \"Int32\",\n                      \"nullable\": false\n                    }\n                  ]\n                }\n              ]\n            }\n          },\n          \"outputs\": [\n            15\n          ]\n        }\n      }\n    },\n    \"optimized\": {\n      \"global\": {\n        \"text\": \"t72:\\n  Finish limit=1 output=[#0]\\n    ArrangeBy keys=[[#0]]\\n      ReadGlobalFromSameDataflow t71\\n\\nt71:\\n  Project (#15)\\n    Map (1)\\n      ReadIndex on=mz_sources mz_sources_ind=[*** full scan ***]\\n\\nTarget cluster: mz_catalog_server\\n\",\n        \"json\": {\n          \"plans\": [\n            {\n              \"id\": \"t72\",\n              \"plan\": {\n                \"ArrangeBy\": {\n                  \"input\": {\n                    \"Get\": {\n                      \"id\": {\n                        \"Global\": {\n                          \"Transient\": 71\n                        }\n                      },\n                      \"typ\": {\n                        \"column_types\": [\n                          {\n                            \"scalar_type\": \"Int32\",\n                            \"nullable\": false\n                          }\n                        ],\n                        \"keys\": []\n                      },\n                      \"access_strategy\": \"SameDataflow\"\n                    }\n                  },\n                  \"keys\": [\n                    [\n                      {\n                        \"Column\": 0\n                      }\n                    ]\n                  ]\n                }\n              }\n            },\n            {\n              \"id\": \"t71\",\n              \"plan\": {\n                \"Project\": {\n                  \"input\": {\n                    \"Map\": {\n                      \"input\": {\n                        \"Get\": {\n                          \"id\": {\n                            \"Global\": {\n                              \"System\": 461\n                            }\n                          },\n                          \"typ\": {\n                            \"column_types\": [\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"Oid\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": {\n                                  \"Array\": \"MzAclItem\"\n                                },\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              }\n                            ],\n                            \"keys\": [\n                              [\n                                0\n                              ],\n                              [\n                                1\n                              ]\n                            ]\n                          },\n                          \"access_strategy\": {\n                            \"Index\": [\n                              [\n                                {\n                                  \"System\": 736\n                                },\n                                \"FullScan\"\n                              ]\n                            ]\n                          }\n                        }\n                      },\n                      \"scalars\": [\n                        {\n                          \"Literal\": [\n                            {\n                              \"Ok\": {\n                                \"data\": [\n                                  45,\n                                  1\n                                ]\n                              }\n                            },\n                            {\n                              \"scalar_type\": \"Int32\",\n                              \"nullable\": false\n                            }\n                          ]\n                        }\n                      ]\n                    }\n                  },\n                  \"outputs\": [\n                    15\n                  ]\n                }\n              }\n            }\n          ],\n          \"sources\": []\n        }\n      },\n      \"fast_path\": {\n        \"text\": \"Explained Query (fast path):\\n  Finish limit=1 output=[#0]\\n    Project (#15)\\n      Map (1)\\n        ReadIndex on=mz_catalog.mz_sources mz_sources_ind=[*** full scan ***]\\n\\nTarget cluster: mz_catalog_server\\n\",\n        \"json\": {\n          \"plans\": [\n            {\n              \"id\": \"Explained Query (fast path)\",\n              \"plan\": {\n                \"PeekExisting\": [\n                  {\n                    \"System\": 461\n                  },\n                  {\n                    \"System\": 736\n                  },\n                  null,\n                  {\n                    \"mfp\": {\n                      \"expressions\": [\n                        {\n                          \"Literal\": [\n                            {\n                              \"Ok\": {\n                                \"data\": [\n                                  45,\n                                  1\n                                ]\n                              }\n                            },\n                            {\n                              \"scalar_type\": \"Int32\",\n                              \"nullable\": false\n                            }\n                          ]\n                        }\n                      ],\n                      \"predicates\": [],\n                      \"projection\": [\n                        15\n                      ],\n                      \"input_arity\": 15\n                    }\n                  }\n                ]\n              }\n            }\n          ],\n          \"sources\": []\n        }\n      }\n    }\n  },\n  \"insights\": {\n    \"imports\": {\n      \"s736\": {\n        \"name\": {\n          \"schema\": \"mz_catalog\",\n          \"item\": \"mz_sources_ind\"\n        },\n        \"type\": \"compute\"\n      }\n    },\n    \"fast_path_clusters\": {},\n    \"fast_path_limit\": null,\n    \"persist_count\": []\n  },\n  \"cluster\": {\n    \"name\": \"mz_catalog_server\",\n    \"id\": {\n      \"System\": 2\n    }\n  },\n  \"redacted_sql\": \"SELECT '<REDACTED>' FROM [s461 AS mz_catalog.mz_sources] LIMIT '<REDACTED>'\"\n}","code":"MZ001","severity":"notice"}}
 {"type":"CommandStarting","payload":{"has_rows":true,"is_streaming":false}}
 {"type":"Rows","payload":{"columns":[{"name":"?column?","type_oid":23,"type_len":4,"type_mod":-1}]}}
 {"type":"Row","payload":["1"]}
@@ -412,7 +412,7 @@ ws-text
 ws-text
 {"query": "SELECT 1 / 0 FROM mz_sources LIMIT 1"}
 ----
-{"type":"Notice","payload":{"message":"{\n  \"plans\": {\n    \"raw\": {\n      \"text\": \"Finish limit=1 output=[#0]\\n  Project (#15)\\n    Map ((1 / 0))\\n      Get mz_catalog.mz_sources\\n\\nTarget cluster: mz_catalog_server\\n\",\n      \"json\": {\n        \"Project\": {\n          \"input\": {\n            \"Map\": {\n              \"input\": {\n                \"Get\": {\n                  \"id\": {\n                    \"Global\": {\n                      \"System\": 460\n                    }\n                  },\n                  \"typ\": {\n                    \"column_types\": [\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"Oid\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": {\n                          \"Array\": \"MzAclItem\"\n                        },\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      }\n                    ],\n                    \"keys\": [\n                      [\n                        0\n                      ],\n                      [\n                        1\n                      ]\n                    ]\n                  }\n                }\n              },\n              \"scalars\": [\n                {\n                  \"CallBinary\": {\n                    \"func\": \"DivInt32\",\n                    \"expr1\": {\n                      \"Literal\": [\n                        {\n                          \"data\": [\n                            45,\n                            1\n                          ]\n                        },\n                        {\n                          \"scalar_type\": \"Int32\",\n                          \"nullable\": false\n                        }\n                      ]\n                    },\n                    \"expr2\": {\n                      \"Literal\": [\n                        {\n                          \"data\": [\n                            44\n                          ]\n                        },\n                        {\n                          \"scalar_type\": \"Int32\",\n                          \"nullable\": false\n                        }\n                      ]\n                    }\n                  }\n                }\n              ]\n            }\n          },\n          \"outputs\": [\n            15\n          ]\n        }\n      }\n    },\n    \"optimized\": {\n      \"global\": {\n        \"text\": \"t75:\\n  Finish limit=1 output=[#0]\\n    ArrangeBy keys=[[#0]]\\n      ReadGlobalFromSameDataflow t74\\n\\nt74:\\n  Project (#15)\\n    Map (error(\\\"division by zero\\\"))\\n      ReadIndex on=mz_sources mz_sources_ind=[*** full scan ***]\\n\\nTarget cluster: mz_catalog_server\\n\",\n        \"json\": {\n          \"plans\": [\n            {\n              \"id\": \"t75\",\n              \"plan\": {\n                \"ArrangeBy\": {\n                  \"input\": {\n                    \"Get\": {\n                      \"id\": {\n                        \"Global\": {\n                          \"Transient\": 74\n                        }\n                      },\n                      \"typ\": {\n                        \"column_types\": [\n                          {\n                            \"scalar_type\": \"Int32\",\n                            \"nullable\": false\n                          }\n                        ],\n                        \"keys\": []\n                      },\n                      \"access_strategy\": \"SameDataflow\"\n                    }\n                  },\n                  \"keys\": [\n                    [\n                      {\n                        \"Column\": 0\n                      }\n                    ]\n                  ]\n                }\n              }\n            },\n            {\n              \"id\": \"t74\",\n              \"plan\": {\n                \"Project\": {\n                  \"input\": {\n                    \"Map\": {\n                      \"input\": {\n                        \"Get\": {\n                          \"id\": {\n                            \"Global\": {\n                              \"System\": 460\n                            }\n                          },\n                          \"typ\": {\n                            \"column_types\": [\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"Oid\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": {\n                                  \"Array\": \"MzAclItem\"\n                                },\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              }\n                            ],\n                            \"keys\": [\n                              [\n                                0\n                              ],\n                              [\n                                1\n                              ]\n                            ]\n                          },\n                          \"access_strategy\": {\n                            \"Index\": [\n                              [\n                                {\n                                  \"System\": 732\n                                },\n                                \"FullScan\"\n                              ]\n                            ]\n                          }\n                        }\n                      },\n                      \"scalars\": [\n                        {\n                          \"Literal\": [\n                            {\n                              \"Err\": \"DivisionByZero\"\n                            },\n                            {\n                              \"scalar_type\": \"Int32\",\n                              \"nullable\": false\n                            }\n                          ]\n                        }\n                      ]\n                    }\n                  },\n                  \"outputs\": [\n                    15\n                  ]\n                }\n              }\n            }\n          ],\n          \"sources\": []\n        }\n      },\n      \"fast_path\": {\n        \"text\": \"Explained Query (fast path):\\n  Finish limit=1 output=[#0]\\n    Project (#15)\\n      Map (error(\\\"division by zero\\\"))\\n        ReadIndex on=mz_catalog.mz_sources mz_sources_ind=[*** full scan ***]\\n\\nTarget cluster: mz_catalog_server\\n\",\n        \"json\": {\n          \"plans\": [\n            {\n              \"id\": \"Explained Query (fast path)\",\n              \"plan\": {\n                \"PeekExisting\": [\n                  {\n                    \"System\": 460\n                  },\n                  {\n                    \"System\": 732\n                  },\n                  null,\n                  {\n                    \"mfp\": {\n                      \"expressions\": [\n                        {\n                          \"Literal\": [\n                            {\n                              \"Err\": \"DivisionByZero\"\n                            },\n                            {\n                              \"scalar_type\": \"Int32\",\n                              \"nullable\": false\n                            }\n                          ]\n                        }\n                      ],\n                      \"predicates\": [],\n                      \"projection\": [\n                        15\n                      ],\n                      \"input_arity\": 15\n                    }\n                  }\n                ]\n              }\n            }\n          ],\n          \"sources\": []\n        }\n      }\n    }\n  },\n  \"insights\": {\n    \"imports\": {\n      \"s732\": {\n        \"name\": {\n          \"schema\": \"mz_catalog\",\n          \"item\": \"mz_sources_ind\"\n        },\n        \"type\": \"compute\"\n      }\n    },\n    \"fast_path_clusters\": {},\n    \"fast_path_limit\": null,\n    \"persist_count\": []\n  },\n  \"cluster\": {\n    \"name\": \"mz_catalog_server\",\n    \"id\": {\n      \"System\": 2\n    }\n  },\n  \"redacted_sql\": \"SELECT '<REDACTED>' / '<REDACTED>' FROM [s460 AS mz_catalog.mz_sources] LIMIT '<REDACTED>'\"\n}","code":"MZ001","severity":"notice"}}
+{"type":"Notice","payload":{"message":"{\n  \"plans\": {\n    \"raw\": {\n      \"text\": \"Finish limit=1 output=[#0]\\n  Project (#15)\\n    Map ((1 / 0))\\n      Get mz_catalog.mz_sources\\n\\nTarget cluster: mz_catalog_server\\n\",\n      \"json\": {\n        \"Project\": {\n          \"input\": {\n            \"Map\": {\n              \"input\": {\n                \"Get\": {\n                  \"id\": {\n                    \"Global\": {\n                      \"System\": 461\n                    }\n                  },\n                  \"typ\": {\n                    \"column_types\": [\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"Oid\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": {\n                          \"Array\": \"MzAclItem\"\n                        },\n                        \"nullable\": false\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      },\n                      {\n                        \"scalar_type\": \"String\",\n                        \"nullable\": true\n                      }\n                    ],\n                    \"keys\": [\n                      [\n                        0\n                      ],\n                      [\n                        1\n                      ]\n                    ]\n                  }\n                }\n              },\n              \"scalars\": [\n                {\n                  \"CallBinary\": {\n                    \"func\": \"DivInt32\",\n                    \"expr1\": {\n                      \"Literal\": [\n                        {\n                          \"data\": [\n                            45,\n                            1\n                          ]\n                        },\n                        {\n                          \"scalar_type\": \"Int32\",\n                          \"nullable\": false\n                        }\n                      ]\n                    },\n                    \"expr2\": {\n                      \"Literal\": [\n                        {\n                          \"data\": [\n                            44\n                          ]\n                        },\n                        {\n                          \"scalar_type\": \"Int32\",\n                          \"nullable\": false\n                        }\n                      ]\n                    }\n                  }\n                }\n              ]\n            }\n          },\n          \"outputs\": [\n            15\n          ]\n        }\n      }\n    },\n    \"optimized\": {\n      \"global\": {\n        \"text\": \"t75:\\n  Finish limit=1 output=[#0]\\n    ArrangeBy keys=[[#0]]\\n      ReadGlobalFromSameDataflow t74\\n\\nt74:\\n  Project (#15)\\n    Map (error(\\\"division by zero\\\"))\\n      ReadIndex on=mz_sources mz_sources_ind=[*** full scan ***]\\n\\nTarget cluster: mz_catalog_server\\n\",\n        \"json\": {\n          \"plans\": [\n            {\n              \"id\": \"t75\",\n              \"plan\": {\n                \"ArrangeBy\": {\n                  \"input\": {\n                    \"Get\": {\n                      \"id\": {\n                        \"Global\": {\n                          \"Transient\": 74\n                        }\n                      },\n                      \"typ\": {\n                        \"column_types\": [\n                          {\n                            \"scalar_type\": \"Int32\",\n                            \"nullable\": false\n                          }\n                        ],\n                        \"keys\": []\n                      },\n                      \"access_strategy\": \"SameDataflow\"\n                    }\n                  },\n                  \"keys\": [\n                    [\n                      {\n                        \"Column\": 0\n                      }\n                    ]\n                  ]\n                }\n              }\n            },\n            {\n              \"id\": \"t74\",\n              \"plan\": {\n                \"Project\": {\n                  \"input\": {\n                    \"Map\": {\n                      \"input\": {\n                        \"Get\": {\n                          \"id\": {\n                            \"Global\": {\n                              \"System\": 461\n                            }\n                          },\n                          \"typ\": {\n                            \"column_types\": [\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"Oid\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": {\n                                  \"Array\": \"MzAclItem\"\n                                },\n                                \"nullable\": false\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              },\n                              {\n                                \"scalar_type\": \"String\",\n                                \"nullable\": true\n                              }\n                            ],\n                            \"keys\": [\n                              [\n                                0\n                              ],\n                              [\n                                1\n                              ]\n                            ]\n                          },\n                          \"access_strategy\": {\n                            \"Index\": [\n                              [\n                                {\n                                  \"System\": 736\n                                },\n                                \"FullScan\"\n                              ]\n                            ]\n                          }\n                        }\n                      },\n                      \"scalars\": [\n                        {\n                          \"Literal\": [\n                            {\n                              \"Err\": \"DivisionByZero\"\n                            },\n                            {\n                              \"scalar_type\": \"Int32\",\n                              \"nullable\": false\n                            }\n                          ]\n                        }\n                      ]\n                    }\n                  },\n                  \"outputs\": [\n                    15\n                  ]\n                }\n              }\n            }\n          ],\n          \"sources\": []\n        }\n      },\n      \"fast_path\": {\n        \"text\": \"Explained Query (fast path):\\n  Finish limit=1 output=[#0]\\n    Project (#15)\\n      Map (error(\\\"division by zero\\\"))\\n        ReadIndex on=mz_catalog.mz_sources mz_sources_ind=[*** full scan ***]\\n\\nTarget cluster: mz_catalog_server\\n\",\n        \"json\": {\n          \"plans\": [\n            {\n              \"id\": \"Explained Query (fast path)\",\n              \"plan\": {\n                \"PeekExisting\": [\n                  {\n                    \"System\": 461\n                  },\n                  {\n                    \"System\": 736\n                  },\n                  null,\n                  {\n                    \"mfp\": {\n                      \"expressions\": [\n                        {\n                          \"Literal\": [\n                            {\n                              \"Err\": \"DivisionByZero\"\n                            },\n                            {\n                              \"scalar_type\": \"Int32\",\n                              \"nullable\": false\n                            }\n                          ]\n                        }\n                      ],\n                      \"predicates\": [],\n                      \"projection\": [\n                        15\n                      ],\n                      \"input_arity\": 15\n                    }\n                  }\n                ]\n              }\n            }\n          ],\n          \"sources\": []\n        }\n      }\n    }\n  },\n  \"insights\": {\n    \"imports\": {\n      \"s736\": {\n        \"name\": {\n          \"schema\": \"mz_catalog\",\n          \"item\": \"mz_sources_ind\"\n        },\n        \"type\": \"compute\"\n      }\n    },\n    \"fast_path_clusters\": {},\n    \"fast_path_limit\": null,\n    \"persist_count\": []\n  },\n  \"cluster\": {\n    \"name\": \"mz_catalog_server\",\n    \"id\": {\n      \"System\": 2\n    }\n  },\n  \"redacted_sql\": \"SELECT '<REDACTED>' / '<REDACTED>' FROM [s461 AS mz_catalog.mz_sources] LIMIT '<REDACTED>'\"\n}","code":"MZ001","severity":"notice"}}
 {"type":"CommandStarting","payload":{"has_rows":false,"is_streaming":false}}
 {"type":"Error","payload":{"message":"division by zero","code":"XX000"}}
 {"type":"ReadyForQuery","payload":"I"}
diff --git a/test/sqllogictest/autogenerated/mz_introspection.slt b/test/sqllogictest/autogenerated/mz_introspection.slt
index 9805b1b825070..b90f36fc63a42 100644
--- a/test/sqllogictest/autogenerated/mz_introspection.slt
+++ b/test/sqllogictest/autogenerated/mz_introspection.slt
@@ -126,6 +126,12 @@ SELECT position, name, type FROM objects WHERE schema = 'mz_introspection' AND o
 4  to_operator_id  uint8
 5  to_operator_address  list
 
+query ITT
+SELECT position, name, type FROM objects WHERE schema = 'mz_introspection' AND object = 'mz_dataflow_global_ids' ORDER BY position
+----
+1  id  uint8
+2  global_id  text
+
 query ITT
 SELECT position, name, type FROM objects WHERE schema = 'mz_introspection' AND object = 'mz_dataflow_operators' ORDER BY position
 ----
@@ -164,6 +170,17 @@ SELECT position, name, type FROM objects WHERE schema = 'mz_introspection' AND o
 7  savings  numeric
 8  hint  double␠precision
 
+query ITT
+SELECT position, name, type FROM objects WHERE schema = 'mz_introspection' AND object = 'mz_lir_mapping' ORDER BY position
+----
+1  global_id           text
+2  lir_id              uint8
+3  operator            text
+4  parent_lir_id       uint8
+5  nesting             uint2
+6  operator_id_start   uint8
+7  operator_id_end     uint8
+
 query ITT
 SELECT position, name, type FROM objects WHERE schema = 'mz_introspection' AND object = 'mz_message_counts' ORDER BY position
 ----
@@ -235,6 +252,7 @@ mz_arrangement_sharing_per_worker
 mz_arrangement_sharing_raw
 mz_arrangement_sizes
 mz_arrangement_sizes_per_worker
+mz_compute_dataflow_global_ids_per_worker
 mz_compute_error_counts
 mz_compute_error_counts_per_worker
 mz_compute_error_counts_raw
@@ -245,6 +263,7 @@ mz_compute_frontiers_per_worker
 mz_compute_hydration_times_per_worker
 mz_compute_import_frontiers
 mz_compute_import_frontiers_per_worker
+mz_compute_lir_mapping_per_worker
 mz_compute_operator_durations_histogram
 mz_compute_operator_durations_histogram_per_worker
 mz_compute_operator_durations_histogram_raw
@@ -255,6 +274,7 @@ mz_dataflow_channel_operators
 mz_dataflow_channel_operators_per_worker
 mz_dataflow_channels
 mz_dataflow_channels_per_worker
+mz_dataflow_global_ids
 mz_dataflow_operator_dataflows
 mz_dataflow_operator_dataflows_per_worker
 mz_dataflow_operator_parents
@@ -270,6 +290,7 @@ mz_dataflow_shutdown_durations_histogram_raw
 mz_dataflows
 mz_dataflows_per_worker
 mz_expected_group_size_advice
+mz_lir_mapping
 mz_message_batch_counts_received_raw
 mz_message_batch_counts_sent_raw
 mz_message_counts
diff --git a/test/sqllogictest/cluster.slt b/test/sqllogictest/cluster.slt
index 8c32bbc093e64..9e8ca980ca70b 100644
--- a/test/sqllogictest/cluster.slt
+++ b/test/sqllogictest/cluster.slt
@@ -210,6 +210,8 @@ bar  mz_arrangement_records_raw  mz_arrangement_records_raw_u7_primary_idx  1  o
 bar  mz_arrangement_records_raw  mz_arrangement_records_raw_u7_primary_idx  2  worker_id  NULL  false
 bar  mz_arrangement_sharing_raw  mz_arrangement_sharing_raw_u7_primary_idx  1  operator_id  NULL  false
 bar  mz_arrangement_sharing_raw  mz_arrangement_sharing_raw_u7_primary_idx  2  worker_id  NULL  false
+bar  mz_compute_dataflow_global_ids_per_worker  mz_compute_dataflow_global_ids_per_worker_u7_primary_idx  1  id  NULL  false
+bar  mz_compute_dataflow_global_ids_per_worker  mz_compute_dataflow_global_ids_per_worker_u7_primary_idx  2  worker_id  NULL  false
 bar  mz_compute_error_counts_raw  mz_compute_error_counts_raw_u7_primary_idx  1  export_id  NULL  false
 bar  mz_compute_error_counts_raw  mz_compute_error_counts_raw_u7_primary_idx  2  worker_id  NULL  false
 bar  mz_compute_exports_per_worker  mz_compute_exports_per_worker_u7_primary_idx  1  export_id  NULL  false
@@ -221,6 +223,9 @@ bar  mz_compute_hydration_times_per_worker  mz_compute_hydration_times_per_worke
 bar  mz_compute_import_frontiers_per_worker  mz_compute_import_frontiers_per_worker_u7_primary_idx  1  export_id  NULL  false
 bar  mz_compute_import_frontiers_per_worker  mz_compute_import_frontiers_per_worker_u7_primary_idx  2  import_id  NULL  false
 bar  mz_compute_import_frontiers_per_worker  mz_compute_import_frontiers_per_worker_u7_primary_idx  3  worker_id  NULL  false
+bar  mz_compute_lir_mapping_per_worker  mz_compute_lir_mapping_per_worker_u7_primary_idx  1  global_id  NULL  false
+bar  mz_compute_lir_mapping_per_worker  mz_compute_lir_mapping_per_worker_u7_primary_idx  2  lir_id  NULL  false
+bar  mz_compute_lir_mapping_per_worker  mz_compute_lir_mapping_per_worker_u7_primary_idx  3  worker_id  NULL  false
 bar  mz_compute_operator_durations_histogram_raw  mz_compute_operator_durations_histogram_raw_u7_primary_idx  1  id  NULL  false
 bar  mz_compute_operator_durations_histogram_raw  mz_compute_operator_durations_histogram_raw_u7_primary_idx  2  worker_id  NULL  false
 bar  mz_compute_operator_durations_histogram_raw  mz_compute_operator_durations_histogram_raw_u7_primary_idx  3  duration_ns  NULL  false
@@ -407,7 +412,7 @@ DROP CLUSTER foo, foo2, foo3, foo4 CASCADE
 query I
 SELECT COUNT(name) FROM mz_indexes WHERE cluster_id = 'u1';
 ----
-29
+31
 
 query I
 SELECT COUNT(name) FROM mz_indexes WHERE cluster_id <> 'u1' AND cluster_id NOT LIKE 's%';
@@ -420,7 +425,7 @@ CREATE CLUSTER test REPLICAS (foo (SIZE '1'));
 query I
 SELECT COUNT(name) FROM mz_indexes;
 ----
-271
+285
 
 statement ok
 DROP CLUSTER test CASCADE
@@ -428,7 +433,7 @@ DROP CLUSTER test CASCADE
 query T
 SELECT COUNT(name) FROM mz_indexes;
 ----
-242
+254
 
 simple conn=mz_system,user=mz_system
 ALTER CLUSTER quickstart OWNER TO materialize
diff --git a/test/sqllogictest/information_schema_tables.slt b/test/sqllogictest/information_schema_tables.slt
index 04758cd6a92ec..4d439cf53f0e6 100644
--- a/test/sqllogictest/information_schema_tables.slt
+++ b/test/sqllogictest/information_schema_tables.slt
@@ -825,6 +825,10 @@ mz_arrangement_sizes_per_worker
 VIEW
 materialize
 mz_introspection
+mz_compute_dataflow_global_ids_per_worker
+SOURCE
+materialize
+mz_introspection
 mz_compute_error_counts
 VIEW
 materialize
@@ -865,6 +869,10 @@ mz_compute_import_frontiers_per_worker
 SOURCE
 materialize
 mz_introspection
+mz_compute_lir_mapping_per_worker
+SOURCE
+materialize
+mz_introspection
 mz_compute_operator_durations_histogram
 VIEW
 materialize
@@ -905,6 +913,10 @@ mz_dataflow_channels_per_worker
 SOURCE
 materialize
 mz_introspection
+mz_dataflow_global_ids
+VIEW
+materialize
+mz_introspection
 mz_dataflow_operator_dataflows
 VIEW
 materialize
@@ -965,6 +977,10 @@ mz_expected_group_size_advice
 VIEW
 materialize
 mz_introspection
+mz_lir_mapping
+VIEW
+materialize
+mz_introspection
 mz_message_batch_counts_received_raw
 SOURCE
 materialize
diff --git a/test/sqllogictest/introspection/attribution_sources.slt b/test/sqllogictest/introspection/attribution_sources.slt
new file mode 100644
index 0000000000000..eb82c2d4dd717
--- /dev/null
+++ b/test/sqllogictest/introspection/attribution_sources.slt
@@ -0,0 +1,208 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# 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.
+
+# Ensure attribution sources function (are created, dropped appropriately)
+
+mode cockroach
+
+# VIEW + INDEX
+
+statement ok
+CREATE TABLE t(x INT NOT NULL, y INT, z TEXT);
+
+statement ok
+CREATE VIEW v AS
+  SELECT t1.x AS x, t1.z AS z1, t2.z AS z2
+  FROM t AS t1, t AS t2
+  WHERE t1.x = t2.y;
+
+statement ok
+CREATE INDEX v_idx_x ON v(x);
+
+# let the introspection sources update
+statement ok
+SELECT mz_unsafe.mz_sleep(2)
+
+query IT
+SELECT id, global_id FROM mz_internal.mz_dataflow_global_ids ORDER BY id, global_id;
+----
+8  u2
+8  u3
+
+query TI
+SELECT global_id, lir_id FROM mz_internal.mz_lir_mapping ORDER BY global_id, lir_id DESC;
+----
+u2  5
+u2  4
+u2  3
+u2  2
+u2  1
+u3  7
+u3  6
+
+## attribution queries
+
+# omitting sum(duration_ns) as duration, sum(count) as count
+query TIIT
+SELECT global_id, lir_id, parent_lir_id, REPEAT(' ', nesting * 2) || operator AS operator
+    FROM           mz_internal.mz_lir_mapping mlm
+         LEFT JOIN mz_introspection.mz_compute_operator_durations_histogram mcodh
+         ON (mlm.operator_id_start <= mcodh.id AND mcodh.id < mlm.operator_id_end)
+GROUP BY global_id, lir_id, operator, parent_lir_id, nesting
+ORDER BY global_id, lir_id DESC;
+----
+u2  5  NULL  Join::Differential␠2»␠4
+u2  4  5  ␠␠Arrange␠3
+u2  3  4  ␠␠␠␠Get::Collection␠u1
+u2  2  5  ␠␠Arrange␠1
+u2  1  2  ␠␠␠␠Get::Collection␠u1
+u3  7  NULL  Arrange␠6
+u3  6  7  ␠␠Get::PassArrangements␠u2
+
+# omitting pg_size_pretty(sum(size)) as size
+query TIIT
+SELECT global_id, lir_id, parent_lir_id, repeat(' ', nesting * 2) || operator AS operator
+    FROM           mz_internal.mz_lir_mapping mlm
+         LEFT JOIN mz_introspection.mz_arrangement_sizes mas
+         ON (mlm.operator_id_start <= mas.operator_id AND mas.operator_id < mlm.operator_id_end)
+GROUP BY global_id, lir_id, operator, parent_lir_id, nesting
+ORDER BY global_id, lir_id DESC;
+----
+u2  5  NULL  Join::Differential␠2»␠4
+u2  4  5  ␠␠Arrange␠3
+u2  3  4  ␠␠␠␠Get::Collection␠u1
+u2  2  5  ␠␠Arrange␠1
+u2  1  2  ␠␠␠␠Get::Collection␠u1
+u3  7  NULL  Arrange␠6
+u3  6  7  ␠␠Get::PassArrangements␠u2
+
+statement ok
+DROP TABLE t CASCADE;
+
+# we need the dataflow to actually drop to see the updates
+statement ok
+SELECT mz_unsafe.mz_sleep(3)
+
+query I
+SELECT COUNT(*) FROM mz_internal.mz_dataflow_global_ids;
+----
+0
+
+query I
+SELECT COUNT(*) FROM mz_internal.mz_lir_mapping;
+----
+0
+
+# MATERIALIZED VIEW
+
+statement ok
+CREATE TABLE u(x INT NOT NULL, y INT, z TEXT);
+
+statement ok
+CREATE MATERIALIZED VIEW w AS
+  SELECT t1.x AS x, t1.z AS z1, t2.z AS z2
+  FROM u AS t1, u AS t2
+  WHERE t1.x = t2.y;
+
+# let the introspection sources update
+statement ok
+SELECT mz_unsafe.mz_sleep(2)
+
+query IT
+SELECT id, global_id FROM mz_internal.mz_dataflow_global_ids ORDER BY id, global_id;
+----
+13  t59
+
+
+query TI
+SELECT global_id, lir_id FROM mz_internal.mz_lir_mapping ORDER BY global_id, lir_id DESC;
+----
+t59  5
+t59  4
+t59  3
+t59  2
+t59  1
+
+## attribution queries
+
+# omitting sum(duration_ns) as duration, sum(count) as count
+query TIIT
+SELECT global_id, lir_id, parent_lir_id, REPEAT(' ', nesting * 2) || operator AS operator
+    FROM           mz_internal.mz_lir_mapping mlm
+         LEFT JOIN mz_introspection.mz_compute_operator_durations_histogram mcodh
+         ON (mlm.operator_id_start <= mcodh.id AND mcodh.id < mlm.operator_id_end)
+GROUP BY global_id, lir_id, operator, parent_lir_id, nesting
+ORDER BY global_id, lir_id DESC;
+----
+t59  5  NULL  Join::Differential␠2»␠4
+t59  4  5  ␠␠Arrange␠3
+t59  3  4  ␠␠␠␠Get::Collection␠u4
+t59  2  5  ␠␠Arrange␠1
+t59  1  2  ␠␠␠␠Get::Collection␠u4
+
+# omitting pg_size_pretty(sum(size)) as size
+query TIIT
+SELECT global_id, lir_id, parent_lir_id, REPEAT(' ', nesting * 2) || operator AS operator
+    FROM           mz_internal.mz_lir_mapping mlm
+         LEFT JOIN mz_introspection.mz_arrangement_sizes mas
+         ON (mlm.operator_id_start <= mas.operator_id AND mas.operator_id < mlm.operator_id_end)
+GROUP BY global_id, lir_id, operator, parent_lir_id, nesting
+ORDER BY global_id, lir_id DESC;
+----
+t59  5  NULL  Join::Differential␠2»␠4
+t59  4  5  ␠␠Arrange␠3
+t59  3  4  ␠␠␠␠Get::Collection␠u4
+t59  2  5  ␠␠Arrange␠1
+t59  1  2  ␠␠␠␠Get::Collection␠u4
+
+statement ok
+DROP TABLE u CASCADE;
+
+# we need the dataflow to actually drop to see the updates
+statement ok
+SELECT mz_unsafe.mz_sleep(3)
+
+query I
+SELECT COUNT(*) FROM mz_internal.mz_dataflow_global_ids;
+----
+0
+
+query I
+SELECT COUNT(*) FROM mz_internal.mz_lir_mapping;
+----
+0
+
+# ATTRIBUTING TOP K HINTS
+
+statement ok
+CREATE TABLE t(x INT NOT NULL, y INT, z TEXT);
+
+statement ok
+CREATE VIEW v2 AS SELECT DISTINCT ON(x, y) * FROM t ORDER BY x, y;
+
+statement ok
+CREATE INDEX v2_idx_x ON v2(x);
+
+statement ok
+SELECT mz_unsafe.mz_sleep(2)
+
+query TIITIIIT
+SELECT mlm.global_id AS global_id, lir_id, parent_lir_id, REPEAT(' ', nesting * 2) || operator AS operator, levels, to_cut, savings, hint
+    FROM           mz_internal.mz_lir_mapping mlm
+         JOIN      mz_introspection.mz_dataflow_global_ids mdgi
+         ON (mlm.global_id = mdgi.global_id)
+         LEFT JOIN mz_introspection.mz_expected_group_size_advice megsa
+         ON (megsa.dataflow_id = mdgi.id AND
+             mlm.operator_id_start <= megsa.region_id AND megsa.region_id < mlm.operator_id_end)
+ORDER BY mlm.global_id, lir_id DESC;
+----
+u7  2  NULL  TopK::Basic␠1  8  7  3584  15.000
+u7  1  2  ␠␠Get::PassArrangements␠u6  NULL  NULL  NULL  NULL
+u8  4  NULL  Arrange␠3  NULL  NULL  NULL  NULL
+u8  3  4  ␠␠Get::PassArrangements␠u7  NULL  NULL  NULL  NULL
diff --git a/test/sqllogictest/mz_catalog_server_index_accounting.slt b/test/sqllogictest/mz_catalog_server_index_accounting.slt
index 4f6adee71c260..ea5e1ff16c220 100644
--- a/test/sqllogictest/mz_catalog_server_index_accounting.slt
+++ b/test/sqllogictest/mz_catalog_server_index_accounting.slt
@@ -37,92 +37,94 @@ mz_arrangement_heap_capacity_raw_s2_primary_idx  CREATE␠INDEX␠"mz_arrangemen
 mz_arrangement_heap_size_raw_s2_primary_idx  CREATE␠INDEX␠"mz_arrangement_heap_size_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_arrangement_heap_size_raw"␠("operator_id",␠"worker_id")
 mz_arrangement_records_raw_s2_primary_idx  CREATE␠INDEX␠"mz_arrangement_records_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_arrangement_records_raw"␠("operator_id",␠"worker_id")
 mz_arrangement_sharing_raw_s2_primary_idx  CREATE␠INDEX␠"mz_arrangement_sharing_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_arrangement_sharing_raw"␠("operator_id",␠"worker_id")
-mz_cluster_replica_history_ind  CREATE␠INDEX␠"mz_cluster_replica_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s577␠AS␠"mz_internal"."mz_cluster_replica_history"]␠("dropped_at")
-mz_cluster_replica_metrics_history_ind  CREATE␠INDEX␠"mz_cluster_replica_metrics_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s491␠AS␠"mz_internal"."mz_cluster_replica_metrics_history"]␠("replica_id")
-mz_cluster_replica_metrics_ind  CREATE␠INDEX␠"mz_cluster_replica_metrics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s490␠AS␠"mz_internal"."mz_cluster_replica_metrics"]␠("replica_id")
-mz_cluster_replica_name_history_ind  CREATE␠INDEX␠"mz_cluster_replica_name_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s578␠AS␠"mz_internal"."mz_cluster_replica_name_history"]␠("id")
-mz_cluster_replica_sizes_ind  CREATE␠INDEX␠"mz_cluster_replica_sizes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s492␠AS␠"mz_catalog"."mz_cluster_replica_sizes"]␠("size")
-mz_cluster_replica_status_history_ind  CREATE␠INDEX␠"mz_cluster_replica_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s494␠AS␠"mz_internal"."mz_cluster_replica_status_history"]␠("replica_id")
-mz_cluster_replica_statuses_ind  CREATE␠INDEX␠"mz_cluster_replica_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s493␠AS␠"mz_internal"."mz_cluster_replica_statuses"]␠("replica_id")
-mz_cluster_replicas_ind  CREATE␠INDEX␠"mz_cluster_replicas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s489␠AS␠"mz_catalog"."mz_cluster_replicas"]␠("id")
-mz_clusters_ind  CREATE␠INDEX␠"mz_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s483␠AS␠"mz_catalog"."mz_clusters"]␠("id")
-mz_columns_ind  CREATE␠INDEX␠"mz_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s456␠AS␠"mz_catalog"."mz_columns"]␠("name")
-mz_comments_ind  CREATE␠INDEX␠"mz_comments_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s506␠AS␠"mz_internal"."mz_comments"]␠("id")
-mz_compute_dependencies_ind  CREATE␠INDEX␠"mz_compute_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s698␠AS␠"mz_internal"."mz_compute_dependencies"]␠("dependency_id")
+mz_cluster_replica_history_ind  CREATE␠INDEX␠"mz_cluster_replica_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s579␠AS␠"mz_internal"."mz_cluster_replica_history"]␠("dropped_at")
+mz_cluster_replica_metrics_history_ind  CREATE␠INDEX␠"mz_cluster_replica_metrics_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s492␠AS␠"mz_internal"."mz_cluster_replica_metrics_history"]␠("replica_id")
+mz_cluster_replica_metrics_ind  CREATE␠INDEX␠"mz_cluster_replica_metrics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s491␠AS␠"mz_internal"."mz_cluster_replica_metrics"]␠("replica_id")
+mz_cluster_replica_name_history_ind  CREATE␠INDEX␠"mz_cluster_replica_name_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s580␠AS␠"mz_internal"."mz_cluster_replica_name_history"]␠("id")
+mz_cluster_replica_sizes_ind  CREATE␠INDEX␠"mz_cluster_replica_sizes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s493␠AS␠"mz_catalog"."mz_cluster_replica_sizes"]␠("size")
+mz_cluster_replica_status_history_ind  CREATE␠INDEX␠"mz_cluster_replica_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s495␠AS␠"mz_internal"."mz_cluster_replica_status_history"]␠("replica_id")
+mz_cluster_replica_statuses_ind  CREATE␠INDEX␠"mz_cluster_replica_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s494␠AS␠"mz_internal"."mz_cluster_replica_statuses"]␠("replica_id")
+mz_cluster_replicas_ind  CREATE␠INDEX␠"mz_cluster_replicas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s490␠AS␠"mz_catalog"."mz_cluster_replicas"]␠("id")
+mz_clusters_ind  CREATE␠INDEX␠"mz_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s484␠AS␠"mz_catalog"."mz_clusters"]␠("id")
+mz_columns_ind  CREATE␠INDEX␠"mz_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s457␠AS␠"mz_catalog"."mz_columns"]␠("name")
+mz_comments_ind  CREATE␠INDEX␠"mz_comments_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s507␠AS␠"mz_internal"."mz_comments"]␠("id")
+mz_compute_dataflow_global_ids_per_worker_s2_primary_idx  CREATE␠INDEX␠"mz_compute_dataflow_global_ids_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_dataflow_global_ids_per_worker"␠("id",␠"worker_id")
+mz_compute_dependencies_ind  CREATE␠INDEX␠"mz_compute_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s700␠AS␠"mz_internal"."mz_compute_dependencies"]␠("dependency_id")
 mz_compute_error_counts_raw_s2_primary_idx  CREATE␠INDEX␠"mz_compute_error_counts_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_error_counts_raw"␠("export_id",␠"worker_id")
 mz_compute_exports_per_worker_s2_primary_idx  CREATE␠INDEX␠"mz_compute_exports_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_exports_per_worker"␠("export_id",␠"worker_id")
 mz_compute_frontiers_per_worker_s2_primary_idx  CREATE␠INDEX␠"mz_compute_frontiers_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_frontiers_per_worker"␠("export_id",␠"worker_id")
 mz_compute_hydration_times_per_worker_s2_primary_idx  CREATE␠INDEX␠"mz_compute_hydration_times_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_hydration_times_per_worker"␠("export_id",␠"worker_id")
 mz_compute_import_frontiers_per_worker_s2_primary_idx  CREATE␠INDEX␠"mz_compute_import_frontiers_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_import_frontiers_per_worker"␠("export_id",␠"import_id",␠"worker_id")
+mz_compute_lir_mapping_per_worker_s2_primary_idx  CREATE␠INDEX␠"mz_compute_lir_mapping_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_lir_mapping_per_worker"␠("global_id",␠"lir_id",␠"worker_id")
 mz_compute_operator_durations_histogram_raw_s2_primary_idx  CREATE␠INDEX␠"mz_compute_operator_durations_histogram_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_operator_durations_histogram_raw"␠("id",␠"worker_id",␠"duration_ns")
-mz_connections_ind  CREATE␠INDEX␠"mz_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s487␠AS␠"mz_catalog"."mz_connections"]␠("schema_id")
-mz_console_cluster_utilization_overview_ind  CREATE␠INDEX␠"mz_console_cluster_utilization_overview_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s702␠AS␠"mz_internal"."mz_console_cluster_utilization_overview"]␠("cluster_id")
-mz_continual_tasks_ind  CREATE␠INDEX␠"mz_continual_tasks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s509␠AS␠"mz_internal"."mz_continual_tasks"]␠("id")
-mz_databases_ind  CREATE␠INDEX␠"mz_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s454␠AS␠"mz_catalog"."mz_databases"]␠("name")
+mz_connections_ind  CREATE␠INDEX␠"mz_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s488␠AS␠"mz_catalog"."mz_connections"]␠("schema_id")
+mz_console_cluster_utilization_overview_ind  CREATE␠INDEX␠"mz_console_cluster_utilization_overview_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s704␠AS␠"mz_internal"."mz_console_cluster_utilization_overview"]␠("cluster_id")
+mz_continual_tasks_ind  CREATE␠INDEX␠"mz_continual_tasks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s510␠AS␠"mz_internal"."mz_continual_tasks"]␠("id")
+mz_databases_ind  CREATE␠INDEX␠"mz_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s455␠AS␠"mz_catalog"."mz_databases"]␠("name")
 mz_dataflow_addresses_per_worker_s2_primary_idx  CREATE␠INDEX␠"mz_dataflow_addresses_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_dataflow_addresses_per_worker"␠("id",␠"worker_id")
 mz_dataflow_channels_per_worker_s2_primary_idx  CREATE␠INDEX␠"mz_dataflow_channels_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_dataflow_channels_per_worker"␠("id",␠"worker_id")
 mz_dataflow_operator_reachability_raw_s2_primary_idx  CREATE␠INDEX␠"mz_dataflow_operator_reachability_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_dataflow_operator_reachability_raw"␠("address",␠"port",␠"worker_id",␠"update_type",␠"time")
 mz_dataflow_operators_per_worker_s2_primary_idx  CREATE␠INDEX␠"mz_dataflow_operators_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_dataflow_operators_per_worker"␠("id",␠"worker_id")
 mz_dataflow_shutdown_durations_histogram_raw_s2_primary_idx  CREATE␠INDEX␠"mz_dataflow_shutdown_durations_histogram_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_dataflow_shutdown_durations_histogram_raw"␠("worker_id",␠"duration_ns")
-mz_frontiers_ind  CREATE␠INDEX␠"mz_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s692␠AS␠"mz_internal"."mz_frontiers"]␠("object_id")
-mz_indexes_ind  CREATE␠INDEX␠"mz_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s457␠AS␠"mz_catalog"."mz_indexes"]␠("id")
-mz_kafka_sources_ind  CREATE␠INDEX␠"mz_kafka_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s452␠AS␠"mz_catalog"."mz_kafka_sources"]␠("id")
-mz_materialized_views_ind  CREATE␠INDEX␠"mz_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s468␠AS␠"mz_catalog"."mz_materialized_views"]␠("id")
+mz_frontiers_ind  CREATE␠INDEX␠"mz_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s694␠AS␠"mz_internal"."mz_frontiers"]␠("object_id")
+mz_indexes_ind  CREATE␠INDEX␠"mz_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s458␠AS␠"mz_catalog"."mz_indexes"]␠("id")
+mz_kafka_sources_ind  CREATE␠INDEX␠"mz_kafka_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s453␠AS␠"mz_catalog"."mz_kafka_sources"]␠("id")
+mz_materialized_views_ind  CREATE␠INDEX␠"mz_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s469␠AS␠"mz_catalog"."mz_materialized_views"]␠("id")
 mz_message_batch_counts_received_raw_s2_primary_idx  CREATE␠INDEX␠"mz_message_batch_counts_received_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_batch_counts_received_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id")
 mz_message_batch_counts_sent_raw_s2_primary_idx  CREATE␠INDEX␠"mz_message_batch_counts_sent_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_batch_counts_sent_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id")
 mz_message_counts_received_raw_s2_primary_idx  CREATE␠INDEX␠"mz_message_counts_received_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_counts_received_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id")
 mz_message_counts_sent_raw_s2_primary_idx  CREATE␠INDEX␠"mz_message_counts_sent_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_counts_sent_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id")
-mz_notices_ind  CREATE␠INDEX␠"mz_notices_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s776␠AS␠"mz_internal"."mz_notices"]␠("id")
-mz_object_dependencies_ind  CREATE␠INDEX␠"mz_object_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s453␠AS␠"mz_internal"."mz_object_dependencies"]␠("object_id")
-mz_object_history_ind  CREATE␠INDEX␠"mz_object_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s517␠AS␠"mz_internal"."mz_object_history"]␠("id")
-mz_object_lifetimes_ind  CREATE␠INDEX␠"mz_object_lifetimes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s518␠AS␠"mz_internal"."mz_object_lifetimes"]␠("id")
-mz_object_transitive_dependencies_ind  CREATE␠INDEX␠"mz_object_transitive_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s530␠AS␠"mz_internal"."mz_object_transitive_dependencies"]␠("object_id")
-mz_objects_ind  CREATE␠INDEX␠"mz_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s514␠AS␠"mz_catalog"."mz_objects"]␠("schema_id")
+mz_notices_ind  CREATE␠INDEX␠"mz_notices_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s780␠AS␠"mz_internal"."mz_notices"]␠("id")
+mz_object_dependencies_ind  CREATE␠INDEX␠"mz_object_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s454␠AS␠"mz_internal"."mz_object_dependencies"]␠("object_id")
+mz_object_history_ind  CREATE␠INDEX␠"mz_object_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s518␠AS␠"mz_internal"."mz_object_history"]␠("id")
+mz_object_lifetimes_ind  CREATE␠INDEX␠"mz_object_lifetimes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s519␠AS␠"mz_internal"."mz_object_lifetimes"]␠("id")
+mz_object_transitive_dependencies_ind  CREATE␠INDEX␠"mz_object_transitive_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s532␠AS␠"mz_internal"."mz_object_transitive_dependencies"]␠("object_id")
+mz_objects_ind  CREATE␠INDEX␠"mz_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s515␠AS␠"mz_catalog"."mz_objects"]␠("schema_id")
 mz_peek_durations_histogram_raw_s2_primary_idx  CREATE␠INDEX␠"mz_peek_durations_histogram_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_peek_durations_histogram_raw"␠("worker_id",␠"type",␠"duration_ns")
-mz_recent_activity_log_thinned_ind  CREATE␠INDEX␠"mz_recent_activity_log_thinned_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s676␠AS␠"mz_internal"."mz_recent_activity_log_thinned"]␠("sql_hash")
-mz_recent_sql_text_ind  CREATE␠INDEX␠"mz_recent_sql_text_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s672␠AS␠"mz_internal"."mz_recent_sql_text"]␠("sql_hash")
-mz_recent_storage_usage_ind  CREATE␠INDEX␠"mz_recent_storage_usage_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s768␠AS␠"mz_catalog"."mz_recent_storage_usage"]␠("object_id")
-mz_roles_ind  CREATE␠INDEX␠"mz_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s476␠AS␠"mz_catalog"."mz_roles"]␠("id")
+mz_recent_activity_log_thinned_ind  CREATE␠INDEX␠"mz_recent_activity_log_thinned_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s678␠AS␠"mz_internal"."mz_recent_activity_log_thinned"]␠("sql_hash")
+mz_recent_sql_text_ind  CREATE␠INDEX␠"mz_recent_sql_text_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s674␠AS␠"mz_internal"."mz_recent_sql_text"]␠("sql_hash")
+mz_recent_storage_usage_ind  CREATE␠INDEX␠"mz_recent_storage_usage_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s772␠AS␠"mz_catalog"."mz_recent_storage_usage"]␠("object_id")
+mz_roles_ind  CREATE␠INDEX␠"mz_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s477␠AS␠"mz_catalog"."mz_roles"]␠("id")
 mz_scheduling_elapsed_raw_s2_primary_idx  CREATE␠INDEX␠"mz_scheduling_elapsed_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_scheduling_elapsed_raw"␠("id",␠"worker_id")
 mz_scheduling_parks_histogram_raw_s2_primary_idx  CREATE␠INDEX␠"mz_scheduling_parks_histogram_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_scheduling_parks_histogram_raw"␠("worker_id",␠"slept_for_ns",␠"requested_ns")
-mz_schemas_ind  CREATE␠INDEX␠"mz_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s455␠AS␠"mz_catalog"."mz_schemas"]␠("database_id")
-mz_secrets_ind  CREATE␠INDEX␠"mz_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s486␠AS␠"mz_catalog"."mz_secrets"]␠("name")
-mz_show_all_objects_ind  CREATE␠INDEX␠"mz_show_all_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s561␠AS␠"mz_internal"."mz_show_all_objects"]␠("schema_id")
-mz_show_cluster_replicas_ind  CREATE␠INDEX␠"mz_show_cluster_replicas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s711␠AS␠"mz_internal"."mz_show_cluster_replicas"]␠("cluster")
-mz_show_clusters_ind  CREATE␠INDEX␠"mz_show_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s563␠AS␠"mz_internal"."mz_show_clusters"]␠("name")
-mz_show_columns_ind  CREATE␠INDEX␠"mz_show_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s562␠AS␠"mz_internal"."mz_show_columns"]␠("id")
-mz_show_connections_ind  CREATE␠INDEX␠"mz_show_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s571␠AS␠"mz_internal"."mz_show_connections"]␠("schema_id")
-mz_show_databases_ind  CREATE␠INDEX␠"mz_show_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s565␠AS␠"mz_internal"."mz_show_databases"]␠("name")
-mz_show_indexes_ind  CREATE␠INDEX␠"mz_show_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s575␠AS␠"mz_internal"."mz_show_indexes"]␠("schema_id")
-mz_show_materialized_views_ind  CREATE␠INDEX␠"mz_show_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s574␠AS␠"mz_internal"."mz_show_materialized_views"]␠("schema_id")
-mz_show_roles_ind  CREATE␠INDEX␠"mz_show_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s570␠AS␠"mz_internal"."mz_show_roles"]␠("name")
-mz_show_schemas_ind  CREATE␠INDEX␠"mz_show_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s566␠AS␠"mz_internal"."mz_show_schemas"]␠("database_id")
-mz_show_secrets_ind  CREATE␠INDEX␠"mz_show_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s564␠AS␠"mz_internal"."mz_show_secrets"]␠("schema_id")
-mz_show_sinks_ind  CREATE␠INDEX␠"mz_show_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s573␠AS␠"mz_internal"."mz_show_sinks"]␠("schema_id")
-mz_show_sources_ind  CREATE␠INDEX␠"mz_show_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s572␠AS␠"mz_internal"."mz_show_sources"]␠("schema_id")
-mz_show_tables_ind  CREATE␠INDEX␠"mz_show_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s567␠AS␠"mz_internal"."mz_show_tables"]␠("schema_id")
-mz_show_types_ind  CREATE␠INDEX␠"mz_show_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s569␠AS␠"mz_internal"."mz_show_types"]␠("schema_id")
-mz_show_views_ind  CREATE␠INDEX␠"mz_show_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s568␠AS␠"mz_internal"."mz_show_views"]␠("schema_id")
-mz_sink_statistics_ind  CREATE␠INDEX␠"mz_sink_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s689␠AS␠"mz_internal"."mz_sink_statistics"]␠("id")
-mz_sink_status_history_ind  CREATE␠INDEX␠"mz_sink_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s661␠AS␠"mz_internal"."mz_sink_status_history"]␠("sink_id")
-mz_sink_statuses_ind  CREATE␠INDEX␠"mz_sink_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s662␠AS␠"mz_internal"."mz_sink_statuses"]␠("id")
-mz_sinks_ind  CREATE␠INDEX␠"mz_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s466␠AS␠"mz_catalog"."mz_sinks"]␠("id")
-mz_source_statistics_ind  CREATE␠INDEX␠"mz_source_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s687␠AS␠"mz_internal"."mz_source_statistics"]␠("id")
-mz_source_statistics_with_history_ind  CREATE␠INDEX␠"mz_source_statistics_with_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s685␠AS␠"mz_internal"."mz_source_statistics_with_history"]␠("id")
-mz_source_status_history_ind  CREATE␠INDEX␠"mz_source_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s663␠AS␠"mz_internal"."mz_source_status_history"]␠("source_id")
-mz_source_statuses_ind  CREATE␠INDEX␠"mz_source_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s680␠AS␠"mz_internal"."mz_source_statuses"]␠("id")
-mz_sources_ind  CREATE␠INDEX␠"mz_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s460␠AS␠"mz_catalog"."mz_sources"]␠("id")
-mz_tables_ind  CREATE␠INDEX␠"mz_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s459␠AS␠"mz_catalog"."mz_tables"]␠("schema_id")
-mz_types_ind  CREATE␠INDEX␠"mz_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s470␠AS␠"mz_catalog"."mz_types"]␠("schema_id")
-mz_views_ind  CREATE␠INDEX␠"mz_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s467␠AS␠"mz_catalog"."mz_views"]␠("schema_id")
-mz_wallclock_global_lag_recent_history_ind  CREATE␠INDEX␠"mz_wallclock_global_lag_recent_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s696␠AS␠"mz_internal"."mz_wallclock_global_lag_recent_history"]␠("object_id")
-mz_webhook_sources_ind  CREATE␠INDEX␠"mz_webhook_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s507␠AS␠"mz_internal"."mz_webhook_sources"]␠("id")
-pg_attrdef_all_databases_ind  CREATE␠INDEX␠"pg_attrdef_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s603␠AS␠"mz_internal"."pg_attrdef_all_databases"]␠("oid",␠"adrelid",␠"adnum",␠"adbin",␠"adsrc")
-pg_attribute_all_databases_ind  CREATE␠INDEX␠"pg_attribute_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s596␠AS␠"mz_internal"."pg_attribute_all_databases"]␠("attrelid",␠"attname",␠"atttypid",␠"attlen",␠"attnum",␠"atttypmod",␠"attnotnull",␠"atthasdef",␠"attidentity",␠"attgenerated",␠"attisdropped",␠"attcollation",␠"database_name",␠"pg_type_database_name")
-pg_class_all_databases_ind  CREATE␠INDEX␠"pg_class_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s584␠AS␠"mz_internal"."pg_class_all_databases"]␠("relname")
-pg_description_all_databases_ind  CREATE␠INDEX␠"pg_description_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s593␠AS␠"mz_internal"."pg_description_all_databases"]␠("objoid",␠"classoid",␠"objsubid",␠"description",␠"oid_database_name",␠"class_database_name")
-pg_namespace_all_databases_ind  CREATE␠INDEX␠"pg_namespace_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s581␠AS␠"mz_internal"."pg_namespace_all_databases"]␠("nspname")
-pg_type_all_databases_ind  CREATE␠INDEX␠"pg_type_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s590␠AS␠"mz_internal"."pg_type_all_databases"]␠("oid")
+mz_schemas_ind  CREATE␠INDEX␠"mz_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s456␠AS␠"mz_catalog"."mz_schemas"]␠("database_id")
+mz_secrets_ind  CREATE␠INDEX␠"mz_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s487␠AS␠"mz_catalog"."mz_secrets"]␠("name")
+mz_show_all_objects_ind  CREATE␠INDEX␠"mz_show_all_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s563␠AS␠"mz_internal"."mz_show_all_objects"]␠("schema_id")
+mz_show_cluster_replicas_ind  CREATE␠INDEX␠"mz_show_cluster_replicas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s715␠AS␠"mz_internal"."mz_show_cluster_replicas"]␠("cluster")
+mz_show_clusters_ind  CREATE␠INDEX␠"mz_show_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s565␠AS␠"mz_internal"."mz_show_clusters"]␠("name")
+mz_show_columns_ind  CREATE␠INDEX␠"mz_show_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s564␠AS␠"mz_internal"."mz_show_columns"]␠("id")
+mz_show_connections_ind  CREATE␠INDEX␠"mz_show_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s573␠AS␠"mz_internal"."mz_show_connections"]␠("schema_id")
+mz_show_databases_ind  CREATE␠INDEX␠"mz_show_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s567␠AS␠"mz_internal"."mz_show_databases"]␠("name")
+mz_show_indexes_ind  CREATE␠INDEX␠"mz_show_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s577␠AS␠"mz_internal"."mz_show_indexes"]␠("schema_id")
+mz_show_materialized_views_ind  CREATE␠INDEX␠"mz_show_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s576␠AS␠"mz_internal"."mz_show_materialized_views"]␠("schema_id")
+mz_show_roles_ind  CREATE␠INDEX␠"mz_show_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s572␠AS␠"mz_internal"."mz_show_roles"]␠("name")
+mz_show_schemas_ind  CREATE␠INDEX␠"mz_show_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s568␠AS␠"mz_internal"."mz_show_schemas"]␠("database_id")
+mz_show_secrets_ind  CREATE␠INDEX␠"mz_show_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s566␠AS␠"mz_internal"."mz_show_secrets"]␠("schema_id")
+mz_show_sinks_ind  CREATE␠INDEX␠"mz_show_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s575␠AS␠"mz_internal"."mz_show_sinks"]␠("schema_id")
+mz_show_sources_ind  CREATE␠INDEX␠"mz_show_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s574␠AS␠"mz_internal"."mz_show_sources"]␠("schema_id")
+mz_show_tables_ind  CREATE␠INDEX␠"mz_show_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s569␠AS␠"mz_internal"."mz_show_tables"]␠("schema_id")
+mz_show_types_ind  CREATE␠INDEX␠"mz_show_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s571␠AS␠"mz_internal"."mz_show_types"]␠("schema_id")
+mz_show_views_ind  CREATE␠INDEX␠"mz_show_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s570␠AS␠"mz_internal"."mz_show_views"]␠("schema_id")
+mz_sink_statistics_ind  CREATE␠INDEX␠"mz_sink_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s691␠AS␠"mz_internal"."mz_sink_statistics"]␠("id")
+mz_sink_status_history_ind  CREATE␠INDEX␠"mz_sink_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s663␠AS␠"mz_internal"."mz_sink_status_history"]␠("sink_id")
+mz_sink_statuses_ind  CREATE␠INDEX␠"mz_sink_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s664␠AS␠"mz_internal"."mz_sink_statuses"]␠("id")
+mz_sinks_ind  CREATE␠INDEX␠"mz_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s467␠AS␠"mz_catalog"."mz_sinks"]␠("id")
+mz_source_statistics_ind  CREATE␠INDEX␠"mz_source_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s689␠AS␠"mz_internal"."mz_source_statistics"]␠("id")
+mz_source_statistics_with_history_ind  CREATE␠INDEX␠"mz_source_statistics_with_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s687␠AS␠"mz_internal"."mz_source_statistics_with_history"]␠("id")
+mz_source_status_history_ind  CREATE␠INDEX␠"mz_source_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s665␠AS␠"mz_internal"."mz_source_status_history"]␠("source_id")
+mz_source_statuses_ind  CREATE␠INDEX␠"mz_source_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s682␠AS␠"mz_internal"."mz_source_statuses"]␠("id")
+mz_sources_ind  CREATE␠INDEX␠"mz_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s461␠AS␠"mz_catalog"."mz_sources"]␠("id")
+mz_tables_ind  CREATE␠INDEX␠"mz_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s460␠AS␠"mz_catalog"."mz_tables"]␠("schema_id")
+mz_types_ind  CREATE␠INDEX␠"mz_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s471␠AS␠"mz_catalog"."mz_types"]␠("schema_id")
+mz_views_ind  CREATE␠INDEX␠"mz_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s468␠AS␠"mz_catalog"."mz_views"]␠("schema_id")
+mz_wallclock_global_lag_recent_history_ind  CREATE␠INDEX␠"mz_wallclock_global_lag_recent_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s698␠AS␠"mz_internal"."mz_wallclock_global_lag_recent_history"]␠("object_id")
+mz_webhook_sources_ind  CREATE␠INDEX␠"mz_webhook_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s508␠AS␠"mz_internal"."mz_webhook_sources"]␠("id")
+pg_attrdef_all_databases_ind  CREATE␠INDEX␠"pg_attrdef_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s605␠AS␠"mz_internal"."pg_attrdef_all_databases"]␠("oid",␠"adrelid",␠"adnum",␠"adbin",␠"adsrc")
+pg_attribute_all_databases_ind  CREATE␠INDEX␠"pg_attribute_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s598␠AS␠"mz_internal"."pg_attribute_all_databases"]␠("attrelid",␠"attname",␠"atttypid",␠"attlen",␠"attnum",␠"atttypmod",␠"attnotnull",␠"atthasdef",␠"attidentity",␠"attgenerated",␠"attisdropped",␠"attcollation",␠"database_name",␠"pg_type_database_name")
+pg_class_all_databases_ind  CREATE␠INDEX␠"pg_class_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s586␠AS␠"mz_internal"."pg_class_all_databases"]␠("relname")
+pg_description_all_databases_ind  CREATE␠INDEX␠"pg_description_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s595␠AS␠"mz_internal"."pg_description_all_databases"]␠("objoid",␠"classoid",␠"objsubid",␠"description",␠"oid_database_name",␠"class_database_name")
+pg_namespace_all_databases_ind  CREATE␠INDEX␠"pg_namespace_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s583␠AS␠"mz_internal"."pg_namespace_all_databases"]␠("nspname")
+pg_type_all_databases_ind  CREATE␠INDEX␠"pg_type_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s592␠AS␠"mz_internal"."pg_type_all_databases"]␠("oid")
 
 # Record all transitive dependencies (tables, sources, views, mvs) of indexes on
 # the mz_catalog_server cluster.
@@ -273,6 +275,9 @@ mz_comments  comment
 mz_comments  id
 mz_comments  object_sub_id
 mz_comments  object_type
+mz_compute_dataflow_global_ids_per_worker  global_id
+mz_compute_dataflow_global_ids_per_worker  id
+mz_compute_dataflow_global_ids_per_worker  worker_id
 mz_compute_dependencies  dependency_id
 mz_compute_dependencies  object_id
 mz_compute_error_counts_raw  count
@@ -298,6 +303,14 @@ mz_compute_import_frontiers_per_worker  export_id
 mz_compute_import_frontiers_per_worker  import_id
 mz_compute_import_frontiers_per_worker  time
 mz_compute_import_frontiers_per_worker  worker_id
+mz_compute_lir_mapping_per_worker  global_id
+mz_compute_lir_mapping_per_worker  lir_id
+mz_compute_lir_mapping_per_worker  nesting
+mz_compute_lir_mapping_per_worker  operator
+mz_compute_lir_mapping_per_worker  operator_id_end
+mz_compute_lir_mapping_per_worker  operator_id_start
+mz_compute_lir_mapping_per_worker  parent_lir_id
+mz_compute_lir_mapping_per_worker  worker_id
 mz_compute_operator_durations_histogram_raw  duration_ns
 mz_compute_operator_durations_histogram_raw  id
 mz_compute_operator_durations_histogram_raw  worker_id
diff --git a/test/sqllogictest/oid.slt b/test/sqllogictest/oid.slt
index ccfee8dc83033..79ab0a04aec60 100644
--- a/test/sqllogictest/oid.slt
+++ b/test/sqllogictest/oid.slt
@@ -1149,3 +1149,7 @@ SELECT oid, name FROM mz_objects WHERE id LIKE 's%' AND oid < 20000 ORDER BY oid
 17041  mz_network_policy_rules
 17042  mz_index_advice
 17043  mz_show_network_policies
+17044  mz_compute_lir_mapping_per_worker
+17045  mz_compute_dataflow_global_ids_per_worker
+17046  mz_lir_mapping
+17047  mz_dataflow_global_ids
diff --git a/test/sqllogictest/pg_catalog_user.slt b/test/sqllogictest/pg_catalog_user.slt
index 703c246f740b5..06f94c639473a 100644
--- a/test/sqllogictest/pg_catalog_user.slt
+++ b/test/sqllogictest/pg_catalog_user.slt
@@ -21,7 +21,7 @@ CREATE ROLE "materialize@foocorp.io"
 query TIBBBBTTT rowsort
 SELECT usename, usesysid, usecreatedb, usesuper, userepl, usebypassrls, passwd, valuntil, useconfig FROM pg_user;
 ----
-materialize@foocorp.io  20178  false  NULL  false  false  ********  NULL  NULL
+materialize@foocorp.io  20190  false  NULL  false  false  ********  NULL  NULL
 mz_support  16662  false  NULL  false  false  ********  NULL  NULL
 mz_system  16661  true  true  false  false  ********  NULL  NULL
 
diff --git a/test/sqllogictest/regclass.slt b/test/sqllogictest/regclass.slt
index 84a67095de2c9..0267ea8995f5d 100644
--- a/test/sqllogictest/regclass.slt
+++ b/test/sqllogictest/regclass.slt
@@ -35,12 +35,12 @@ CREATE MATERIALIZED VIEW s.m AS SELECT * FROM s.t;
 query T
 SELECT 't'::regclass::oid::int
 ----
-20178
+20190
 
 query T
 SELECT 's.t'::regclass::oid::int
 ----
-20179
+20191
 
 query T
 SELECT 't'::regclass = 's.t'::regclass
@@ -73,7 +73,7 @@ t
 query T
 SELECT 't'::regclass::oid::int
 ----
-20179
+20191
 
 query T
 SELECT 'public.t'::regclass::text;
@@ -101,12 +101,12 @@ d.public.t
 query T
 SELECT 'm'::regclass::oid::int
 ----
-20183
+20195
 
 query T
 SELECT 's.m'::regclass::oid::int
 ----
-20184
+20196
 
 query T
 SELECT 'm'::regclass = 's.m'::regclass
@@ -296,7 +296,7 @@ true
 query T
 SELECT 'materialize.public.t'::regclass::oid::int
 ----
-20178
+20190
 
 query error db error: ERROR: relation "t" does not exist
 SELECT 't'::regclass::oid::int
diff --git a/test/sqllogictest/regtype.slt b/test/sqllogictest/regtype.slt
index f8d336c220c55..f6a434f7e0d20 100644
--- a/test/sqllogictest/regtype.slt
+++ b/test/sqllogictest/regtype.slt
@@ -142,12 +142,12 @@ CREATE TYPE d.public.t AS LIST (ELEMENT TYPE = int4);
 query T
 SELECT 't'::regtype::oid::int
 ----
-20179
+20191
 
 query T
 SELECT 's.t'::regtype::oid::int
 ----
-20180
+20192
 
 query T
 SELECT 't'::regtype = 's.t'::regtype
diff --git a/test/testdrive-old-kafka-src-syntax/indexes.td b/test/testdrive-old-kafka-src-syntax/indexes.td
index b16708bbe6a97..0547741dd4e7e 100644
--- a/test/testdrive-old-kafka-src-syntax/indexes.td
+++ b/test/testdrive-old-kafka-src-syntax/indexes.td
@@ -303,11 +303,13 @@ mz_clusters_ind                                             mz_clusters
 mz_columns_ind                                              mz_columns                                   mz_catalog_server    {name}                                      ""
 mz_comments_ind                                             mz_comments                                  mz_catalog_server    {id}                                        ""
 mz_compute_dependencies_ind                                 mz_compute_dependencies                      mz_catalog_server    {dependency_id}                             ""
+mz_compute_dataflow_global_ids_per_worker_s2_primary_idx   mz_compute_dataflow_global_ids_per_worker     mz_catalog_server    {id,worker_id}                             ""
 mz_compute_error_counts_raw_s2_primary_idx                  mz_compute_error_counts_raw                  mz_catalog_server    {export_id,worker_id}                       ""
 mz_compute_exports_per_worker_s2_primary_idx                mz_compute_exports_per_worker                mz_catalog_server    {export_id,worker_id}                       ""
 mz_compute_frontiers_per_worker_s2_primary_idx              mz_compute_frontiers_per_worker              mz_catalog_server    {export_id,worker_id}                       ""
 mz_compute_hydration_times_per_worker_s2_primary_idx        mz_compute_hydration_times_per_worker        mz_catalog_server    {export_id,worker_id}                       ""
 mz_compute_import_frontiers_per_worker_s2_primary_idx       mz_compute_import_frontiers_per_worker       mz_catalog_server    {export_id,import_id,worker_id}             ""
+mz_compute_lir_mapping_per_worker_s2_primary_idx            mz_compute_lir_mapping_per_worker            mz_catalog_server    {global_id,lir_id,worker_id}                ""
 mz_compute_operator_durations_histogram_raw_s2_primary_idx  mz_compute_operator_durations_histogram_raw  mz_catalog_server    {id,worker_id,duration_ns}                  ""
 mz_connections_ind                                          mz_connections                               mz_catalog_server    {schema_id}                                 ""
 mz_console_cluster_utilization_overview_ind                 mz_console_cluster_utilization_overview      mz_catalog_server    {cluster_id}                                ""
diff --git a/test/testdrive/catalog.td b/test/testdrive/catalog.td
index 801712c53dfef..044f37c5aa0bb 100644
--- a/test/testdrive/catalog.td
+++ b/test/testdrive/catalog.td
@@ -715,11 +715,13 @@ mz_arrangement_heap_capacity_raw             log   <null>    ""
 mz_arrangement_heap_size_raw                 log   <null>    ""
 mz_arrangement_records_raw                   log   <null>    ""
 mz_arrangement_sharing_raw                   log   <null>    ""
+mz_compute_dataflow_global_ids_per_worker     log   <null>    ""
 mz_compute_error_counts_raw                  log   <null>    ""
 mz_compute_exports_per_worker                log   <null>    ""
 mz_compute_frontiers_per_worker              log   <null>    ""
 mz_compute_hydration_times_per_worker        log   <null>    ""
 mz_compute_import_frontiers_per_worker       log   <null>    ""
+mz_compute_lir_mapping_per_worker            log   <null>    ""
 mz_compute_operator_durations_histogram_raw  log   <null>    ""
 mz_dataflow_addresses_per_worker             log   <null>    ""
 mz_dataflow_channels_per_worker              log   <null>    ""
@@ -754,6 +756,7 @@ mz_dataflow_arrangement_sizes ""
 mz_dataflow_channel_operators ""
 mz_dataflow_channel_operators_per_worker ""
 mz_dataflow_channels ""
+mz_dataflow_global_ids ""
 mz_dataflow_operator_dataflows ""
 mz_dataflow_operator_dataflows_per_worker ""
 mz_dataflow_operator_parents ""
@@ -766,6 +769,7 @@ mz_dataflow_shutdown_durations_histogram_per_worker ""
 mz_dataflows ""
 mz_dataflows_per_worker ""
 mz_expected_group_size_advice ""
+mz_lir_mapping ""
 mz_message_counts ""
 mz_message_counts_per_worker ""
 mz_peek_durations_histogram ""
@@ -796,7 +800,7 @@ test_table ""
 
 # There is one entry in mz_indexes for each field_number/expression of the index.
 > SELECT COUNT(id) FROM mz_indexes WHERE id LIKE 's%'
-242
+254
 
 # Create a second schema with the same table name as above
 > CREATE SCHEMA tester2
diff --git a/test/testdrive/distinct-arrangements.td b/test/testdrive/distinct-arrangements.td
index 51ed3ed963f6e..8f044dab6771f 100644
--- a/test/testdrive/distinct-arrangements.td
+++ b/test/testdrive/distinct-arrangements.td
@@ -781,8 +781,10 @@ ReduceMinsMaxes
 "Arrange Compute(ArrangementHeapCapacity)"
 "Arrange Compute(ArrangementHeapSize)"
 "Arrange Compute(DataflowCurrent)"
+"Arrange Compute(DataflowGlobal)"
 "Arrange Compute(ErrorCount)"
 "Arrange Compute(FrontierCurrent)"
+"Arrange Compute(LirMapping)"
 "Arrange Compute(HydrationTime)"
 "Arrange Compute(ImportFrontierCurrent)"
 "Arrange Compute(PeekCurrent)"
diff --git a/test/testdrive/divergent-dataflow-cancellation.td b/test/testdrive/divergent-dataflow-cancellation.td
index 6672101715380..eaed19d1eb818 100644
--- a/test/testdrive/divergent-dataflow-cancellation.td
+++ b/test/testdrive/divergent-dataflow-cancellation.td
@@ -88,13 +88,13 @@ contains: canceling statement due to statement timeout
 > SELECT count(*)
   FROM mz_introspection.mz_compute_exports_per_worker
   WHERE worker_id = 0
-29
+31
 
 # One frontier for each introspection arrangement.
 > SELECT count(*)
   FROM mz_introspection.mz_compute_frontiers_per_worker
   WHERE worker_id = 0
-29
+31
 
 > SELECT count(*) FROM mz_introspection.mz_compute_import_frontiers_per_worker
 0
diff --git a/test/testdrive/indexes.td b/test/testdrive/indexes.td
index 4e7799538fcf5..620c8a3de8d61 100644
--- a/test/testdrive/indexes.td
+++ b/test/testdrive/indexes.td
@@ -320,11 +320,13 @@ mz_clusters_ind                                             mz_clusters
 mz_columns_ind                                              mz_columns                                   mz_catalog_server    {name}                                      ""
 mz_comments_ind                                             mz_comments                                  mz_catalog_server    {id}                                        ""
 mz_compute_dependencies_ind                                 mz_compute_dependencies                      mz_catalog_server    {dependency_id}                             ""
+mz_compute_dataflow_global_ids_per_worker_s2_primary_idx   mz_compute_dataflow_global_ids_per_worker     mz_catalog_server    {id,worker_id}                             ""
 mz_compute_error_counts_raw_s2_primary_idx                  mz_compute_error_counts_raw                  mz_catalog_server    {export_id,worker_id}                       ""
 mz_compute_exports_per_worker_s2_primary_idx                mz_compute_exports_per_worker                mz_catalog_server    {export_id,worker_id}                       ""
 mz_compute_frontiers_per_worker_s2_primary_idx              mz_compute_frontiers_per_worker              mz_catalog_server    {export_id,worker_id}                       ""
 mz_compute_hydration_times_per_worker_s2_primary_idx        mz_compute_hydration_times_per_worker        mz_catalog_server    {export_id,worker_id}                       ""
 mz_compute_import_frontiers_per_worker_s2_primary_idx       mz_compute_import_frontiers_per_worker       mz_catalog_server    {export_id,import_id,worker_id}             ""
+mz_compute_lir_mapping_per_worker_s2_primary_idx            mz_compute_lir_mapping_per_worker            mz_catalog_server    {global_id,lir_id,worker_id}                ""
 mz_compute_operator_durations_histogram_raw_s2_primary_idx  mz_compute_operator_durations_histogram_raw  mz_catalog_server    {id,worker_id,duration_ns}                  ""
 mz_connections_ind                                          mz_connections                               mz_catalog_server    {schema_id}                                 ""
 mz_console_cluster_utilization_overview_ind                 mz_console_cluster_utilization_overview      mz_catalog_server    {cluster_id}                                ""