Merge pull request #34246 from ggevay/frontend-peek-slow-path-peeks

ggevay · web-flow · commit eda01f483e51 · 2025-11-25T14:07:16.000+01:00
Frontend peek sequencing -- slow-path peeks
diff --git a/src/adapter/src/client.rs b/src/adapter/src/client.rs
@@ -1015,7 +1015,8 @@ impl SessionClient {
                 | Command::Dump { .. }
                 | Command::GetComputeInstanceClient { .. }
                 | Command::GetOracle { .. }
-                | Command::DetermineRealTimeRecentTimestamp { .. } => {}
+                | Command::DetermineRealTimeRecentTimestamp { .. }
+                | Command::ExecuteSlowPathPeek { .. } => {}
             };
             cmd
         });
diff --git a/src/adapter/src/command.rs b/src/adapter/src/command.rs
@@ -18,15 +18,17 @@ use enum_kinds::EnumKind;
 use futures::Stream;
 use mz_adapter_types::connection::{ConnectionId, ConnectionIdType};
 use mz_auth::password::Password;
+use mz_cluster_client::ReplicaId;
 use mz_compute_types::ComputeInstanceId;
+use mz_expr::RowSetFinishing;
 use mz_ore::collections::CollectionExt;
 use mz_ore::soft_assert_no_log;
 use mz_ore::tracing::OpenTelemetryContext;
 use mz_persist_client::PersistClient;
 use mz_pgcopy::CopyFormatParams;
 use mz_repr::global_id::TransientIdGen;
 use mz_repr::role_id::RoleId;
-use mz_repr::{CatalogItemId, ColumnIndex, GlobalId, RowIterator};
+use mz_repr::{CatalogItemId, ColumnIndex, GlobalId, RowIterator, SqlRelationType};
 use mz_sql::ast::{FetchDirection, Raw, Statement};
 use mz_sql::catalog::ObjectType;
 use mz_sql::optimizer_metrics::OptimizerMetrics;
@@ -43,7 +45,8 @@ use crate::catalog::Catalog;
 use crate::coord::ExecuteContextExtra;
 use crate::coord::appends::BuiltinTableAppendNotify;
 use crate::coord::consistency::CoordinatorInconsistencies;
-use crate::coord::peek::PeekResponseUnary;
+use crate::coord::peek::{PeekDataflowPlan, PeekResponseUnary};
+use crate::coord::timestamp_selection::TimestampDetermination;
 use crate::error::AdapterError;
 use crate::session::{EndTransactionAction, RowBatchStream, Session};
 use crate::statement_logging::{StatementEndedExecutionReason, StatementExecutionStrategy};
@@ -182,6 +185,20 @@ pub enum Command {
         real_time_recency_timeout: Duration,
         tx: oneshot::Sender<Result<Option<mz_repr::Timestamp>, AdapterError>>,
     },
+
+    ExecuteSlowPathPeek {
+        dataflow_plan: Box<PeekDataflowPlan<mz_repr::Timestamp>>,
+        determination: TimestampDetermination<mz_repr::Timestamp>,
+        finishing: RowSetFinishing,
+        compute_instance: ComputeInstanceId,
+        target_replica: Option<ReplicaId>,
+        intermediate_result_type: SqlRelationType,
+        source_ids: BTreeSet<GlobalId>,
+        conn_id: ConnectionId,
+        max_result_size: u64,
+        max_query_result_size: Option<u64>,
+        tx: oneshot::Sender<Result<ExecuteResponse, AdapterError>>,
+    },
 }
 
 impl Command {
@@ -204,7 +221,8 @@ impl Command {
             | Command::Dump { .. }
             | Command::GetComputeInstanceClient { .. }
             | Command::GetOracle { .. }
-            | Command::DetermineRealTimeRecentTimestamp { .. } => None,
+            | Command::DetermineRealTimeRecentTimestamp { .. }
+            | Command::ExecuteSlowPathPeek { .. } => None,
         }
     }
 
@@ -227,7 +245,8 @@ impl Command {
             | Command::Dump { .. }
             | Command::GetComputeInstanceClient { .. }
             | Command::GetOracle { .. }
-            | Command::DetermineRealTimeRecentTimestamp { .. } => None,
+            | Command::DetermineRealTimeRecentTimestamp { .. }
+            | Command::ExecuteSlowPathPeek { .. } => None,
         }
     }
 }
