feat: cloud_upload(<filename>) table func

.github/workflows/ci.yaml

@@ -593,10 +593,10 @@ jobs:
           - name: PostgreSQL
             path: "sqllogictests_postgres/*"
             prepare: |
-               ./scripts/prepare-testdata.sh
-               export POSTGRES_TEST_DB=$(./scripts/create-test-postgres-db.sh)
-               export POSTGRES_CONN_STRING=$(echo "$POSTGRES_TEST_DB" | sed -n 1p)
-               export POSTGRES_TUNNEL_SSH_CONN_STRING=$(echo "$POSTGRES_TEST_DB" | sed -n 2p)
+              ./scripts/prepare-testdata.sh
+              export POSTGRES_TEST_DB=$(./scripts/create-test-postgres-db.sh)
+              export POSTGRES_CONN_STRING=$(echo "$POSTGRES_TEST_DB" | sed -n 1p)
+              export POSTGRES_TUNNEL_SSH_CONN_STRING=$(echo "$POSTGRES_TEST_DB" | sed -n 2p)
     runs-on: ubuntu-latest
     needs: ["sql-logic-tests"]
     steps:
@@ -629,10 +629,9 @@ jobs:
 
           just slt-bin --protocol=flightsql ${{matrix.settings.path}}
 
-
   docker-push:
     name: Build and Push Docker Image
-    if: github.event_name == 'push' && github.repository == 'GlareDB/glaredb' && github.ref == 'refs/heads/main'
+    # if: github.event_name == 'push' && github.repository == 'GlareDB/glaredb' && github.ref == 'refs/heads/main'
     needs: ["sql-logic-tests"]
     runs-on: ubuntu-latest-8-cores
     permissions:

crates/datasources/src/bson/table.rs

