Skip to content
Open
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
15 changes: 11 additions & 4 deletions src_cpp/include/py_database.h
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
#pragma once

#include <memory>
#include <optional>

#include "main/lbug.h"
#include "main/storage_driver.h"
Expand All @@ -19,10 +20,16 @@ class PyDatabase {

static uint64_t getStorageVersion();

explicit PyDatabase(const std::string& databasePath, uint64_t bufferPoolSize,
uint64_t maxNumThreads, bool compression, bool readOnly, uint64_t maxDBSize,
bool autoCheckpoint, int64_t checkpointThreshold, bool throwOnWalReplayFailure = true,
bool enableChecksums = true, bool enableMultiWrites = false);
explicit PyDatabase(const std::string& databasePath,
std::optional<uint64_t> bufferPoolSize = std::nullopt,
std::optional<uint64_t> maxNumThreads = std::nullopt,
std::optional<bool> compression = std::nullopt, std::optional<bool> readOnly = std::nullopt,
std::optional<uint64_t> maxDBSize = std::nullopt,
std::optional<bool> autoCheckpoint = std::nullopt,
std::optional<int64_t> checkpointThreshold = std::nullopt,
std::optional<bool> throwOnWalReplayFailure = std::nullopt,
std::optional<bool> enableChecksums = std::nullopt,
std::optional<bool> enableMultiWrites = std::nullopt);

~PyDatabase();

Expand Down
70 changes: 51 additions & 19 deletions src_cpp/py_database.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

#include <memory>
#include <mutex>
#include <optional>

#include "common/exception/runtime.h"
#include "extension/extension.h"
Expand All @@ -13,14 +14,17 @@ using namespace lbug::common;

void PyDatabase::initialize(py::handle& m) {
py::class_<PyDatabase>(m, "Database")
.def(py::init<const std::string&, uint64_t, uint64_t, bool, bool, uint64_t, bool, int64_t,
bool, bool, bool>(),
py::arg("database_path"), py::arg("buffer_pool_size") = 0,
py::arg("max_num_threads") = 0, py::arg("compression") = true,
py::arg("read_only") = false, py::arg("max_db_size") = -1u,
py::arg("auto_checkpoint") = true, py::arg("checkpoint_threshold") = -1,
py::arg("throw_on_wal_replay_failure") = true, py::arg("enable_checksums") = true,
py::arg("enable_multi_writes") = false)
.def(py::init<const std::string&, std::optional<uint64_t>, std::optional<uint64_t>,
std::optional<bool>, std::optional<bool>, std::optional<uint64_t>,
std::optional<bool>, std::optional<int64_t>, std::optional<bool>,
std::optional<bool>, std::optional<bool>>(),
py::arg("database_path"), py::arg("buffer_pool_size") = py::none(),
py::arg("max_num_threads") = py::none(), py::arg("compression") = py::none(),
py::arg("read_only") = py::none(), py::arg("max_db_size") = py::none(),
py::arg("auto_checkpoint") = py::none(), py::arg("checkpoint_threshold") = py::none(),
py::arg("throw_on_wal_replay_failure") = py::none(),
py::arg("enable_checksums") = py::none(),
py::arg("enable_multi_writes") = py::none())
.def("scan_node_table_as_int64", &PyDatabase::scanNodeTable<std::int64_t>,
py::arg("table_name"), py::arg("prop_name"), py::arg("indices"), py::arg("np_array"),
py::arg("num_threads"))
Expand Down Expand Up @@ -49,18 +53,46 @@ uint64_t PyDatabase::getStorageVersion() {
return Version::getStorageVersion();
}

PyDatabase::PyDatabase(const std::string& databasePath, uint64_t bufferPoolSize,
uint64_t maxNumThreads, bool compression, bool readOnly, uint64_t maxDBSize,
bool autoCheckpoint, int64_t checkpointThreshold, bool throwOnWalReplayFailure,
bool enableChecksums, bool enableMultiWrites) {
auto systemConfig = SystemConfig(bufferPoolSize, maxNumThreads, compression, readOnly,
maxDBSize, autoCheckpoint);
if (checkpointThreshold >= 0) {
systemConfig.checkpointThreshold = static_cast<uint64_t>(checkpointThreshold);
PyDatabase::PyDatabase(const std::string& databasePath, std::optional<uint64_t> bufferPoolSize,
std::optional<uint64_t> maxNumThreads, std::optional<bool> compression,
std::optional<bool> readOnly, std::optional<uint64_t> maxDBSize,
std::optional<bool> autoCheckpoint, std::optional<int64_t> checkpointThreshold,
std::optional<bool> throwOnWalReplayFailure, std::optional<bool> enableChecksums,
std::optional<bool> enableMultiWrites) {
// Start from the engine defaults and only apply explicitly set settings,
// so engine-side default changes propagate automatically.
auto systemConfig = SystemConfig();
if (bufferPoolSize.has_value()) {
systemConfig.bufferPoolSize = bufferPoolSize.value();
}
if (maxNumThreads.has_value()) {
systemConfig.maxNumThreads = maxNumThreads.value();
}
if (compression.has_value()) {
systemConfig.enableCompression = compression.value();
}
if (readOnly.has_value()) {
systemConfig.readOnly = readOnly.value();
}
if (maxDBSize.has_value()) {
systemConfig.maxDBSize = maxDBSize.value();
}
if (autoCheckpoint.has_value()) {
systemConfig.autoCheckpoint = autoCheckpoint.value();
}
if (checkpointThreshold.has_value() && checkpointThreshold.value() >= 0) {
systemConfig.checkpointThreshold =
static_cast<uint64_t>(checkpointThreshold.value());
}
if (throwOnWalReplayFailure.has_value()) {
systemConfig.throwOnWalReplayFailure = throwOnWalReplayFailure.value();
}
if (enableChecksums.has_value()) {
systemConfig.enableChecksums = enableChecksums.value();
}
if (enableMultiWrites.has_value()) {
systemConfig.enableMultiWrites = enableMultiWrites.value();
}
systemConfig.throwOnWalReplayFailure = throwOnWalReplayFailure;
systemConfig.enableChecksums = enableChecksums;
systemConfig.enableMultiWrites = enableMultiWrites;
state = std::make_shared<PyDatabaseState>();
state->database = std::make_unique<Database>(databasePath, systemConfig);
lbug::extension::ExtensionUtils::addTableFunc<lbug::PandasScanFunction>(*state->database);
Expand Down
46 changes: 28 additions & 18 deletions src_py/_lbug_capi.py
Original file line number Diff line number Diff line change
Expand Up @@ -1197,33 +1197,43 @@ class Database:
def __init__(
self,
database_path: str,
buffer_pool_size: int = 0,
max_num_threads: int = 0,
compression: bool = True,
read_only: bool = False,
buffer_pool_size: int | None = None,
max_num_threads: int | None = None,
compression: bool | None = None,
read_only: bool | None = None,
max_db_size: int | None = None,
auto_checkpoint: bool = True,
checkpoint_threshold: int = -1,
throw_on_wal_replay_failure: bool = True,
enable_checksums: bool = True,
enable_multi_writes: bool = False,
auto_checkpoint: bool | None = None,
checkpoint_threshold: int | None = None,
throw_on_wal_replay_failure: bool | None = None,
enable_checksums: bool | None = None,
enable_multi_writes: bool | None = None,
):
self._database = _LbugDatabase()
# Start from the engine defaults and only apply explicitly set
# settings, so engine-side default changes propagate automatically.
config = _LIB.lbug_default_system_config()
config.buffer_pool_size = buffer_pool_size
config.max_num_threads = max_num_threads
config.enable_compression = compression
config.read_only = read_only
if buffer_pool_size is not None:
config.buffer_pool_size = buffer_pool_size
if max_num_threads is not None:
config.max_num_threads = max_num_threads
if compression is not None:
config.enable_compression = compression
if read_only is not None:
config.read_only = read_only

if max_db_size is not None:
config.max_db_size = max_db_size

config.auto_checkpoint = auto_checkpoint
if checkpoint_threshold >= 0:
if auto_checkpoint is not None:
config.auto_checkpoint = auto_checkpoint
if checkpoint_threshold is not None and checkpoint_threshold >= 0:
config.checkpoint_threshold = checkpoint_threshold
config.throw_on_wal_replay_failure = throw_on_wal_replay_failure
config.enable_checksums = enable_checksums
config.enable_multi_writes = enable_multi_writes
if throw_on_wal_replay_failure is not None:
config.throw_on_wal_replay_failure = throw_on_wal_replay_failure
if enable_checksums is not None:
config.enable_checksums = enable_checksums
if enable_multi_writes is not None:
config.enable_multi_writes = enable_multi_writes

state = _LIB.lbug_database_init(
database_path.encode("utf-8"), config, ctypes.byref(self._database)
Expand Down
115 changes: 63 additions & 52 deletions src_py/database.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,17 +34,17 @@ def __init__(
self,
database_path: str | Path | None = None,
*,
buffer_pool_size: int = 0,
max_num_threads: int = 0,
compression: bool = True,
buffer_pool_size: int | None = None,
max_num_threads: int | None = None,
compression: bool | None = None,
lazy_init: bool = False,
read_only: bool = False,
read_only: bool | None = None,
max_db_size: int | None = None,
auto_checkpoint: bool = True,
checkpoint_threshold: int = -1,
throw_on_wal_replay_failure: bool = True,
enable_checksums: bool = True,
enable_multi_writes: bool = False,
auto_checkpoint: bool | None = None,
checkpoint_threshold: int | None = None,
throw_on_wal_replay_failure: bool | None = None,
enable_checksums: bool | None = None,
enable_multi_writes: bool | None = None,
backend: str = "auto",
):
"""
Expand All @@ -54,54 +54,60 @@ def __init__(
The path to database files. If the path is not specified, or empty, or equal to `:memory:`, the database
will be created in memory.

buffer_pool_size : int
The maximum size of buffer pool in bytes. Defaults to ~80% of system memory.
buffer_pool_size : int, optional
The maximum size of buffer pool in bytes. If None (default), the
engine default (~80% of system memory) is used.

max_num_threads : int
The maximum number of threads to use for executing queries.
max_num_threads : int, optional
The maximum number of threads to use for executing queries. If None
(default), the engine default is used.

compression : bool
Enable database compression.
compression : bool, optional
Enable database compression. If None (default), the engine default
is used.

lazy_init : bool
If True, the database will not be initialized until the first query.
This is useful when the database is not used in the main thread or
when the main process is forked.
Default to False.

read_only : bool
read_only : bool, optional
If true, the database is opened read-only. No write transactions is
allowed on the `Database` object. Multiple read-only `Database`
objects can be created with the same database path. However, there
cannot be multiple `Database` objects created with the same
database path.
Default to False.
If None (default), the engine default is used.

max_db_size : int, optional
The maximum size of the database in bytes. Note that this is introduced
temporarily for now to get around with the default 8TB mmap address
space limit some environment. This will be removed once we implement
a better solution later. If not specified, the backend's default is used.

auto_checkpoint: bool
auto_checkpoint: bool, optional
If true, the database will automatically checkpoint when the size of
the WAL file exceeds the checkpoint threshold.
the WAL file exceeds the checkpoint threshold. If None (default),
the engine default is used.

checkpoint_threshold: int
checkpoint_threshold: int, optional
The threshold of the WAL file size in bytes. When the size of the
WAL file exceeds this threshold, the database will checkpoint if autoCheckpoint is true.
If None (default), the engine default is used.

throw_on_wal_replay_failure: bool
throw_on_wal_replay_failure: bool, optional
If true, any WAL replaying failure when loading the database will throw an error.
Otherwise, Lbug will silently ignore the failure and replay up to where the error
occured.
occured. If None (default), the engine default is used.

enable_checksums: bool
enable_checksums: bool, optional
If true, the database will use checksums to detect corruption in the
WAL file.
WAL file. If None (default), the engine default is used.

enable_multi_writes: bool
If true, multiple concurrent write transactions are allowed. Default to False.
enable_multi_writes: bool, optional
If true, multiple concurrent write transactions are allowed. If None
(default), the engine default is used.

backend : {"auto", "capi", "pybind"}
Backend to use for database/query execution.
Expand Down Expand Up @@ -213,6 +219,35 @@ def __getstate__(self) -> dict[str, Any]:
}
return state

def _engine_kwargs(self) -> dict[str, Any]:
"""Return only the explicitly set engine settings.

Unset (None) settings are omitted so the engine default applies,
instead of this binding pinning its own copy of the default.
"""
kwargs: dict[str, Any] = {}
if self.buffer_pool_size is not None:
kwargs["buffer_pool_size"] = self.buffer_pool_size
if self.max_num_threads is not None:
kwargs["max_num_threads"] = self.max_num_threads
if self.compression is not None:
kwargs["compression"] = self.compression
if self.read_only is not None:
kwargs["read_only"] = self.read_only
if self.max_db_size is not None:
kwargs["max_db_size"] = self.max_db_size
if self.auto_checkpoint is not None:
kwargs["auto_checkpoint"] = self.auto_checkpoint
if self.checkpoint_threshold is not None:
kwargs["checkpoint_threshold"] = self.checkpoint_threshold
if self.throw_on_wal_replay_failure is not None:
kwargs["throw_on_wal_replay_failure"] = self.throw_on_wal_replay_failure
if self.enable_checksums is not None:
kwargs["enable_checksums"] = self.enable_checksums
if self.enable_multi_writes is not None:
kwargs["enable_multi_writes"] = self.enable_multi_writes
return kwargs

def init_database(self) -> None:
"""Initialize the database."""
self.check_for_database_close()
Expand All @@ -221,17 +256,7 @@ def init_database(self) -> None:
self._database = self.init_pybind_database()
else:
self._database = get_capi_module().Database(
self.database_path,
self.buffer_pool_size,
self.max_num_threads,
self.compression,
self.read_only,
self.max_db_size,
self.auto_checkpoint,
self.checkpoint_threshold,
self.throw_on_wal_replay_failure,
self.enable_checksums,
self.enable_multi_writes,
self.database_path, **self._engine_kwargs()
)

def init_pybind_database(self) -> Any | None:
Expand All @@ -241,21 +266,7 @@ def init_pybind_database(self) -> Any | None:
if pybind_module is None:
return None
if self._pybind_database is None:
kwargs = {
"database_path": self.database_path,
"buffer_pool_size": self.buffer_pool_size,
"max_num_threads": self.max_num_threads,
"compression": self.compression,
"read_only": self.read_only,
"auto_checkpoint": self.auto_checkpoint,
"checkpoint_threshold": self.checkpoint_threshold,
"throw_on_wal_replay_failure": self.throw_on_wal_replay_failure,
"enable_checksums": self.enable_checksums,
"enable_multi_writes": self.enable_multi_writes,
}

if self.max_db_size is not None:
kwargs["max_db_size"] = self.max_db_size
kwargs = {"database_path": self.database_path, **self._engine_kwargs()}

self._pybind_database = pybind_module.Database(**kwargs)

Expand Down
10 changes: 8 additions & 2 deletions test/test_database.py
Original file line number Diff line number Diff line change
Expand Up @@ -177,9 +177,15 @@ def test_database_enable_checksums_config(tmp_path: Path) -> None:
conn.execute("call force_checkpoint_on_close=false")
conn.execute("create node table test(id int64 primary key)")

# running again with enable_checksums=False should give an error
# running again with enable_checksums=False should give an error, as long as
# WAL replay failures are configured to throw (the engine default is to
# silently ignore them, and the binding no longer overrides that default).
with pytest.raises(RuntimeError) as check:
db = lb.Database(database_path=database_path, enable_checksums=False)
db = lb.Database(
database_path=database_path,
enable_checksums=False,
throw_on_wal_replay_failure=True,
)


def test_database_close_order() -> None:
Expand Down
Loading