feat: support system.parquet_files table

Cargo.toml

@@ -86,6 +86,7 @@ sha2 = "0.10.8"
 snap = "1.0.0"
 sqlparser = "0.41.0"
 sysinfo = "0.30.8"
+test-log = { version = "0.2.16", features = ["trace"] }
 thiserror = "1.0"
 tokio = { version = "1.35", features = ["full"] }
 tokio-util = "0.7.9"

influxdb3/src/commands/serve.rs

@@ -156,6 +156,17 @@ pub struct Config {
  action
  )]
  pub segment_duration: SegmentDuration,
+
+ // TODO - tune this default:
+ /// The size of the query log. Up to this many queries will remain in the log before
+ /// old queries are evicted to make room for new ones.
+ #[clap(
+ long = "query-log-size",
+ env = "INFLUXDB3_QUERY_LOG_SIZE",
+ default_value = "1000",
+ action
+ )]
+ pub query_log_size: usize,
 }
 
 /// If `p` does not exist, try to create it as a directory.
@@ -275,6 +286,7 @@ pub async fn command(config: Config) -> Result<()> {
  Arc::clone(&metrics),
  Arc::new(config.datafusion_config),
  10,
+ config.query_log_size,
  ));
 
  let builder = ServerBuilder::new(common_state)

influxdb3/tests/server/flight.rs

@@ -108,16 +108,18 @@ async fn flight() -> Result<(), influxdb3_client::Error> {
 
  assert_batches_sorted_eq!(
  [
- "+--------------+--------------------+-------------+------------+",
- "| catalog_name | db_schema_name | table_name | table_type |",
- "+--------------+--------------------+-------------+------------+",
- "| public | information_schema | columns | VIEW |",
- "| public | information_schema | df_settings | VIEW |",
- "| public | information_schema | schemata | VIEW |",
- "| public | information_schema | tables | VIEW |",
- "| public | information_schema | views | VIEW |",
- "| public | iox | cpu | BASE TABLE |",
- "+--------------+--------------------+-------------+------------+",
+ "+--------------+--------------------+---------------+------------+",
+ "| catalog_name | db_schema_name | table_name | table_type |",
+ "+--------------+--------------------+---------------+------------+",
+ "| public | information_schema | columns | VIEW |",
+ "| public | information_schema | df_settings | VIEW |",
+ "| public | information_schema | schemata | VIEW |",
+ "| public | information_schema | tables | VIEW |",
+ "| public | information_schema | views | VIEW |",
+ "| public | iox | cpu | BASE TABLE |",
+ "| public | system | parquet_files | BASE TABLE |",
+ "| public | system | queries | BASE TABLE |",
+ "+--------------+--------------------+---------------+------------+",
  ],
  &batches
  );

influxdb3/tests/server/main.rs

@@ -17,6 +17,7 @@ mod flight;
 mod limits;
 mod ping;
 mod query;
+mod system_tables;
 mod write;
 
 /// Configuration for a [`TestServer`]
@@ -122,7 +123,6 @@ impl TestServer {
  .expect("connect to gRPC client");
  let mut client = FlightSqlClient::new(channel);
  client.add_header("database", database).unwrap();
- client.add_header("iox-debug", "true").unwrap();
  client
  }
 

influxdb3/tests/server/system_tables.rs

@@ -0,0 +1,92 @@
+use arrow_util::assert_batches_sorted_eq;
+use influxdb3_client::Precision;
+
+use crate::{collect_stream, TestServer};
+
+#[tokio::test]
+async fn queries_table() {
+ let server = TestServer::spawn().await;
+
+ server
+ .write_lp_to_db(
+ "foo",
+ "cpu,host=s1,region=us-east usage=0.9 1\n\
+ cpu,host=s1,region=us-east usage=0.89 2\n\
+ cpu,host=s1,region=us-east usage=0.85 3",
+ Precision::Nanosecond,
+ )
+ .await
+ .expect("write some lp");
+
+ let mut client = server.flight_sql_client("foo").await;
+
+ // Check queries table for completed queries, will be empty:
+ {
+ let response = client
+ .query("SELECT COUNT(*) FROM system.queries WHERE running = false")
+ .await
+ .unwrap();
+
+ let batches = collect_stream(response).await;
+ assert_batches_sorted_eq!(
+ [
+ "+----------+",
+ "| COUNT(*) |",
+ "+----------+",
+ "| 0 |",
+ "+----------+",
+ ],
+ &batches
+ );
+ }
+
+ // Do some queries, to produce some query logs:
+ {
+ let queries = [
+ "SELECT * FROM cpu", // valid
+ "SELECT * FROM mem", // not valid table, will fail, and not be logged
+ "SELECT usage, time FROM cpu", // specific columns
+ ];
+ for q in queries {
+ let response = client.query(q).await;
+ // collect the stream to make sure the query completes:
+ if let Ok(stream) = response {
+ let _batches = collect_stream(stream).await;
+ }
+ }
+ }
+
+ // Now check the log:
+ {
+ // A sub-set of columns is selected in this query, because the queries
+ // table contains may columns whose values change in susequent test runs
+ let response = client
+ .query(
+ "SELECT \
+ phase, \
+ query_type, \
+ query_text, \
+ success, \
+ running, \
+ cancelled \
+ FROM system.queries \
+ WHERE success = true",
+ )
+ .await
+ .unwrap();
+
+ let batches = collect_stream(response).await;
+ assert_batches_sorted_eq!(
+ [
+ "+---------+------------+--------------------------------------------------------------------------------+---------+---------+-----------+",
+ "| phase | query_type | query_text | success | running | cancelled |",
+ "+---------+------------+--------------------------------------------------------------------------------+---------+---------+-----------+",
+ "| success | flightsql | CommandStatementQuerySELECT * FROM cpu | true | false | false |",
+ "| success | flightsql | CommandStatementQuerySELECT COUNT(*) FROM system.queries WHERE running = false | true | false | false |",
+ "| success | flightsql | CommandStatementQuerySELECT usage, time FROM cpu | true | false | false |",
+ "+---------+------------+--------------------------------------------------------------------------------+---------+---------+-----------+",
+ ],
+ &batches
+ );
+ }
+}

influxdb3_server/Cargo.toml

@@ -16,6 +16,7 @@ iox_http.workspace = true
 iox_query.workspace = true
 iox_query_params.workspace = true
 iox_query_influxql.workspace = true
+iox_system_tables.workspace = true
 iox_time.workspace = true
 metric.workspace = true
 metric_exporters.workspace = true
@@ -70,6 +71,7 @@ parquet.workspace = true
 parquet_file.workspace = true
 test_helpers.workspace = true
 test_helpers_end_to_end.workspace = true
+test-log.workspace = true
 
 # crates.io crates
 http.workspace = true

influxdb3_server/src/lib.rs

@@ -290,6 +290,7 @@ mod tests {
  Arc::clone(&metrics),
  Arc::new(HashMap::new()),
  10,
+ 10,
  ));
 
  let server = ServerBuilder::new(common_state)
@@ -449,6 +450,7 @@ mod tests {
  Arc::clone(&metrics),
  Arc::new(HashMap::new()),
  10,
+ 10,
  );
 
  let server = ServerBuilder::new(common_state)
@@ -655,6 +657,7 @@ mod tests {
  Arc::clone(&metrics),
  Arc::new(HashMap::new()),
  10,
+ 10,
  );
 
  let server = ServerBuilder::new(common_state)