Examples of dynamic queries with diesel-rs

Also see examples of dynamic filters

Dynamic Queries ?

diesel-rs is a great and powerful tool, however extending it and doing anything dynamic seems impossible. I wanted to show how you build rich SQL queries while still maintaining some type safety and keeping the power of diesel dsl.

Motivation

SQL is powerful and engines are optimised for aggregations. Quite often you would want to leave the heavy duty calculations to SQL engines to improve performance. Rust is also powerful and fast, but serialisation has it's costs. It's great when you can ask SQL engine to aggregate data for 100k rows and output 20 rows of displayable data, but that requires use of recursive of window functions, which quite often require more flexibility in query building then just base diesel. This tutorial will help you unchain from contraints you may have taken for granted when using diesel-rs

Disclaimer

Don't forget to test your code, diesel offers great type safety, however when extending diesel your rust compiler (Mum) won't tell you off when you've missed a column in your SQL, so make sure to have a good set of test for every query. Saying that, if you follow and understand the principles in this tutorial, you should still have pretty high confidence in type safety.

Content

This tutorial is based on this part of diesel guide.

It's example based with step by step increments in complexity. In the end there is a practical example of querying aggregate temperature data grouped by time series, you can run it with cargo run to see it in a table, also check out cargo run -- --help

SQL

? variables are substituted with diesel typed parameters

WITH RECURSIVE inner_query AS (
    SELECT
        *
    FROM (
        SELECT
            `temperature`.`id`,
            `temperature`.`location`,
            `temperature`.`time`,
            `temperature`.`value`
        FROM
            `temperature`
        ORDER BY
            `temperature`.`value`)
),
variables AS (
    SELECT
        datetime(?) AS start_datetime,
        datetime(?) AS end_datetime,
        ? AS interval
),
time_series AS (
    SELECT
        start_datetime AS start,
        datetime(start_datetime, interval) AS end,
        interval
    FROM
        variables
    UNION ALL
    SELECT
        datetime(start, interval) AS start,
        datetime(end, interval) AS end,
        interval
    FROM
        time_series
    WHERE end < (SELECT end_datetime FROM variables)
    )
SELECT
    start,
    end,
    location,
    MAX(value) AS max_value,
    MIN(value) AS min_value,
    AVG(value) AS avg_value
FROM
    time_series
    LEFT JOIN inner_query ON time >= start
        AND time < end
GROUP BY
    1,
    2,
    3

Sections

• The simplest of queries
• Query with parameter
• Using source table
• Practical example with aggregation and time series

The simples of queries

This is a starting point of a very simple query 'SELECT 1', it uses the minimum code from this guide, implementing three traits on a struct to allow you to use load method with your connection, and still being type safe with the result.

Here we specify that BigInt is returned:

diesel-rs-dynamic-queries/src/basic_query.rs

Lines 13 to 15 in da7da2a

	impl Query for BasicQuery {
	type SqlType = BigInt;
	}

The actual statement

diesel-rs-dynamic-queries/src/basic_query.rs

Line 8 in da7da2a

out.push_sql("SELECT 1");

Boilerplate

diesel-rs-dynamic-queries/src/basic_query.rs

Line 17 in da7da2a

impl RunQueryDsl<SqliteConnection> for BasicQuery {}

And how to consume this extension

diesel-rs-dynamic-queries/src/basic_query.rs

Lines 22 to 24 in da7da2a

	let result: Vec<i64> = BasicQuery.load(&mut connection).unwrap();
	assert_eq!(result, vec![1]);

With parameter

In this example we are binding a parameter instead of 1, letting diesel convert it.

push_bind_param is added

diesel-rs-dynamic-queries/src/with_parameter.rs

Lines 8 to 9 in da7da2a

	out.push_sql("SELECT ");
	out.push_bind_param::<BigInt, _>(&self.0)?;

Which uses parameter in struct

diesel-rs-dynamic-queries/src/with_parameter.rs

Line 4 in da7da2a

struct BasicQuery(i64);

And how to consume this

diesel-rs-dynamic-queries/src/with_parameter.rs

Lines 23 to 25 in da7da2a

	let result: Vec<i64> = BasicQuery(2).load(&mut connection).unwrap();
	assert_eq!(result, vec![2]);

Using source table

