[dnm] extra logging to debug #10086

misc/python/materialize/checks/all_checks/materialized_views.py

@@ -310,7 +310,6 @@ def validate(self) -> Testdrive:
         )
 
 
-@disabled("due to https://github.com/MaterializeInc/database-issues/issues/10086")
 class MaterializedViewReplacement(Check):
     def _can_run(self, e: Executor) -> bool:
         return self.base_version >= MzVersion.parse_mz("v26.8.0")

src/compute-client/src/as_of_selection.rs

@@ -771,9 +771,13 @@ impl<'a, T: TimestampManipulation> Context<'a, T> {
     /// impose as-of constraints on other compute collections.
     fn prune_sealed_persist_sinks(&mut self) {
         self.collections.retain(|id, _| {
-            self.storage_collections
+            let retain = self.storage_collections
                 .collection_frontiers(*id)
-                .map_or(true, |f| !f.write_frontier.is_empty())
+                .map_or(true, |f| !f.write_frontier.is_empty());
+            if !retain {
+                tracing::info!("prune sealed persist sink: {id}");
+            }
+            retain
         });
     }
 

src/compute-client/src/controller/instance.rs

@@ -1543,6 +1543,7 @@ where
                         .collection_metadata(id)
                         .map_err(|_| CollectionMissing(id))?
                         .clone();
+                    tracing::info!("MV dataflow export: {id} {metadata:?}");
                     let conn = MaterializedViewSinkConnection {
                         value_desc: conn.value_desc,
                         storage_metadata: metadata,
@@ -1619,6 +1620,7 @@ where
                 export_ids = %augmented_dataflow.display_export_ids(),
                 as_of = ?augmented_dataflow.as_of.as_ref().unwrap().elements(),
                 until = ?augmented_dataflow.until.elements(),
+                sink_exports = ?augmented_dataflow.sink_exports,
                 "creating dataflow",
             );
         }
@@ -1897,6 +1899,10 @@ where
             return;
         }
 
+        if id.is_user() && new_frontier.is_empty() {
+            tracing::info!("advanced to empty write frontier: {id}");
+        }
+
         // Relax the implied read hold according to the read policy.
         let new_since = match &collection.read_policy {
             Some(read_policy) => {

src/compute/src/sink/materialized_view.rs

@@ -1338,6 +1338,9 @@ mod append {
             let (lower, upper) = (desc.lower, desc.upper);
             let mut to_append: Vec<_> = self.batches.iter_mut().collect();
 
+            if upper.is_empty() && self.sink_id.is_user() {
+                tracing::info!("appending batch with empty frontier: {}", self.sink_id);
+            }
             loop {
                 let result = self
                     .persist_writer

src/storage-client/src/storage_collections.rs

@@ -2733,6 +2733,9 @@ where
                     if id.is_user() {
                         trace!("new upper for collection {id}: {:?}", upper);
                     }
+                    if id.is_user() && upper.is_empty() {
+                        info!("new empty upper for collection {id}");
+                    }
                     let current_shard = self.shard_by_id.get(&id);
                     if let Some(shard_id) = current_shard {
                         if shard_id == &handle.shard_id() {
@@ -3071,7 +3074,7 @@ async fn finalize_shards_task<T>(
                     debug!(%shard_id, "shard is already finalized!");
                     Some(shard_id)
                 } else {
-                    debug!(%shard_id, "finalizing shard");
+                    info!(%shard_id, "finalizing shard");
                     let finalize = || async move {
                         // TODO: thread the global ID into the shard finalization WAL
                         let diagnostics = Diagnostics::from_purpose("finalizing shards");