Fix result_scan columns for empty records batch

crates/core-executor/Cargo.toml

@@ -10,6 +10,7 @@ core-metastore = { path = "../core-metastore" }
 core-history = {  path = "../core-history" }
 df-catalog = { path = "../df-catalog" }
 embucket-functions = { path = "../embucket-functions" }
+arrow-schema = { version = "55", features = ["serde"] }
 async-trait = { workspace = true }
 aws-config = { workspace = true }
 aws-credential-types = { workspace = true }

crates/core-executor/src/models.rs

@@ -69,7 +69,7 @@ impl QueryResult {
 
     #[must_use]
     pub fn column_info(&self) -> Vec<ColumnInfo> {
-        ColumnInfo::from_batch(&self.schema)
+        ColumnInfo::from_schema(&self.schema)
     }
 }
 
@@ -108,7 +108,7 @@ impl ColumnInfo {
     }
 
     #[must_use]
-    pub fn from_batch(schema: &Arc<Schema>) -> Vec<Self> {
+    pub fn from_schema(schema: &Arc<Schema>) -> Vec<Self> {
         let mut column_infos = Vec::new();
         for field in schema.fields() {
             column_infos.push(Self::from_field(field));

crates/core-executor/src/query.rs

@@ -1322,7 +1322,7 @@ impl UserQuery {
                     let is_table_func = self
                         .session
                         .ctx
-                        .table_function(table_name.to_string().as_str())
+                        .table_function(table_name.to_string().to_lowercase().as_str())
                         .is_ok();
                     if !cte_names.contains(&table_name.to_string()) && !is_table_func {
                         match self.resolve_table_object_name(table_name.0.clone()) {

crates/core-executor/src/utils.rs

@@ -518,10 +518,13 @@ pub fn query_result_to_result_set(query_result: &QueryResult) -> ExecutionResult
         })
         .collect();
 
+    // Serialize original Schema into a JSON string
+    let schema = serde_json::to_string(&query_result.schema).context(SerdeParseSnafu)?;
     Ok(ResultSet {
         columns,
         rows,
         data_format: data_format.to_string(),
+        schema,
     })
 }
 

crates/core-history/src/entities/result_set.rs

@@ -22,4 +22,5 @@ pub struct ResultSet {
     pub columns: Vec<Column>,
     pub rows: Vec<Row>,
     pub data_format: String,
+    pub schema: String,
 }

crates/embucket-functions/Cargo.toml

@@ -5,6 +5,7 @@ edition = "2024"
 license-file.workspace = true
 
 [dependencies]
+arrow-schema = { version = "55", features = ["serde"] }
 chrono = { workspace = true }
 core-history = { path = "../core-history" }
 datafusion = { workspace = true }

crates/embucket-functions/src/table/result_scan.rs

@@ -4,12 +4,13 @@ use core_history::{GetQueriesParams, HistoryStore, QueryRecord};
 use datafusion::arrow;
 use datafusion::arrow::array::RecordBatch;
 use datafusion::arrow::datatypes::SchemaRef;
-use datafusion::arrow::json::reader::{ReaderBuilder, infer_json_schema};
+use datafusion::arrow::json::StructMode;
+use datafusion::arrow::json::reader::ReaderBuilder;
 use datafusion::catalog::{TableFunctionImpl, TableProvider};
 use datafusion::datasource::MemTable;
 use datafusion_common::{DataFusionError, Result as DFResult, ScalarValue, exec_err};
 use datafusion_expr::Expr;
-use std::io::{BufReader, Cursor};
+use std::io::Cursor;
 use std::sync::Arc;
 
 #[derive(Debug, Clone)]
@@ -85,12 +86,14 @@ impl ResultScanFunc {
             .map_err(|e| DataFusionError::External(Box::new(e)))?;
 
         let arrow_json = convert_resultset_to_arrow_json_lines(&result_set)?;
-        let mut buf_reader = BufReader::new(Cursor::new(&arrow_json));
-        let (inferred_schema, _) = infer_json_schema(&mut buf_reader, None)
+
+        // Parse schema from serialized JSON
+        let schema_value = serde_json::from_str(&result_set.schema)
             .map_err(|e| DataFusionError::External(Box::new(e)))?;
 
-        let schema_ref: SchemaRef = Arc::new(inferred_schema);
+        let schema_ref: SchemaRef = Arc::new(schema_value);
         let json_reader = ReaderBuilder::new(schema_ref.clone())
+            .with_struct_mode(StructMode::ListOnly)
             .build(Cursor::new(&arrow_json))
             .map_err(|e| DataFusionError::External(Box::new(e)))?;
 
@@ -149,17 +152,14 @@ fn utf8_val(val: impl Into<String>) -> ScalarValue {
 fn convert_resultset_to_arrow_json_lines(
     result_set: &ResultSet,
 ) -> Result<String, DataFusionError> {
-    let mut output = String::new();
-
+    let mut lines = String::new();
     for row in &result_set.rows {
-        let mut record = serde_json::Map::with_capacity(result_set.columns.len());
-        for (col, value) in result_set.columns.iter().zip(&row.0) {
-            record.insert(col.name.clone(), value.clone());
-        }
-        let json_line =
-            serde_json::to_string(&record).map_err(|e| DataFusionError::External(Box::new(e)))?;
-        output.push_str(&json_line);
-        output.push('\n');
+        let json_value = serde_json::Value::Array(row.0.clone());
+        lines.push_str(
+            &serde_json::to_string(&json_value)
+                .map_err(|e| DataFusionError::External(Box::new(e)))?,
+        );
+        lines.push('\n');
     }
-    Ok(output)
+    Ok(lines)
 }

crates/embucket-functions/src/tests/utils.rs

@@ -44,9 +44,9 @@ pub fn history_store_mock() -> Arc<dyn HistoryStore> {
         {
             "columns": [{"name":"a","type":"text"},{"name":"b","type":"text"},{"name":"c","type":"text"}],
             "rows": [[1,"2",true],[2.0,"4",false]],
-            "data_format": "arrow"
-        }
-        "#;
+            "data_format": "arrow",
+            "schema": "{\"fields\":[{\"name\":\"a\",\"data_type\":\"Float64\",\"nullable\":false,\"dict_id\":0,\"dict_is_ordered\":false,\"metadata\":{}},{\"name\":\"b\",\"data_type\":\"Utf8\",\"nullable\":false,\"dict_id\":0,\"dict_is_ordered\":false,\"metadata\":{}},{\"name\":\"c\",\"data_type\":\"Boolean\",\"nullable\":false,\"dict_id\":0,\"dict_is_ordered\":false,\"metadata\":{}}],\"metadata\":{}}"
+        }"#;
         record.result = Some(buf.to_string());
         if id == 500 {
             return Err(HistoryStoreError::ExecutionResult {