Use existing schema when querying local data. (#86)

Datafusion needs a valid schema to execute a query. If no associated schema
found for registered table then datafusion tries to infer that schema. If
there are no available listings from which schema can be derived then it fails
and returns an error.

This causes entire query to fail and return error response. This is fixed by
giving it a proper schema to work with that is already part of metadata.
There are however ways that this schema is might not be available
( maybe first event has happened yet), then we simply return doing no
changes to RecordBatch.

Fixes parseablehq/console#14

Author: trueleo

server/src/query.rs

@@ -17,13 +17,15 @@
  */
 
 use chrono::{DateTime, Utc};
+use datafusion::arrow::datatypes::Schema;
 use datafusion::arrow::record_batch::RecordBatch;
 use datafusion::datasource::file_format::parquet::ParquetFormat;
 use datafusion::datasource::listing::ListingOptions;
 use datafusion::prelude::*;
 use serde_json::Value;
 use std::sync::Arc;
 
+use crate::metadata::STREAM_INFO;
 use crate::option::CONFIG;
 use crate::storage;
 use crate::storage::ObjectStorage;
@@ -92,6 +94,14 @@ impl Query {
             target_partitions: 1,
         };
 
+        let schema = &STREAM_INFO.schema(&self.stream_name)?;
+
+        if schema.is_empty() {
+            return Ok(());
+        }
+
+        let schema: Arc<Schema> = Arc::new(serde_json::from_str(schema)?);
+
         ctx.register_listing_table(
             &self.stream_name,
             CONFIG
@@ -100,7 +110,7 @@ impl Query {
                 .to_str()
                 .unwrap(),
             listing_options,
-            None,
+            Some(schema),
         )
         .await?;