Data control from scripts

The data control tool (dhctl) is a command-line tool for managing intraday data. The same functionality is available from any Core+ worker with sufficient permissions, in both Groovy and Python.

Truncate and delete intraday partitions

Truncate and delete operations target intraday partitions. Truncating removes the data and marks the partition as permanently truncated. Deleting removes the partition directories entirely; partitions must be truncated before they can be deleted.

Simple usage

The following patterns use default authentication and target all Data Import Servers configured for the table.

import io.deephaven.configuration.IntradayControl

// truncate all internal partitions for a column partition
opts = IntradayControl.Options.builder().key("MyNamespace", "MyTable", "2025-03-19").build()
result = IntradayControl.truncateIntradayPartition(opts)
println result

// truncate a single internal partition
opts = IntradayControl.Options.builder()
        .key("MyNamespace", "MyTable", "MyInternalPartition", "2025-03-19").build()
result = IntradayControl.truncateIntradayPartition(opts)

// delete (must truncate first)
opts = IntradayControl.Options.builder().key("MyNamespace", "MyTable", "2025-03-19").build()
result = IntradayControl.deleteIntradayPartition(opts)

from deephaven_enterprise.intraday_control import truncate_partition, delete_partition

# truncate all internal partitions for a column partition
result = truncate_partition("MyNamespace", "MyTable", "2025-03-19")
print(result)

# truncate a single internal partition
result = truncate_partition(
    "MyNamespace", "MyTable", "2025-03-19", "MyInternalPartition"
)
print(result)

# delete (must truncate first)
result = delete_partition("MyNamespace", "MyTable", "2025-03-19")
print(result)

Tip

Always check the results of these commands.

Options

For full control over parameters - including dry runs and DIS filtering - pass additional keyword arguments in Python, or use the Options.builder() in Groovy.

import io.deephaven.configuration.IntradayControl

builder = IntradayControl.Options.builder()
        .key("MyNamespace", "MyTable", "columnPartition")
// or with an internal partition:
//   builder.key("MyNamespace", "MyTable", "internalPartition", "columnPartition")

Dry run options

builder.dryRun(true)   // perform a dry run - do not actually truncate or delete
builder.dryRun(false)  // perform the actual operation

# perform a dry run - do not actually truncate or delete
result = truncate_partition("MyNamespace", "MyTable", "2025-03-19", dry_run=True)

DIS include/exclude

// one DIS at a time
builder.addDisInclude("dis_name")   // only send to this DIS
builder.addDisExclude("dis_name")   // do not send to this DIS

// or add several at once from any Iterable<String>
builder.addAllDisIncludes(["dis_a", "dis_b"])
builder.addAllDisExcludes(["dis_c", "dis_d"])

result = truncate_partition(
    "MyNamespace",
    "MyTable",
    "2025-03-19",
    include_dis=("dis_name",),  # only send to these DISes
    exclude_dis=("other_dis",),  # do not send to these DISes
)

Check the results

Truncate, delete, and rescan operations return a DisCommandResult containing detailed information about the operation. The result has a summary status and a map of per-DIS results, each of which may contain per-location details.

Check overall success

import io.deephaven.configuration.DisCommandResult

opts = IntradayControl.Options.builder().key("MyNamespace", "MyTable", "2025-03-19").build()
result = IntradayControl.truncateIntradayPartition(opts)

if (result.getSummaryResult() == DisCommandResult.Status.SUCCESS) {
    println "Success"
} else {
    println "Failed: " + result
}

from deephaven_enterprise.intraday_control import truncate_partition, DisCommandStatus

result = truncate_partition("MyNamespace", "MyTable", "2025-03-19")

if result.summary_status is DisCommandStatus.SUCCESS:
    print("Success")
else:
    print(f"Failed: {result}")

Check per-DIS results

result.getDisResultsMap().each { disName, disResult ->
    println "DIS: ${disName} -> ${disResult.getStatus()}"
    if (disResult.getException() != null) {
        println "  Exception: ${disResult.getException()}"
    }
}

for dis_result in result.dis_results.values():
    print(f"DIS: {dis_result.dis_name} -> {dis_result.status.name}")
    if dis_result.exception_message is not None:
        print(
            f"  Exception: {dis_result.exception_class_name}: {dis_result.exception_message}"
        )

Check per-location results

For truncate and delete operations, each DIS result contains location-level detail:

result.getDisResultsMap().each { disName, disResult ->
    disResult.getLocationResults().each { loc ->
        println "  Location: ${loc.getKeyString()} -> ${loc.status()}"
        println "    Active processor: ${loc.hasActiveProcessor()}"
    }
}

for dis_result in result.dis_results.values():
    print(f"DIS: {dis_result.dis_name}")
    for loc in dis_result.location_results:
        key = ".".join(
            [
                loc.namespace,
                loc.table_name,
                loc.internal_partition or "<all>",
                loc.column_partition,
            ]
        )
        print(f"  Location: {key} -> {loc.status.name}")
        print(f"    Active processor: {loc.has_active_processor}")

