@Rocky is it possible to add kafka offset column to table via routine loading from kafka with avro schema?

Currently, StarRocks does not support adding Kafka metadata columns (such as

offset

,

partition

, or

timestamp

) to a table directly via Routine Load. Routine Load focuses on mapping fields from the message payload (CSV, JSON, or Avro) to your StarRocks table columns. There are no built-in system variables (like

*partition

or

*offset

) available in the

COLUMNS

or

jsonpaths

mapping syntax to capture these Kafka-specific properties. Alternatives If you must have the Kafka offset in your table, consider the following options: 1. Include Metadata in Payload: Modify your Kafka producer to include the partition and offset as fields within the Avro/JSON record itself. You can then map these fields like any other data column. 2. StarRocks Kafka Connector: If you use the StarRocks Kafka Connector (a Kafka Connect Sink connector) instead of Routine Load, you may have more flexibility through Kafka Connect transformations (SMTs) to pull metadata into the payload before it reaches StarRocks. 3. Audit Columns: While Routine Load can't capture the offset, you can still add a "load time" column using a default value or a function in the

COLUMNS

clause:

sql
    -- Example adding a load timestamp, but NOT the Kafka offset
    COLUMNS (col1, col2, load_time = now())

For now, the Routine Load

COLUMNS

parameter only allows mapping from the data source's content or using SQL functions to transform that content. References • quick_start/routine-load.md • loading/RoutineLoad.md • sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md

@Rocky give an example with generated load time column

To add a generated load time column when using Routine Load with Avro, you include the column in your table definition and then use the

COLUMNS

parameter in the

CREATE ROUTINE LOAD

statement to assign a value using the

now()

or

current_timestamp()

function. 1. Create the Target Table Include a column for the load timestamp (e.g.,

load_time

).

sql
CREATE TABLE sensor_data (
    id BIGINT,
    sensor_name VARCHAR(64),
    reading DOUBLE,
    load_time DATETIME -- The generated column
)
DISTRIBUTED BY HASH(id);

2. Create the Routine Load Job In the

COLUMNS

clause, list the fields that match your Avro schema keys, and then define the generated column with an assignment.

sql
CREATE ROUTINE LOAD sensor_load_job ON sensor_data
COLUMNS (
    id, 
    sensor_name, 
    reading, 
    load_time = now() -- Assigns the current time at the moment of loading
)
PROPERTIES (
    "format" = "avro"
)
FROM KAFKA (
    "kafka_broker_list" = "broker_host:9092",
    "confluent.schema.registry.url" = "<http://registry_host:8081>",
    "kafka_topic" = "sensor_topic"
);

How it works * Column Mapping: StarRocks maps the variable names in

COLUMNS

(like

id

,

sensor_name

) to the keys in your Avro record. * Transformation: The expression

load_time = now()

tells StarRocks that the

load_time

column in the table should not be pulled from the Kafka message, but should instead be populated by the result of the

now()

function for every row. * Avro Specifics: Since StarRocks v3.0.1, Avro loading uses the same column mapping and transformation logic as JSON. If your Avro field names differ from your table column names, you should also use the

jsonpaths

property to map the Avro schema paths to the variables used in the

COLUMNS

clause. References • sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md • loading/RoutineLoad.md References • sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md • loading/RoutineLoad.md

@Rocky generated column available in 3.3?

Yes, Generated Columns (the table schema feature) are available in StarRocks 3.3. This feature was introduced in version 3.1. However, there is an important distinction between a Generated Column (in the table definition) and the loading transformation example I gave previously. 1. Generated Columns (Table Schema) A Generated Column is defined when you create the table. Its value is automatically calculated based on other columns in the same row. * Availability: Since v3.1 (available in 3.3). * Limitation: The expression can only reference other regular columns in the table. It cannot reference Kafka metadata like

offset

because that metadata is not a column in the table. * Example:

sql
    CREATE TABLE test (
        price DOUBLE,
        quantity INT,
        total_cost DOUBLE AS price * quantity -- Generated column
    ) DISTRIBUTED BY HASH(price);

2. Loading Transformations (Routine Load) The example provided in the previous response (

load_time = now()

) is technically a loading transformation defined within the

CREATE ROUTINE LOAD

statement, not a "Generated Column" in the schema sense. * This is the standard way to "generate" a load timestamp if you don't want to change your table schema. * Limitation: Like Generated Columns, this transformation still cannot access Kafka

offset

or

partition

because StarRocks does not currently expose those metadata fields to the loading engine. 3. Comparison for your use case Feature | Supported in 3.3? | Can it get Kafka Offset? | Best Use Case Generated Column (

AS expr

) | Yes | No | Calculations based on other payload fields (e.g., extracting a field from a JSON string). Routine Load Mapping (

col = func()

) | Yes | No | Simple metadata like

now()

or constant values. Default Value (

DEFAULT now()

) | Yes | No | Standard auditing where every row gets a timestamp. Summary: While you can use Generated Columns in 3.3 to process your data, they do not solve the problem of capturing the Kafka offset. To get the offset into StarRocks, it must be part of the Kafka message payload sent by your producer. References • sql-reference/sql-statements/generated_columns.md • loading/RoutineLoad.md References • faq/Others.md • sql-reference/sql-statements/generated_columns.md • knowledge/usage/best-practice.md • sql-reference/sql-functions/date-time-functions/current_timestamp.md