AsyncExecutor API¶

Recommended usage

For application code, prefer async SQL/OpenCypher via async_exec.command(...) and async_exec.query(...). Record-level helpers remain available for lower-level workflows and tests.

Bulk ingest guidance

For bulk table/document ingest in this repository, keep async SQL on a single worker unless you have workload-specific evidence to do otherwise. Multi-threaded async insert has not been safe or reliable in the current Python benchmarks.

The AsyncExecutor provides low-level async operations for parallel processing, automatic batching, and optimized WAL operations.

Using Context Managers

For automatic resource cleanup, prefer using context managers:

with arcadedb.create_database("./mydb") as db:
    async_exec = db.async_executor()
    async_exec.set_parallel_level(1)
    # Use for bulk operations...
    async_exec.wait_completion()
# Database automatically closed

Examples below show explicit db.close() for clarity, but context managers are recommended in production.

Overview¶

The AsyncExecutor class enables:

Parallel Execution: 1-16 worker threads for concurrent operations
Automatic Batching: Auto-commit every N operations
Optimized WAL: Configurable Write-Ahead Log settings
High Performance: 50,000-200,000 records/sec throughput
Fluent Interface: Method chaining for configuration

Getting AsyncExecutor¶

import arcadedb_embedded as arcadedb

db = arcadedb.create_database("./mydb")

# Get async executor
async_exec = db.async_executor()

# Configure (all methods return self for chaining)
async_exec.set_parallel_level(8)       # 8 worker threads
async_exec.set_commit_every(5000)      # Auto-commit every 5K ops
async_exec.set_back_pressure(75)       # Queue back-pressure at 75%

# Use for bulk SQL operations
for i in range(100000):
    async_exec.command(
        "sql",
        "INSERT INTO User SET userId = :id, name = :name",
        callback=lambda rs: None,
        id=i,
        name=f"User {i}",
    )

# Wait for completion
async_exec.wait_completion()

# Clean up worker threads
async_exec.close()

db.close()

Configuration Methods¶

All configuration methods return self for method chaining.

set_parallel_level¶

async_exec.set_parallel_level(level: int) -> AsyncExecutor

Set number of parallel worker threads (1-16).

Parameters:

level (int): Number of worker threads

Returns:

AsyncExecutor: Self for chaining

Guidelines:

CPU-bound: Match CPU cores (4-8)
I/O-bound: Can exceed cores (8-16)
Default: Number of CPU cores
Raises ValueError if level is not between 1 and 16

Example:

# Configure for 8-core CPU
async_exec = db.async_executor().set_parallel_level(8)

set_commit_every¶

async_exec.set_commit_every(count: int) -> AsyncExecutor

Set auto-commit batch size. Commits transaction every N operations.

Parameters:

count (int): Number of operations before commit (0 = no auto-commit)

Returns:

AsyncExecutor: Self for chaining

Guidelines:

Small datasets (< 10K): 1000-2000
Medium datasets (10K-100K): 5000
Large datasets (> 100K): 10000-20000

Example:

# Auto-commit every 10K operations
async_exec = db.async_executor().set_commit_every(10000)

set_transaction_use_wal¶

async_exec.set_transaction_use_wal(use_wal: bool) -> AsyncExecutor

Enable or disable Write-Ahead Log for transactions.

Parameters:

use_wal (bool): True to enable WAL (durability), False for speed

Returns:

AsyncExecutor: Self for chaining

Note: Disabling WAL increases speed but reduces durability.

Example:

# Disable WAL for maximum speed (less durable)
async_exec = db.async_executor().set_transaction_use_wal(False)

set_back_pressure¶

async_exec.set_back_pressure(percentage: int) -> AsyncExecutor

Set queue back-pressure threshold (0-100).

Parameters:

percentage (int): Percentage (0-100). Raises ValueError if outside 0-100

Returns:

AsyncExecutor: Self for chaining

How it works:

Queue fills up → Back-pressure kicks in
Slows down enqueue operations
Prevents memory overflow
0 = disabled, 50-75 = recommended

Example:

# Set back-pressure at 75% full
async_exec = db.async_executor().set_back_pressure(75)

Method Chaining¶

# Chain all configurations
async_exec = (db.async_executor()
    .set_parallel_level(8)
    .set_commit_every(10000)
    .set_transaction_use_wal(True)
    .set_back_pressure(75)
)

Operation Methods¶

The async executor schedules SQL/OpenCypher work and a small set of record-level graph and time-series operations. There are no create_record/update_record/delete_record methods; perform inserts, updates, and deletes through command(...) with SQL.

command¶

