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()