Dry run then truncate

This example executes a dry run, verifies no locations are actively being tailed, and then performs the actual truncate:

import io.deephaven.configuration.IntradayControl
import io.deephaven.configuration.DisCommandResult

namespace = "MyNamespace"
tableName = "MyTable"
date = today()

dryRunOpts = IntradayControl.Options.builder()
        .key(namespace, tableName, date)
        .dryRun(true)
        .build()

dryRunResult = IntradayControl.truncateIntradayPartition(dryRunOpts)

// verify overall success
okToTruncate = dryRunResult.getSummaryResult() == DisCommandResult.Status.SUCCESS

// verify all DIS results are successful
okToTruncate = okToTruncate &&
    dryRunResult.getDisResultsMap().values().every { disResult ->
        disResult.getStatus() == DisCommandResult.Status.SUCCESS
    }

// check for locations being actively tailed
hasActiveProcessor = dryRunResult.getDisResultsMap().values().any { disResult ->
    disResult.getLocationResults().any { loc -> loc.hasActiveProcessor() }
}
okToTruncate = okToTruncate && !hasActiveProcessor

println "OK to truncate: ${okToTruncate}"

if (okToTruncate) {
    realOpts = IntradayControl.Options.builder()
            .key(namespace, tableName, date)
            .build()
    commandResult = IntradayControl.truncateIntradayPartition(realOpts)
    println commandResult
}

from deephaven_enterprise.intraday_control import truncate_partition, DisCommandStatus

namespace = "MyNamespace"
table_name = "MyTable"
date = "2025-03-19"

dry_run_result = truncate_partition(namespace, table_name, date, dry_run=True)

ok_to_truncate = (
    dry_run_result.summary_status is DisCommandStatus.SUCCESS
    and not dry_run_result.has_active_processor()
)
print(f"OK to truncate: {ok_to_truncate}")

if ok_to_truncate:
    result = truncate_partition(namespace, table_name, date)
    print(result)

Truncate then delete

import io.deephaven.configuration.IntradayControl
import io.deephaven.configuration.DisCommandResult

namespace = "MyNamespace"
tableName = "MyTable"
date = "2025-03-19"
opts = IntradayControl.Options.builder().key(namespace, tableName, date).build()

truncateResult = IntradayControl.truncateIntradayPartition(opts)
printf "Truncate %s.%s partition %s:\n", namespace, tableName, date
println truncateResult

if (truncateResult.getSummaryResult() == DisCommandResult.Status.SUCCESS &&
    truncateResult.getDisResultsMap().values().every { it.getStatus() == DisCommandResult.Status.SUCCESS }) {
    deleteResult = IntradayControl.deleteIntradayPartition(opts)
    printf "Delete %s.%s partition %s:\n", namespace, tableName, date
    println deleteResult
}

from deephaven_enterprise.intraday_control import (
    truncate_partition,
    delete_partition,
    DisCommandStatus,
)

namespace = "MyNamespace"
table_name = "MyTable"
date = "2025-03-19"

truncate_result = truncate_partition(namespace, table_name, date)
print(f"Truncate {namespace}.{table_name} partition {date}:")
print(truncate_result)

if truncate_result.summary_status is DisCommandStatus.SUCCESS:
    delete_result = delete_partition(namespace, table_name, date)
    print(f"Delete {namespace}.{table_name} partition {date}:")
    print(delete_result)

Rescan tables

A rescan instructs the Data Import Server to look for new data.

import io.deephaven.configuration.IntradayControl

// rescan a specific table
result = IntradayControl.rescan(IntradayControl.RescanOptions.builder().table("MyNamespace", "MyTable").build())
println result

// rescan all tables on all DIS instances
result = IntradayControl.rescan(IntradayControl.RescanOptions.builder().build())
println result

from deephaven_enterprise.intraday_control import rescan_table, rescan_all

# rescan a specific table
result = rescan_table("MyNamespace", "MyTable")
print(result)

# rescan all tables on all DIS instances
result = rescan_all()
print(result)

Caveats

These methods make a best-effort attempt to perform the operation on all appropriate Data Import Servers. This cannot be atomic, so the operation might have only partial success. Make sure you check all the results.
Truncated partitions are marked as permanently truncated, meaning no further data can be ingested into them. This prevents confusion if loggers produce new data or if tailers have not yet processed all existing files.
Before logging new data for truncated partitions, remove any existing data files (bin files), and then delete the partitions.
It is possible to delete data on one Data Import Server and leave it on another (e.g., a backup). Be extremely careful with this, as it can create confusion.

Data control tool (dhctl) - command-line interface for the same operations
Data control tool (Legacy) - Legacy Groovy worker version
Intraday data
Intraday partitions
Historical metadata indexes
Data Import Server
Data routing configuration
Binary log files