Improve Sqlite support for sub-queries and CTE's (#1816)

* reproduce incorrect nullability for materialized views

* split ephemeral/index-only table handling from real table handling

* add test for literal null, expect nullability to be identified from table information

* gather interpreter state into a struct, no change in behaviour

* prevent infinite loops that could arise once branching is supported

* track nullability alongside the datatype instead of in a separate lookup

* implement basic comprehension of branching opcodes

* fix datatype calculation of aggregates which are never 'stepped' through

* implement coroutine and return operations, including tracking of 'actual' integer value stored in the register by Integer/InitCoroutine/Yield operations.

* strip unnecessary history field out

* Modify variable test to expect bind-variable outputs to be nullable, rather than unknown

* add partially commented-out union tests, simplify code to satisfy simplest union case

* fix unit test incorrectly expecting primary keys to be implicitly not-null

* add failing test for recursive tables

* add logging of query explain plan

* track explain plan execution history

* broken RowData implementation (doesn't alias)

* Implement OpenPseudo tables as an alias of a register value

* fix comment

* clean up logging code warnings

* use cfg to omit QueryPlanLogger unless sqlite feature is used

sqlx-core/src/logger.rs

@@ -1,4 +1,10 @@
 use crate::connection::LogSettings;
+#[cfg(feature = "sqlite")]
+use std::collections::HashSet;
+#[cfg(feature = "sqlite")]
+use std::fmt::Debug;
+#[cfg(feature = "sqlite")]
+use std::hash::Hash;
 use std::time::Instant;
 
 pub(crate) struct QueryLogger<'q> {
@@ -78,6 +84,86 @@ impl<'q> Drop for QueryLogger<'q> {
     }
 }
 
