coord: initialize read policy for log indexes

src/adapter/src/catalog.rs

@@ -3165,7 +3165,7 @@ impl<S: Append> Catalog<S> {
                 Op::CreateComputeInstance {
                     name,
                     config,
-                    introspection_sources,
+                    introspection_source_indexes: introspection_sources,
                 } => {
                     if is_reserved_name(&name) {
                         return Err(AdapterError::Catalog(Error::new(
@@ -4030,7 +4030,7 @@ pub enum Op {
     CreateComputeInstance {
         name: String,
         config: Option<ComputeInstanceIntrospectionConfig>,
-        introspection_sources: Vec<(&'static BuiltinLog, GlobalId)>,
+        introspection_source_indexes: Vec<(&'static BuiltinLog, GlobalId)>,
     },
     CreateComputeInstanceReplica {
         name: String,

src/adapter/src/coord/sequencer.rs

@@ -579,15 +579,19 @@ impl<S: Append + 'static> Coordinator<S> {
         }: CreateComputeInstancePlan,
     ) -> Result<ExecuteResponse, AdapterError> {
         tracing::debug!("sequence_create_compute_instance");
-        let introspection_sources = if compute_instance_config.is_some() {
+        let introspection_source_indexes = if compute_instance_config.is_some() {
             self.catalog.allocate_introspection_source_indexes().await
         } else {
             Vec::new()
         };
+        let introspection_source_index_ids: Vec<_> = introspection_source_indexes
+            .iter()
+            .map(|(_, id)| *id)
+            .collect();
         let mut ops = vec![catalog::Op::CreateComputeInstance {
             name: name.clone(),
             config: compute_instance_config.clone(),
-            introspection_sources,
+            introspection_source_indexes,
         }];
 
         let azs = self.catalog.state().availability_zones();
@@ -690,6 +694,14 @@ impl<S: Append + 'static> Coordinator<S> {
                 .unwrap();
         }
 
+        if !introspection_source_index_ids.is_empty() {
+            self.initialize_compute_read_policies(
+                introspection_source_index_ids,
+                instance.id,
+                DEFAULT_LOGICAL_COMPACTION_WINDOW_MS,
+            )
+            .await;
+        }
         if !introspection_collection_ids.is_empty() {
             self.initialize_storage_read_policies(
                 introspection_collection_ids,

test/sqllogictest/cluster_log_compaction.slt

@@ -0,0 +1,29 @@
+# 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 correct log creation on `CREATE CLUSTER` initializes read policies for introspection source indexes.
+
+mode cockroach
+
+statement ok
+CREATE CLUSTER c1 REPLICAS (r (SIZE '1'));
+
+statement ok
+SET CLUSTER TO c1;
+
+statement ok
+BEGIN
+
+# Transaction will force a read hold on this index.
+query T rowsort
+SELECT * FROM mz_arrangement_batches_internal_2;
+----
+
+statement ok
+COMMIT