async_exec.command(
    language: str,
    command_text: str,
    callback: Optional[Callable[[Any], None]] = None,
    args: Optional[Sequence[Any]] = None,
    error_callback: Optional[Callable[[Exception], None]] = None,
    **params,
)

Execute an async command (INSERT/UPDATE/DELETE/DDL). The callback is optional.

Parameters:

language (str): Command language ("sql", "opencypher", etc.)
command_text (str): Command string
callback (Optional[Callable]): Optional callback invoked with each result row
args (Optional[Sequence]): Positional parameters (use ? placeholders)
error_callback (Optional[Callable]): Optional per-operation error callback
**params: Named parameters (use :name placeholders)

args vs. params

Pass either positional args or named **params, not both. Mixing them raises ValueError.

Example:

async_exec = db.async_executor()

# Async inserts via SQL (no create_record helper exists)
for i in range(10000):
    async_exec.command(
        "sql",
        "INSERT INTO User SET userId = :id, name = :name",
        id=i,
        name=f"User {i}",
    )

# Async update
async_exec.command("sql", "UPDATE User SET active = true WHERE active = false")

# Async delete with positional args
async_exec.command(
    "sql",
    "DELETE FROM LogEntry WHERE timestamp < ?",
    args=[cutoff_date],
)

async_exec.wait_completion()
async_exec.close()

query¶

async_exec.query(
    language: str,
    query_text: str,
    callback: Callable[[Any], None],
    args: Optional[Sequence[Any]] = None,
    error_callback: Optional[Callable[[Exception], None]] = None,
    **params,
)

Execute an async query with a callback invoked for each result row.

Parameters:

language (str): Query language ("sql", "opencypher", etc.)
query_text (str): Query string
callback (Callable): Callback receiving each result row
args (Optional[Sequence]): Positional parameters
error_callback (Optional[Callable]): Optional per-operation error callback
**params: Named parameters

Example:

def process_row(row):
    print(row.get("name"))

async_exec = db.async_executor()
async_exec.query("sql", "SELECT FROM User WHERE age > 18", process_row)
async_exec.wait_completion()
async_exec.close()

new_edge¶

async_exec.new_edge(
    source_vertex,
    edge_type: str,
    destination_vertex_or_rid,
    light: bool = False,
    callback: Optional[Callable[[Any, bool, bool], None]] = None,
    **properties,
)

Asynchronously create an edge between an existing source vertex and a destination vertex (or RID string). The optional callback receives (edge, created_source_vertex, created_dest_vertex).

transaction¶

async_exec.transaction(
    tx_block: Callable[[], None],
    retries: Optional[int] = None,
    ok_callback: Optional[Callable[[], None]] = None,
    error_callback: Optional[Callable[[Exception], None]] = None,
    slot: Optional[int] = None,
)

Run tx_block inside an async transaction scope, optionally with automatic retries and completion callbacks.

scan_type¶

async_exec.scan_type(
    type_name: str,
    callback: Callable[[Any], bool],
    polymorphic: bool = True,
    error_callback: Optional[Callable[[Any, Exception], bool]] = None,
)

Asynchronously scan all records of a type, invoking callback per record. Returning False from the callback stops the scan.

Status Methods¶

wait_completion¶

async_exec.wait_completion(timeout_ms: Optional[int] = None)

Wait for all pending operations to complete.

Parameters:

timeout_ms (Optional[int]): Max wait time in milliseconds (None = forever)

Raises:

TimeoutError: If the timeout elapses before completion

Note: Always call before closing executor or database.

Example:

async_exec = db.async_executor()

# Queue operations via SQL
for i in range(10000):
    async_exec.command("sql", "INSERT INTO User SET userId = :id", id=i)

# Wait for all to complete (wait forever, or pass milliseconds)
async_exec.wait_completion()
async_exec.wait_completion(30000)  # wait at most 30 seconds

# Now safe to close
async_exec.close()

is_pending¶

async_exec.is_pending() -> bool

Check if operations are still pending.

Returns:

bool: True if operations in progress

Example:

while async_exec.is_pending():
    print("Still processing...")
    time.sleep(1)

close¶

async_exec.close()

Shutdown worker threads and clean up resources.

Note: Always call after wait_completion().

Example:

try:
    async_exec = db.async_executor()
    # Operations
    async_exec.wait_completion()
finally:
    async_exec.close()

Complete Example¶

import arcadedb_embedded as arcadedb
import time

# Create database
db = arcadedb.create_database("./async_demo")