Now we are going to do something that might be useful, and combining diesel dsl with our query. The layer diesel generate query with sql that uses variable we manually bind from rust.

The result query will be:

SQL

? variables are substituted with diesel typed parameters

WITH inner_query AS (
    SELECT
        *
    FROM (
        SELECT
            `value`.`id`,
            `value`.`data`
        FROM
            `value`
        ORDER BY
            `value`.`data`)
),
variables AS (
    SELECT
        ? AS threshold
)
SELECT
    *,
    CASE WHEN data > (SELECT threshold FROM variables) THEN
        'higher then threshold'
    ELSE
        'lower then threshold'
    END AS stat
FROM
    inner_query,
    variables

After defining our base (inner query table)

diesel-rs-dynamic-queries/src/source_table.rs

Lines 3 to 16 in da7da2a

	table! {
	value (id) {
	id -> Text,
	data -> BigInt
	}
	}
	#[derive(Queryable)]
	#[diesel(table_name = value)]
	#[allow(unused)]
	struct Value {
	id: String,
	data: i64,
	}

We make sure our BasicQuery struct can store the query

diesel-rs-dynamic-queries/src/source_table.rs

Line 27 in da7da2a

struct BasicQuery<T>(i64, T);

For it to be accessed with walk_ast (notice self.1)

diesel-rs-dynamic-queries/src/source_table.rs

Lines 62 to 64 in da7da2a

	out.push_sql("WITH inner_query AS (SELECT * FROM (");
	self.1.walk_ast(out.reborrow())?;
	out.push_sql("))");

Then we substitue the rest of the SQL and bind our variable

diesel-rs-dynamic-queries/src/source_table.rs

Lines 67 to 80 in da7da2a

	out.push_sql(", variables AS (SELECT ");
	out.push_bind_param::<BigInt, _>(&self.0)?;
	out.push_sql(" AS threshold)");
	out.push_sql(
	r#"
	SELECT *,
	CASE WHEN data > (SELECT threshold FROM variables)
	THEN 'higher then threshold'
	ELSE 'lower then threshold'
	END AS stat
	FROM inner_query
	"#,
	);

Return type of the query will now be SELECT *, stat, and stat is a string, so we need to adjust Query implementation

diesel-rs-dynamic-queries/src/source_table.rs

Lines 85 to 88 in da7da2a

	impl<T: Query> Query for BasicQuery<T> {
	type SqlType = (T::SqlType, Text);
	}

That's it, now it can be consumed as follows:

diesel-rs-dynamic-queries/src/source_table.rs

Lines 120 to 125 in da7da2a

	let query = BasicQuery(2, value::table.order(value::data));
	// println!("{}", diesel::debug_query::<Sqlite, _>(&query).to_string());
	let result: Vec<(Value, String)> = query.load(&mut connection).unwrap();

Notice that this is different to the guide, I wanted to simplify things and avoid creating and implementing extra trait with the trade of just passing diesel dsl query as parameter rather then dot notation on diesel dsl query.

A few things to note:

We can use filter, etc.. in the inner query, order was used for the first test result.

diesel-rs-dynamic-queries/src/source_table.rs

Line 157 in 3e15759

let query = BasicQuery(2, value::table.filter(value::data.ge(3)));

Doing push_sql and push_bind_param when building the query can obfuscate your sql and make things less readable, thus I opted for 'variables' at the top, so that the rest of the query can be seen as a whole.

SELECT * FROM (diesel query) is used because if you pass just value::table to BasicQuery, walk_ast will just return the table name.

Practical example with aggregation and time series

Now we come to the juicy bits, we combine what we've learned so far and apply to a practical example of time series aggregation of temperature records. The resulting query will look like this:

SQL

? variables are substituted with diesel typed parameters

WITH RECURSIVE inner_query AS (
    SELECT
        *
    FROM (
        SELECT
            `temperature`.`id`,
            `temperature`.`location`,
            `temperature`.`time`,
            `temperature`.`value`
        FROM
            `temperature`
        ORDER BY
            `temperature`.`value`)
),
variables AS (
    SELECT
        datetime(?) AS start_datetime,
        datetime(?) AS end_datetime,
        ? AS interval
),
time_series AS (
    SELECT
        start_datetime AS start,
        datetime(start_datetime, interval) AS end,
        interval
    FROM
        variables
    UNION ALL
    SELECT
        datetime(start, interval) AS start,
        datetime(end, interval) AS end,
        interval
    FROM
        time_series
    WHERE end < (SELECT end_datetime FROM variables)
    )
