Add in-memory store

.github/workflows/ci.yml

@@ -154,7 +154,7 @@ jobs:
             /o:"npiontko" \
             /d:sonar.token="$SONAR_TOKEN" \
             /d:sonar.cs.vscoveragexml.reportsPaths=coverage.xml \
-            /d:sonar.coverage.exclusions="docs/**,examples/**,native/**""
+            /d:sonar.coverage.exclusions="docs/**,examples/**,native/**"
 
           # Build the project without building the native library, it should be already built in the check job and cached
           dotnet build -c Release -p:BuildNativeLib=false

DataFusionSharp.slnx

@@ -70,6 +70,7 @@
   </Folder>
   <Folder Name="/examples/">
     <Project Path="examples\QueryData\QueryData.csproj" />
+    <Project Path="examples\QueryInMemoryData\QueryInMemoryData.csproj" />
     <Project Path="examples/QueryS3DataWithDapper/QueryS3DataWithDapper.csproj" />
     <Project Path="examples\QueryDataWithDapper\QueryDataWithDapper.csproj" />
     <Project Path="examples\QueryS3Data\QueryS3Data.csproj" />

README.md

@@ -11,8 +11,8 @@
 
 ## Features
 
-| Component        | Feature                                      | Status  | Notes                                             |
-|------------------|----------------------------------------------|---------|---------------------------------------------------|
+| Component        | Feature                                      | Status | Notes                                             |
+|------------------|----------------------------------------------|--------|---------------------------------------------------|
 | **Runtime**      | Tokio runtime                                | ✅      | Configurable threads, supports multiple instances |
 |                  | Logger                                       | ✅      | Configurable with log levels                      |
 | **Session**      | Create session context                       | ✅      |                                                   |
@@ -25,13 +25,13 @@
 |                  | Amazon S3                                    | ✅      |                                                   |
 |                  | Azure Blob Storage                           | ✅      |                                                   |
 |                  | Google Cloud Storage                         | ✅      |                                                   |
-|                  | InMemory                                     | ❌      |                                                   |
+|                  | InMemory                                     | ✅      |                                                   |
 | **DataFrame**    | Count rows                                   | ✅      | `CountAsync()`                                    |
 |                  | Get schema                                   | ✅      | `GetSchema()` → Arrow Schema                      |
 |                  | Collect all data                             | ✅      | `CollectAsync()` → RecordBatches                  |
 |                  | Stream results                               | ✅      | `ExecuteStreamAsync()` → IAsyncEnumerable         |
 |                  | Show/print                                   | ✅      | `ShowAsync()`, `ToStringAsync()`                  |
-|                  | Select, Aggregate, Join, Filter, Limit, Sort | 🟡      | Use SQL instead                                   |
+|                  | Select, Aggregate, Join, Filter, Limit, Sort | 🟡     | Use SQL instead                                   |
 | **Arrow**        | Apache Arrow support                         | ✅      | Via Apache.Arrow nuget package                    |
 |                  | Zero copy support                            | ✅      |                                                   |
 | **Advanced**     | UDF registration                             | ❌      |                                                   |

examples/QueryInMemoryData/Program.cs

@@ -0,0 +1,58 @@
+using Dapper;
+using DataFusionSharp;
+using DataFusionSharp.Data;
+using Microsoft.Extensions.Logging;
+
+// Setup logging
+var loggerFactory = LoggerFactory.Create(builder =>
+{
+    builder
+        .SetMinimumLevel(LogLevel.Information)
+        .AddSimpleConsole(o => o.IncludeScopes = true);
+});
+DataFusionNativeLogger.ConfigureLogger(loggerFactory.CreateLogger("DataFusionSharp"), LogLevel.Information);
+
+// Init runtime and single session
+using var runtime = DataFusionRuntime.Create();
+using var session = runtime.CreateSessionContext();
+
+// Load customers CSV into an in-memory object store
+using var store = runtime.CreateInMemoryStore();
+
+var customersCsvBytes = await File.ReadAllBytesAsync(Path.Combine("Data", "orders", "csv", "customers.csv"));
+await store.PutAsync("customers.csv", customersCsvBytes);
+
+session.RegisterInMemoryObjectStore("memory://", store);
+await session.RegisterCsvAsync("customers", "memory:///customers.csv");
+
+// Register orders from a regular file on disk
+await session.RegisterCsvAsync("orders", Path.Combine("Data", "orders", "csv", "orders.csv"));
+
+// Create a connection wrapper around the SessionContext for ADO.NET operations.
+await using var connection = session.AsConnection();
+
+
+// QueryAsync<T> – map rows to a strongly-typed record
+Console.WriteLine("=== Completed orders per customer (in-memory JOIN file) ===");
+
+var summaries = await connection.QueryAsync<OrderSummary>(
+    """
+    SELECT
+        c.customer_name     AS CustomerName,
+        c.country           AS Country,
+        COUNT(*)            AS OrderCount,
+        SUM(o.order_amount) AS TotalAmount
+    FROM customers AS c
+        JOIN orders AS o ON c.customer_id = o.customer_id
+    WHERE o.order_status = @status
+    GROUP BY c.customer_name, c.country
+    ORDER BY TotalAmount DESC
+    """,
+    new { status = "Completed" });
+
+foreach (var s in summaries)
+    Console.WriteLine($"  {s.CustomerName,-25} {s.Country,-10}  orders: {s.OrderCount,3}  total: {s.TotalAmount,10:N0}");
+
+// Model types
+
+record OrderSummary(string CustomerName, string Country, long OrderCount, long TotalAmount);

