@Rocky I'm trying to understand schema evolution in starrocks. If my starrocks table has 4 fields and incoming data ingesting to starrocks has 5 fields, is there a way currently which automatically updates schema by adding new field and write the data to starrocks table?

Hi! I'm Rocky. Dealing with changing data schemas is a common challenge, and here is how StarRocks handles this: Currently, StarRocks does not automatically update its native table schema (i.e., add a new column) just by detecting extra fields in an incoming data stream via standard ingestion methods like Stream Load or Routine Load. By default, if the data has more fields than the table, those extra fields are typically ignored (or might cause the load to fail if strict mode is enabled and mappings are misaligned). However, there are three primary ways to achieve the behavior you're looking for: 1. Flink CDC 3.0 (Recommended for Automation) The most robust way to get automatic schema evolution in StarRocks is by using the Flink CDC 3.0 framework with the StarRocks Pipeline Connector. * How it works: It can monitor the source database (like MySQL, PostgreSQL, etc.) or Kafka and automatically synchronize schema changes—including adding new columns—directly to your StarRocks tables without manual intervention. * Benefit: This is the "set it and forget it" solution for end-to-end schema synchronization. 2. Using the JSON Data Type If your incoming data is frequently changing or has "extra" fields you didn't predict, a common design pattern is to use a JSON column. * Strategy: Map the 4 known fields to dedicated columns and store the entire record (or just the extra fields) in a

JSON

type column. * Benefit: Since JSON is semi-structured, StarRocks can store the 5th, 6th, or Nth field inside that JSON blob without any

ALTER TABLE

commands. You can still query these fields efficiently using functions like

get_json_string()

or

json_query()

. StarRocks also has a "Flat JSON" optimization that provides performance close to native columns for JSON fields. 3. Fast Schema Evolution (Manual but Instant) If you prefer fixed columns, StarRocks supports Fast Schema Evolution (introduced in v3.2/v3.3). * How it works: While it doesn't trigger automatically, it makes the

ALTER TABLE ADD COLUMN

operation a metadata-only change. In many cases (especially in shared-data/cloud-native clusters), this is nearly instantaneous. * Workflow: You would detect the schema change in your orchestration layer (like Airflow or a custom script) and issue an

ALTER TABLE

command before the next load. Summary * For fully automated ingestion from a source: Use Flink CDC 3.0. * For high-flexibility without schema changes: Use the JSON data type. * For standard loading: You must manually (or via an external script) run

ALTER TABLE ADD COLUMN

to update the schema before the data is ingested into that specific column. Do you have a specific ingestion tool (like Kafka, Flink, or a Python script) in mind? I can give you more specific advice if you do! References • sql-reference/data-types/semi_structured/VARIANT.md • sql-reference/sql-statements/table_bucket_part_index/ALTER_TABLE.md • using_starrocks/Flat_json.md

I have kafka and I'm loading data to starrocks with sink connector. If fields are added it should add new fields to tables before loading? how can I do this? if not possible from sink connector atleast with some external script? and table wise control on this @Rocky

As of now, the StarRocks Kafka Sink Connector does not natively support "Auto Schema Evolution" (the ability to automatically detect new Kafka fields and run

ALTER TABLE ADD COLUMN

on the fly). This is currently a requested feature in the StarRocks community. However, you can achieve this functionality using one of the following three patterns, depending on how much automation vs. control you need. 1. The "JSON / semi-structured" approach (No Schema Changes Needed) This is the most popular way to handle unpredictable incoming fields without ever needing to run

ALTER TABLE

. * How it works: Instead of mapping every Kafka field to a dedicated StarRocks column, you create a column with the JSON data type. * The Loading: Configure your Sink Connector to load the entire JSON payload (or just the "extra" fields) into that column. * The Querying: StarRocks has highly optimized JSON processing ("Flat JSON"). You can query new fields immediately using

json_column->'new_field'

. * Control: You can decide on a per-table basis which tables should be "Strict" (dedicated columns) and which should be "Flexible" (JSON column). 2. External Script (Manual Automation) If you require physical columns for every field, you can use a sidecar script (e.g., in Python) to bridge the gap. * The Logic: 1. The script polls your Kafka Schema Registry (if you use one) or samples a few messages from the topic. 2. It queries StarRocks for the current table schema using

DESCRIBE <table_name>;

. 3. It compares the two. If a new field exists in Kafka but not in StarRocks, it issues:

sql
        ALTER TABLE <table_name> ADD COLUMN <new_field_name> <data_type>;