SELECT
    start,
    end,
    location,
    MAX(value) AS max_value,
    MIN(value) AS min_value,
    AVG(value) AS avg_value
FROM
    time_series
    LEFT JOIN inner_query ON time >= start
        AND time < end
GROUP BY
    1,
    2,
    3

There is another addition, of also using diesel dsl in the output, let's start with that.

The result of this query can be typed to a diesel dsl temperature_aggregate, so we define it.

diesel-rs-dynamic-queries/src/aggregate_query.rs

Lines 160 to 181 in 3e15759

	table! {
	temperature_aggregate(start, location) {
	start -> Timestamp,
	end -> Timestamp,
	location -> Text,
	max_value -> BigInt,
	min_value -> BigInt,
	avg_value -> Double,
	}
	}
	#[derive(Queryable, Debug, PartialEq)]
	#[diesel(table_name = temperature_aggregate)]
	pub struct TemperatureAggregate {
	pub start: NaiveDateTime,
	pub end: NaiveDateTime,
	pub location: String,
	pub max_value: i64,
	pub min_value: i64,
	pub avg_value: f64,
	}

Which has to match to the sql result:

diesel-rs-dynamic-queries/src/aggregate_query.rs

Lines 135 to 141 in 3e15759

	temperature_aggregate AS (SELECT
	start,
	end,
	location,
	MAX(value) AS max_value,
	MIN(value) AS min_value,
	AVG(value) AS avg_value

Generated diesel dsl of temperature_aggregate is inserted at the end of the query

diesel-rs-dynamic-queries/src/aggregate_query.rs

Lines 152 to 154 in 3e15759

	out.push_sql("SELECT * FROM (");
	self.select_query.walk_ast(out.reborrow())?;
	out.push_sql(")");

And the return is typed as:

diesel-rs-dynamic-queries/src/aggregate_query.rs

Lines 182 to 185 in 3e15759

	impl<IQ: QueryFragment<Sqlite>, SQ: QueryFragment<Sqlite>> Query for AggregateQuery<IQ, SQ> {
	type SqlType = temperature_aggregate::SqlType;
	}

Now AggregateQuery takes a few more parameters, including the inner_query and select query:

diesel-rs-dynamic-queries/src/aggregate_query.rs

Lines 22 to 41 in 3e15759

	impl<IQ, SQ> AggregateQuery<IQ, SQ> {
	pub fn new(
	start: NaiveDateTime,
	end: NaiveDateTime,
	number_of_points: i64,
	inner_query: IQ,
	select_query: SQ,
	) -> Self {
	let interval_seconds = (end - start).num_seconds() / number_of_points;
	let interval = format!("{} seconds", interval_seconds);
	Self {
	start,
	end,
	interval,
	inner_query,
	select_query,
	}
	}
	}

This is all added to the start of the query:

diesel-rs-dynamic-queries/src/aggregate_query.rs

Lines 102 to 113 in 3e15759

	out.push_sql("WITH RECURSIVE inner_query AS (SELECT * FROM (");
	self.inner_query.walk_ast(out.reborrow())?;
	out.push_sql("))");
	// Variables
	out.push_sql(", variables AS (SELECT datetime(");
	out.push_bind_param::<Timestamp, _>(&self.start)?;
	out.push_sql(") AS start_datetime, datetime(");
	out.push_bind_param::<Timestamp, _>(&self.end)?;
	out.push_sql(") AS end_datetime, ");
	out.push_bind_param::<Text, _>(&self.interval)?;
	out.push_sql(" AS interval)");

To be used in generation of time series

diesel-rs-dynamic-queries/src/aggregate_query.rs

Lines 116 to 134 in 3e15759

	out.push_sql(
	r#"
	,
	time_series AS (
	SELECT
	start_datetime AS start,
	datetime(start_datetime, interval) AS end,
	interval
	FROM
	variables
	UNION ALL
	SELECT
	datetime(start, interval) AS start,
	datetime(end, interval) AS end,
	interval
	FROM
	time_series
	WHERE datetime(end, interval) <= (SELECT end_datetime FROM variables)
	),

And lastly the aggregation

diesel-rs-dynamic-queries/src/aggregate_query.rs

Lines 135 to 149 in 3e15759

	temperature_aggregate AS (SELECT
	start,
	end,
	location,
	MAX(value) AS max_value,
	MIN(value) AS min_value,
	AVG(value) AS avg_value
	FROM
	time_series
	LEFT JOIN inner_query ON time >= start
	AND time < end
	GROUP BY
	1,
	2,
	3)

It can be consumed as

diesel-rs-dynamic-queries/src/aggregate_query.rs

Lines 319 to 328 in 3e15759

	let query = AggregateQuery::new(
	NaiveDateTime::parse_from_str("2023-10-01 00:00:00", fmt).unwrap(),
	NaiveDateTime::parse_from_str("2023-10-01 17:00:00", fmt).unwrap(),
	3,
	temperature::table,
	temperature_aggregate::table.order((
	temperature_aggregate::location,
	temperature_aggregate::start,
	)),
	);

Notice the sorting is done through diesel dsl in temperature_aggregate

diesel-rs-dynamic-queries/src/aggregate_query.rs

Lines 324 to 327 in 3e15759

	temperature_aggregate::table.order((
	temperature_aggregate::location,
	temperature_aggregate::start,
	)),

And pre filtering can be performed before aggregation using temperature table diesel dsl

diesel-rs-dynamic-queries/src/aggregate_query.rs

Line 393 in 3e15759

temperature::table.filter(temperature::location.eq("indoors")),

I think in this case filter with temperature or temperature_aggregate would have the same effect, but if you had a complex window function, it maybe more performant to filter early on.

Check this out in the example in main.rs where I ordered result differently to allow for faster restructuring for display in a table

diesel-rs-dynamic-queries/src/main.rs

Lines 83 to 86 in 3e15759

	temperature_aggregate::table.order((
	temperature_aggregate::start,
	temperature_aggregate::location,
	)),

You can cargo run and cargo run -- --help for more controls.

Summary

I hope you found this tutorial useful, you can create an issue if you need some clarification or found an error, etc..

(For anyone that's interested, the original trigger for writing this tutorial came from a more demanding days out of stock aggregation requirement from omSupply project, see complex query in this issue)

SQL

WITH starting_stock AS (
  SELECT
    item_id,
    store_id,
    SUM(quantity) AS running_balance,
    '2019-10-03 13:00:29' AS datetime
  FROM
    stock_movement
  WHERE
    datetime <= '2019-10-03 13:00:29'
  GROUP BY
    item_id,
    store_id
),
ending_stock AS (
  SELECT
    item_id,
    store_id,
    SUM(quantity) AS running_balance,
    '2019-11-03 13:00:29' AS datetime
  FROM
    stock_movement
  WHERE
    datetime <= '2019-11-03 13:00:29'
  GROUP BY
    item_id,
    store_id
),
ledger AS (
  SELECT
    *,
    DATE(datetime) AS date
  FROM
    starting_stock
  UNION
  SELECT
    *,
    DATE(datetime) AS date
  FROM
    ending_stock
  UNION
  SELECT
    item_id,
    store_id,
    running_balance,
    datetime,
    DATE(datetime) AS date
  FROM
    item_ledger
  WHERE
    datetime > '2019-10-03 13:00:29'
    AND datetime < '2019-11-03 13:00:29'
),
daily_stock AS (
  SELECT DISTINCT
    item_id,
    store_id,
    date,
    MAX(running_balance) OVER (PARTITION BY store_id,
      item_id,
      date) AS max_stock,
    FIRST_VALUE(running_balance) OVER (PARTITION BY store_id,
      item_id,
      date ORDER BY datetime DESC) AS running_balance
  FROM
    ledger
),
with_lag AS (
  SELECT
    *,
    LAG(running_balance) OVER (PARTITION BY store_id,
      item_id ORDER BY date) AS pr,
    LAG(date) OVER (PARTITION BY store_id,
      item_id ORDER BY date) AS pd
  FROM
    daily_stock
  ORDER BY
    store_id,
    item_id
)
SELECT
  item_id,
  store_id,
  sum(julianday(date) - julianday(pd))
FROM
  with_lag
WHERE
  pr = 0
GROUP BY
  1,
  2
ORDER BY
  store_id,
  item_id