feat: support SHOW queries for /query API

* SHOW DATABASES
* SHOW RETENTION POLICIES

Includes tests for the lower level implementation of these interfaces
from iox_query

Author: hiltontj

Cargo.lock

@@ -84,8 +84,8 @@ criterion = { version = "0.5", features = ["html_reports"] }
 crossbeam-channel = "0.5.11"
 csv = "1.3.0"
 # See https://github.com/influxdata/arrow-datafusion/pull/73 for contents
-datafusion = { git = "https://github.com/influxdata/arrow-datafusion.git", rev = "a9cf9aca9ebf0d6c04e0861d2baebffa0ba77dbc" }
-datafusion-proto = { git = "https://github.com/influxdata/arrow-datafusion.git", rev = "a9cf9aca9ebf0d6c04e0861d2baebffa0ba77dbc" }
+datafusion = { git = "https://github.com/influxdata/arrow-datafusion.git", rev = "ee81b1cc652bde6c131973d091b178836692112d" }
+datafusion-proto = { git = "https://github.com/influxdata/arrow-datafusion.git", rev = "ee81b1cc652bde6c131973d091b178836692112d" }
 dashmap = "6.1.0"
 dotenvy = "0.15.7"
 flate2 = "1.0.27"
@@ -112,25 +112,25 @@ mockito = { version = "1.4.0", default-features = false }
 mockall = { version = "0.13.0" }
 non-empty-string = "0.2.5"
 num_cpus = "1.16.0"
-object_store = { version = "0.12.3", features = ["aws", "azure", "gcp"] }
+object_store = { version = "0.12.4", features = ["aws", "azure", "gcp"] }
 parking_lot = { version = "0.12.1", features = ["serde"] }
 paste = "1.0.15"
 parquet = {version = "55", features = ["object_store"]}
-pbjson = "0.6.0"
-pbjson-build = "0.6.2"
-pbjson-types = "0.6.0"
+pbjson = "0.8"
+pbjson-build = "0.8"
+pbjson-types = "0.7"
 pin-project-lite = "0.2"
 pretty_assertions = { version = "1.4.0", features = ["unstable"] }
-prost = "0.12.6"
-prost-build = "0.12.6"
-prost-types = "0.12.6"
+prost = "0.13"
+prost-build = "0.13"
+prost-types = "0.13"
 proptest = { version = "1", default-features = false, features = ["std"] }
 pyo3 = { version = "0.24.1", features = ["experimental-async"]}
 rand = "0.8.5"
 rcgen = "0.13.2"
 regex = "1.11.1"
-reqwest = { version = "0.11.27", default-features = false, features = ["rustls-tls", "stream", "json"] }
-rstest = "0.18"
+reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "stream", "json"] }
+rstest = "0.26"
 rustls = { version = "0.23", default-features = false, features = ["logging", "ring", "std", "tls12"] }
 rustls-webpki = { version = "0.103", default-features = false, features = ["ring", "std"] }
 rustls-pemfile = "2.2.0"
@@ -157,7 +157,7 @@ tonic = { version = "0.12.3", features = ["tls", "tls-roots"] }
 tonic-build = "0.12.3"
 tonic-health = "0.12.3"
 tonic-reflection = "0.12.3"
-tower = "0.4.13"
+tower = "0.5"
 twox-hash = "2.1.0"
 unicode-segmentation = "1.11.0"
 url = "2.5.0"

Cargo.toml