* Fast Schema Evolution: In StarRocks v3.2 and later, ensure Fast Schema Evolution is enabled (set

fast_schema_evolution = TRUE

in table properties). This makes the

ALTER

operation a metadata-only change, so it happens almost instantly without blocking data ingestion. * Table-wise Control: You can maintain a config file for your script that lists which table/topic pairs should be monitored for evolution. 3. Flink CDC 3.0 (Full Automation) If you are open to moving away from the Kafka Sink Connector to a Flink-based pipeline, Flink CDC 3.0 is the official solution for this. * How it works: The StarRocks Pipeline Connector in Flink CDC 3.0 is designed specifically to handle end-to-end schema evolution. * Benefit: When a column is added to the source (or Kafka schema), Flink automatically pauses the sink, executes the

ALTER TABLE

on StarRocks, and resumes the load. Summary Recommendation Requirement | Recommended Solution I want zero maintenance | Use a JSON column for the evolving data. I need physical columns + Sink Connector | Use an external Python script to poll Schema Registry and run

ALTER TABLE

. I want the most "Enterprise" automation | Switch to Flink CDC 3.0. Note on Table-wise Control: If you go with the Script or Flink approach, you have full control. You can enable evolution for

table_a

and

table_b

while leaving

table_c

strictly locked to a specific schema. Do you currently use a Schema Registry (like Confluent or Apicurio)? If so, I can provide a logic snippet for a monitoring script. Python snippet for the "External Script" approach. References • administration/management/FE_configuration.md • sql-reference/sql-statements/table_bucket_part_index/CREATE_TABLE.md • faq/operation_maintenance_faq.md • best_practices/query_tuning/schema_tuning.md • release_notes/release-3.3.md

I use schema registry @Rocky. Also, on JSON type, if I give specific column as JSON and individual columns coming from source, how does it ingest to JSON column?

To handle your scenario with the Kafka Sink Connector and a Schema Registry, here is the breakdown of how to map fields and automate the schema updates. 1. Ingesting Individual Fields into a JSON Column If you have 4 fixed columns and want the "extra" fields to land in a JSON column, you have two main strategies using the Sink Connector: Option A: Map the Entire Record to a JSON Column (Recommended) The simplest way is to store the entire incoming record as a JSON object in one column. You can then use StarRocks' Generated Columns to automatically pull out the 4 fields you care about into their own high-performance columns. * Table Schema:

sql
    CREATE TABLE my_table (
      col1 INT AS (data_json->'col1'), -- Automatically extracted
      col2 INT AS (data_json->'col2'),
      data_json JSON                   -- The entire raw message
    ) ...

* Connector Config: In your

connect-starrocks-sink.properties

, you don't need complex mapping. Just ensure

sink.properties.format=json

. The connector will send the whole record, and StarRocks will put it into the

data_json

column (assuming the keys match or you use

jsonpaths

). Option B: Partial Mapping with

jsonpaths

If you want to pick specific source fields to go into a single JSON column, the Sink Connector uses Stream Load headers. You can pass these via the

sink.properties.

prefix: * `sink.properties.jsonpaths`:

["$.col1", "$.col2", "$.col3", "$.col4", "$"]

* `sink.properties.columns`:

col1, col2, col3, col4, other_fields

* Result: The first 4 fields go to their columns, and the

$

(root) maps the entire JSON object into the

other_fields

column. ────────── 2. External Script with Schema Registry (Table-wise Control) Since the Sink Connector doesn't auto-run

ALTER TABLE

, you can use a Python sidecar script. This is the most common way to get "Physical Column Evolution" with the Kafka Connector. Conceptual Workflow: 1. Poll Registry: The script checks the Confluent Schema Registry for the latest version of the topic's schema. 2. Compare: It fetches the StarRocks schema (

DESCRIBE table_name

). 3. Alter: If a new field exists in the Registry but not in StarRocks, it executes the

ALTER TABLE

. 4. Control: You can maintain a small metadata table in StarRocks or a JSON config to tell the script which tables to "auto-evolve." Example Logic (Python Snippet):

python
import requests
import pymysql # StarRocks is MySQL protocol compatible

# Configuration
TABLES_TO_EVOLVE = ["user_logs", "orders"]
REGISTRY_URL = "<http://schema-registry:8081/subjects/{topic}-value/versions/latest>"

