Profile DIS ingestion performance

The Data Import Server (DIS) exposes detailed, per-column ingestion metrics through its profiling system. When enabled, the DIS records nanosecond-resolution timing statistics for each column it processes and writes them to an internal process metrics table. These metrics help you identify which columns are the bottleneck in your ingestion pipeline and investigate latency or throughput problems at a very granular level.

Enable DIS profiling

Both properties must be set as JVM arguments on the DIS — either the standalone db_dis process or the Persistent Query that runs the DIS.

Enable process metrics

The DIS must write process metrics. Add the following JVM argument to the DIS:

-DIrisLogDefaults.writeDatabaseProcessMetrics=true

For the standalone db_dis process, add this to the process's JVM arguments in your system configuration. For a DIS Persistent Query, add it to the Extra JVM Arguments field in the Settings tab (under Show Advanced).

Caution

Enabling process metrics generates a large volume of data. Enable this only for the specific DIS process you are investigating and disable it when you are done.

Enable column-level profiling

To activate per-column profiling for a specific table, set the following JVM argument on the DIS:

-DDataImportStreamProcessor.<NAMESPACE>.<TABLE_NAME>.profilingLevel=CATEGORY_WITH_COLUMNS

Replace <NAMESPACE> and <TABLE_NAME> with the actual namespace and table name. For example, to profile the Trades table in the Market namespace:

-DDataImportStreamProcessor.Market.Trades.profilingLevel=CATEGORY_WITH_COLUMNS

The CATEGORY_WITH_COLUMNS level enables full per-column statistics when the DIS uses batched row processing. You can profile multiple tables by adding a separate property for each one.

Note

Profiling is supported for Tailer-based connections only. DIS Persistent Queries for Kafka and Solace feeds do not currently support profiling.

Query the metrics

Once both properties are set, the DIS writes metrics to an internal table under the DbInternal namespace:

DIS Persistent Query (in-worker DIS running in a Core+ worker): Metrics are written to ProcessMetricsLogCoreV2.
Standalone db_dis process: Metrics are written to ProcessMetrics (the legacy table).

Metric names follow the pattern:

DataImport-<NAMESPACE>/<TABLE_NAME>/<TABLE_TYPE>/<INTERNAL_PARTITION>/<COLUMN_PARTITION>/columnProcessingDurationNanos-<ColumnName>

The <TABLE_TYPE> segment is the table type descriptor — I for intraday tables, which is the type handled by the DIS. The internal partition and column partition segments identify the specific DIS partition the data was recorded for. Each metric is recorded at multiple time intervals (e.g., 1m, 15m, 1h).

The following query loads the metrics table and produces two result tables: a per-column summary filtered to 1-minute intervals, and a broader view of all DataImport metrics for the target table. Use ProcessMetricsLogCoreV2 for DIS Persistent Queries or ProcessMetrics for standalone db_dis.

// For DIS Persistent Query (Core+ worker):
process_metrics = db.liveTable("DbInternal", "ProcessMetricsLogCoreV2").where("Date=today()")
// For standalone db_dis:
// process_metrics = db.liveTable("DbInternal", "ProcessMetrics").where("Date=today()")

def analyzeColumnProcessing = { table, namespace, tableName ->
    def prefix = "DataImport-${namespace}/${tableName}/I/"

    columnProcessing_1m = table
        .where("Interval=`1m`", "Name.startsWith(`${prefix}`)", "Name.contains(`columnProcessingDurationNanos-`)")
        .update(
            "Partition = Name.split(`columnProcessingDurationNanos-`)[0]",
            "Column = Name.split(`columnProcessingDurationNanos-`)[1]"
        )
        .moveColumnsUp("Partition", "Column", "Sum")

    dataimport = table.where("Name.startsWith(`DataImport-${namespace}/${tableName}`)")

    [columnProcessing_1m: columnProcessing_1m, dataimport: dataimport]
}

result = analyzeColumnProcessing(process_metrics, "<NAMESPACE>", "<TABLE_NAME>")

# For DIS Persistent Query (Core+ worker):
process_metrics = db.live_table("DbInternal", "ProcessMetricsLogCoreV2").where(
    "Date=today()"
)
# For standalone db_dis:
# process_metrics = db.live_table("DbInternal", "ProcessMetrics").where("Date=today()")


def analyze_column_processing(table, namespace, table_name):
    prefix = f"DataImport-{namespace}/{table_name}/I/"

    column_processing_1m = (
        table.where(
            [
                "Interval=`1m`",
                f"Name.startsWith(`{prefix}`)",
                "Name.contains(`columnProcessingDurationNanos-`)",
            ]
        )
        .update(
            [
                "Partition = Name.split(`columnProcessingDurationNanos-`)[0]",
                "Column = Name.split(`columnProcessingDurationNanos-`)[1]",
            ]
        )
        .move_columns_up(["Partition", "Column", "Sum"])
    )

    dataimport = table.where(f"Name.startsWith(`DataImport-{namespace}/{table_name}`)")

    return {"column_processing_1m": column_processing_1m, "dataimport": dataimport}


result = analyze_column_processing(process_metrics, "<NAMESPACE>", "<TABLE_NAME>")

Interpret the results

`columnProcessing_1m`

This table has one row per column per 1-minute interval. The key fields are:

Column — the name of the table column being processed.
Partition — the partition key prefix extracted from the metric name, identifying which DIS partition processed the data.
Sum — total nanoseconds spent processing that column during the 1-minute interval. This is the most useful field for identifying expensive columns.
N — the number of processing events recorded in the interval.
Max — the longest single processing event, useful for spotting latency spikes.
Avg — the average nanoseconds per event (Sum / N).

Sort by Sum descending to find which columns consume the most processing time. Columns with high Sum values relative to others are candidates for optimization.

`dataimport`

This table includes all DataImport-<NAMESPACE>/<TABLE_NAME> metrics — not only column-level but also higher-level ingestion categories such as total row processing time, flush durations, and checkpoint activity. Use this table to understand the full ingestion picture alongside the per-column breakdown.

Filter dataimport by Interval (e.g., 1m, 15m) and sort by Sum descending to identify the most time-consuming ingestion phases.

Symbol table caching considerations

String columns that use symbol tables are often the most expensive columns in the columnProcessing_1m output. Each incoming string value must be looked up or inserted into the symbol manager cache, and a cache miss — where the value is not found and a new ID must be assigned — is significantly more expensive than a cache hit.

If a string column shows disproportionately high Sum values:

Check whether the column has a symbol table (see the symbol table audit script).
If efficiency is low (well below 1.0), the bounded symbol manager cache may be too small for the column's cardinality or repetition interval. Increasing the cache size with LocalAppendableTableComponentFactory.boundedSymbolManagerSize or configuring a column-specific cache hint on the DIS can reduce per-row processing time.
For columns where values do not repeat, disabling the symbol table entirely with symbolTable="None" in the schema removes the lookup overhead entirely.

See Symbol tables and caching for guidance on tuning.