@@ -75,7 +75,7 @@ pub async fn bson_streaming_table(
                     // field name against the schema, and only pull out the
                     // fields that match. This is easier in the short term
                     // but less performant for large documents where the
-                    // docuemnts are a superset of the schema, we'll end up
+                    // documents are a superset of the schema, we'll end up
                     // doing much more parsing work than is actually needed
                     // for the bson documents.
                     |bt: Result<BytesMut, std::io::Error>| -> Result<RawDocumentBuf, BsonError> {
@@ -112,7 +112,7 @@ pub async fn bson_streaming_table(
         readers.pop_front();
     }
 
-    // infer the sechema; in the future we can allow users to specify the
+    // infer the schema; in the future we can allow users to specify the
     // schema directly; in the future users could specify the schema (kind
     // of as a base-level projection, but we'd need a schema specification
     // language). Or have some other strategy for inference rather than

crates/datasources/src/native/access.rs

@@ -41,11 +41,13 @@ use crate::native::insert::NativeTableInsertExec;
 #[derive(Debug, Clone)]
 pub struct NativeTableStorage {
     db_id: Uuid,
-    /// URL pointing to the bucket and/or directory which is the root of the native storage.
+
+    /// URL pointing to the bucket and/or directory which is the root of the
+    /// native storage, for example `gs://<bucket-name>`.
     ///
-    /// In other words this is the location to which the the table prefix is applied to get
-    /// a full table URL.
-    root_url: Url,
+    /// In other words this is the location to which the the table prefix is
+    /// applied to get a full table URL.
+    pub root_url: Url,
 
     /// Tables are only located in one bucket which the provided service account
     /// should have access to.
@@ -56,12 +58,15 @@ pub struct NativeTableStorage {
     /// what this type is for.
     ///
     /// Arcs all the way down...
-    store: SharedObjectStore,
+    pub store: SharedObjectStore,
 }
-// Deltalake is expecting a factory that implements `ObjectStoreFactory` and `LogStoreFactory`.
-// Since we already have an object store, we don't need to do anything here,
-// but we still need to register the url with delta-rs so it does't error when it tries to validate the object-store.
-// So we just create a fake factory that returns the object store we already have and register it with the root url.
+
+/// Deltalake is expecting a factory that implements [`ObjectStoreFactory`] and
+/// [`LogStoreFactory`]. Since we already have an object store, we don't need to
+/// do anything here, but we still need to register the url with delta-rs so it
+/// does't error when it tries to validate the object-store. So we just create a
+/// fake factory that returns the object store we already have and register it
+/// with the root url.
 struct FakeStoreFactory {
     pub store: ObjectStoreRef,
 }
@@ -88,6 +93,10 @@ impl LogStoreFactory for FakeStoreFactory {
         Ok(default_logstore(store, location, options))
     }
 }
+
+/// DeltaField represents data types as stored in Delta Lake, with additional
+/// metadata for indicating the 'real' (original) type, for cases when
+/// downcasting occurs.
 struct DeltaField {
     data_type: DeltaDataType,
     metadata: Option<HashMap<String, Value>>,
@@ -142,8 +151,8 @@ fn arrow_to_delta_safe(arrow_type: &DataType) -> DeltaResult<DeltaField> {
 }
 
 impl NativeTableStorage {
-    /// Create a native table storage provider from a URL and an object store instance
-    /// rooted at that location.
+    /// Create a native table storage provider from a URL and an object store
+    /// instance rooted at that location.
     pub fn new(db_id: Uuid, root_url: Url, store: Arc<dyn ObjectStore>) -> NativeTableStorage {
         // register the default handlers
         // TODO, this should only happen once
@@ -172,6 +181,7 @@ impl NativeTableStorage {
         self.db_id
     }
 
+    /// Returns the location of 'native' Delta Lake tables.
     fn table_prefix(&self, tbl_id: u32) -> String {
         format!("databases/{}/tables/{}", self.db_id, tbl_id)
     }

crates/datasources/src/object_store/mod.rs

@@ -284,6 +284,24 @@ pub struct ObjStoreTableProvider {
     file_format: Arc<dyn FileFormat>,
 }
 
+impl ObjStoreTableProvider {
+    pub fn new(
+        store: Arc<dyn ObjectStore>,
+        arrow_schema: SchemaRef,
+        base_url: ObjectStoreUrl,
+        objects: Vec<ObjectMeta>,
+        file_format: Arc<dyn FileFormat>,
+    ) -> ObjStoreTableProvider {
+        ObjStoreTableProvider {
+            store,
+            arrow_schema,
+            base_url,
+            objects,
+            file_format,
+        }
+    }
+}
+
 #[async_trait]
 impl TableProvider for ObjStoreTableProvider {
     fn as_any(&self) -> &dyn Any {

crates/sqlbuiltins/src/functions/mod.rs

@@ -35,13 +35,11 @@ use table::{BuiltinTableFuncs, TableFunc};
 use self::alias_map::AliasMap;
 use crate::functions::scalars::openai::OpenAIEmbed;
 
-/// All builtin functions available for all sessions.
+/// FUNCTION_REGISTRY provides all implementations of [`BuiltinFunction`]
 pub static FUNCTION_REGISTRY: Lazy<FunctionRegistry> = Lazy::new(FunctionRegistry::new);
 
-/// A builtin function.
-/// This trait is implemented by all builtin functions.
-/// This is used to derive catalog entries for all supported functions.
-/// Any new function MUST implement this trait.
+/// BuiltinFunction **MUST** be implemented by all builtin functions, including
+/// new ones. This is used to derive catalog entries for all supported functions.
 pub trait BuiltinFunction: Sync + Send {
     /// The name for this function. This name will be used when looking up
     /// function implementations.
@@ -87,6 +85,7 @@ pub trait ConstBuiltinFunction: Sync + Send {
         None
     }
 }
+
 /// The namespace of a function.
 ///
 /// Optional -> "namespace.function" || "function"

crates/sqlbuiltins/src/functions/table/mod.rs

@@ -47,15 +47,15 @@ use self::json::JsonScan;
 use self::lance::LanceScan;
 use self::mongodb::ReadMongoDb;
 use self::mysql::ReadMysql;
-use self::object_store::{READ_CSV, READ_JSON, READ_PARQUET};
+use self::object_store::{GlareDBUpload, READ_CSV, READ_JSON, READ_PARQUET};
 use self::postgres::ReadPostgres;
 use self::snowflake::ReadSnowflake;
 use self::sqlite::ReadSqlite;
 use self::sqlserver::ReadSqlServer;
 use self::system::cache_external_tables::CacheExternalDatabaseTables;
 use self::virtual_listing::{ListColumns, ListSchemas, ListTables};
-use super::alias_map::AliasMap;
-use super::BuiltinFunction;
+use crate::functions::alias_map::AliasMap;
+use crate::functions::BuiltinFunction;
 
 /// A builtin table function.
 /// Table functions are ones that are used in the FROM clause.
@@ -103,6 +103,7 @@ impl BuiltinTableFuncs {
             Arc::new(READ_JSON),
             Arc::new(BsonScan),
             Arc::new(JsonScan),
+            Arc::new(GlareDBUpload),
             // Data lakes
             Arc::new(DeltaScan),
             Arc::new(IcebergScan),

crates/sqlbuiltins/src/functions/table/object_store.rs

@@ -1,5 +1,6 @@
 use std::collections::HashMap;
 use std::marker::PhantomData;
+use std::path::Path;
 use std::sync::Arc;
 use std::vec;
 
@@ -16,19 +17,21 @@ use datafusion::logical_expr::{Signature, TypeSignature, Volatility};
 use datafusion_ext::errors::{ExtensionError, Result};
 use datafusion_ext::functions::{FuncParamValue, IdentValue, TableFuncContextProvider};
 use datasources::common::url::{DatasourceUrl, DatasourceUrlType};
+use datasources::native::access::NativeTableStorage;
 use datasources::object_store::gcs::GcsStoreAccess;
 use datasources::object_store::generic::GenericStoreAccess;
 use datasources::object_store::http::HttpStoreAccess;
 use datasources::object_store::local::LocalStoreAccess;
 use datasources::object_store::s3::S3StoreAccess;
-use datasources::object_store::{MultiSourceTableProvider, ObjStoreAccess};
-use futures::TryStreamExt;
+use datasources::object_store::{MultiSourceTableProvider, ObjStoreAccess, ObjStoreTableProvider};
+use futures::{StreamExt, TryStreamExt};
 use object_store::azure::AzureConfigKey;
+use object_store::path::Path as ObjectStorePath;
+use object_store::{ObjectMeta, ObjectStore};
 use protogen::metastore::types::catalog::{FunctionType, RuntimePreference};
 use protogen::metastore::types::options::{CredentialsOptions, StorageOptions};
 
-use super::TableFunc;
-use crate::functions::BuiltinFunction;
+use crate::functions::{BuiltinFunction, ConstBuiltinFunction, TableFunc};
 
 #[derive(Debug, Clone, Copy)]
 pub struct ParquetOptionsReader;
@@ -123,7 +126,7 @@ pub trait OptionReader: Sync + Send + Sized {
     /// List of options and their expected data types.
     const OPTIONS: &'static [(&'static str, DataType)];
 
-    /// Read user provided options, and construct a file format usign those options.
+    /// Read user provided options, and construct a file format using those options.
     fn read_options(opts: &HashMap<String, FuncParamValue>) -> Result<Self::Format>;
 }
 
@@ -582,3 +585,122 @@ fn create_azure_store_access(
         storage_options: opts,
     }))
 }
+
+#[derive(Debug, Clone, Copy, Default)]
+pub struct GlareDBUpload;
+
+impl ConstBuiltinFunction for GlareDBUpload {
+    const NAME: &'static str = "glaredb_upload";
+    const DESCRIPTION: &'static str = "Reads a file that was uploaded to GlareDB Cloud.";
+    const EXAMPLE: &'static str = "SELECT * FROM glaredb_upload('my_upload.csv')";
+    const FUNCTION_TYPE: FunctionType = FunctionType::TableReturning;
+
+    // signature for GlareUpload is a single filename. The filename may
+    // optionally contain an extension, though it is not required. Filename
+    // should not be a path.
+    fn signature(&self) -> Option<Signature> {
+        Some(Signature::uniform(
+            1,
+            vec![DataType::Utf8],
+            Volatility::Stable,
+        ))
+    }
+}
+
+#[async_trait]
+impl TableFunc for GlareDBUpload {
+    fn detect_runtime(
+        &self,
+        _args: &[FuncParamValue],
+        _parent: RuntimePreference,
+    ) -> Result<RuntimePreference> {
+        // Uploads can only exist remotely; this operation is not meaningful
+        // when not connected to remote/hybrid.
+        Ok(RuntimePreference::Remote)
+    }
+
+    async fn create_provider(
+        &self,
+        ctx: &dyn TableFuncContextProvider,
+        args: Vec<FuncParamValue>,
+        _opts: HashMap<String, FuncParamValue>,
+    ) -> Result<Arc<dyn TableProvider>> {
+        if args.len() != 1 {
+            return Err(ExtensionError::InvalidNumArgs);
+        }
+
+        let session_state = ctx.get_session_state();
+
+        // NativeTableStorage is available on Remote ctx. Use store already
+        // constructed there.
+        let storage = ctx
+            .get_session_state()
+            .config()
+            .get_extension::<NativeTableStorage>()
+            .ok_or_else(|| {
+                ExtensionError::String(
+                    format!(
+                        "access unavailable, {} is not supported in local environments",
+                        self.name(),
+                    )
+                    .to_string(),
+                )
+            })?;
+        let store = storage.store.clone(); // Cheap to clone
+
+        let file_name: String = args.into_iter().next().unwrap().try_into()?;
+        let ext = Path::new(&file_name)
+            .extension()
+            .ok_or_else(|| {
+                ExtensionError::String(
+                    "missing file extension, supported: [.csv, .json, .parquet]".to_string(),
+                )
+            })?
+            .to_str()
+            .ok_or_else(|| {
+                ExtensionError::String(format!("unsupported file extension: {file_name}"))
+            })?
+            .to_lowercase();
+        let file_format: Arc<dyn FileFormat> = match ext.as_str() {
+            "csv" => Arc::new(CsvFormat::default().with_schema_infer_max_rec(Some(20480))),
+            "json" => Arc::new(JsonFormat::default()),
+            "parquet" => Arc::new(ParquetFormat::default()),
+            ext => {
+                return Err(ExtensionError::String(format!(
+                    "unsupported file extension: {ext}"
+                )))
+            }
+        };
+
+        // This is required to read the object meta, but we unfortunately also
+        // need the base_url below for ObjStoreTableProvider impl. Maybe there's
+        // a refactor opportunity.
+        let prefix: ObjectStorePath =
+            format!("databases/{}/uploads/{}", storage.db_id(), file_name).into();
+        let base_url = format!("{}", storage.root_url);
+        let base_url = ObjectStoreUrl::parse(base_url)?;
+
+        // glaredb_upload currently does not support globbing, and therefore we
+        // do not need to iterate: there should only be one.
+        let mut objects = store.list(Some(&prefix));
+        let meta = objects
+            .next()
+            .await
+            .ok_or_else(|| ExtensionError::String(format!("file not found: {}", file_name)))?
+            .or_else(|_| return Err(ExtensionError::ObjectStore("read file".to_string())))?;
+        let objects: Vec<ObjectMeta> = vec![meta];
+
+        // Infer schema
+        let arrow_schema = file_format
+            .infer_schema(&session_state, &store.inner, &objects)
+            .await?;
+
+        return Ok(Arc::new(ObjStoreTableProvider::new(
+            store.inner.clone(),
+            arrow_schema,
+            base_url,
+            objects,
+            file_format,
+        )));
+    }
+}