Core+ Java Client

Deephaven's Core+ Java client provides communication to the PQ controller as well as the Authentication server. This communication enables you to create new workers, discover the address of existing workers, and access the internal database via a session.

The underlying session object is a wrapper around the Deephaven Community Core Java client. Thus, all table operations and Community Core functionalities are inherited from the Community client API.

Note

This guide focuses on Core+ specific features like Persistent Query management, cluster connections, and enterprise authentication. For table operations, listeners, and general Java client usage, see the Community Core Java client guide.

Setup

Setup for the Core+ Java client is typically done through Gradle. For instructions, see Core+ Java client setup.

While there are multiple ways to set up the client with tools other than Gradle, this guide uses Gradle for its simplicity and interoperability.

Note

Maven Core+ Java Client Example provides a full project example using the Barrage Java Client with Maven as an alternative to using Gradle.

Get started

After completing setup, you must configure Deephaven to load properties from your Deephaven server's URL. This typically uses port 8000 when your system is configured with Envoy or port 8123 without Envoy. You must set the properties URL before using any other Deephaven classes; otherwise, static fields in classes will fail to initialize.

import io.deephaven.enterprise.config.HttpUrlPropertyInputStreamLoader;

final String serverUrl = "https://deephaven-host:8000";

// Tell the configuration library to load configuration from the specified URL
HttpUrlPropertyInputStreamLoader.setServerUrl(serverUrl);

There are two ways you can interact with data via the Core+ Java client:

client-barrage: Unlocks the full power of the ticking Deephaven engine through Deephaven's Barrage extension. Supports live, ticking table subscriptions. Requires Java 11 or later.
client-flight: Provides the ability to get static snapshots of tables using the Apache Arrow Flight protocol. Does not support live updates. Requires Java 8 or later.

Tip

Which should I use? Use client-barrage if you need live, ticking data or are building applications that react to real-time updates. Use client-flight only if you need Java 8 compatibility or only require static snapshots.

Next, create a session that loads the connection configuration from the URL. You can use either client-barrage or client-flight:

import io.deephaven.enterprise.dnd.client.DndSessionFactoryBarrage;

// Create the session factory for Barrage. This creates the connections to the server and downloads the
// configuration.
final DndSessionFactoryBarrage sessionFactory = new DndSessionFactoryBarrage(serverUrl + "/iris/connection.json");

import io.deephaven.enterprise.dnd.client.DndSessionFactoryFlight;

// Create the session factory for Flight. This creates the connections to the server and downloads the
// configuration.
final DndSessionFactoryFlight sessionFactory = new DndSessionFactoryFlight(serverUrl + "/iris/connection.json");

Then, authenticate with the server using a username and password:

sessionFactory.password("user", "password");

Alternatively, you can use a private key for better security:

sessionFactory.privateKey("/path/to/key_base64.txt");

Now you can connect to an existing PQ. The session object provides basic methods for interacting with tables and also gives you access to the underlying SessionImpl for direct worker interaction.

import io.deephaven.enterprise.dnd.client.DndSessionBarrage;

final DndSessionBarrage session = sessionFactory.persistentQuery("MyTestQuery");

Alternatively, you could create a new temporary Persistent Query to host your worker:

final DndSessionBarrage session = sessionFactory.newWorker(
    "ExampleTestWorker",  // Worker name
    2,                    // Heap size in GB
    600_000,              // Timeout in milliseconds (10 minutes)
    60_000                // Startup wait in milliseconds (60 seconds)
);

import io.deephaven.enterprise.dnd.client.DndSessionFlight;

final DndSessionFlight session = sessionFactory.persistentQuery("MyTestQuery");

Alternatively, you could create a new temporary Persistent Query to host your worker:

final DndSessionFlight session = sessionFactory.newWorker(
    "ExampleTestWorker",  // Worker name
    2,                    // Heap size in GB
    600_000,              // Timeout in milliseconds (10 minutes)
    60_000                // Startup wait in milliseconds (60 seconds)
);

For more precise control over Persistent Query options, you can create a configuration object builder directly:

import io.deephaven.proto.controller.PersistentQueryConfigMessage;

final PersistentQueryConfigMessage.Builder builder = sessionFactory.createPersistentQueryConfig("ExampleTestWorker", 2, 600_000);

// Set viewer and admin permissions
builder.addViewerGroups("MyViewers");
builder.addAdminGroups("SomeUser");

// Add custom JVM arguments
builder.addExtraJvmArguments("-DMy.Extra.Param=27");
builder.addExtraJvmArguments("-XX:MaxGCPauseMillis=200");

