(
instance: Option<&mut RefInstance<I>>,
db: Arc<RelationalDB>,
sql_text: String,
auth: AuthCtx,
subs: Option<ModuleSubscriptions>,
head: &mut Vec<(RawIdentifier, AlgebraicType
| 78 | } |
| 79 | |
| 80 | fn run_inner<I: WasmInstance>( |
| 81 | instance: Option<&mut RefInstance<I>>, |
| 82 | db: Arc<RelationalDB>, |
| 83 | sql_text: String, |
| 84 | auth: AuthCtx, |
| 85 | subs: Option<ModuleSubscriptions>, |
| 86 | head: &mut Vec<(RawIdentifier, AlgebraicType)>, |
| 87 | ) -> Result<(SqlResult, bool), DBError> { |
| 88 | // We parse the sql statement in a mutable transaction. |
| 89 | // If it turns out to be a query, we downgrade the tx. |
| 90 | let (tx, stmt) = db.with_auto_rollback(db.begin_mut_tx(IsolationLevel::Serializable, Workload::Sql), |tx| { |
| 91 | compile_sql_stmt(&sql_text, &SchemaViewer::new(tx, &auth), &auth) |
| 92 | })?; |
| 93 | |
| 94 | let mut metrics = ExecutionMetrics::default(); |
| 95 | |
| 96 | match stmt { |
| 97 | Statement::Select(stmt) => { |
| 98 | // Materialize views and downgrade to a read-only transaction |
| 99 | let (tx, trapped) = match instance { |
| 100 | Some(instance) => ModuleHost::materialize_views(tx, instance, &stmt, auth.caller(), Workload::Sql)?, |
| 101 | None => (tx, false), |
| 102 | }; |
| 103 | |
| 104 | let (tx_data, tx_metrics_mut, tx) = db.commit_tx_downgrade(tx, Workload::Sql); |
| 105 | |
| 106 | let (tx_offset_send, tx_offset) = oneshot::channel(); |
| 107 | // Release the tx on drop, so that we record metrics |
| 108 | // and set the transaction offset. |
| 109 | let mut tx = scopeguard::guard(tx, |tx| { |
| 110 | let (offset, tx_metrics_downgrade, reducer) = db.release_tx(tx); |
| 111 | let _ = tx_offset_send.send(offset); |
| 112 | db.report_tx_metrics(reducer, Some(tx_data), Some(tx_metrics_mut), Some(tx_metrics_downgrade)); |
| 113 | }); |
| 114 | |
| 115 | // Compute the header for the result set |
| 116 | stmt.for_each_return_field(|col_name, col_type| { |
| 117 | head.push((col_name.clone(), col_type.clone())); |
| 118 | }); |
| 119 | |
| 120 | // Evaluate the query |
| 121 | let rows = execute_select_stmt(&auth, stmt, &DeltaTx::from(&*tx), &mut metrics, |plan| { |
| 122 | check_row_limit( |
| 123 | &[&plan], |
| 124 | &db, |
| 125 | &tx, |
| 126 | |plan, tx| plan.plan_iter().map(|plan| estimate_rows_scanned(tx, plan)).sum(), |
| 127 | &auth, |
| 128 | )?; |
| 129 | Ok(plan) |
| 130 | })?; |
| 131 | |
| 132 | // Update transaction metrics |
| 133 | tx.metrics.merge(metrics); |
| 134 | |
| 135 | Ok(( |
| 136 | SqlResult { |
| 137 | tx_offset, |
nothing calls this directly
no test coverage detected
searching dependent graphs…