Skip to content

Commit fd0b046

Browse files
feat: add in-memory store (#72)
1 parent f71a223 commit fd0b046

20 files changed

Lines changed: 773 additions & 30 deletions

.github/workflows/ci.yml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -154,7 +154,7 @@ jobs:
154154
/o:"npiontko" \
155155
/d:sonar.token="$SONAR_TOKEN" \
156156
/d:sonar.cs.vscoveragexml.reportsPaths=coverage.xml \
157-
/d:sonar.coverage.exclusions="docs/**,examples/**,native/**""
157+
/d:sonar.coverage.exclusions="docs/**,examples/**,native/**"
158158
159159
# Build the project without building the native library, it should be already built in the check job and cached
160160
dotnet build -c Release -p:BuildNativeLib=false

DataFusionSharp.slnx

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,7 @@
7070
</Folder>
7171
<Folder Name="/examples/">
7272
<Project Path="examples\QueryData\QueryData.csproj" />
73+
<Project Path="examples\QueryInMemoryData\QueryInMemoryData.csproj" />
7374
<Project Path="examples/QueryS3DataWithDapper/QueryS3DataWithDapper.csproj" />
7475
<Project Path="examples\QueryDataWithDapper\QueryDataWithDapper.csproj" />
7576
<Project Path="examples\QueryS3Data\QueryS3Data.csproj" />

README.md

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -11,8 +11,8 @@
1111
1212
## Features
1313

14-
| Component | Feature | Status | Notes |
15-
|------------------|----------------------------------------------|---------|---------------------------------------------------|
14+
| Component | Feature | Status | Notes |
15+
|------------------|----------------------------------------------|--------|---------------------------------------------------|
1616
| **Runtime** | Tokio runtime || Configurable threads, supports multiple instances |
1717
| | Logger || Configurable with log levels |
1818
| **Session** | Create session context || |
@@ -25,13 +25,13 @@
2525
| | Amazon S3 || |
2626
| | Azure Blob Storage || |
2727
| | Google Cloud Storage || |
28-
| | InMemory | | |
28+
| | InMemory | | |
2929
| **DataFrame** | Count rows || `CountAsync()` |
3030
| | Get schema || `GetSchema()` → Arrow Schema |
3131
| | Collect all data || `CollectAsync()` → RecordBatches |
3232
| | Stream results || `ExecuteStreamAsync()` → IAsyncEnumerable |
3333
| | Show/print || `ShowAsync()`, `ToStringAsync()` |
34-
| | Select, Aggregate, Join, Filter, Limit, Sort | 🟡 | Use SQL instead |
34+
| | Select, Aggregate, Join, Filter, Limit, Sort | 🟡 | Use SQL instead |
3535
| **Arrow** | Apache Arrow support || Via Apache.Arrow nuget package |
3636
| | Zero copy support || |
3737
| **Advanced** | UDF registration || |
Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
using Dapper;
2+
using DataFusionSharp;
3+
using DataFusionSharp.Data;
4+
using Microsoft.Extensions.Logging;
5+
6+
// Setup logging
7+
var loggerFactory = LoggerFactory.Create(builder =>
8+
{
9+
builder
10+
.SetMinimumLevel(LogLevel.Information)
11+
.AddSimpleConsole(o => o.IncludeScopes = true);
12+
});
13+
DataFusionNativeLogger.ConfigureLogger(loggerFactory.CreateLogger("DataFusionSharp"), LogLevel.Information);
14+
15+
// Init runtime and single session
16+
using var runtime = DataFusionRuntime.Create();
17+
using var session = runtime.CreateSessionContext();
18+
19+
// Load customers CSV into an in-memory object store
20+
using var store = runtime.CreateInMemoryStore();
21+
22+
var customersCsvBytes = await File.ReadAllBytesAsync(Path.Combine("Data", "orders", "csv", "customers.csv"));
23+
await store.PutAsync("customers.csv", customersCsvBytes);
24+
25+
session.RegisterInMemoryObjectStore("memory://", store);
26+
await session.RegisterCsvAsync("customers", "memory:///customers.csv");
27+
28+
// Register orders from a regular file on disk
29+
await session.RegisterCsvAsync("orders", Path.Combine("Data", "orders", "csv", "orders.csv"));
30+
31+
// Create a connection wrapper around the SessionContext for ADO.NET operations.
32+
await using var connection = session.AsConnection();
33+
34+
35+
// QueryAsync<T> – map rows to a strongly-typed record
36+
Console.WriteLine("=== Completed orders per customer (in-memory JOIN file) ===");
37+
38+
var summaries = await connection.QueryAsync<OrderSummary>(
39+
"""
40+
SELECT
41+
c.customer_name AS CustomerName,
42+
c.country AS Country,
43+
COUNT(*) AS OrderCount,
44+
SUM(o.order_amount) AS TotalAmount
45+
FROM customers AS c
46+
JOIN orders AS o ON c.customer_id = o.customer_id
47+
WHERE o.order_status = @status
48+
GROUP BY c.customer_name, c.country
49+
ORDER BY TotalAmount DESC
50+
""",
51+
new { status = "Completed" });
52+
53+
foreach (var s in summaries)
54+
Console.WriteLine($" {s.CustomerName,-25} {s.Country,-10} orders: {s.OrderCount,3} total: {s.TotalAmount,10:N0}");
55+
56+
// Model types
57+
58+
record OrderSummary(string CustomerName, string Country, long OrderCount, long TotalAmount);
Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
<Project Sdk="Microsoft.NET.Sdk">
2+
3+
<PropertyGroup>
4+
<OutputType>Exe</OutputType>
5+
<TargetFramework>net10.0</TargetFramework>
6+
<LangVersion>14</LangVersion>
7+
<ImplicitUsings>enable</ImplicitUsings>
8+
<Nullable>enable</Nullable>
9+
<IsPackable>false</IsPackable>
10+
</PropertyGroup>
11+
12+
<ItemGroup>
13+
<ProjectReference Include="..\..\src\DataFusionSharp.Data\DataFusionSharp.Data.csproj" />
14+
</ItemGroup>
15+
16+
<ItemGroup>
17+
<PackageReference Include="Dapper" Version="2.1.72" />
18+
<PackageReference Include="Microsoft.Extensions.Logging" Version="10.0.5" />
19+
<PackageReference Include="Microsoft.Extensions.Logging.Console" Version="10.0.5" />
20+
</ItemGroup>
21+
22+
<ItemGroup>
23+
<None Include="..\..\datasets\**\*.*"
24+
LinkBase="Data\"
25+
CopyToOutputDirectory="PreserveNewest" />
26+
</ItemGroup>
27+
28+
</Project>

native/Cargo.toml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ panic = "unwind"
1616
[dependencies]
1717
anyhow = "1.0.102"
1818
arrow-array = { version = "57.3.0", features = ["ffi"] }
19+
bytes = "1.11.1"
1920
datafusion = "52.4.0"
2021
datafusion-proto = "52.4.0"
2122
futures = "0.3.32"

native/src/common.rs

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,16 +20,30 @@ impl BytesData {
2020
}
2121

2222
pub(crate) fn as_slice(&self) -> &[u8] {
23-
unsafe { std::slice::from_raw_parts( self.data, self.len as usize) }
23+
unsafe {
24+
std::slice::from_raw_parts( self.data, self.len as usize)
25+
}
26+
}
27+
28+
pub(crate) fn as_slice_static(&self) -> &'static [u8] {
29+
unsafe {
30+
std::slice::from_raw_parts( self.data, self.len as usize)
31+
}
2432
}
2533