// Set engine type and DB server
builder.setWorkerKind("DeephavenCommunity"); // Use Core+ engine
builder.setServerName("Query_1");

final DndSessionBarrage session = sessionFactory.newWorker(builder.build(), 60_000);

The following examples are self-contained Java clients that connect to a running Persistent Query and table.

The Gradle setup instructions provided here include the required parameters to run the included examples. A Core+ Persistent Query should be running that creates a table (for example, with db.liveTable), and the Java files should be updated to use that PQ's name as well as a valid username/password or a keyfile.

Cluster connections and PQ management

For programmatic Persistent Query management and cluster-level operations, create a DeephavenClusterConnection:

import io.deephaven.enterprise.dnd.client.DeephavenClusterConnection;
import io.deephaven.enterprise.controller.client.ControllerClientGrpc;
import io.deephaven.enterprise.auth.UserContext;
import io.deephaven.proto.controller.PersistentQueryInfoMessage;
import io.deephaven.proto.controller.PersistentQueryStatusEnum;

final String connectionUrl = serverUrl + "/iris/connection.json";
final DeephavenClusterConnection clusterConnection = new DeephavenClusterConnection(connectionUrl);

// Authenticate the cluster connection. Both methods return false if authentication fails.
final boolean authenticated = (keyFile != null)
        ? clusterConnection.privateKey(keyFile)
        : clusterConnection.password(userName, password);
if (!authenticated) {
    throw new RuntimeException("Authentication failed");
}

// Access cluster services
final UserContext userContext = clusterConnection.getUserContext();
final ControllerClientGrpc controllerClient = clusterConnection.getControllerClient();

// Get effective user
final String effectiveUser = userContext.getEffectiveUser();
System.out.println("Connected as: " + effectiveUser);

Check PQ status

Query PQ information and status:

// Get information about a specific PQ
final PersistentQueryInfoMessage pqInfo = sessionFactory.getPQ("MyPersistentQuery");

if (pqInfo != null) {
    System.out.println("PQ Name: " + pqInfo.getConfig().getName());

    final PersistentQueryStatusEnum status = pqInfo.getState().getStatus();

    switch (status) {
        case PQS_RUNNING:
            System.out.println("PQ is running");
            break;
        case PQS_STOPPED:
            System.out.println("PQ is stopped");
            break;
        case PQS_FAILED:
            System.out.println("PQ has failed");
            break;
        default:
            System.out.println("PQ status: " + status);
    }
}

Start or restart a PQ

Programmatically control PQ lifecycle:

final PersistentQueryInfoMessage pqInfo = sessionFactory.getPQ("MyPQ");

if (pqInfo != null) {
    controllerClient.restartQuery(pqInfo.getConfig());
}

Monitor PQ state changes

Subscribe to PQ state updates:

import io.deephaven.proto.controller.PersistentQueryStateMessage;
import org.jetbrains.annotations.NotNull;
import java.util.concurrent.atomic.AtomicReference;

final AtomicReference<PersistentQueryStatusEnum> statusRef = new AtomicReference<>(null);

final ControllerClientGrpc.ObserverImpl observer = new ControllerClientGrpc.ObserverImpl() {
    @Override
    public void handlePut(@NotNull final PersistentQueryInfoMessage value) {
        final PersistentQueryStateMessage state = value.getState();
        System.out.println("PQ update: " + value.getConfig().getName() +
                         " -> " + (state != null ? state.getStatus() : "null"));

        synchronized (statusRef) {
            statusRef.set(state != null ? state.getStatus() : null);
            statusRef.notify();
        }
    }
};

// Register the observer before subscribing so no initial snapshot events are missed
controllerClient.addObserver(observer);
controllerClient.subscribeToAll();

// Wait for running status with timeout
synchronized (statusRef) {
    final long maxWaitMillis = 60_000; // 60 second timeout
    long currentTime = System.currentTimeMillis();
    final long endTime = currentTime + maxWaitMillis;

    while (statusRef.get() != PersistentQueryStatusEnum.PQS_RUNNING && currentTime < endTime) {
        final long waitTime = endTime - currentTime;
        if (waitTime <= 0) {
            break;
        }
        try {
            statusRef.wait(waitTime);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            controllerClient.shutdown();
            throw new RuntimeException("Interrupted while waiting for PQ to start", e);
        }
        currentTime = System.currentTimeMillis();
    }

    if (statusRef.get() != PersistentQueryStatusEnum.PQS_RUNNING) {
        throw new RuntimeException("PQ did not reach running state within timeout");
    }
}