+#[cfg(feature = "sqlite")]
+pub(crate) struct QueryPlanLogger<'q, O: Debug + Hash + Eq, R: Debug + Hash + Eq, P: Debug> {
+    sql: &'q str,
+    unknown_operations: HashSet<O>,
+    results: HashSet<R>,
+    program: Vec<P>,
+    settings: LogSettings,
+}
+
+#[cfg(feature = "sqlite")]
+impl<'q, O: Debug + Hash + Eq, R: Debug + Hash + Eq, P: Debug> QueryPlanLogger<'q, O, R, P> {
+    pub(crate) fn new(sql: &'q str, settings: LogSettings) -> Self {
+        Self {
+            sql,
+            unknown_operations: HashSet::new(),
+            results: HashSet::new(),
+            program: Vec::new(),
+            settings,
+        }
+    }
+
+    pub(crate) fn add_result(&mut self, result: R) {
+        self.results.insert(result);
+    }
+
+    pub(crate) fn add_program(&mut self, program: Vec<P>) {
+        self.program = program;
+    }
+
+    pub(crate) fn add_unknown_operation(&mut self, operation: O) {
+        self.unknown_operations.insert(operation);
+    }
+
+    pub(crate) fn finish(&self) {
+        let lvl = self.settings.statements_level;
+
+        if let Some(lvl) = lvl
+            .to_level()
+            .filter(|lvl| log::log_enabled!(target: "sqlx::explain", *lvl))
+        {
+            let mut summary = parse_query_summary(&self.sql);
+
+            let sql = if summary != self.sql {
+                summary.push_str(" …");
+                format!(
+                    "\n\n{}\n",
+                    sqlformat::format(
+                        &self.sql,
+                        &sqlformat::QueryParams::None,
+                        sqlformat::FormatOptions::default()
+                    )
+                )
+            } else {
+                String::new()
+            };
+
+            log::logger().log(
+                &log::Record::builder()
+                    .args(format_args!(
+                        "{}; program:{:?}, unknown_operations:{:?}, results: {:?}{}",
+                        summary, self.program, self.unknown_operations, self.results, sql
+                    ))
+                    .level(lvl)
+                    .module_path_static(Some("sqlx::explain"))
+                    .target("sqlx::explain")
+                    .build(),
+            );
+        }
+    }
+}
+
+#[cfg(feature = "sqlite")]
+impl<'q, O: Debug + Hash + Eq, R: Debug + Hash + Eq, P: Debug> Drop
+    for QueryPlanLogger<'q, O, R, P>
+{
+    fn drop(&mut self) {
+        self.finish();
+    }
+}
+
 fn parse_query_summary(sql: &str) -> String {
     // For now, just take the first 4 words
     sql.split_whitespace()

sqlx-core/src/sqlite/type_info.rs

@@ -7,7 +7,7 @@ use libsqlite3_sys::{SQLITE_BLOB, SQLITE_FLOAT, SQLITE_INTEGER, SQLITE_NULL, SQL
 use crate::error::BoxDynError;
 use crate::type_info::TypeInfo;
 
-#[derive(Debug, Copy, Clone, Eq, PartialEq)]
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)]
 #[cfg_attr(feature = "offline", derive(serde::Serialize, serde::Deserialize))]
 pub(crate) enum DataType {
     Null,
@@ -29,7 +29,7 @@ pub(crate) enum DataType {
 }
 
 /// Type information for a SQLite type.
-#[derive(Debug, Clone, Eq, PartialEq)]
+#[derive(Debug, Clone, Eq, PartialEq, Hash)]
 #[cfg_attr(feature = "offline", derive(serde::Serialize, serde::Deserialize))]
 pub struct SqliteTypeInfo(pub(crate) DataType);
 

tests/sqlite/describe.rs

@@ -44,13 +44,13 @@ async fn it_describes_variables() -> anyhow::Result<()> {
     let info = conn.describe("SELECT ?1").await?;
 
     assert_eq!(info.columns()[0].type_info().name(), "NULL");
-    assert_eq!(info.nullable(0), None); // unknown
+    assert_eq!(info.nullable(0), Some(true)); // nothing prevents the value from being bound to null
 
     // context can be provided by using CAST(_ as _)
     let info = conn.describe("SELECT CAST(?1 AS REAL)").await?;
 
     assert_eq!(info.columns()[0].type_info().name(), "REAL");
-    assert_eq!(info.nullable(0), None); // unknown
+    assert_eq!(info.nullable(0), Some(true)); // nothing prevents the value from being bound to null
 
     Ok(())
 }
@@ -60,7 +60,7 @@ async fn it_describes_expression() -> anyhow::Result<()> {
     let mut conn = new::<Sqlite>().await?;
 
     let d = conn
-        .describe("SELECT 1 + 10, 5.12 * 2, 'Hello', x'deadbeef'")
+        .describe("SELECT 1 + 10, 5.12 * 2, 'Hello', x'deadbeef', null")
         .await?;
 
     let columns = d.columns();
@@ -81,6 +81,10 @@ async fn it_describes_expression() -> anyhow::Result<()> {
     assert_eq!(columns[3].name(), "x'deadbeef'");
     assert_eq!(d.nullable(3), Some(false)); // literal constant
 
+    assert_eq!(columns[4].type_info().name(), "NULL");
+    assert_eq!(columns[4].name(), "null");
+    assert_eq!(d.nullable(4), Some(true)); // literal null
+
     Ok(())
 }
 
@@ -99,10 +103,10 @@ async fn it_describes_expression_from_empty_table() -> anyhow::Result<()> {
     assert_eq!(d.nullable(0), Some(false)); // COUNT(*)
 
     assert_eq!(d.columns()[1].type_info().name(), "INTEGER");
-    assert_eq!(d.nullable(1), None); // `a + 1` is potentially nullable but we don't know for sure currently
+    assert_eq!(d.nullable(1), Some(true)); // `a+1` is nullable, because a is nullable
 
     assert_eq!(d.columns()[2].type_info().name(), "TEXT");
-    assert_eq!(d.nullable(2), Some(false)); // `name` is not nullable
+    assert_eq!(d.nullable(2), Some(true)); // `name` is not nullable, but the query can be null due to zero rows
 
     assert_eq!(d.columns()[3].type_info().name(), "REAL");
     assert_eq!(d.nullable(3), Some(false)); // literal constant
@@ -256,3 +260,123 @@ async fn it_describes_left_join() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+#[sqlx_macros::test]
+async fn it_describes_literal_subquery() -> anyhow::Result<()> {
+    async fn assert_literal_described(
+        conn: &mut sqlx::SqliteConnection,
+        query: &str,
+    ) -> anyhow::Result<()> {
+        let info = conn.describe(query).await?;
+
+        assert_eq!(info.column(0).type_info().name(), "TEXT", "{}", query);
+        assert_eq!(info.nullable(0), Some(false), "{}", query);
+        assert_eq!(info.column(1).type_info().name(), "NULL", "{}", query);
+        assert_eq!(info.nullable(1), Some(true), "{}", query);
+
+        Ok(())
+    }
+
+    let mut conn = new::<Sqlite>().await?;
+    assert_literal_described(&mut conn, "SELECT 'a', NULL").await?;
+    assert_literal_described(&mut conn, "SELECT * FROM (SELECT 'a', NULL)").await?;
+    assert_literal_described(
+        &mut conn,
+        "WITH cte AS (SELECT 'a', NULL) SELECT * FROM cte",
+    )
+    .await?;
+    assert_literal_described(
+        &mut conn,
+        "WITH cte AS MATERIALIZED (SELECT 'a', NULL) SELECT * FROM cte",
+    )
+    .await?;
+    assert_literal_described(
+        &mut conn,
+        "WITH RECURSIVE cte(a,b) AS (SELECT 'a', NULL UNION ALL SELECT a||a, NULL FROM cte WHERE length(a)<3) SELECT * FROM cte",
+    )
+    .await?;
+
+    Ok(())
+}
+
+#[sqlx_macros::test]
+async fn it_describes_table_subquery() -> anyhow::Result<()> {
+    async fn assert_tweet_described(
+        conn: &mut sqlx::SqliteConnection,
+        query: &str,
+    ) -> anyhow::Result<()> {
+        let info = conn.describe(query).await?;
+        let columns = info.columns();
+
+        assert_eq!(columns[0].name(), "id", "{}", query);
+        assert_eq!(columns[1].name(), "text", "{}", query);
+        assert_eq!(columns[2].name(), "is_sent", "{}", query);
+        assert_eq!(columns[3].name(), "owner_id", "{}", query);
+
+        assert_eq!(columns[0].ordinal(), 0, "{}", query);
+        assert_eq!(columns[1].ordinal(), 1, "{}", query);
+        assert_eq!(columns[2].ordinal(), 2, "{}", query);
+        assert_eq!(columns[3].ordinal(), 3, "{}", query);
+
+        assert_eq!(info.nullable(0), Some(false), "{}", query);
+        assert_eq!(info.nullable(1), Some(false), "{}", query);
+        assert_eq!(info.nullable(2), Some(false), "{}", query);
+        assert_eq!(info.nullable(3), Some(true), "{}", query);
+
+        assert_eq!(columns[0].type_info().name(), "INTEGER", "{}", query);
+        assert_eq!(columns[1].type_info().name(), "TEXT", "{}", query);
+        assert_eq!(columns[2].type_info().name(), "BOOLEAN", "{}", query);
+        assert_eq!(columns[3].type_info().name(), "INTEGER", "{}", query);
+
+        Ok(())
+    }
+
+    let mut conn = new::<Sqlite>().await?;
+    assert_tweet_described(&mut conn, "SELECT * FROM tweet").await?;
+    assert_tweet_described(&mut conn, "SELECT * FROM (SELECT * FROM tweet)").await?;
+    assert_tweet_described(
+        &mut conn,
+        "WITH cte AS (SELECT * FROM tweet) SELECT * FROM cte",
+    )
+    .await?;
+    assert_tweet_described(
+        &mut conn,
+        "WITH cte AS MATERIALIZED (SELECT * FROM tweet) SELECT * FROM cte",
+    )
+    .await?;
+
+    Ok(())
+}
+
+#[sqlx_macros::test]
+async fn it_describes_union() -> anyhow::Result<()> {
+    async fn assert_union_described(
+        conn: &mut sqlx::SqliteConnection,
+        query: &str,
+    ) -> anyhow::Result<()> {
+        let info = conn.describe(query).await?;
+
+        assert_eq!(info.column(0).type_info().name(), "TEXT", "{}", query);
+        assert_eq!(info.nullable(0), Some(false), "{}", query);
+        assert_eq!(info.column(1).type_info().name(), "TEXT", "{}", query);
+        assert_eq!(info.nullable(1), Some(true), "{}", query);
+        assert_eq!(info.column(2).type_info().name(), "INTEGER", "{}", query);
+        assert_eq!(info.nullable(2), Some(true), "{}", query);
+        //TODO: mixed type columns not handled correctly
+        //assert_eq!(info.column(3).type_info().name(), "NULL", "{}", query);
+        //assert_eq!(info.nullable(3), Some(false), "{}", query);
+
+        Ok(())
+    }
+
+    let mut conn = new::<Sqlite>().await?;
+    assert_union_described(
+        &mut conn,
+        "SELECT 'txt','a',null,'b' UNION ALL SELECT 'int',NULL,1,2 ",
+    )
+    .await?;
+    //TODO: insert into temp-table not merging datatype/nullable of all operations - currently keeping last-writer
+    //assert_union_described(&mut conn, "SELECT 'txt','a',null,'b' UNION     SELECT 'int',NULL,1,2 ").await?;
+
+    Ok(())
+}