def sync_schema(table_name):
    # 1. Get Schema from Registry
    res = requests.get(REGISTRY_URL.format(topic=table_name)).json()
    kafka_fields = res['schema']['fields'] # Simplified for Avro
    
    # 2. Get StarRocks Columns
    db = pymysql.connect(host='fe_host', user='root', password='', database='db')
    cursor = db.cursor()
    cursor.execute(f"DESCRIBE {table_name}")
    sr_fields = [row[0] for row in cursor.fetchall()]

    # 3. Find New Fields & Alter
    for field in kafka_fields:
        name = field['name']
        if name not in sr_fields:
            # Map Avro types to StarRocks types (int -> INT, string -> VARCHAR, etc.)
            sr_type = map_type(field['type']) 
            cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {name} {sr_type}")
            print(f"Added {name} to {table_name}")

# Run this on a cron job or a listener

────────── 3. The New "VARIANT" Type (The "Lazy" Pro Way) Starting in StarRocks v3.2, there is a new

VARIANT

data type. It is designed exactly for your problem. * How it works: You define a column as type

VARIANT

. * The Magic: When you load data into a

VARIANT

column, StarRocks automatically detects the sub-fields and types. It stores them in a columnar format (for speed) but keeps the schema flexible.

@Rocky what is actual use of fast schema evolution? v2

To understand "Fast Schema Evolution v2" (introduced in StarRocks v4.1 for cloud-native tables), it helps to contrast it with the older, standard way StarRocks handled schema changes. In the past, adding or dropping a column was an asynchronous process. When you ran

ALTER TABLE

, StarRocks would start a background job to update metadata across every single "tablet" (data shard) in the cluster. If your table had thousands of partitions and tablets, this could take minutes or even hours, and you had to wait for it to finish before the new schema was fully active. The Actual Use of Fast Schema Evolution (v2) The primary goal of Fast Schema Evolution v2 is to make schema changes instant and synchronous, behaving more like a traditional operational database (like MySQL or Postgres). 1. Instant Success (Sub-second Latency) * Old Way (Async): You run

ALTER TABLE

, get a "job started" message, and have to poll

SHOW ALTER TABLE COLUMN

to see if it’s done. * New Way (v2 Synchronous): You run

ALTER TABLE

, and the command returns "OK" almost immediately (sub-second). The change is applied immediately to the Frontend (FE) metadata. 2. No Tablet Metadata Updates In v2, StarRocks only updates the global metadata in the Frontend. It does not go out and update the individual metadata files stored on S3/Object Storage for every tablet. * Why this matters: Even if you have a 100PB table with 100,000 tablets, the

ALTER TABLE

still takes the same amount of time (milliseconds). 3. Zero Data Rewrite Fast Schema Evolution (both v1 and v2) ensures that StarRocks never rewrites your actual data files when you add or drop a column. * Adding a column: It simply notes the new column in the metadata with its default value. * Dropping a column: It marks the column as "hidden" in the metadata so it’s ignored during queries. ────────── How this solves your "Kafka + Schema Registry" problem Earlier, you asked about automatically adding fields from Kafka. Fast Schema Evolution v2 is the missing piece for that automation: 1. Safety for Automation: Because the change is now synchronous and instant, your external Python script can run

ALTER TABLE

, wait for the "OK" (which happens in 50ms), and then immediately tell the Kafka Sink to resume. There is no "waiting period" where your ingestion might be in an inconsistent state. 2. Zero Resource Impact: Since it only touches FE metadata, running an

ALTER TABLE

doesn't spike your CPU or IO on your compute nodes. You can evolve your schema 10 times a day without impacting query performance. How to enable it * For Cloud-Native (Shared-Data) Tables: Since v4.1, it is enabled by default for new tables. For existing tables, you can enable it via:

sql
    ALTER TABLE my_table SET ("cloud_native_fast_schema_evolution_v2" = "true");

* For Shared-Nothing (Local Storage) Tables: This is controlled by the

fast_schema_evolution

property (introduced in v3.2). While not called "v2" in the property name, it provides the same "no data rewrite" benefit, though it remains an asynchronous background process compared to the synchronous Cloud-Native v2. Summary: You use Fast Schema Evolution v2 when you need predictable, instant schema changes that don't depend on the size of your data, and when you want to automate your data pipeline without worrying about long-running background DDL jobs. References • administration/management/FE_configuration.md • sql-reference/sql-statements/table_bucket_part_index/CREATE_TABLE.md • best_practices/query_tuning/schema_tuning.md • release_notes/release-3.2.md