diff --git a/src/adapter/src/coord.rs b/src/adapter/src/coord.rs
@@ -366,6 +366,7 @@ impl Message {
                 Command::DetermineRealTimeRecentTimestamp { .. } => {
                     "determine-real-time-recent-timestamp"
                 }
+                Command::ExecuteSlowPathPeek { .. } => "execute-slow-path-peek",
             },
             Message::ControllerReady {
                 controller: ControllerReadiness::Compute,
diff --git a/src/adapter/src/coord/command_handler.rs b/src/adapter/src/coord/command_handler.rs
@@ -313,6 +313,37 @@ impl Coordinator {
                         }
                     }
                 }
+
+                Command::ExecuteSlowPathPeek {
+                    dataflow_plan,
+                    determination,
+                    finishing,
+                    compute_instance,
+                    target_replica,
+                    intermediate_result_type,
+                    source_ids,
+                    conn_id,
+                    max_result_size,
+                    max_query_result_size,
+                    tx,
+                } => {
+                    let result = self
+                        .implement_slow_path_peek(
+                            *dataflow_plan,
+                            determination,
+                            finishing,
+                            compute_instance,
+                            target_replica,
+                            intermediate_result_type,
+                            source_ids,
+                            conn_id,
+                            max_result_size,
+                            max_query_result_size,
+                        )
+                        .await;
+
+                    let _ = tx.send(result);
+                }
             }
         }
         .instrument(debug_span!("handle_command"))
diff --git a/src/adapter/src/coord/peek.rs b/src/adapter/src/coord/peek.rs
@@ -1189,6 +1189,55 @@ impl crate::coord::Coordinator {
         pending_peek
     }
 
+    /// Implements a slow-path peek by creating a transient dataflow.
+    /// This is called from the command handler for ExecuteSlowPathPeek.
+    ///
+    /// (For now, this method simply delegates to implement_peek_plan by constructing
+    /// the necessary PlannedPeek structure and a minimal ExecuteContext.)
+    pub(crate) async fn implement_slow_path_peek(
+        &mut self,
+        dataflow_plan: PeekDataflowPlan<mz_repr::Timestamp>,
+        determination: TimestampDetermination<mz_repr::Timestamp>,
+        finishing: RowSetFinishing,
+        compute_instance: ComputeInstanceId,
+        target_replica: Option<ReplicaId>,
+        intermediate_result_type: SqlRelationType,
+        source_ids: BTreeSet<GlobalId>,
+        conn_id: ConnectionId,
+        max_result_size: u64,
+        max_query_result_size: Option<u64>,
+    ) -> Result<ExecuteResponse, AdapterError> {
+        let source_arity = intermediate_result_type.arity();
+
+        let planned_peek = PlannedPeek {
+            plan: PeekPlan::SlowPath(dataflow_plan),
+            determination,
+            conn_id,
+            intermediate_result_type,
+            source_arity,
+            source_ids,
+        };
+
+        // Create a minimal ExecuteContext
+        // TODO(peek-seq): Use the real context once we have statement logging.
+        let mut ctx_extra = ExecuteContextExtra::default();
+
+        // Call the old peek sequencing's implement_peek_plan for now.
+        // TODO(peek-seq): After the old peek sequencing is completely removed, we should merge the
+        // relevant parts of the old `implement_peek_plan` into this method, and remove the old
+        // `implement_peek_plan`.
+        self.implement_peek_plan(
+            &mut ctx_extra,
+            planned_peek,
+            finishing,
+            compute_instance,
+            target_replica,
+            max_result_size,
+            max_query_result_size,
+        )
+        .await
+    }
+
     /// Constructs an [`ExecuteResponse`] that that will send some rows to the
     /// client immediately, as opposed to asking the dataflow layer to send along
     /// the rows after some computation.
diff --git a/src/adapter/src/frontend_peek.rs b/src/adapter/src/frontend_peek.rs
@@ -731,16 +731,6 @@ impl PeekClient {
                     session.add_notice(AdapterNotice::PlanInsights(insights));
                 }
 