examples/QueryInMemoryData/QueryInMemoryData.csproj

@@ -0,0 +1,28 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <OutputType>Exe</OutputType>
+        <TargetFramework>net10.0</TargetFramework>
+        <LangVersion>14</LangVersion>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+        <IsPackable>false</IsPackable>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\..\src\DataFusionSharp.Data\DataFusionSharp.Data.csproj" />
+    </ItemGroup>
+
+    <ItemGroup>
+        <PackageReference Include="Dapper" Version="2.1.72" />
+        <PackageReference Include="Microsoft.Extensions.Logging" Version="10.0.5" />
+        <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="10.0.5" />
+    </ItemGroup>
+
+    <ItemGroup>
+        <None Include="..\..\datasets\**\*.*"
+              LinkBase="Data\"
+              CopyToOutputDirectory="PreserveNewest" />
+    </ItemGroup>
+
+</Project>

native/Cargo.toml

@@ -16,6 +16,7 @@ panic = "unwind"
 [dependencies]
 anyhow = "1.0.102"
 arrow-array = { version = "57.3.0", features = ["ffi"] }
+bytes = "1.11.1"
 datafusion = "52.4.0"
 datafusion-proto = "52.4.0"
 futures = "0.3.32"

native/src/common.rs

@@ -20,16 +20,30 @@ impl BytesData {
     }
 
     pub(crate) fn as_slice(&self) -> &[u8] {
-        unsafe { std::slice::from_raw_parts( self.data, self.len as usize) }
+        unsafe {
+            std::slice::from_raw_parts( self.data, self.len as usize)
+        }
+    }
+
+    pub(crate) fn as_slice_static(&self) -> &'static [u8] {
+        unsafe {
+            std::slice::from_raw_parts( self.data, self.len as usize)
+        }
     }
 
     pub(crate) fn as_opt_slice(&self) -> Option<&[u8]> {
         if self.data.is_null() {
             None
         } else {
-            Some(unsafe { std::slice::from_raw_parts( self.data, self.len as usize) })
+            Some(unsafe {
+                std::slice::from_raw_parts( self.data, self.len as usize)
+            })
         }
     }
+
+    pub(crate) fn len(&self) -> usize {
+        self.len as usize
+    }
 }
 
 #[repr(C)]

native/src/context.rs

@@ -623,6 +623,39 @@ pub unsafe extern "C" fn datafusion_context_register_object_store_http(
     ErrorCode::Ok
 }
 
+/// Registers an in-memory store.
+///
+/// This is a synchronous operation. The callback is invoked with the result.
+///
+/// # Safety
+/// - `context_ptr` must be a valid pointer returned by `datafusion_context_new`
+/// - `base_url_ptr` must be a valid null-terminated UTF-8 string (e.g. "memory://")
+/// - `store_ptr` must be a valid pointer returned by `datafusion_in_memory_store_new`
+/// - `callback` must be valid to call from any thread
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn datafusion_context_register_object_store_in_memory(
+    context_ptr: *mut SessionContextWrapper,
+    base_url_ptr: *const std::ffi::c_char,
+    store_ptr: *const crate::memory_store::InMemoryStoreWrapper,
+    callback: crate::Callback,
+    user_data: u64,
+) -> ErrorCode {
+    let context = ffi_ref!(context_ptr);
+
+    let base_url = ffi_cstr_to_string!(base_url_ptr);
+    let Ok(url) = url::Url::parse(&base_url) else { return ErrorCode::InvalidArgument };
+
+    let store = ffi_ref!(store_ptr);
+
+    debug!("Registering in-memory object store {store_ptr:p} for '{url}' on session {context_ptr:p}");
+
+    context.inner.register_object_store(&url, store.inner());
+
+    crate::invoke_callback_null_result(callback, user_data);
+
+    ErrorCode::Ok
+}
+
 /// Deregisters an object store for the given URL.
 ///
 /// This is a synchronous operation. The callback is invoked with the result.

native/src/error.rs

@@ -8,7 +8,8 @@ pub enum ErrorCode {
     RuntimeShutdownFailed = 4,
     TableRegistrationFailed = 5,
     SqlError = 6,
-    DataFrameError = 7
+    DataFrameError = 7,
+    ObjectStoreError = 8
 }
 
 #[derive(Debug, Clone)]

native/src/lib.rs

@@ -12,6 +12,7 @@ pub mod logger;
 pub mod runtime;
 pub mod context;
 pub mod dataframe;
+pub mod memory_store;
 
 pub use proto::*;
 pub use error::*;