Use register_table for execute on cache (#128)

Registering table with `register_listing_table` uses the listing feature of the
object store to find the files to be processed. Internally this call
`ListingTable::list_files_for_scan` in datafusion which panics when
table path is empty. This code change is similar to how query works for s3.

Also remove hostname from listing options, so querying files
works properly on S3. This is for cases where one server posted 
data and another server is trying to query the data.

Author: trueleo

server/src/main.rs

@@ -114,10 +114,6 @@ fn startup_sync() {
 
     for stream in metadata::STREAM_INFO.list_streams() {
         let dir = StorageDir::new(stream.clone());
-        // if data.records file is not present then skip this stream
-        if !dir.local_data_exists() {
-            continue;
-        }
 
         if let Err(e) = dir.create_temp_dir() {
             log::error!(
@@ -127,6 +123,12 @@ fn startup_sync() {
             );
             continue;
         }
+
+        // if data.records file is not present then skip this stream
+        if !dir.local_data_exists() {
+            continue;
+        }
+
         // create prefix for this file from its last modified time
         let path = dir.data_path.join("data.records");
 

server/src/query.rs

@@ -21,6 +21,9 @@ 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::datasource::listing::ListingTable;
+use datafusion::datasource::listing::ListingTableConfig;
+use datafusion::datasource::listing::ListingTableUrl;
 use datafusion::prelude::*;
 use serde_json::Value;
 use std::sync::Arc;
@@ -29,6 +32,7 @@ use crate::metadata::STREAM_INFO;
 use crate::option::CONFIG;
 use crate::storage;
 use crate::storage::ObjectStorage;
+use crate::storage::ObjectStorageError;
 use crate::utils::TimePeriod;
 use crate::validator;
 use crate::Error;
@@ -102,17 +106,26 @@ impl Query {
             None => return Ok(()),
         };
 
-        ctx.register_listing_table(
-            &self.stream_name,
-            CONFIG
-                .parseable
-                .get_cache_path(&self.stream_name)
-                .to_str()
-                .unwrap(),
-            listing_options,
-            Some(schema),
-        )
-        .await?;
+        let cache_path = CONFIG.parseable.get_cache_path(&self.stream_name);
+
+        let table_path = match ListingTableUrl::parse(
+            cache_path.to_str().expect("path should is valid unicode"),
+        ) {
+            Ok(table_path) => table_path,
+            Err(e) => {
+                log::warn!("could not parse local filesystem path. Maybe directory does not exist. Error {}", e);
+                return Ok(());
+            }
+        };
+
+        let config = ListingTableConfig::new(table_path)
+            .with_listing_options(listing_options)
+            .with_schema(schema);
+
+        let table = ListingTable::try_new(config)?;
+
+        ctx.register_table(&*self.stream_name, Arc::new(table))
+            .map_err(ObjectStorageError::DataFusionError)?;
 
         // execute the query and collect results
         let df = ctx.sql(self.query.as_str()).await?;

server/src/s3.rs

@@ -29,7 +29,6 @@ use crate::metadata::Stats;
 use crate::option::{StorageOpt, CONFIG};
 use crate::query::Query;
 use crate::storage::{LogStream, ObjectStorage, ObjectStorageError};
-use crate::utils::hostname_unchecked;
 
 // Default object storage currently is DO Spaces bucket
 // Any user who starts the Parseable server with default configuration
@@ -427,7 +426,7 @@ impl ObjectStorage for S3 {
 
         let file_format = ParquetFormat::default().with_enable_pruning(true);
         let listing_options = ListingOptions {
-            file_extension: format!("{}.data.parquet", hostname_unchecked()),
+            file_extension: ".data.parquet".to_string(),
             format: Arc::new(file_format),
             table_partition_cols: vec![],
             collect_stat: true,