# Create schema (ArcadeDB SQL DDL)
db.command("sql", "CREATE VERTEX TYPE Product")
db.command("sql", "CREATE PROPERTY Product.productId LONG")
db.command("sql", "CREATE PROPERTY Product.name STRING")
db.command("sql", "CREATE PROPERTY Product.price DECIMAL")
db.command("sql", "CREATE INDEX ON Product (productId) UNIQUE")

# Prepare async executor
async_exec = (db.async_executor()
    .set_parallel_level(8)
    .set_commit_every(10000)
    .set_back_pressure(75)
)

# Measure performance
start = time.time()

# Create 100K vertices asynchronously via SQL
for i in range(100000):
    async_exec.command(
        "sql",
        "INSERT INTO Product SET productId = :id, name = :name, price = :price",
        id=i,
        name=f"Product {i}",
        price=i * 10.5,
    )

# Wait for completion
async_exec.wait_completion()

elapsed = time.time() - start
throughput = 100000 / elapsed

print(f"✅ Created 100,000 vertices")
print(f"⏱️  Time: {elapsed:.2f}s")
print(f"🚀 Throughput: {throughput:,.0f} records/sec")

# Clean up
async_exec.close()
db.close()

Performance Comparison¶

import time

# Synchronous (baseline)
start = time.time()
with db.transaction():
    for i in range(10000):
        vertex = db.new_vertex("User")
        vertex.set("userId", i)
        vertex.save()
sync_time = time.time() - start

# Asynchronous
start = time.time()
async_exec = db.async_executor().set_parallel_level(8)
for i in range(10000):
    async_exec.command("sql", "INSERT INTO User SET userId = :id", id=i)
async_exec.wait_completion()
async_exec.close()
async_time = time.time() - start

print(f"Synchronous: {10000 / sync_time:,.0f} records/sec")
print(f"Asynchronous: {10000 / async_time:,.0f} records/sec")
print(f"Speedup: {sync_time / async_time:.1f}x")

Typical Results: - Synchronous: 15,000-30,000 records/sec - Asynchronous: 50,000-200,000 records/sec - Speedup: 3-5x

Best Practices¶

0. Set a Commit Cadence¶

async_exec = db.async_executor()
async_exec.set_commit_every(500)  # Ensures async writes are persisted transactionally

Configure set_commit_every() for every async workload so writes are grouped into transactions.
Tune the batch size to balance commit overhead and memory.

1. Always Close the Executor¶

# ✅ Good: Use try/finally
async_exec = db.async_executor()
try:
    # Operations
    async_exec.wait_completion()
finally:
    async_exec.close()

2. Wait Before Closing¶

# ✅ Good: Wait first
async_exec.wait_completion()
async_exec.close()

# ❌ Bad: Close without waiting
async_exec.close()  # Operations may be lost!

3. Use Appropriate Batch Size¶

# ✅ Good: Tune for dataset size
if record_count < 10000:
    async_exec.set_commit_every(2000)
elif record_count < 100000:
    async_exec.set_commit_every(5000)
else:
    async_exec.set_commit_every(20000)

4. Match Parallelism to Hardware¶

import os

# ✅ Good: Match CPU cores
cpu_count = os.cpu_count() or 4
async_exec.set_parallel_level(min(cpu_count, 16))

Troubleshooting¶

Out of Memory Errors¶

# Reduce back-pressure threshold
async_exec.set_back_pressure(50)  # Slow down enqueue

# Or reduce parallel level
async_exec.set_parallel_level(4)  # Fewer workers

Slow Performance¶

# Increase parallelism
async_exec.set_parallel_level(16)

# Increase batch size
async_exec.set_commit_every(20000)

# Consider disabling WAL (less durable!)
async_exec.set_transaction_use_wal(False)

Operations Not Completing¶

# Always call wait_completion()
async_exec.wait_completion()

# Check for pending operations
if async_exec.is_pending():
    print("Still processing...")

AsyncExecutor API¶

Overview¶

Getting AsyncExecutor¶

Configuration Methods¶

set_parallel_level¶

set_commit_every¶

set_transaction_use_wal¶

set_back_pressure¶

Method Chaining¶

Operation Methods¶

command¶

query¶

new_edge¶

transaction¶

scan_type¶

Status Methods¶

wait_completion¶

is_pending¶

close¶

Complete Example¶

Performance Comparison¶

Best Practices¶

0. Set a Commit Cadence¶

1. Always Close the Executor¶

2. Wait Before Closing¶

3. Use Appropriate Batch Size¶

4. Match Parallelism to Hardware¶

Troubleshooting¶

Out of Memory Errors¶

Slow Performance¶

Operations Not Completing¶

See Also¶