http://starrocks.io logo
Join Slack
Powered by
<@U0A71G31CDV> I'm trying partial update from star...
# questions-and-troubleshooting
m
@Rocky I'm trying partial update from starrocks sink kafka connector, it is mandatory asking for sink.properties.columns - If I specify all cols here and if I get only two keys in kafka topic (with id as pk mandatory), the same id it is not doing partial update, it is adding other keys as null which are not part of kafka message. Am I missing something?
r
To perform a partial update via the StarRocks Kafka Connector, you must ensure that your configuration correctly identifies which columns are being updated. If you specify all columns in 
sink.properties.columns
, StarRocks expects values for all of them. For any column listed in that property but missing from the Kafka message, StarRocks will insert a 
NULL
(or the column's default value), effectively overwriting your existing data. How to fix this 1. Restrict the column list: In your Kafka connector configuration, set 
sink.properties.columns
to only include the Primary Key and the specific columns you intend to update. * Example: If your table has 10 columns but you only want to update 
id
and 
price
, set: 
"sink.properties.columns": "id, price"
2. Enable Partial Update: Ensure you have the following properties set: * 
"sink.properties.partial_update": "true"
* 
"sink.properties.partial_update_mode": "row"
(This is the default and usually best for real-time Kafka streams). Key points to remember * Column Matching: StarRocks matches the keys in your Kafka JSON message to the names provided in 
sink.properties.columns
. If a column is listed in the configuration but absent in the JSON, it is treated as a 
NULL
update. * Primary Key Table: Partial updates are only supported on the Primary Key model. * Static Configuration: Because 
sink.properties.columns
is a static configuration in the Kafka connector, all messages in that specific topic/task should ideally contain the same set of columns. If different messages contain different subsets of columns, you may need to use multiple connectors/topics or a transformation layer (like Flink or a Kafka Streams app) to standardize the schema. Example Configuration Snippet
Copy code
json
{
    "name": "starrocks-sink",
    "config": {
        "connector.class": "com.starrocks.connector.kafka.StarRocksSinkConnector",
        "starrocks.database.name": "your_db",
        "starrocks.table.name": "your_table",
        "sink.properties.partial_update": "true",
        "sink.properties.partial_update_mode": "row",
        "sink.properties.columns": "id,col1,col2" 
    }
}
In this example, even if the table has 
col3
and 
col4
, they will remain untouched because they are not listed in 
sink.properties.columns
. References • loading/Kafka-connector-starrocks.md • loading/Flink-connector-starrocks.md • loading/Spark-connector-starrocks.md
m
@Rocky here is my connector config
Copy code
{
  "connector.class": "com.starrocks.connector.kafka.StarRocksSinkConnector",
  "starrocks.password": "******",
  "starrocks.topic2table.map": "mohan-test:mohan_partial_update",
  "tasks.max": "1",
  "transforms.kafkaTransform.convert.fields.to.snake_case": "true",
  "transforms.kafkaTransform.type": "com.nobroker.kafka.connect.transforms.KafkaStarRocksTransformPassNullableDates",
  "transforms": "kafkaTransform",
  "sink.properties.strip_outer_array": "true",
  "sink.properties.ignore_json_size": "true",
  "transforms.kafkaTransform.topic2table.phonenumber.fields": "",
  "starrocks.database.name": "test",
  "sink.properties.partial_update_mode": "row",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "transforms.kafkaTransform.topic2epochmillis.fields": "",
  "transforms.kafkaTransform.topic2epochseconds.fields": "",
  "sink.properties.columns": "id,name,status,date",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "starrocks.username": "connector",
  "bufferflush.intervalms": "600000",
  "topics": "mohan-test",
  "bufferflush.maxbytes": "3073741824",
  "transforms.kafkaTransform.topic2table.phonenumber.skip_country_code.fields": "",
  "transforms.kafkaTransform.topic2ISODate.fields": "",
  "sink.properties.partial_update": "true",
  "value.converter.schemas.enable": "false",
  "name": "mohan-test-partial-updates",
  "starrocks.http.url": "10.142.0.11:8080",
  "transforms.kafkaTransform.topic2table.other.sensitive.fields": ""
}
and my table CREATE TABLE 
mohan_partial_update
( 
id
varchar(100) NOT NULL COMMENT "", 
name
varchar(100) NULL COMMENT "", 
status
varchar(50) NULL COMMENT "", 
date
datetime NULL COMMENT "" ) ENGINE=OLAP PRIMARY KEY(
id
) COMMENT "OLAP" DISTRIBUTED BY HASH(
id
) BUCKETS 1 PROPERTIES ( "compression" = "LZ4", "datacache.enable" = "true", "enable_async_write_back" = "false", "enable_persistent_index" = "true", "file_bundling" = "true", "persistent_index_type" = "CLOUD_NATIVE", "replication_num" = "1", "storage_volume" = "nob_starrocks_prod" ); I already have this record user_1003 | NULL | PENDING | NULL My new kafka message { "id": "user_1003", "name": "Jai" } Now, I'm expecting Jai should get added to name for same id user_1003 without touching the status and other cols.
r
According to the StarRocks documentation and the behavior of the Primary Key model's partial updates, here is what is happening and how to address it. Why you are seeing 
NULL
overwrites When you perform a partial update using the Kafka Connector (which uses Stream Load internally), StarRocks defines the "update scope" based on the 
columns
parameter in your configuration. 1. Fixed Update Scope: By setting 
"sink.properties.columns": "id,name,status,date"
, you are telling StarRocks: "For every record in this batch, update these four columns." 2. Missing Fields = NULL: If a specific Kafka message only contains 
id
and 
name
, StarRocks sees that 
status
and 
date
are missing from the JSON for that record. Because they are explicitly listed in your 
columns
header, StarRocks defaults them to 
NULL
(or the column's default value) and performs the update. This overwrites your existing 
PENDING
status with 
NULL
. 3. Static Header: The Kafka Connector sends data in batches. It uses a single HTTP header for the entire batch. It does not dynamically change the 
columns
header on a per-message basis. How to achieve your goal To update only the 
name
column and leave 
status
and 
date
untouched, you must remove them from the 
columns
list:
Copy code
json
"sink.properties.columns": "id,name",
"sink.properties.partial_update": "true"
With this configuration: * Only 
id
and 
name
are in the update scope. * Existing values for 
status
and 
date
will remain unchanged. The "Dynamic Schema" Limitation If your Kafka topic contains mixed messages (e.g., some messages update 
name
, while others update 
status
), you will run into a limitation: * Because 
sink.properties.columns
is a static configuration, you cannot currently update different sets of columns using a single connector instance if you want to avoid 
NULL
overwrites for the missing fields. * StarRocks Stream Load requires that any column you do NOT want to touch must be omitted from the 
columns
header. Recommended Workarounds If your messages have varying fields, consider these options: 1. Multiple Connectors/Topics: Route messages with different schemas to different topics, each with its own Kafka Sink Connector configured with the appropriate 
sink.properties.columns
. 2. Use a Transformation Layer (Flink): If you use the StarRocks Flink Connector, it is better equipped to handle dynamic partial updates or record-level filtering. 3. Standardize the Payload: Ensure that the system producing the Kafka messages always includes the "current" values for all columns if you must list them all in the connector. 4. Remove 
columns
property?: You mentioned it feels mandatory. In StarRocks, if you omit 
columns
for a JSON load, it defaults to all columns in the table. This would still result in the same problem (missing fields becoming 
NULL
) because the update scope would effectively be every column in your table. Summary: To stop the 
NULL
overwrite, change your 
sink.properties.columns
to include only the primary key and the specific fields you are sending in that stream. References • loading/Kafka-connector-starrocks.md • sql-reference/sql-statements/loading_unloading/STREAM_LOAD.md • loading/Json_loading.md References • loading/Json_loading.md • sql-reference/sql-statements/loading_unloading/STREAM_LOAD.md
m
@Rocky If I give only id,name in the sink columns but if I receive some messages with all four keys, I want to insert it into table. here also, it is inserting only id,name and rest all null. There is no partial update here, it is working like normal pk table.
r
The behavior you are experiencing is a known design characteristic of how the StarRocks Kafka Connector (and the underlying Stream Load process) handles partial updates. Why this is happening 1. Static Update Scope: The 
sink.properties.columns
configuration defines the update scope for every message in the batch. By setting it to 
id, name, status, date
, you are instructing StarRocks to update all four columns for every record it processes. 2. JSON Handling of Missing Fields: When StarRocks parses a JSON message (like your 
{id, name}
payload) against a fixed column list (
id, name, status, date
), it finds that 
status
and 
date
are missing. In StarRocks Stream Load, missing fields for columns included in the 
columns
list are automatically treated as 
NULL
. 3. The Result: Because 
status
and 
date
are in the "update scope" but "missing" in the payload, they are updated to 
NULL
, overwriting your existing data. Why you can't just send all 4 keys As you noticed, if you restrict the 
columns
property to 
id, name
, then even if you send a message with all four keys, StarRocks will ignore 
status
and 
date
because they are outside the update scope defined in the connector's header. How to solve this The Kafka Connector currently uses a static header for an entire batch of messages. To achieve "dynamic" partial updates (where you only update what is present in the JSON), you have a few options: 1. Standardize the Payload (Recommended for Performance) If possible, modify your upstream producer to send the "full" current state of the record. This avoids the need for partial updates entirely and is the most performant way to use StarRocks. 2. Use Multiple Connectors If your Kafka topic contains a predictable variety of schemas (e.g., one type of message updates 
name
, another updates 
status
), you can route these to different topics (or use a SMT to filter) and create separate Sink Connectors for each, with specific 
sink.properties.columns
for each. 3. Transition to the Flink Connector The StarRocks Flink Connector is significantly more flexible for these scenarios. In Flink, you can define your sink logic to handle record-level differences more gracefully, and it has better support for dynamic partial updates where you only send the fields that have actually changed. 4. Force Single-Record Batches (Not recommended for high volume) You could theoretically set 
bufferflush.maxrows: 1
and 
bufferflush.maxbytes: 1
, and then create a custom Kafka Connect Transform (SMT) that dynamically sets the 
sink.properties.columns
for each record. However, this is highly discouraged as it would create a massive number of tiny Stream Load transactions, which will likely crash or severely slow down your StarRocks cluster. Summary of the "Missing" Piece You aren't missing a setting; rather, you've hit a limitation of the static nature of the Kafka Connector's Stream Load headers. For a column to not be overwritten by a partial update, it must be omitted from the 
sink.properties.columns
list. Since the connector's list is static for the whole topic, it currently cannot support "Update only the keys present in this specific JSON" if those keys vary from message to message. References • loading/Kafka-connector-starrocks.md • loading/Load_to_Primary_Key_tables.md • sql-reference/sql-statements/loading_unloading/STREAM_LOAD.md References • loading/Json_loading.md • loading/StreamLoad.md • sql-reference/sql-statements/loading_unloading/STREAM_LOAD.md
m
@Rocky with flink, how this can be handled?
r
With Apache Flink, you encounter a similar challenge because Flink SQL also relies on a fixed schema (DDL), but you have more advanced patterns available to solve the problem of "dynamic" partial updates. Why Flink has the same limitation by default When you define a StarRocks Sink in Flink SQL, the DDL schema determines the 
columns
header sent to StarRocks. * If your Flink DDL includes 
id, name, status, date
, the connector will send all four fields in every batch. * If a Flink record only has values for 
id
and 
name
, the 
status
and 
date
fields will be 
NULL
in the Flink row and will overwrite your existing StarRocks data. To handle cases where you want to update only the fields present in the message (and those fields vary from message to message), you can use the following strategies: 1. The Lookup Join Pattern (Recommended) This is the most common way to handle dynamic partial updates in Flink. Instead of doing a partial update in StarRocks, you "complete" the record in Flink before sending it. 1. Lookup: For every incoming Kafka message, Flink performs a lookup against the StarRocks table using the Primary Key. 2. Merge: In Flink, you compare the incoming fields with the existing fields. If the incoming field is missing or null, you keep the value from the lookup. 3. Sink: Send the "full" record to StarRocks. Pros: No 
NULL
overwrites; works with any StarRocks version. Cons: Adds latency and load to StarRocks due to lookups (mitigated by Flink's lookup cache). 2. Multiple Sinks with Filtering If you only have a few common "update types" (e.g., some messages only update 
name
, some only update 
status
), you can split the stream in Flink: * Path A: Filter for messages that have 
name
. Sink to a StarRocks table defined with only 
id
and 
name
. * Path B: Filter for messages that have 
status
. Sink to a StarRocks table defined with only 
id
and 
status
. Set 
'sink.properties.partial_update' = 'true'
on both. Because the DDLs are different, the 
columns
header will only include the relevant fields for each path. 3. Use the StarRocks JSON Data Type If your schema is highly volatile (keys change constantly), consider changing your StarRocks table to use the 
JSON
type for the data fields:
Copy code
sql
CREATE TABLE users (
    id VARCHAR(100) PRIMARY KEY,
    data JSON
);
In Flink, you can then use 
json_set()
or 
json_merge()
logic. This allows you to update specific keys within the JSON object without touching others, but it requires changing your table structure. 4. Wait for / Use "Flexible Partial Update" (v3.4+) Starting from StarRocks v3.4, there is improved support for partial updates via 
INSERT INTO
 and better handling of JSON. If you are on a very recent version, check if you can use the 
partial_update_mode
= 
column
which, combined with specific ingest methods, can be more flexible. However, for most Flink/Kafka connector use cases, the behavior remains that any column included in the ingest scope but missing in the record will be treated as NULL. Summary for Flink Implementation The most robust "Flink-native" way to solve this is to define your Sink DDL to match only the fields you know you will always update, or use a Lookup Join to fetch existing values so you never send a 
NULL
for a field you want to preserve. References • loading/Flink-connector-starrocks.md • loading/Load_to_Primary_Key_tables.md • sql-reference/sql-statements/loading_unloading/STREAM_LOAD.md • using_starrocks/Flat_json.md References • loading/Load_to_Primary_Key_tables.md
m
@Rocky I'm using 4.0.8 version of starrocks, to implement this, suggest the best possible approach, should work dyamically for any col and should handle scale as well.
r
To implement dynamic partial updates that handle scale and varying columns in StarRocks (specifically for version 3.3+ and the version 4.0.x you are using), the best approach is to switch from Row Mode to Column Mode. The Solution: Column-Mode Partial Update In Row Mode (which you are currently using), StarRocks requires a fixed set of columns to update. Any field missing from the JSON but listed in your config is explicitly overwritten with 
NULL
. In Column Mode, StarRocks dynamically determines the update scope based on the keys present in each JSON message. If a key (like 
status
) is missing from the JSON, StarRocks will simply not touch that column in the database. Updated Connector Config You need to make three specific changes to your configuration: 1. Change 
partial_update_mode
to 
column
. 2. Remove the 
sink.properties.columns
property (so it doesn't force a static schema). 3. Ensure 
partial_update
is 
true
.
Copy code
json
{
    "connector.class": "com.starrocks.connector.kafka.StarRocksSinkConnector",
    "starrocks.topic2table.map": "mohan-test:mohan_partial_update",
    "starrocks.database.name": "test",
    "sink.properties.partial_update": "true",
    "sink.properties.partial_update_mode": "column", 
    "sink.properties.format": "json",
    "sink.properties.strip_outer_array": "true",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    ... (rest of your settings)
}
Why this works for your case: * Dynamic Column Handling: When you send 
{"id": "user_1003", "name": "Jai"}
, StarRocks sees only 
id
and 
name
. Because you are in 
column
mode, it performs a partial update only on the 
name
column. The 
status
and 
date
columns for 
user_1003
are left exactly as they were. * Scale: Column mode is highly optimized for performance. While it is slightly more CPU-intensive on the storage engine (BE) during the write phase compared to row mode, it scales better for tables with many columns where you only update a small subset at a time. * Automatic Schema Mapping: By removing the 
sink.properties.columns
header, the connector relies on StarRocks' ability to map JSON keys directly to table column names. Important Requirements 1. Primary Key Model: Your table is already a Primary Key model, which is required for this feature. 2. Persistent Index: You already have 
enable_persistent_index
= 
true
, which is highly recommended for maintaining performance at scale during partial updates. 3. Column Alignment: Ensure the keys in your Kafka JSON exactly match the column names in your StarRocks table (case-sensitive). Summary: Switch to 
"sink.properties.partial_update_mode": "column"
and remove the static 
columns
definition. This will make your ingest process "key-aware" and prevent 
NULL
overwrites. References • loading/Load_to_Primary_Key_tables.md • loading/Json_loading.md • sql-reference/sql-statements/table_bucket_part_index/UPDATE.md • sql-reference/sql-statements/loading_unloading/BROKER_LOAD.md • loading/Kafka-connector-starrocks.md • sql-reference/sql-functions/json-functions/json-query-and-processing-functions/json_set.md • sql-reference/System_variable.md
m
@Rocky If I remove sink columns in connector with column mode as well, giving same error responseBody: { "Status": "ANALYSIS_ERROR", "Message": "key column id not in partial update columns", "Db": "test", "Table": "mohan_partial_update" }
r
The error 
ANALYSIS_ERROR: key column id not in partial update columns
occurs because StarRocks requires an explicit list of columns for any partial update request to validate that the Primary Key is present. You cannot remove the 
columns
property entirely when using partial updates. To achieve your goal in StarRocks 4.0.8 (where different messages update different fields without overwriting others with 
NULL
), you must use Column-Mode Partial Update. The Solution: "Full" Column List + Column Mode In Column Mode, StarRocks 3.3+ (including your 4.0.x version) treats missing fields in a JSON record differently than Row Mode. Instead of setting them to 
NULL
, it simply skips those columns for that specific record, preserving the existing data in the table. Update your configuration with these specific settings: 1. 
sink.properties.columns
: List all columns that could potentially be updated across your different message types (including the primary key 
id
). 2. 
sink.properties.partial_update
: 
true
3. 
sink.properties.partial_update_mode
: 
column
(This is the critical switch). Example Connector Config:
Copy code
json
{
    "connector.class": "com.starrocks.connector.kafka.StarRocksSinkConnector",
    "starrocks.topic2table.map": "mohan-test:mohan_partial_update",
    "starrocks.database.name": "test",
    "sink.properties.partial_update": "true",
    "sink.properties.partial_update_mode": "column",
    "sink.properties.columns": "id, name, status, date", 
    "sink.properties.format": "json",
    "sink.properties.strip_outer_array": "true",
    ...
}
Why this works for your case: * Message A (
{id, name}
): StarRocks matches 
id
and 
name
. Because you are in 
column
mode, it performs a partial update only on the 
name
column. Even though 
status
and 
date
are in the 
columns
header, StarRocks sees they are missing in the JSON and does not touch them. * Message B (
{id, name, status, date}
): StarRocks sees all four keys and updates all four columns. * The Error: By including 
id
in the 
sink.properties.columns
list, you satisfy the FE's requirement that the Primary Key must be part of the update scope. Important Note on Column Mode Column Mode is specifically designed for scenarios where you update a subset of columns. It is highly efficient for tables with many columns. However, please ensure that your table is a Primary Key model and that 
enable_persistent_index
is set to 
true
(which is the default in newer versions) for the best performance at scale. References • loading/Load_to_Primary_Key_tables.md • loading/Kafka-connector-starrocks.md • sql-reference/sql-statements/loading_unloading/STREAM_LOAD.md • administration/management/monitoring/metrics.md References • loading/Load_to_Primary_Key_tables.md • sql-reference/sql-functions/json-functions/json-query-and-processing-functions/json_set.md • loading/Json_loading.md • using_starrocks/Flat_json.md • faq/Others.md
m
@Rocky here is the updated config, { "connector.class": "com.starrocks.connector.kafka.StarRocksSinkConnector", "starrocks.password": "******", "starrocks.topic2table.map": "mohan-test:mohan_partial_update", "tasks.max": "1", "transforms.kafkaTransform.convert.fields.to.snake_case": "true", "transforms.kafkaTransform.type": "com.nobroker.kafka.connect.transforms.KafkaStarRocksTransformPassNullableDates", "transforms": "kafkaTransform", "sink.properties.strip_outer_array": "true", "sink.properties.ignore_json_size": "true", "transforms.kafkaTransform.topic2table.phonenumber.fields": "", "starrocks.database.name": "test", "sink.properties.partial_update_mode": "column", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "sink.properties.columns": "id,name,status,date", "transforms.kafkaTransform.topic2epochmillis.fields": "", "transforms.kafkaTransform.topic2epochseconds.fields": "", "key.converter": "org.apache.kafka.connect.storage.StringConverter", "starrocks.username": "connector", "bufferflush.intervalms": "600000", "topics": "mohan-test", "bufferflush.maxbytes": "3073741824", "transforms.kafkaTransform.topic2table.phonenumber.skip_country_code.fields": "", "transforms.kafkaTransform.topic2ISODate.fields": "", "sink.properties.partial_update": "true", "value.converter.schemas.enable": "false", "name": "mohan-test-partial-updates", "starrocks.http.url": "10.142.0.11:8080", "transforms.kafkaTransform.topic2table.other.sensitive.fields": "" }, same is happening like how it happened for partial udpate mode - row. If I pass all cols, inserting all. If I pass only 2, it is not partial updating, it is just adding these 2keys in table.
r
The error 
key column id not in partial update columns
confirms that for any partial update (even dynamic ones), StarRocks requires the 
columns
header to at least contain the Primary Key. However, since you are on StarRocks 4.0.8, the behavior you are seeing—where missing fields are still being overwritten with 
NULL
even in 
column
mode—is almost certainly due to one of the following two reasons. 1. The "Explicit Null" Problem (Most Likely) In StarRocks 
partial_update_mode = column
, the logic is: * Key is missing from the JSON: The column is not updated (existing data is preserved). * Key is present but value is 
null
: The column is updated to 
NULL
. Because you are using a custom Kafka transform (
KafkaStarRocksTransformPassNullableDates
), it is very likely that the transform or the 
JsonConverter
is producing a JSON payload that looks like this: 
{"id": "123", "name": "Jai", "status": null, "date": null}
Even though you only "passed 2 keys" into your Kafka record, the Kafka Connect pipeline often fills the rest of the schema with 
null
. When StarRocks receives an explicit 
null
for 
status
, it overwrites the previous value. To fix this: You must ensure the JSON actually sent to StarRocks is missing the keys. You may need to add a transform to your Kafka Connect config to drop null fields, such as the community-standard 
org.apache.kafka.connect.transforms.ReplaceField$Value
(though this is usually for specific names) or a custom SMT that removes entries from the Map where the value is null. 2. The "New Record" Behavior Check if the record you are updating already exists in the table. * If the 
id
exists, 
column
mode will preserve other fields. * If the 
id
does not exist, StarRocks will insert a new row. Since your JSON only has 
id
and 
name
, the other columns (
status
, 
date
) have no data to pull from, so StarRocks fills them with their Default Values (which is usually 
NULL
unless you defined the table otherwise). 3. Verification of Configuration Your current config is correct for version 4.0.8, but ensure these two things are true: 1. Check the Table Definition: Ensure the table is a Primary Key table. 2. Remove 
jsonpaths
: Ensure you are not using 
sink.properties.jsonpaths
. If 
jsonpaths
is defined, it forces a static mapping for every row, which breaks the "dynamic" nature of column mode. Recommended "Scale-Ready" Setup for 4.0.8: To handle any column dynamically without 
NULL
overwrites: 1. Table:
Copy code
sql
    CREATE TABLE mohan_partial_update (
        id VARCHAR(100) PRIMARY KEY,
        name VARCHAR(100),
        status VARCHAR(100),
        date DATETIME
    ) DISTRIBUTED BY HASH(id)
    PROPERTIES (
        "enable_persistent_index" = "true"
    );
2. Connector Config: * 
"sink.properties.partial_update": "true"
* 
"sink.properties.partial_update_mode": "column"
* 
"sink.properties.columns": "id,name,status,date"
(List all potential columns here). 3. The Payload: Ensure the JSON hitting StarRocks is actually 
{"id": "1", "name": "Jai"}
and not 
{"id": "1", "name": "Jai", "status": null}
. How to verify what the connector is sending: Check the StarRocks FE/BE logs or look at the "Load Profile" for one of the Sink tasks in the StarRocks web UI. If you see the "Rows" showing 
null
values for those columns in the source data, the issue is in your Kafka Connect transformation layer, not StarRocks. References • loading/Load_to_Primary_Key_tables.md • loading/Kafka-connector-starrocks.md • loading/Json_loading.md • sql-reference/sql-statements/loading_unloading/STREAM_LOAD.md
m
@Rocky kafka connect is not adding null values for keys. it is giving the same record without extra keys as null like 884 DEBUG [mohan-test-partial-updates|task-0] after updating value {date=2026-10-15 184500, name=Mohan, id=user_1002} (com.nobroker.kafka.connect.transforms.KafkaStarRocksTransformPassNullableDates) [task-thread-mohan-test-partial-updates-0]
r
To resolve the 
ANALYSIS_ERROR: key column id not in partial update columns
while achieving dynamic updates that don't overwrite missing fields with 
NULL
, you need to leverage the specific behavior of Column-Mode in StarRocks 3.3 and higher (including your version 4.0.8). The "Hybrid" Solution In StarRocks, the 
columns
header in a partial update request serves two purposes: 1. FE Validation: It must contain the Primary Key column to satisfy the safety check. 2. Scope Definition: In 
row
mode, it defines the exact list of columns; anything missing in the data is set to 
NULL
. However, in 
column
mode + JSON format, StarRocks uses the JSON keys as the source of truth for which columns to update, while the 
columns
header is used primarily to validate the presence of the Primary Key. Recommended Config Change: Update your 
sink.properties.columns
to only include the Primary Key. This satisfies the FE's requirement while preventing the Kafka Connector from potentially padding your records with 
null
values for the other columns.
Copy code
json
{
    "connector.class": "com.starrocks.connector.kafka.StarRocksSinkConnector",
    "starrocks.topic2table.map": "mohan-test:mohan_partial_update",
    "starrocks.database.name": "test",
    "sink.properties.partial_update": "true",
    "sink.properties.partial_update_mode": "column",
    "sink.properties.columns": "id", 
    "sink.properties.format": "json",
    "sink.properties.strip_outer_array": "true",
    ...
}
Why this is the best approach for scale and any column: 1. Dynamic Update: Since 
partial_update_mode
is 
column
, StarRocks will look at the keys in each individual JSON record. If a record has 
{"id": 1, "name": "Mohan"}
, it updates 
name
. If the next record has 
{"id": 1, "status": "Active"}
, it updates 
status
. 2. Prevents Overwrites: By removing 
name
, 
status
, and 
date
from the 
sink.properties.columns
header, you prevent the Kafka Connector from seeing a "schema" that it needs to fill. This ensures that the JSON sent to StarRocks is truly partial. 3. Handles Scale: Column mode is designed for high-performance updates. Coupled with 
enable_persistent_index = true
(which you have), this handles large-scale updates efficiently by only touching the modified columns on disk. Critical Verification Steps: If you still see 
NULL
overwrites after this change, please check the following: * Explicit Nulls: Ensure your source Kafka message or SMT (the 
KafkaStarRocksTransformPassNullableDates
transform) is not explicitly adding 
"status": null
to the record. If the key exists with a 
null
value, StarRocks will respect it and write a 
NULL
. * Case Sensitivity: Ensure the keys in your JSON (e.g., 
id
, 
name
) exactly match the column names in StarRocks. * Check the Load Profile: To be 100% sure what's happening, run a load and then check the profile in the StarRocks FE (e.g., 
SHOW LOAD
or via the web UI). Look at the 
StreamLoad
parameters to confirm 
partial_update_mode
is indeed 
column
and the 
Columns
header is what you expect. Summary: Set 
sink.properties.columns
to just your primary key (
id
) and keep 
partial_update_mode
as 
column
. This allows any other key present in the JSON to be updated dynamically without affecting columns not present in the message. References • loading/Load_to_Primary_Key_tables.md • loading/Kafka-connector-starrocks.md • loading/Json_loading.md • sql-reference/sql-statements/loading_unloading/STREAM_LOAD.md • using_starrocks/Flat_json.md References
m
@Rocky on adding only primary key column, it is trying to insert only that id column so no values are updating in the record of table.
r
To achieve a partial update that updates only the columns present in your Kafka message (and preserves existing values for missing fields), you must use Column-Mode Partial Update correctly. The reason you are seeing the behavior where adding only the Primary Key (
id
) updates nothing else, while adding all columns overwrites with 
NULL
, is because of how the 
columns
header interacts with JSON mapping in StarRocks. The Correct Configuration for Version 4.0.8 To perform a dynamic partial update where any column can be updated without affecting others, you must use the following configuration: 1. 
sink.properties.partial_update
: 
true
2. 
sink.properties.partial_update_mode
: 
column
(Required to prevent 
NULL
overwrites). 3. 
sink.properties.columns
: You must list every column that could potentially be updated in your pipeline. 4. Table Property: Ensure your table is created with 
"enable_persistent_index" = "true"
. This is a requirement for 
column
mode. Example Config:
Copy code
json
{
    ...
    "sink.properties.partial_update": "true",
    "sink.properties.partial_update_mode": "column",
    "sink.properties.columns": "id,name,status,date", 
    "sink.properties.format": "json",
    "sink.properties.strip_outer_array": "true",
    ...
}
Why your previous attempts failed: * Attempt with only 
id
in 
columns
: The 
columns
header in StarRocks acts as a filter/selector. If you only put 
id
there, StarRocks will only extract the 
id
field from your JSON and ignore 
name
, 
status
, etc., even if they are in the message. * Attempt with all columns in 
columns
(behaving like 
row
mode): If you provide the full list but StarRocks still overwrites with 
NULL
, it usually means the system has fallen back to 
row
mode. This happens if: * The Table is not a Primary Key model. * Persistent Index is disabled: Check your table DDL. If 
enable_persistent_index
is 
false
, 
column
mode is not supported and it defaults to 
row
mode (which fills missing fields with 
NULL
). * JSON Keys Case Sensitivity: Ensure the keys in your Kafka message (
id
, 
name
) match the case in the 
columns
header exactly. Troubleshooting & Verification 1. Check Table DDL: Run 
SHOW CREATE TABLE mohan_partial_update;
. Verify it is a 
PRIMARY KEY
table and 
enable_persistent_index
is 
true
. 2. Verify Headers: Check your StarRocks FE logs (
fe.audit.log
) for the 
StreamLoad
entry generated by the connector. Ensure the headers include 
partial_update: true
and 
partial_update_mode: column
. 3. JSON Payload: Ensure your Kafka transformation isn't accidentally sending explicit 
null
values (e.g., 
{"id":1, "status": null}
). In 
column
mode, a missing key preserves the old value, but an explicit 
null
 will overwrite the column with 
NULL
. Summary: Use the full list in 
sink.properties.columns
and ensure the table has 
enable_persistent_index = true
. This will tell StarRocks "I might update these 4 columns," and the 
column
mode will ensure only the ones found in the JSON are actually touched. References • loading/Load_to_Primary_Key_tables.md • loading/Kafka-connector-starrocks.md • sql-reference/sql-statements/loading_unloading/STREAM_LOAD.md • sql-reference/System_variable.md • table_design/table_types/primary_key_table.md References • deployment/shared_data/feature-support-shared-data.md • loading/Load_to_Primary_Key_tables.md
m
@Rocky any blogs that show handling partial updates in starrocks using flink or any other way?
r
Several blogs and technical guides demonstrate how to handle partial updates in StarRocks using various tools. Below is a curated list of resources, categorized by the ingestion method: 1. Flink & Kafka Ingestion Blogs * The Real-Time Data Journey: Connecting Flink, Airflow, and StarRocks (Medium) This blog specifically discusses the architecture for streaming data from Kafka to StarRocks using Flink. It details how to set 
"partial_update" = "true"
and use SMTs (Simple Message Transforms) like 
ReplaceField
to handle partial column updates. * Real-Time Analytics Using Apache Kafka and Apache Flink (StarRocks Official Blog) A high-level overview and webinar summary showing how StarRocks handles high-concurrency updates from Flink and Kafka topics, particularly useful for "upsert" scenarios. 2. General Best Practices & Ingestion * StarRocks Best Practices: Data Ingestion (StarRocks Official Blog) Written by a StarRocks contributor, this is part of a 5-part series. It covers the Primary Key model in depth and explains when to use Row mode (small batches, many columns) vs. Column mode (large batches, fewer columns) for partial updates. * Real-Time Data Enrichment with RisingWave and StarRocks (RisingWave Blog) This post explains how to use StarRocks as a sink for enriched streams. It highlights how StarRocks’ partial update feature allows you to join multiple streams into a single wide table without needing complex state management in the stream processor. 3. Developer & Documentation "How-To" Guides While not traditional blogs, these official "Continuous Loading" guides are structured as step-by-step tutorials: * Flink Connector Partial Update Guide: Shows the exact SQL/Properties needed to enable partial updates in Flink SQL. * Spark Connector Partial Update: Recently updated (as of v1.1.1) to support partial updates on Primary Key tables. Key Takeaway for your configuration: If you are transitioning from your current Kafka Connect setup to Flink, the Flink connector (v1.2.8+) can automatically detect partial updates based on the DDL schema you define in Flink. You don't necessarily have to manually list the columns in 
sink.properties.columns
if the Flink table schema already matches the subset of columns you want to update. Remember: For Column Mode (which you are using in 4.0.8), always ensure your table has 
"enable_persistent_index" = "true"
in its properties, or it will default to Row mode behavior.
Open in Slack
PreviousNext
Powered by