// Clean up
controllerClient.removeObserver(observer);

Manipulate tables

You can run table operations and queries on either the server or the client. Depending on your needs, you can run queries directly on the server or distribute workloads across multiple clients. The Core+ Java client supports both approaches.

Manipulate tables on the server

To fetch and manipulate tables from the query, create a TableSpec object using one of the session's four root methods, then subscribe to it using the methods below.

Method	Description
`catalogTable`	A table containing the full catalog of available database tables.
`scopeTable`	A named table in the query scope.
`historicalTable`	A table from the historical table space.
`liveTable`	A table from the intraday table space.

For example:

import io.deephaven.qst.table.TableSpec;
import io.deephaven.enterprise.dnd.client.DndSession;

// This is an example of a batch operation based on a table in the existing query's scope.
final TableSpec testTable1 = DndSession.scopeTable("TestTable").where("Value > 4");

The TableSpec does not send any data to the client. The following list includes just a few table operations that can be applied to a TableSpec:

The example above runs a where on the TestTable in the worker directly.

When the subscribeTo or snapshotOf methods are called, then the server performs the operations described in the TableSpec and returns the resultant data.

import io.deephaven.engine.table.Table;

// This creates a ticking (Live) barrage subscription
final Table testTableSubscription = session.subscribeTo(testTable1);

// This creates a static barrage snapshot
final Table testTableSnapshot = session.snapshotOf(testTable1);

When streamOf is called, the server performs the operations described in the TableSpec and returns the resultant data as a FlightStream.

import org.apache.arrow.flight.FlightStream;
import org.apache.arrow.vector.VectorSchemaRoot;

// This creates a FlightStream of the table
final FlightStream stream = session.streamOf(testTable1);

// Print schema and iterate through data
System.out.println("Schema: " + stream.getSchema());

while (stream.next()) {
    final VectorSchemaRoot root = stream.getRoot();
    System.out.println(root.contentToTSVString());
}

stream.close();

This snippet produces a FlightStream, which is a static snapshot of the source table.

Manipulate tables in the client

The Table is a client-side object that contains a copy of all data from the server. The Barrage protocol replicates consistent updates to the client. The engine runs locally within the client, allowing table operations like where, sort, or update to execute outside the server, distributing workloads across multiple machines.

For details on table listeners and client-side operations, see the Community Core Java client guide.

Execute scripts on the server

You can also execute script code blocks on the worker using the executeCode method. After the script runs, retrieve its results using the method appropriate for your client:

import io.deephaven.engine.table.Table;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;

try {
    session.executeCode("pel = db.live_table(namespace=\"DbInternal\", table_name=\"ProcessEventLog\").where(\"Date=today()\").tail(2)");
    final Table fromExCode = session.snapshotOf(DndSession.scopeTable("pel"));
} catch (ExecutionException | InterruptedException | TimeoutException e) {
    throw new RuntimeException("executeCode failed", e);
}

import org.apache.arrow.flight.FlightStream;
import org.apache.arrow.vector.VectorSchemaRoot;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;

try {
    session.executeCode("pel = db.live_table(namespace=\"DbInternal\", table_name=\"ProcessEventLog\").where(\"Date=today()\").tail(2)");
    try (final FlightStream stream = session.streamOf(DndSession.scopeTable("pel"))) {
        while (stream.next()) {
            final VectorSchemaRoot root = stream.getRoot();
            System.out.println(root.contentToTSVString());
        }
    }
} catch (ExecutionException | InterruptedException | TimeoutException e) {
    throw new RuntimeException("executeCode failed", e);
}

Note

The executeCode method permits arbitrary code execution and is unstructured and unvalidated. Use TableSpec methods where possible.

Access database tables

Note

The following examples use client-barrage. For client-flight, replace snapshotOf and subscribeTo calls with streamOf, which returns a FlightStream instead of a Table.

Use liveTable to access intraday data and historicalTable to access historical data. You can use snapshot or subscribe on either type of data:

// Intraday table from the live table space
final TableSpec intradayOrders = DndSession.liveTable("MyNamespace", "Orders")
    .where("Date = today()")
    .sort("Timestamp");

final Table orders = session.subscribeTo(intradayOrders);

// Historical table from the historical table space
final TableSpec historicalTrades = DndSession.historicalTable("MyNamespace", "Trades")
    .where("Date = parseLocalDate(`2024-01-15`)")
    .lastBy("Symbol");

final Table trades = session.snapshotOf(historicalTrades);

Cleanup