2634
pub(crate) fn as_opt_slice(&self) -> Option<&[u8]> {
2735
if self.data.is_null() {
2836
None
2937
} else {
30-
Some(unsafe { std::slice::from_raw_parts( self.data, self.len as usize) })
38+
Some(unsafe {
39+
std::slice::from_raw_parts( self.data, self.len as usize)
40+
})
3141
}
3242
}
43+
44+
pub(crate) fn len(&self) -> usize {
45+
self.len as usize
46+
}
3347
}
3448

3549
#[repr(C)]

native/src/context.rs

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -623,6 +623,39 @@ pub unsafe extern "C" fn datafusion_context_register_object_store_http(
623623
ErrorCode::Ok
624624
}
625625

626+
/// Registers an in-memory store.
627+
///
628+
/// This is a synchronous operation. The callback is invoked with the result.
629+
///
630+
/// # Safety
631+
/// - `context_ptr` must be a valid pointer returned by `datafusion_context_new`
632+
/// - `base_url_ptr` must be a valid null-terminated UTF-8 string (e.g. "memory://")
633+
/// - `store_ptr` must be a valid pointer returned by `datafusion_in_memory_store_new`
634+
/// - `callback` must be valid to call from any thread
635+
#[unsafe(no_mangle)]
636+
pub unsafe extern "C" fn datafusion_context_register_object_store_in_memory(
637+
context_ptr: *mut SessionContextWrapper,
638+
base_url_ptr: *const std::ffi::c_char,
639+
store_ptr: *const crate::memory_store::InMemoryStoreWrapper,
640+
callback: crate::Callback,
641+
user_data: u64,
642+
) -> ErrorCode {
643+
let context = ffi_ref!(context_ptr);
644+
645+
let base_url = ffi_cstr_to_string!(base_url_ptr);
646+
let Ok(url) = url::Url::parse(&base_url) else { return ErrorCode::InvalidArgument };
647+
648+
let store = ffi_ref!(store_ptr);
649+
650+
debug!("Registering in-memory object store {store_ptr:p} for '{url}' on session {context_ptr:p}");
651+
652+
context.inner.register_object_store(&url, store.inner());
653+
654+
crate::invoke_callback_null_result(callback, user_data);
655+
656+
ErrorCode::Ok
657+
}
658+
626659
/// Deregisters an object store for the given URL.
627660
///
628661
/// This is a synchronous operation. The callback is invoked with the result.

native/src/error.rs

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,8 @@ pub enum ErrorCode {
88
RuntimeShutdownFailed = 4,
99
TableRegistrationFailed = 5,
1010
SqlError = 6,
11-
DataFrameError = 7
11+
DataFrameError = 7,
12+
ObjectStoreError = 8
1213
}
1314

1415
#[derive(Debug, Clone)]

native/src/lib.rs

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ pub mod logger;
1212
pub mod runtime;
1313
pub mod context;
1414
pub mod dataframe;
15+
pub mod memory_store;
1516

1617
pub use proto::*;
1718
pub use error::*;

0 commit comments

Comments
 (0)