Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
890 changes: 60 additions & 830 deletions Cargo.lock

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ uuid = { version = "1.11.0", features = ["v4", "v5", "v7", "serde"] }
tower-http = { version = "0.6", features = ["trace", "cors"] }
http = "1.1"
base64 = "0.22.1"
duckdb = { git = "https://github.com/altertable-ai/duckdb-rs", rev = "3dc963e9d80ddf6403a1114d20a0f9a3de7833d4", features = ["serde_json", "uuid", "chrono", "appender-arrow"] }
duckdb = { git = "https://github.com/altertable-ai/duckdb-rs", rev = "88f673c5cdf09e287c010a910db53b97e523a9fa", features = ["serde_json", "uuid", "chrono", "appender-arrow"] }
tonic-async-interceptor = "0.14.1"
anyhow = "1.0"
serde = { version = "1", features = ["derive"] }
Expand Down
22 changes: 22 additions & 0 deletions Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ RUN DUCKDB_DOWNLOAD_LIB=1 cargo build --release --locked

FROM debian:bookworm-slim

ARG TARGETARCH

WORKDIR /app

RUN apt-get update && apt-get -y install --no-install-recommends \
Expand All @@ -23,13 +25,33 @@ RUN apt-get update && apt-get -y install --no-install-recommends \
libcurl4-openssl-dev \
libstdc++6 \
procps \
curl \
gzip \
&& apt-get autoclean && rm -rf /var/lib/apt/lists/*

COPY --from=builder /app/target/release/deps/libduckdb.so /usr/local/lib/libduckdb.so
RUN ldconfig

COPY --from=builder /app/target/release/altertable-mock /usr/local/bin/altertable-mock

# Install DuckLake for the linked DuckDB version into the default extension directory.
RUN set -eu; \
case "${TARGETARCH:-}" in \
amd64) platform=linux_amd64 ;; \
arm64) platform=linux_arm64 ;; \
"") \
case "$(uname -m)" in \
x86_64) platform=linux_amd64 ;; \
aarch64|arm64) platform=linux_arm64 ;; \
*) echo "unsupported architecture: $(uname -m)" >&2; exit 1 ;; \
esac ;; \
*) echo "unsupported architecture: ${TARGETARCH}" >&2; exit 1 ;; \
esac; \
dir="/root/.duckdb/extensions/v1.5.5/${platform}"; \
mkdir -p "$dir"; \
curl -fsSL "http://extensions.duckdb.org/v1.5.5/${platform}/ducklake.duckdb_extension.gz" \
| gzip -dc > "$dir/ducklake.duckdb_extension"

EXPOSE 15000
EXPOSE 15002
ENV RUST_LOG=info
Expand Down
8 changes: 4 additions & 4 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
[![GHCR](https://img.shields.io/badge/ghcr.io-altertable--mock-blue?logo=github)](https://github.com/altertable-ai/altertable-mock/pkgs/container/altertable-mock)
[![CI](https://github.com/altertable-ai/altertable-mock/actions/workflows/ci.yml/badge.svg)](https://github.com/altertable-ai/altertable-mock/actions/workflows/ci.yml)

A mock server that implements the full [Altertable](https://altertable.ai) API surface — [Arrow Flight SQL](https://arrow.apache.org/docs/format/FlightSql.html) (gRPC), Lakehouse REST, and Product Analytics REST — backed by an in-memory [DuckDB](https://duckdb.org) instance. Intended for local development and testing.
A mock server that implements the full [Altertable](https://altertable.ai) API surface — [Arrow Flight SQL](https://arrow.apache.org/docs/format/FlightSql.html) (gRPC), Lakehouse REST, and Product Analytics REST — backed by [DuckLake](https://ducklake.select) on DuckDB 1.5.5. Intended for local development and testing.

## What it is

Expand All @@ -16,7 +16,7 @@ Altertable exposes three interfaces:

This server speaks all three protocols and behaves similarly to the real service, without requiring a real Altertable account or any external infrastructure.

Each authenticated user gets their own isolated in-memory DuckDB database. Sessions within a user share that database. The server supports:
Each authenticated user gets their own DuckLake. Its catalog is an in-memory DuckDB database and its data files live in a temporary directory for the life of that lakehouse. The default catalog name remains `memory`. Sessions within a user share that lakehouse. The server supports:

- Ad-hoc SQL queries and DML statements
- Prepared statements with parameter binding
Expand All @@ -28,7 +28,7 @@ Each authenticated user gets their own isolated in-memory DuckDB database. Sessi

- Run integration tests against a real Flight SQL endpoint without hitting production
- Develop locally without network access or credentials
- Deterministic, ephemeral state — the in-memory database resets when the server stops
- Deterministic, ephemeral state — DuckLake metadata is discarded when the server stops
- Fast and self-contained — no external services required

## Ports
Expand Down Expand Up @@ -213,7 +213,7 @@ All endpoints return `{"ok": true}` on success or `{"ok": false, "error_code": "

## Building from source

DuckDB is compiled in via the `bundled` feature, so no external shared library is required.
The build links DuckDB 1.5.5: set `DUCKDB_DOWNLOAD_LIB=1` to download it, or point `DUCKDB_LIB_DIR` at a local copy. Lakehouse connections run `LOAD ducklake`, so the extension must already be installed for DuckDB 1.5.5 (`INSTALL ducklake;`). The Docker image installs it at build time.

```bash
cargo build --release
Expand Down
150 changes: 150 additions & 0 deletions src/ducklake.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,150 @@
use std::path::{Path, PathBuf};
use std::sync::Mutex;

use anyhow::Context;
use duckdb::Connection;

use crate::utils::{MEMORY_DB, escape_identifier, escape_literal};

/// Keeps the temporary DuckLake directory alive for as long as it is held.
///
/// Dropping the last owner removes the session database file and data files.
#[must_use = "dropping this deletes the DuckLake data directory"]
pub struct DataDir {
path: PathBuf,
}

impl Drop for DataDir {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.path);
}
}

/// A lakehouse connection plus the temp directory its files live in.
pub struct SharedLakehouse {
pub connection: std::sync::Arc<Mutex<Connection>>,
_data_dir: DataDir,
}

impl SharedLakehouse {
pub fn open() -> anyhow::Result<Self> {
let (connection, data_dir) = open()?;
Ok(Self {
connection: std::sync::Arc::new(Mutex::new(connection)),
_data_dir: data_dir,
})
}
}

/// Opens a DuckDB session and attaches a DuckLake named `memory`.
///
/// The catalog metadata is an in-memory DuckDB database (`ducklake::memory:`).
/// Table data is written under a temporary directory. The session database itself
/// is a normal file in that directory: a path such as `:memory:session` is not
/// an in-memory database and DuckDB creates a file with that name.
pub fn open() -> anyhow::Result<(Connection, DataDir)> {
let root = std::env::temp_dir().join(format!("altertable-mock-{}", uuid::Uuid::now_v7()));
let data_path = root.join("data");
std::fs::create_dir_all(&data_path).with_context(|| {
format!(
"failed to create DuckLake data directory {}",
data_path.display()
)
})?;

let shell_path = root.join("session.duckdb");
let connection = Connection::open(&shell_path).with_context(|| {
format!(
"failed to open DuckDB session database {}",
shell_path.display()
)
})?;

if let Err(err) = attach(&connection, &data_path) {
drop(connection);
let _ = std::fs::remove_dir_all(&root);
return Err(err);
}

Ok((connection, DataDir { path: root }))
}

fn attach(connection: &Connection, data_path: &Path) -> anyhow::Result<()> {
load_ducklake(connection)?;

let catalog = escape_identifier(MEMORY_DB);
let data_path = escape_literal(&data_path.to_string_lossy());
connection
.execute_batch(&format!(
"ATTACH 'ducklake::memory:' AS \"{catalog}\" (DATA_PATH '{data_path}', DATA_INLINING_ROW_LIMIT 0);
USE \"{catalog}\";"
))
.context("failed to attach in-memory DuckLake")?;
Ok(())
}

fn load_ducklake(connection: &Connection) -> anyhow::Result<()> {
connection
.execute_batch("LOAD ducklake;")
.context("failed to load the DuckLake extension (run `INSTALL ducklake;` first)")?;
Ok(())
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn keeps_files_in_the_temp_directory() {
let cwd = std::env::current_dir().unwrap();
let (connection, data_dir) = open().unwrap();
connection
.execute_batch("CREATE TABLE t (id INTEGER); INSERT INTO t VALUES (1);")
.unwrap();

assert!(data_dir.path.starts_with(std::env::temp_dir()));
assert!(data_dir.path.join("session.duckdb").is_file());
assert!(data_dir.path.join("data").is_dir());
for name in [":memory:session", ":memory:engine", ":memory:"] {
assert!(
!cwd.join(name).exists(),
"{name} was created in {}",
cwd.display()
);
}
}

#[test]
fn cloned_connection_sees_ducklake_rows() {
let (connection, _data_dir) = open().unwrap();
connection
.execute_batch("CREATE TABLE shared (id INTEGER); INSERT INTO shared VALUES (7);")
.unwrap();

let clone = connection.try_clone().unwrap();
clone.execute_batch("USE memory;").unwrap();
let id: i32 = clone
.query_row("SELECT id FROM shared", [], |row| row.get(0))
.unwrap();
assert_eq!(id, 7);
}

#[test]
fn rejects_primary_key_and_unique() {
let (connection, _data_dir) = open().unwrap();
let primary_key = connection
.execute_batch("CREATE TABLE pk_t (id INTEGER PRIMARY KEY);")
.unwrap_err()
.to_string();
assert!(
primary_key.contains("not supported in DuckLake"),
"{primary_key}"
);

let unique = connection
.execute_batch("CREATE TABLE uniq_t (name VARCHAR UNIQUE);")
.unwrap_err()
.to_string();
assert!(unique.contains("not supported in DuckLake"), "{unique}");
}
}
31 changes: 18 additions & 13 deletions src/flight/layers/session.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,24 +22,25 @@ pub fn layer() -> AsyncInterceptorLayer<

pub fn interceptor()
-> impl AsyncInterceptor<Future = BoxFuture<'static, Result<Request<()>, Status>>> + Clone {
let lakehouse_store = Arc::new(RwLock::new(HashMap::new()));
let session_store = Arc::new(RwLock::new(HashMap::new()));
let lakehouse_store = Arc::new(RwLock::new(HashMap::new()));
SessionMiddleware {
lakehouse_store,
session_store,
lakehouse_store,
}
}

type LakehouseKey = Identity;
type LakehouseStore = Arc<RwLock<HashMap<LakehouseKey, Arc<Mutex<Connection>>>>>;
type LakehouseStore = Arc<RwLock<HashMap<LakehouseKey, crate::ducklake::SharedLakehouse>>>;

type SessionKey = (Identity, SessionID);
type SessionStore = Arc<RwLock<HashMap<SessionKey, Session>>>;

#[derive(Clone)]
struct SessionMiddleware {
lakehouse_store: LakehouseStore,
// Drop sessions before lakehouses so DuckLake files outlive cloned connections.
session_store: SessionStore,
lakehouse_store: LakehouseStore,
}

impl AsyncInterceptor for SessionMiddleware {
Expand Down Expand Up @@ -78,7 +79,7 @@ impl AsyncInterceptor for SessionMiddleware {
);

let lakehouse =
get_or_create_lakehouse(key.0.clone(), lakehouse_store.clone()).await;
get_or_create_lakehouse(key.0.clone(), lakehouse_store.clone()).await?;
#[allow(clippy::result_large_err)]
let new_conn = tokio::task::spawn_blocking(move || {
lakehouse
Expand Down Expand Up @@ -106,13 +107,17 @@ impl AsyncInterceptor for SessionMiddleware {
async fn get_or_create_lakehouse(
key: LakehouseKey,
store: LakehouseStore,
) -> Arc<Mutex<Connection>> {
if let Some(connection) = store.read().await.get(&key).cloned() {
connection
} else {
info!("Creating new lakehouse for user: {}", key.username);
let connection = Arc::new(Mutex::new(Connection::open_in_memory().unwrap()));
store.write().await.insert(key, connection.clone());
connection
) -> Result<Arc<Mutex<Connection>>, Status> {
if let Some(lakehouse) = store.read().await.get(&key) {
return Ok(lakehouse.connection.clone());
}

info!("Creating new lakehouse for user: {}", key.username);
let lakehouse = tokio::task::spawn_blocking(crate::ducklake::SharedLakehouse::open)
.await
.map_err(|e| Status::internal(format!("Failed to open DuckLake: {e}")))?
.map_err(|e| Status::internal(format!("Failed to open DuckLake: {e}")))?;
let connection = lakehouse.connection.clone();
store.write().await.insert(key, lakehouse);
Ok(connection)
}
46 changes: 44 additions & 2 deletions src/flight/schema_extractor.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
use std::{ffi::CString, ptr, sync::Arc};

use std::sync::Mutex;

use arrow_schema::{DataType, Field, Schema, TimeUnit};
use duckdb::ffi::{
DUCKDB_TYPE, DUCKDB_TYPE_DUCKDB_TYPE_BIGINT, DUCKDB_TYPE_DUCKDB_TYPE_BLOB,
Expand Down Expand Up @@ -34,13 +36,30 @@ impl Drop for PreparedStatementGuard {
}
}

// duckdb-rs 1.10505 removed `Connection::raw_connection`. `InterruptHandle` still
// stores that same `duckdb_connection`, and its only field is the mutex around it.
fn raw_connection(conn: &duckdb::Connection) -> duckdb::ffi::duckdb_connection {
struct InterruptHandleLayout {
conn: Mutex<duckdb::ffi::duckdb_connection>,
}

let handle = conn.interrupt_handle();
unsafe {
let layout = &*std::sync::Arc::as_ptr(&handle).cast::<InterruptHandleLayout>();
*layout
.conn
.lock()
.expect("DuckDB interrupt handle lock poisoned")
}
}

// Since `duckdb::Statement` does not provide a way to get the schema without executing the statement,
// we need to use the low-level FFI API
pub fn extract_schema(conn: &duckdb::Connection, sql: &str) -> anyhow::Result<Schema> {
let c_sql = CString::new(sql).map_err(|e| anyhow::anyhow!("Invalid SQL: {e}"))?;

unsafe {
let raw_conn = conn.raw_connection();
let raw_conn = raw_connection(conn);

let mut prep: duckdb_prepared_statement = ptr::null_mut();
let _guard = PreparedStatementGuard { stmt: &mut prep };
Expand Down Expand Up @@ -87,7 +106,7 @@ pub fn extract_parameter_schema(
let c_sql = CString::new(sql).map_err(|e| anyhow::anyhow!("Invalid SQL: {e}"))?;

unsafe {
let raw_conn = conn.raw_connection();
let raw_conn = raw_connection(conn);

let mut prep: duckdb_prepared_statement = ptr::null_mut();
let _guard = PreparedStatementGuard { stmt: &mut prep };
Expand Down Expand Up @@ -237,3 +256,26 @@ unsafe fn logical_type_to_arrow_type(ltyp: duckdb_logical_type) -> DataType {
}
}
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn extracts_result_columns_without_executing() {
let (conn, _data_dir) = crate::ducklake::open().unwrap();
conn.execute_batch("CREATE TABLE people (id INTEGER, name VARCHAR)")
.unwrap();

let schema = extract_schema(&conn, "SELECT id, name FROM people WHERE id = $id").unwrap();
assert_eq!(schema.fields().len(), 2);
assert_eq!(schema.field(0).name(), "id");
assert_eq!(schema.field(1).name(), "name");

let parameters =
extract_parameter_schema(&conn, "SELECT id, name FROM people WHERE id = $id")
.unwrap()
.unwrap();
assert_eq!(parameters.fields().len(), 1);
}
}
2 changes: 1 addition & 1 deletion src/lakehouse/explain.rs
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ mod tests {

#[test]
fn test_explain_statement_against_duckdb() {
let conn = duckdb::Connection::open_in_memory().unwrap();
let (conn, _data_dir) = crate::ducklake::open().unwrap();
conn.execute("CREATE TABLE events (id INTEGER, category VARCHAR)", [])
.unwrap();
conn.execute(
Expand Down
Loading
Loading