feat(utils): add get_line_schema, split_row, infer_type and public catalog constants

[caterryan]

Rows from process_writes and from query() do not say which columns are tags or what type each field is. Plugins currently read information_schema themselves to find out, each in a slightly different way. This adds that lookup to utils, plus a helper that turns a row into a typed line. Bumps the package to 0.5.0.

New APIs

• introspection.get_line_schema(influxdb3_local, table) returns {"tags": [...], "fields": {name: line_type}}. Cached like get_schema; pass refresh=True to re-read the catalog.
• write.split_row(row, schema) returns (tags, typed_fields, time_ns) ready for build_line_typed. Keys the schema does not know are typed from their value.
• write.infer_type(value) is now public.
• introspection.tag_data_type, numeric_types, line_types, numeric_line_types are now public.

Usage

Before

TAG_DATA_TYPE = "Dictionary(Int32, Utf8)"
LINE_TYPES = {"Int64": "int", "Int32": "int", "UInt64": "uint", "Float64": "float",
              "Float32": "float", "Boolean": "bool", "Utf8": "string"}


def infer_line_type(value) -> str:
    if isinstance(value, bool):
        return "bool"
    if isinstance(value, int):
        return "int"
    if isinstance(value, float):
        return "float"
    return "string"


def resolve_schema(influxdb3_local, measurement) -> dict:
    columns = influxdb3_local.query(
        "SELECT column_name, data_type FROM information_schema.columns "
        "WHERE table_name = $table",
        {"table": measurement},
    )
    if not columns:
        raise Exception(f"Table '{measurement}' not found.")
    return {
        "tags": [c["column_name"] for c in columns if c["data_type"] == TAG_DATA_TYPE],
        "fields": {
            c["column_name"]: LINE_TYPES.get(c["data_type"])
            for c in columns
            if c["data_type"] != TAG_DATA_TYPE and c["column_name"] != "time"
        },
    }


def build_enriched_line(row, measurement, schema):
    tags = {name: row[name] for name in schema["tags"] if row.get(name) is not None}
    typed_fields = {}
    for name, line_type in schema["fields"].items():
        value = row.get(name)
        if value is None:
            continue
        typed_fields[name] = (value, line_type or infer_line_type(value))
    typed_fields["enriched"] = (True, "bool")
    return build_line_typed(
        LineBuilder, measurement, tags=tags, typed_fields=typed_fields, time_ns=int(row["time"])
    )


def process_writes(influxdb3_local, table_batches, args=None):
    for batch in table_batches:
        schema = resolve_schema(influxdb3_local, batch["table_name"])
        lines = [build_enriched_line(row, "cpu_enriched", schema) for row in batch["rows"]]
        write_data(influxdb3_local, lines)

After

from influxdata_plugin_utils.introspection import get_line_schema
from influxdata_plugin_utils.write import build_line_typed, split_row, write_data


def process_writes(influxdb3_local, table_batches, args=None):
    for batch in table_batches:
        schema = get_line_schema(influxdb3_local, batch["table_name"])
        lines = []
        for row in batch["rows"]:
            tags, typed_fields, time_ns = split_row(row, schema)
            typed_fields["enriched"] = (True, "bool")
            lines.append(build_line_typed(LineBuilder, "cpu_enriched", tags=tags, typed_fields=typed_fields, time_ns=time_ns))
        write_data(influxdb3_local, lines)

If a batch can carry a column added after the schema was cached, check for unknown keys and call get_line_schema(..., refresh=True) before splitting. The README shows this.

Plugins that will use this

Adoption follows in a separate PR once 0.5.0 is tagged.

Use get_line_schema and split_row:

• geo_enrichment
• gapfill
• downsampler
• basic_transformation
• nori_regression

Use the schema lookup or the public typing helpers:

• resampler
• synthefy_forecasting
• sagemaker
• schema_validator
• simple_data_replicator
• influxdb_to_iceberg

Drop a local copy of add_field_with_type or a redundant tag re-read:

• kafka_subscriber
• amqp_subscriber
• mqtt_subscriber
• opcua
• state_change
• stateless_adtk_detector
• threshold_deadman_checks

🤖 Generated with Claude Code