Connections, sessions, and tables fetched from a Deephaven query all consume resources and must be properly closed when you finish using them. Every Table, Session, and SessionFactory has a close method that you should call when finished with each object.

// Close tables first, then close the session
myTable.close();
session.close();

Using try-finally

Always close sessions when done to free resources:

import java.io.IOException;

DndSessionBarrage session = null;
try {
    session = sessionFactory.persistentQuery("MyPQ");
    final TableSpec spec = DndSession.scopeTable("MyTable");

    final Table table = session.snapshotOf(spec);

    // Use the table
    System.out.println("Table size: " + table.size());

    table.close();

} catch (IOException e) {
    System.err.println("Connection error: " + e.getMessage());
} finally {
    if (session != null) {
        session.close();
    }
}

Explicit cleanup pattern for Barrage subscriptions

DndSessionBarrage session = null;
Table subscription = null;
try {
    session = sessionFactory.persistentQuery("MyPQ");
    subscription = session.subscribeTo(DndSession.scopeTable("MyTable"));

    // Use the subscription
    System.out.println("Table size: " + subscription.size());

} finally {
    if (subscription != null) {
        subscription.close();
    }
    if (session != null) {
        try {
            session.close();
        } catch (IOException e) {
            // Handle close error
        }
    }
}

Complete example

Here's a complete end-to-end example:

import io.deephaven.enterprise.config.HttpUrlPropertyInputStreamLoader;
import io.deephaven.enterprise.dnd.client.DndSessionFactoryBarrage;
import io.deephaven.enterprise.dnd.client.DndSessionBarrage;
import io.deephaven.enterprise.dnd.client.DndSession;
import io.deephaven.qst.table.TableSpec;
import io.deephaven.engine.table.Table;
import java.io.IOException;

public class DeephavenClientExample {
    public static void main(String[] args) {
        // 1. Configure server URL
        final String serverUrl = "https://deephaven:8000";
        HttpUrlPropertyInputStreamLoader.setServerUrl(serverUrl);

        // 2. Create and authenticate session factory
        final DndSessionFactoryBarrage factory;
        try {
            factory = new DndSessionFactoryBarrage(serverUrl + "/iris/connection.json");
            factory.privateKey("/path/to/keyfile.txt");
        } catch (IOException e) {
            System.err.println("Failed to create session factory: " + e.getMessage());
            return;
        }

        // 3. Connect to PQ and fetch data; always close the factory on exit
        DndSessionBarrage session = null;
        try {
            session = factory.persistentQuery("MyPQ");

            // 4. Get table from query scope
            final TableSpec tableSpec = DndSession.scopeTable("MyTable")
                .where("Value > 100")
                .sort("Timestamp");

            final Table table = session.snapshotOf(tableSpec);

            // 5. Use the table
            System.out.println("Table has " + table.size() + " rows");

        } catch (IOException e) {
            System.err.println("Error: " + e.getMessage());
        } finally {
            // Close the session and factory to release resources
            if (session != null) {
                session.close();
            }
            factory.close();
        }
    }
}

Best practices

Connection pooling

Reuse session factories across your application:

public class ConnectionPool {
    private static DndSessionFactoryBarrage factory;

    public static synchronized DndSessionFactoryBarrage getFactory() throws IOException {
        if (factory == null) {
            final String serverUrl = System.getenv("DH_SERVER_URL");
            HttpUrlPropertyInputStreamLoader.setServerUrl(serverUrl);
            factory = new DndSessionFactoryBarrage(serverUrl + "/iris/connection.json");
            factory.privateKey(System.getenv("DH_KEY_FILE"));
        }
        return factory;
    }
}

Performance

Choose the right approach for your use case:

Server-side operations: Use TableSpec operations for heavy filtering, joins, and aggregations.
Client-side operations: Distribute light operations across multiple clients.
Snapshots vs. subscriptions: Use snapshotOf for static data and subscribeTo for live updates.
Resource cleanup: Always close tables and sessions to free memory.

Access the full Deephaven Community API

If you want more sophisticated interactions with the query worker, you may retrieve the underlying SessionImpl from the session using the .session method.

See the Community Documentation and Java client guide for more details on table listeners, advanced operations, and client-side table manipulation.

Core+ Java Client

Setup

Get started

Cluster connections and PQ management

Check PQ status

Start or restart a PQ

Monitor PQ state changes

Manipulate tables

Manipulate tables on the server

Manipulate tables in the client

Execute scripts on the server

Access database tables

Cleanup

Using try-finally

Complete example

Best practices

Connection pooling

Performance

Access the full Deephaven Community API

Related documentation