Memory datasource protobuf support (#17290)

* Add proto

* fix proto

* gen proto code

* exec to proto

* gen proto code

* impl (de)serialization

* Add test

* Update submodules

* gen proto

* Set parquet-testing back to main

---------

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>
Co-authored-by: Tim Saucer <timsaucer@gmail.com>

Author: 3 people

datafusion/proto/proto/datafusion.proto

@@ -730,6 +730,7 @@ message PhysicalPlanNode {
     CooperativeExecNode cooperative = 32;
     GenerateSeriesNode generate_series = 33;
     SortMergeJoinExecNode sort_merge_join = 34;
+    MemoryScanExecNode memory_scan = 35;
   }
 }
 
@@ -1041,6 +1042,15 @@ message AvroScanExecNode {
   FileScanExecConf base_conf = 1;
 }
 
+message MemoryScanExecNode {
+  repeated bytes partitions = 1;
+  datafusion_common.Schema schema = 2;
+  repeated uint32 projection = 3;
+  repeated PhysicalSortExprNodeCollection sort_information = 4;
+  bool show_sizes = 5;
+  optional uint32 fetch = 6;
+}
+
 message CooperativeExecNode {
   PhysicalPlanNode input = 1;
 }

datafusion/proto/src/generated/pbjson.rs

@@ -19,8 +19,10 @@
 
 use std::sync::Arc;
 
+use arrow::array::RecordBatch;
 use arrow::compute::SortOptions;
 use arrow::datatypes::Field;
+use arrow::ipc::reader::StreamReader;
 use chrono::{TimeZone, Utc};
 use datafusion_expr::dml::InsertOp;
 use object_store::path::Path;
@@ -556,6 +558,18 @@ pub fn parse_protobuf_file_scan_config(
     Ok(config)
 }
 
+pub fn parse_record_batches(buf: &[u8]) -> Result<Vec<RecordBatch>> {
+    if buf.is_empty() {
+        return Ok(vec![]);
+    }
+    let reader = StreamReader::try_new(buf, None)?;
+    let mut batches = Vec::new();
+    for batch in reader {
+        batches.push(batch?);
+    }
+    Ok(batches)
+}
+
 impl TryFrom<&protobuf::PartitionedFile> for PartitionedFile {
     type Error = DataFusionError;