@@ -1040,6 +1040,7 @@ pub async fn command(config: Config, user_params: HashMap<String, String>) -> Re
     let write_buffer: Arc<dyn WriteBuffer> = write_buffer_impl;
 
     let common_state = CommonServerState::new(
+        Arc::clone(&catalog),
         Arc::clone(&metrics),
         trace_exporter,
         trace_header_parser,

influxdb3/src/commands/serve.rs

@@ -761,13 +761,7 @@ impl Drop for TestServer {
 }
 
 impl TestServer {
-    /// Write some line protocol to the server
-    pub async fn write_lp_to_db(
-        &self,
-        database: &str,
-        lp: impl ToString,
-        precision: Precision,
-    ) -> Result<(), influxdb3_client::Error> {
+    fn maybe_authorized_client(&self) -> influxdb3_client::Client {
         let mut client = influxdb3_client::Client::new(
             self.client_addr(),
             Some("../testing-certs/rootCA.pem".into()),
@@ -776,6 +770,17 @@ impl TestServer {
         if let Some(token) = &self.auth_token {
             client = client.with_auth_token(token);
         }
+        client
+    }
+
+    /// Write some line protocol to the server
+    pub async fn write_lp_to_db(
+        &self,
+        database: &str,
+        lp: impl ToString,
+        precision: Precision,
+    ) -> Result<(), influxdb3_client::Error> {
+        let client = self.maybe_authorized_client();
         client
             .api_v3_write_lp(database)
             .body(lp.to_string())
@@ -784,21 +789,25 @@ impl TestServer {
             .await
     }
 
+    pub async fn api_v3_create_database(
+        &self,
+        database: &str,
+        retention_period: Option<Duration>,
+    ) -> Result<(), influxdb3_client::Error> {
+        let client = self.maybe_authorized_client();
+        client
+            .api_v3_configure_db_create(database, retention_period)
+            .await
+    }
+
     pub async fn api_v3_create_table(
         &self,
         database: &str,
         table: &str,
         tags: Vec<String>,
         fields: Vec<(String, FieldType)>,
     ) -> Result<(), influxdb3_client::Error> {
-        let mut client = influxdb3_client::Client::new(
-            self.client_addr(),
-            Some("../testing-certs/rootCA.pem".into()),
-        )
-        .unwrap();
-        if let Some(token) = &self.auth_token {
-            client = client.with_auth_token(token);
-        }
+        let client = self.maybe_authorized_client();
 
         client
             .api_v3_configure_table_create(database, table, tags, fields)

influxdb3/tests/server/mod.rs

@@ -1,4 +1,5 @@
 use core::str;
+use std::time::Duration;
 
 use crate::server::TestServer;
 use futures::StreamExt;
@@ -1799,6 +1800,69 @@ async fn api_v1_query_group_by_with_nulls() {
     }
 }
 
+#[tokio::test]
+async fn api_v1_query_api_show_databases_and_retention_policies() {
+    let server = TestServer::spawn().await;
+
+    server.api_v3_create_database("foo", None).await.unwrap();
+    server
+        .api_v3_create_database("bar", Some(Duration::from_secs(30 * 24 * 60 * 60)))
+        .await
+        .unwrap();
+
+    let response = server
+        .api_v1_query(&[("q", "SHOW DATABASES")], None)
+        .await
+        .text()
+        .await
+        .unwrap();
+
+    insta::with_settings!({
+        description => "SHOW DATABASES",
+    }, {
+        insta::assert_snapshot!(response);
+    });
+
+    let response = server
+        .api_v1_query(&[("q", "SHOW RETENTION POLICIES"), ("db", "foo")], None)
+        .await
+        .text()
+        .await
+        .unwrap();
+
+    insta::with_settings!({
+        description => "SHOW RETENTION POLICIES on foo db",
+    }, {
+        insta::assert_snapshot!(response);
+    });
+
+    let response = server
+        .api_v1_query(&[("q", "SHOW RETENTION POLICIES"), ("db", "bar")], None)
+        .await
+        .text()
+        .await
+        .unwrap();
+
+    insta::with_settings!({
+        description => "SHOW RETENTION POLICIES on bar db",
+    }, {
+        insta::assert_snapshot!(response);
+    });
+
+    let response = server
+        .api_v1_query(&[("q", "SHOW RETENTION POLICIES"), ("db", "frodo")], None)
+        .await
+        .text()
+        .await
+        .unwrap();
+
+    insta::with_settings!({
+        description => "SHOW RETENTION POLICIES on non-existent db",
+    }, {
+        insta::assert_snapshot!(response);
+    });
+}
+
 #[tokio::test]
 async fn api_v3_query_sql_distinct_cache() {
     let server = TestServer::spawn().await;

influxdb3/tests/server/query.rs

@@ -0,0 +1,6 @@
+---
+source: influxdb3/tests/server/query.rs
+description: SHOW DATABASES
+expression: response
+---
+{"results":[{"statement_id":0,"series":[{"name":"databases","columns":["name"],"values":[["_internal"],["foo"],["bar"]]}]}]}

influxdb3/tests/server/snapshots/lib__server__query__api_v1_query_api_show_databases.snap

@@ -0,0 +1,6 @@
+---
+source: influxdb3/tests/server/query.rs
+description: SHOW RETENTION POLICIES on foo db
+expression: response
+---
+{"results":[{"statement_id":0,"series":[{"name":"retention_policies","columns":["name","duration","default"],"values":[["foo",null,true]]}]}]}

influxdb3/tests/server/snapshots/lib__server__query__api_v1_query_api_show_databases_and_retention_policies-2.snap

@@ -0,0 +1,6 @@
+---
+source: influxdb3/tests/server/query.rs
+description: SHOW RETENTION POLICIES on bar db
+expression: response
+---
+{"results":[{"statement_id":0,"series":[{"name":"retention_policies","columns":["name","duration","default"],"values":[["bar","30days",true]]}]}]}

influxdb3/tests/server/snapshots/lib__server__query__api_v1_query_api_show_databases_and_retention_policies-3.snap

@@ -0,0 +1,6 @@
+---
+source: influxdb3/tests/server/query.rs
+description: SHOW RETENTION POLICIES on non-existent db
+expression: response
+---
+{"results":[{"statement_id":0,"error":"datafusion error: Error during planning: database not found: frodo"}]}

influxdb3/tests/server/snapshots/lib__server__query__api_v1_query_api_show_databases_and_retention_policies-4.snap

@@ -0,0 +1,6 @@
+---
+source: influxdb3/tests/server/query.rs
+description: SHOW DATABASES
+expression: response
+---
+{"results":[{"statement_id":0,"series":[{"name":"databases","columns":["name"],"values":[["_internal"],["foo"],["bar"]]}]}]}