-                let fast_path_plan = match peek_plan {
-                    PeekPlan::SlowPath(_) => {
-                        debug!(
-                            "Bailing out from try_frontend_peek_inner, because it's a slow-path peek"
-                        );
-                        return Ok(None);
-                    }
-                    PeekPlan::FastPath(p) => p,
-                };
-
                 // Warning: Do not bail out from the new peek sequencing after this point, because the
                 // following has side effects. TODO(peek-seq): remove this comment once we never
                 // bail out to the old sequencing.
@@ -798,35 +788,57 @@ impl PeekClient {
 
                 let max_result_size = catalog.system_config().max_result_size();
 
-                let row_set_finishing_seconds =
-                    session.metrics().row_set_finishing_seconds().clone();
-
-                let peek_stash_read_batch_size_bytes =
-                    mz_compute_types::dyncfgs::PEEK_RESPONSE_STASH_READ_BATCH_SIZE_BYTES
-                        .get(catalog.system_config().dyncfgs());
-                let peek_stash_read_memory_budget_bytes =
-                    mz_compute_types::dyncfgs::PEEK_RESPONSE_STASH_READ_MEMORY_BUDGET_BYTES
-                        .get(catalog.system_config().dyncfgs());
-
-                // Implement the peek, and capture the response.
-                let resp = self
-                    .implement_fast_path_peek_plan(
-                        fast_path_plan,
-                        determination.timestamp_context.timestamp_or_default(),
-                        select_plan.finishing.clone(),
-                        target_cluster_id,
-                        target_replica,
-                        typ,
-                        max_result_size,
-                        max_query_result_size,
-                        row_set_finishing_seconds,
-                        read_holds,
-                        peek_stash_read_batch_size_bytes,
-                        peek_stash_read_memory_budget_bytes,
-                    )
-                    .await?;
+                match peek_plan {
+                    PeekPlan::FastPath(fast_path_plan) => {
+                        let row_set_finishing_seconds =
+                            session.metrics().row_set_finishing_seconds().clone();
 
-                Ok(Some(resp))
+                        let peek_stash_read_batch_size_bytes =
+                            mz_compute_types::dyncfgs::PEEK_RESPONSE_STASH_READ_BATCH_SIZE_BYTES
+                                .get(catalog.system_config().dyncfgs());
+                        let peek_stash_read_memory_budget_bytes =
+                            mz_compute_types::dyncfgs::PEEK_RESPONSE_STASH_READ_MEMORY_BUDGET_BYTES
+                                .get(catalog.system_config().dyncfgs());
+
+                        let resp = self
+                            .implement_fast_path_peek_plan(
+                                fast_path_plan,
+                                determination.timestamp_context.timestamp_or_default(),
+                                select_plan.finishing,
+                                target_cluster_id,
+                                target_replica,
+                                typ,
+                                max_result_size,
+                                max_query_result_size,
+                                row_set_finishing_seconds,
+                                read_holds,
+                                peek_stash_read_batch_size_bytes,
+                                peek_stash_read_memory_budget_bytes,
+                            )
+                            .await?;
+
+                        Ok(Some(resp))
+                    }
+                    PeekPlan::SlowPath(dataflow_plan) => {
+                        let response = self
+                            .call_coordinator(|tx| Command::ExecuteSlowPathPeek {
+                                dataflow_plan: Box::new(dataflow_plan),
+                                determination,
+                                finishing: select_plan.finishing,
+                                compute_instance: target_cluster_id,
+                                target_replica,
+                                intermediate_result_type: typ,
+                                source_ids,
+                                conn_id: session.conn_id().clone(),
+                                max_result_size,
+                                max_query_result_size,
+                                tx,
+                            })
+                            .await?;
+
+                        Ok(Some(response))
+                    }
+                }
             }
         }
     }

Original file line number	Diff line number	Diff line change
`@@ -366,6 +366,7 @@ impl Message {`
`366`	`366`	`Command::DetermineRealTimeRecentTimestamp { .. } => {`
`367`	`367`	`"determine-real-time-recent-timestamp"`
`368`	`368`	`}`
	`369`	`+ Command::ExecuteSlowPathPeek { .. } => "execute-slow-path-peek",`
`369`	`370`	`},`
`370`	`371`	`Message::ControllerReady {`
`371`	`372`	`controller: ControllerReadiness::Compute,`