Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 54 additions & 7 deletions crates/utopia-server/src/query_engine/trino.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

use super::conn::TrinoConn;
use super::{
rows_to_json_lines, sql_literal, truncate_rows, wrap_limit, QueryEngine, QueryResult,
coerce, rows_to_json_lines, sql_literal, truncate_rows, wrap_limit, QueryEngine, QueryResult,
SchemaColumn, HTTP_POLL_BUDGET, STATEMENT_TIMEOUT_SECS,
};
use base64::Engine as _;
Expand All @@ -22,6 +22,8 @@ pub struct TrinoEngine {
#[derive(Deserialize)]
struct Column {
name: String,
#[serde(rename = "type", default)]
ty: String,
}

#[derive(Deserialize)]
Expand Down Expand Up @@ -86,18 +88,15 @@ impl TrinoEngine {
.json()
.await?;
let started = Instant::now();
let mut columns: Option<Vec<String>> = None;
let mut columns: Option<Vec<Column>> = None;
let mut rows = Vec::new();
loop {
if let Some(e) = page.error {
let name = e.error_name.map(|n| format!("{n}: ")).unwrap_or_default();
anyhow::bail!("{name}{}", e.message);
}
if columns.is_none() {
columns = page
.columns
.take()
.map(|cs| cs.into_iter().map(|c| c.name).collect());
columns = page.columns.take();
}
if let Some(d) = page.data.take() {
rows.extend(d);
Expand All @@ -120,7 +119,16 @@ impl TrinoEngine {
.json()
.await?;
}
Ok((columns.unwrap_or_default(), rows))
let columns = columns.unwrap_or_default();
// 协议把 decimal 写成字符串(其余的数都是 JSON 数),按列类型还原
for row in &mut rows {
for (v, c) in row.iter_mut().zip(&columns) {
if c.ty.starts_with("decimal") {
*v = coerce("DECIMAL", v);
}
}
}
Ok((columns.into_iter().map(|c| c.name).collect(), rows))
}
}

Expand Down Expand Up @@ -311,6 +319,45 @@ mod tests {
assert!(!out.truncated);
}

/// Trino 的 REST 协议把 decimal 写成字符串(`SqlDecimal` 走 `writeString`),
/// 别的数是 JSON 数
#[tokio::test]
async fn a_decimal_reads_as_a_number() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/statement"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"id": "q3",
"columns": [
{ "name": "region", "type": "varchar(5)" },
{ "name": "total", "type": "decimal(12,2)" },
{ "name": "orders", "type": "bigint" }
],
"data": [ ["east", "1234.50", 7], ["west", null, 0] ],
"stats": { "state": "FINISHED" }
})))
.mount(&server)
.await;
let conn = TrinoConn::parse(&format!(
"trino://alice@{}/hive?ssl=false",
server.uri().trim_start_matches("http://")
))
.unwrap();
let out = TrinoEngine::new(conn)
.execute(
"SELECT region, SUM(amount) AS total, COUNT(*) AS orders FROM orders GROUP BY 1",
)
.await
.unwrap();
assert_eq!(
out.rows,
vec![
r#"{"region":"east","total":1234.5,"orders":7}"#,
r#"{"region":"west","total":null,"orders":0}"#
]
);
}

#[tokio::test]
async fn a_trino_error_page_becomes_an_error() {
let server = MockServer::start().await;
Expand Down
2 changes: 1 addition & 1 deletion docs/decisions/0018-the-lakehouse-is-one-protocol-away.md
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ Left out on purpose: Snowflake key-pair JWT (local RSA signing, a dependency for

0011 set up defense in depth: parse and admit only SELECT, wrap a LIMIT, a read-only session with a timeout, JSON Lines out. The HTTP family **has no session**, so the third gate is a timeout alone (Trino's `query_max_execution_time` session property, Databricks' `wait_timeout`, Snowflake's `timeout`), and read-only rests entirely on the first gate. The first gate therefore parses with each engine's dialect: `DatabricksDialect`, `SnowflakeDialect`, and `GenericDialect` for Trino (sqlparser has no Trino dialect; Generic is a superset). One test runs the same three checks under all four dialects: SELECT passes, DELETE fails, two statements fail.

The fourth gate is assembled here for the HTTP family. Databricks' `JSON_ARRAY` and Snowflake's `data` return every value as a string; `coerce` restores numbers and booleans from the manifest / rowType column types, otherwise a model handed `"42"` stops doing arithmetic. Snowflake dates, times and timestamps are also restored from their documented epoch encodings. Their protocol-replay tests validate that conversion, not acceptance against a live Snowflake cluster; the live-cluster gate remains open (#241, #242). Key order is assembled by hand instead of through `serde_json::Map`, which sorts keys unless `preserve_order` is on, and column order is the order the query wrote.
The fourth gate is assembled here for the HTTP family. Databricks' `JSON_ARRAY` and Snowflake's `data` return every value as a string; `coerce` restores numbers and booleans from the manifest / rowType column types, otherwise a model handed `"42"` stops doing arithmetic. Snowflake dates, times and timestamps are also restored from their documented epoch encodings. Their protocol-replay tests validate that conversion, not acceptance against a live Snowflake cluster; the live-cluster gate remains open (#241, #242). Trino writes decimal cells as strings too, and they go through the same `coerce`: short decimals reach JSON Lines as numbers, and decimals beyond f64 precision stay strings rather than rounded values. Key order is assembled by hand instead of through `serde_json::Map`, which sorts keys unless `preserve_order` is on, and column order is the order the query wrote.

## Loopback goes direct

Expand Down
Loading