http://starrocks.io logo
Join Slack
Powered by
<@U0A71G31CDV> is there any different in using DUP...
# questions-and-troubleshooting
d
@Rocky is there any different in using DUPLICATE KEY or ORDER BY? I understand in v3.3.0, ORDER BY became supported - just curious if there's any difference 🙂
r
Great question! Since StarRocks v3.3.0, you can use the 
ORDER BY
clause to define sort keys for Duplicate Key, Aggregate, and Unique Key tables (Primary Key tables have supported this since v3.0). While they both define the columns used for sorting data and building the Prefix Index, there are a few important differences in terms of flexibility and precedence. 1. Functional Difference: Precedence If you happen to specify both in your 
CREATE TABLE
statement, the 
ORDER BY
clause takes precedence. * Behavior: When 
ORDER BY
is present, the 
DUPLICATE KEY
clause is effectively ignored for sorting purposes. * Default: If neither is specified, StarRocks defaults to using the first three columns of the table as the sort key. 2. Management Difference: Ability to ALTER This is the biggest practical reason to use 
ORDER BY
. * `DUPLICATE KEY`: In older versions, the sort key was tightly coupled to the table model definition. Changing it usually required re-creating the table. * `ORDER BY`: By using the 
ORDER BY
syntax, you "decouple" the sorting logic from the table model. This allows you to change the sort key later without re-creating the table:
Copy code
sql
    ALTER TABLE my_table ORDER BY (new_col1, new_col2);
This is much more flexible as your query patterns evolve over time. 3. Syntax Consistency The introduction of 
ORDER BY
for Duplicate Key tables brings them into alignment with the Primary Key model. This provides a unified way to handle "Sort Keys" across all table types in StarRocks. 4. Constraints (Aggregate & Unique Models) While you asked about Duplicate Key tables specifically, it's worth noting the difference for other models: * Duplicate Key tables: The 
ORDER BY
can be any combination of any columns. * Aggregate/Unique Key tables: The 
ORDER BY
columns must include all the key columns (the ones defined in 
AGGREGATE KEY
or 
UNIQUE KEY
), but the columns in the 
ORDER BY
clause can be in a different order than they appear in the key definition. Summary Feature | 
DUPLICATE KEY
 | 
ORDER BY
 Primary Purpose | Defines table model & sort key | Defines sort key (decoupled) Precedence | Lower | Higher (Overrides Duplicate Key) Modifiable? | No (requires table recreation) | Yes (via 
ALTER TABLE
) Recommended? | Legacy/Standard | Recommended for v3.3+ Bottom line: In v3.3.0+, it is recommended to use 
ORDER BY
 because it gives you the freedom to optimize your prefix index later via 
ALTER TABLE
if your query patterns change! 🙂 change. References • release_notes/release-3.3.md
d
@Rocky here's a raw ingest table we have which is where all records pushed into our SR instance land. Should we discard BUCKETS 4 and let SR carry out random bucketing, or is it best to still define the buckets ourselves? /* Bronze: raw_ingest (append-only -> DUPLICATE KEY) */ CREATE TABLE IF NOT EXISTS raw_ingest ( integration_id VARCHAR(36) NOT NULL, integration_type VARCHAR(255) NOT NULL, record_type VARCHAR(255) NOT NULL, remote_id STRING NOT NULL, ingested_at DATETIME NOT NULL, payload STRING NOT NULL ) DISTRIBUTED BY HASH(integration_id) BUCKETS 4 ORDER BY (integration_id, integration_type, record_type, remote_id, ingested_at) PROPERTIES ("replication_num" = "1");
r
For a Bronze/Raw Ingest table in StarRocks v3.3.x, the short answer is: Yes, you should generally discard the hardcoded 
BUCKETS 4
 and let StarRocks manage it. Hardcoding a small number like 
BUCKETS 4
is often a bottleneck because it limits the ingestion and query parallelism to only 4 tablets, regardless of how many Backends (BEs) you have. Here are your two best paths forward: Option 1: Automatic Hash Bucketing (Recommended if you query by 
integration_id
) If your downstream ETL or troubleshooting queries frequently filter by 
integration_id
, stick with Hash distribution but remove the 
BUCKETS
keyword.
Copy code
sql
CREATE TABLE raw_ingest (
    ...
)
DISTRIBUTED BY HASH(integration_id) -- No BUCKETS specified
ORDER BY (...)
PROPERTIES ("replication_num" = "1");
* Why? Since v2.5.7, StarRocks uses Automatic Bucketing. It will calculate the ideal number of buckets based on your cluster size (BE count) and historical data volume. * Benefit: You keep Bucket Pruning (queries for a specific 
integration_id
only hit relevant tablets) while letting SR scale the parallelism for you. Option 2: Random Bucketing with Auto-Scaling (Recommended for pure landing tables) If this table is strictly for high-speed ingestion and you don't care about point-query performance (e.g., you just read everything for a batch ETL), use Random Bucketing with a 
bucket_size
property.
Copy code
sql
CREATE TABLE raw_ingest (
    ...
)
DISTRIBUTED BY RANDOM
ORDER BY (...)
PROPERTIES (
    "replication_num" = "1",
    "bucket_size" = "1073741824" -- 1GB target size per bucket
);
* Why? This is the most "hands-off" approach. By setting 
bucket_size
(available since v3.2), StarRocks will dynamically increase the number of buckets as the table grows. * Strengths: * Zero Skew: Even if one 
integration_id
has 90% of your data, it will be spread evenly across all BEs. * Elasticity: The table starts small and adds tablets automatically as data accumulates. * Weakness: Every query will perform a full scan of all tablets in a partition (no bucket pruning). Comparison for your use case: Feature | Manual (
BUCKETS 4
) | Auto Hash | Random + 
bucket_size
 Ingest Parallelism | Very Low (Fixed) | High (Cluster-aware) | Highest (Uniform spread) Data Skew Risk | High | Medium | None Bucket Pruning | Yes | Yes | No Maintenance | Manual | Automatic | Fully Elastic Final Recommendation: For a Bronze table, Option 2 (Random Bucketing + 
bucket_size
) is often the winner. It guarantees the most stable write performance and prevents "hot spots" in your cluster during heavy ingestion, which is usually the priority for raw landing zones. If you find yourself needing to do "point lookups" on 
integration_id
in this table, go with Option 1. Either way, delete 
BUCKETS 4
!** References • knowledge/usage/best-practice.md • table_design/data_distribution/feature-support-data-distribution.md • table_design/data_distribution/Data_distribution.md • best_practices/bucketing.md • administration/management/FE_configuration.md
d
@Rocky the primary query on this table in the MVs we use is by integration type + record type. i.e. we grab all 
deals
records from 
hubspot
, all 
opportunities
records from 
salesforce
, and push them into a 
dim_opportunities
table. Should we be therefore using 
ORDER BY (integration_type, record_type, integration_id, remote_id, ingested_at)
and distribute the has by integration type?
r
For your use case, your instinct on the 
ORDER BY
 change is spot on, but you should be very careful with the 
DISTRIBUTED BY
 change. Here is the breakdown of how to optimize this Bronze table for your Materialized Views (MVs): 1. `ORDER BY`: Yes, change the order You should definitely move 
integration_type
and 
record_type
to the front of your 
ORDER BY
clause. * Why: StarRocks uses the first few columns of the 
ORDER BY
clause to build a Prefix Index (a sparse index). * Performance: When your MVs run queries like 
WHERE integration_type = 'hubspot' AND record_type = 'deals'
, StarRocks can use the prefix index to skip 90%+ of the data on disk without even reading it. * Updated Order: 
ORDER BY (integration_type, record_type, integration_id, ...)
is perfect for your pattern. 2. `DISTRIBUTED BY`: No, do NOT hash by 
integration_type
 Distributing by 
integration_type
is risky and will likely lead to Data Skew. * The Cardinality Problem: Distribution keys should have high cardinality (thousands or millions of unique values). 
integration_type
likely only has a few dozen values (hubspot, salesforce, etc.). * Skewed Buckets: If 80% of your records come from 'hubspot', then 80% of your table's data will live in the specific buckets assigned to that hash value. One or two nodes in your cluster will do all the work, while the others sit idle. * Query Performance: You don't need to distribute by a column to filter by it efficiently. The 
ORDER BY
handles the filtering. The 
DISTRIBUTED BY
is strictly for spreading the load evenly across the cluster. Recommended Configuration For a Bronze ingest table where you need fast filtering for downstream MVs, use this:
Copy code
sql
CREATE TABLE raw_ingest (
    integration_id      VARCHAR(36)     NOT NULL,
    integration_type    VARCHAR(255)    NOT NULL,
    record_type         VARCHAR(255)    NOT NULL,
    remote_id           STRING          NOT NULL,
    ingested_at         DATETIME        NOT NULL,
    payload             STRING          NOT NULL
)
-- OPTION A: Best for even distribution (High cardinality)
DISTRIBUTED BY HASH(integration_id) 

-- OPTION B: Best if you want SR to handle everything automatically (v3.2+)
-- DISTRIBUTED BY RANDOM 

ORDER BY (integration_type, record_type, integration_id)
PROPERTIES (
    "replication_num" = "1",
    "bucket_size" = "1073741824" -- Optional: targets 1GB per bucket
);
Why this works best: 1. Filtering: Your MVs will be incredibly fast because 
integration_type
and 
record_type
are the first columns in the sort key. 2. Parallelism: By hashing on 
integration_id
(high cardinality) or using 
RANDOM
, you ensure that every node in your StarRocks cluster participates in the ingest and the query, preventing "hot" nodes. 3. Efficiency: Hashing on 
integration_id
is still useful if you ever need to do a join on that ID later, as it enables "Colocated Joins." Summary: Change the 
ORDER BY
to match your query filters, but keep the 
DISTRIBUTED BY
on a high-cardinality column (like 
integration_id
) to keep the cluster balanced! 🚀 References • knowledge/usage/best-practice.md • table_design/indexes/Prefix_index_sort_key.md • best_practices/query_tuning/schema_tuning.md
d
@Rocky so this is your recommendation? CREATE TABLE IF NOT EXISTS raw_ingest ( integration_id VARCHAR(36) NOT NULL, integration_type VARCHAR(255) NOT NULL, record_type VARCHAR(255) NOT NULL, remote_id STRING NOT NULL, ingested_at DATETIME NOT NULL, payload STRING NOT NULL ) DISTRIBUTED BY RANDOM ORDER BY (integration_type, record_type, integration_id, remote_id, ingested_at) PROPERTIES ("replication_num" = "1"); What's replication_num for here?
r
That DDL looks excellent for your use case! It solves your filtering performance (via 
ORDER BY
) and ensures even data distribution (via 
RANDOM
). To answer your question about 
replication_num
: What is 
replication_num
? It defines how many copies (replicas) of your data StarRocks stores across different Backend (BE) nodes. * The Default is 
3
: This is the StarRocks standard for production. If one node fails, your data remains available on the other two, and the cluster automatically heals itself by creating a new 3rd copy on a healthy node. * Your setting of 
"1"
: This means there is only one copy of the data. Should you use 
"1"
? For a Bronze (Raw Ingest) table, it depends on your tolerance for re-ingesting data: * When to use 
1
: * Development/Testing: If you are running a single-node StarRocks cluster for testing. * Easily Replaceable Data: If this table is just a temporary landing zone and you can easily re-run the ingest from your source (S3, Kafka, etc.) if a node goes down. It saves 66% on disk space! * When to use 
3
: * Production: If this table is a critical part of your pipeline and you don't want your MVs or downstream jobs to fail if a single hardware node has an issue. One final "Pro-Tip" for your DDL: Since you are using 
DISTRIBUTED BY RANDOM
in StarRocks v3.2+, I highly recommend adding the 
bucket_size
 property. Without it, StarRocks will pick a fixed number of buckets when you create the table. With it, the table will auto-scale (add more buckets) as your data grows:
Copy code
sql
CREATE TABLE IF NOT EXISTS raw_ingest (
    ...
)
DISTRIBUTED BY RANDOM
ORDER BY (integration_type, record_type, integration_id, remote_id, ingested_at)
PROPERTIES (
    "replication_num" = "3", -- Changed to 3 for safety, stick to 1 if disk is tight
    "bucket_size" = "1073741824" -- Targets ~1GB per bucket for auto-scaling
);
Summary: Change 
replication_num
to 
3
if this is a production environment. Keep the 
ORDER BY
and 
DISTRIBUTED BY RANDOM
as they are—they are a perfect fit for your MV patterns where you're headed! 🙂 want to go! smiling face References • sql-reference/sql-statements/table_bucket_part_index/CREATE_TABLE.md • administration/management/resource_management/Replica.md • administration/management/FE_configuration.md • knowledge/usage/best-practice.md
d
@Rocky Okay, similar evaluation for one of our later tables - dim_users. CREATE TABLE IF NOT EXISTS dim_users ( integration_id VARCHAR(36) NOT NULL, user_id STRING NOT NULL, created_at DATETIME NULL, modified_at DATETIME NOT NULL, deleted_at DATETIME NULL, first_name STRING NULL, last_name STRING NULL, email STRING NULL, job_title STRING NULL, properties STRING NOT NULL, custom_fields STRING NOT NULL, ingested_at DATETIME NOT NULL, materialized_at DATETIME NOT NULL ) PRIMARY KEY(integration_id, user_id) DISTRIBUTED BY HASH(integration_id) BUCKETS 4 ORDER BY (integration_id, user_id) PROPERTIES ("replication_num" = "1"); Based on your comments previously, we should remove 
BUCKETS 4
but set a max bucket size, and leave replication num at 1 (since the data is easily repopulated from raw_ingest)
r
For your 
dim_users
table (which uses the Primary Key model), the evaluation is slightly different than the Bronze table because Primary Key tables have stricter requirements. Here is the breakdown for your 
dim_users
table: 1. `DISTRIBUTED BY`: Must be HASH Unlike the Bronze table (Duplicate Key), Primary Key tables do not support 
RANDOM
bucketing. You must use 
HASH
. * Recommendation: Change the distribution key to include both 
integration_id
and 
user_id
. * Why? Hashing only on 
integration_id
creates a risk of Data Skew. If one integration has 1 million users and another has 10, those 1 million users will all end up in the same bucket. HASHing on both columns ensures a perfectly even spread across your cluster. 2. `BUCKETS`: Remove it (Auto-Bucketing) You should definitely remove 
BUCKETS 4
. * What happens? StarRocks will use Automatic Bucketing (v2.5.7+). For a non-partitioned table like this, it will typically set the bucket count based on your Backend (BE) count (e.g., 
2 * number of BEs
). This ensures that every node in your cluster helps process the data. 3. 
bucket_size
Property * Note: The 
bucket_size
property (which allows buckets to grow dynamically) is currently only supported for 
RANDOM
distribution (Duplicate Key tables). * For Primary Key tables: This property will be ignored. However, because you are removing the hardcoded 
BUCKETS 4
, StarRocks' auto-bucketing is smart enough to pick a much better default than 4. 4. 
ORDER BY
and 
PRIMARY KEY
 In StarRocks v3.0+, the Primary Key and Sort Key are decoupled. * Your 
ORDER BY (integration_id, user_id)
is perfectly fine. It tells StarRocks to use those columns for the Prefix Index, making lookups by 
integration_id
very fast. 5. 
replication_num
 * Since you mentioned this data is easily repopulated from 
raw_ingest
, leaving this at 
1
is a valid way to save 66% on disk space and speed up writes. Just be aware that if a node fails, this table will be unavailable until the node returns or you re-ingest. ────────── Final Recommended DDL
Copy code
sql
CREATE TABLE IF NOT EXISTS dim_users (
    integration_id      VARCHAR(36)     NOT NULL,
    user_id             STRING          NOT NULL,
    created_at          DATETIME        NULL,
    modified_at         DATETIME        NOT NULL,
    deleted_at          DATETIME        NULL,
    first_name          STRING          NULL,
    last_name           STRING          NULL,
    email               STRING          NULL,
    job_title           STRING          NULL,
    properties          STRING          NOT NULL,
    custom_fields       STRING          NOT NULL,
    ingested_at         DATETIME        NOT NULL,
    materialized_at     DATETIME        NOT NULL
)
PRIMARY KEY(integration_id, user_id)
-- Removed BUCKETS 4, Hashed on full PK to prevent skew
DISTRIBUTED BY HASH(integration_id, user_id) 
ORDER BY (integration_id, user_id)
PROPERTIES (
    "replication_num" = "1",
    "enable_persistent_index" = "true" -- Recommended for PK tables to save RAM
);
Key Takeaways: 1. Discard 
BUCKETS 4
 to let StarRocks scale the table to your cluster size. 2. Hash on both PK columns to avoid data skew. 3. Enable Persistent Index (
"enable_persistent_index" = "true"
) is the default in newer versions, but it's good practice to keep it. It keeps the primary key index on disk (with an SSD cache) so you don't run out of RAM as your user base grows. References • table_design/table_types/primary_key_table.md • table_design/data_distribution/Data_distribution.md • sql-reference/sql-statements/table_bucket_part_index/CREATE_TABLE.md
d
@Rocky Doesn't using DISTRIBUTED BY HASH(integration, user_id) basically eliminate grouping (since user_ids are mostly unique)?
r
It’s important to distinguish between Physical Distribution (Bucketing) and Logical Sorting (Ordering). You are correct that hashing on 
(integration_id, user_id)
will spread a single integration's data across many nodes, effectively "breaking up" that specific integration at the hardware level. Whether that is "good" or "bad" depends on your query volume and the size of your integrations. 1. The "Grouping" Tradeoff Strategy | 
DISTRIBUTED BY HASH(integration_id)
 | 
DISTRIBUTED BY HASH(integration_id, user_id)
 Data Locality | High. All users for "Hubspot" live on one node. | Low. "Hubspot" users are scattered across the cluster. Colocated Joins | Enabled. You can join 
dim_users
and 
dim_opportunities
on 
integration_id
without moving data. | Disabled. (Unless the other table also hashes on both). Concurrency | High. A query for one integration only hits one node, leaving other nodes free for other queries. | Low. Every query for one integration hits every node in the cluster. Data Skew Risk | High. If Hubspot has 90% of your data, one node will be 90% full and do 90% of the work. | Zero. Data is perfectly balanced. 2. When to use which? * Use 
HASH(integration_id)
if: You have thousands of unique `integration_id`s (high cardinality) and they are roughly the same size. This gives you the best of both worlds: even distribution across the cluster, but individual integration queries stay local to one node. * Use 
HASH(integration_id, user_id)
if: You only have a handful of integrations (e.g., just Hubspot and Salesforce) and one is significantly larger than the others. In this "low cardinality" scenario, hashing on only the ID will lead to Data Skew, which is the #1 cause of performance issues in StarRocks. 3. Does it break the Index? No. Your 
ORDER BY (integration_id, user_id)
ensures that within each bucket, the data is perfectly sorted and grouped. * If a query for 
WHERE integration_id = 'hubspot'
hits a node, the Prefix Index will still allow StarRocks to jump straight to the 'hubspot' records in that bucket. * The only difference is that instead of reading from 1 node, it reads from all nodes in parallel. Recommendation for `dim_users`: If you are a multi-tenant app with 10,000+ customers:
Copy code
sql
DISTRIBUTED BY HASH(integration_id) -- Best for concurrency & colocated joins
If you are an internal data team with only 5–10 big data sources (Hubspot, Salesforce, etc.):
Copy code
sql
DISTRIBUTED BY HASH(integration_id, user_id) -- Best for preventing skew and maximizing scan speed
Summary: You aren't losing the "grouping" for the index (that's what 
ORDER BY
is for), but you are trading "Data Locality" for "Parallelism." For a few large data sources, Parallelism is usually the winner. References • table_design/table_types/primary_key_table.md • table_design/data_distribution/Data_distribution.md
d
@Rocky roughly 500 integration instances where the most popular one (Salesforce for arguments sake) has about 130. Sounds like the first option is best
r
That logic is solid, but since you are dealing with a Primary Key table and specifically mentioned "130" (I assume that's 130 million rows or a similar large weight for one integration?), let's refine that "First Option" choice. Is 
HASH(integration_id)
really best for you? Given your setup (~500 unique integration instances), here is the final check to confirm that choice: 1. The Cardinality Rule: StarRocks generally prefers the number of unique hash values to be significantly higher than your number of Backend (BE) nodes (ideally 100x to 1000x). With 500 unique IDs, you are fine if you have a small/medium cluster (e.g., 3–10 nodes). 2. The Row-Count Skew (The "Salesforce" Problem): Even if you have 500 unique IDs, if that "Salesforce" instance (or the set of Salesforce instances) has 80% of the total rows in the table, hashing on 
integration_id
will make the BE nodes holding those IDs much "heavier" than the others. This leads to: * Uneven disk usage. * Slow queries (because the "heavy" node becomes the bottleneck). 3. Concurrency vs. Parallelism: * Choose 
HASH(integration_id)
 if you expect to have many users querying different integrations at the same time. This is "Concurrency." Each query stays on one node, and they don't interfere with each other. * Choose 
HASH(integration_id, user_id)
 if you are the main user and you want one query for "Salesforce" to run as fast as humanly possible by using every CPU on every node in your cluster. This is "Parallelism." Recommendation If your 500 integrations are roughly the same size (or at least no single one is 50%+ of the data), 
HASH(integration_id)
is the winner. It allows for Colocated Joins later (joining Users to Tickets/Opportunities without moving data over the network). Revised DDL for `dim_users`:
Copy code
sql
CREATE TABLE IF NOT EXISTS dim_users (
    integration_id      VARCHAR(36)     NOT NULL,
    user_id             STRING          NOT NULL,
    ... -- other columns
)
PRIMARY KEY(integration_id, user_id)
-- 1. Use HASH(integration_id) for locality/colocated joins
-- 2. Remove "BUCKETS 4" to let StarRocks auto-bucket based on node count
DISTRIBUTED BY HASH(integration_id) 
ORDER BY (integration_id, user_id)
PROPERTIES (
    "replication_num" = "1",
    "enable_persistent_index" = "true" 
);
One important correction on your bucket comment: You mentioned setting a "max bucket size" for this table. Just a heads-up: the 
bucket_size
property (dynamic auto-scaling) is currently only for Duplicate Key tables using RANDOM distribution. For this Primary Key table, simply omitting the 
BUCKETS
keyword is the way to go. StarRocks will automatically set the number of buckets based on your cluster size (usually 10-20 buckets per BE node) to ensure the data is balanced. References • table_design/data_distribution/Data_distribution.md • best_practices/bucketing.md • table_design/table_types/primary_key_table.md • sql-reference/sql-statements/table_bucket_part_index/CREATE_TABLE.md References • table_design/data_distribution/Data_distribution.md • best_practices/bucketing.md
d
@Rocky no sorry, I meant we have 500 integration instances, and of that 500, approximately 130 are of the same type (Salesforce). Distribution might look like 130 Salesforce integrations, 80 HubSpot ones, 50 Pipedrive ones, and so on. 
integration_id
is a unique value for each of those 500 instances.
r
Got it! That clarification makes a huge difference. Since you have 500 unique `integration_id`s, even if 130 of them are "Salesforce type," they are still 130 distinct hash values. The Verdict: 
DISTRIBUTED BY HASH(integration_id)
is the way to go. Here is why your specific distribution makes this the better choice: 1. Sufficient Cardinality: With 500 unique IDs, StarRocks can easily balance these across your Backend (BE) nodes. Even if you have a 10-node cluster, each node would handle roughly 50 integrations. The "130 Salesforce" integrations will be spread across those nodes based on their unique IDs, not clumped together. 2. Point Lookup Efficiency: If a user views a dashboard for one specific 
integration_id
, StarRocks only has to talk to one bucket (on one node). If you hashed on 
(integration_id, user_id)
, that same query would have to hit every node in the cluster to find the data, which wastes CPU cycles and reduces your total system capacity (concurrency). 3. Future-Proofing for Joins: You likely have other tables (like 
dim_opportunities
or 
dim_tickets
) that also have an 
integration_id
. If you bucket both by 
integration_id
, you can perform Colocated Joins. This allows StarRocks to join the tables locally on each node without sending massive amounts of data over the network. The "Salesforce Skew" Check The only remaining risk is Data Volume Skew. If one specific Salesforce customer has 50 million users and every other customer has only 1,000 users, the node holding that one big ID will work harder than the others. * If your integrations are "roughly" similar in size: Stick with 
HASH(integration_id)
. * If you have one "Mega-Customer" that is 100x larger than everyone else: You might consider 
HASH(integration_id, user_id)
just for this table, but you lose the Join optimizations mentioned above. Given most B2B datasets, the locality benefit usually outweighs the risk. Final Recommended DDL I’ve kept the Primary Key model and enabled the Persistent Index, which is highly recommended for Primary Key tables to keep RAM usage low as you scale to 500+ integrations.
Copy code
sql
CREATE TABLE IF NOT EXISTS dim_users (
    integration_id      VARCHAR(36)     NOT NULL,
    user_id             STRING          NOT NULL,
    created_at          DATETIME        NULL,
    modified_at         DATETIME        NOT NULL,
    deleted_at          DATETIME        NULL,
    first_name          STRING          NULL,
    last_name           STRING          NULL,
    email               STRING          NULL,
    job_title           STRING          NULL,
    properties          STRING          NOT NULL,
    custom_fields       STRING          NOT NULL,
    ingested_at         DATETIME        NOT NULL,
    materialized_at     DATETIME        NOT NULL
)
PRIMARY KEY(integration_id, user_id)
-- Hashing only on integration_id to enable Colocated Joins and Tablet Pruning
DISTRIBUTED BY HASH(integration_id) 
ORDER BY (integration_id, user_id)
PROPERTIES (
    "replication_num" = "1",
    "enable_persistent_index" = "true"
);
Key Reminders: * Buckets: By omitting 
BUCKETS X
, StarRocks will use Auto-Bucketing to pick the ideal count for your current cluster size. * Replication: Since you are at 
replication_num = 1
, ensure your 
raw_ingest
table is durable (e.g., has replication 3) so you can actually rebuild this if a node fails! References • table_design/data_distribution/Data_distribution.md • best_practices/bucketing.md • table_design/table_types/primary_key_table.md • sql-reference/sql-statements/table_bucket_part_index/CREATE_TABLE.md References
d
@Rocky great, thanks. How does using replication_num=2 fair? Sounds like a happy middleground for particular large tables.
r
Using 
replication_num=2
is often considered an "unhappy middle ground" in StarRocks. While it sounds like a logical compromise to save disk space, it actually creates a more fragile system for writes than 
replication_num=3
. Here is how it fairs in practice: 1. The "Majority" Math Problem (Availability) StarRocks uses a Write Quorum to ensure data consistency. By default, this is set to 
MAJORITY
. * For 3 replicas: Majority is 2. If one node goes down, you can still successfully load data because 2/3 replicas are healthy. * For 2 replicas: Majority is 2. This means if one node goes down, your writes (loads) will fail because you cannot reach the required majority. In this scenario, 
replication_num=2
is actually less available than 
replication_num=3
. Even though your data is "safe" on the remaining node, your pipeline stops. 2. The Durability Trap To fix the write issue above, you could set 
"write_quorum" = "ONE"
. This allows writes to succeed as long as one replica is alive. However: * If you write to only one replica and that node's disk fails before it can sync to the second node, you lose data. * With 
replication_num=3
and 
write_quorum=MAJORITY
, you always have at least 2 copies before a load is considered "finished." 3. Disk Space vs. Operational Overhead Strategy | Disk Usage | Load Availability | Data Safety Replication 1 | 100% | High (until the node dies) | None (Risk of total loss) Replication 2 | 200% | Low (Fails if 1 node is down) | Moderate Replication 3 | 300% | High (Tolerates 1 node down) | High When to use what? * Use `replication_num=1`: Only for staging tables or "disposable" data that you can easily re-ingest from an upstream source (like S3/Iceberg) if a node fails. * Use `replication_num=3`: For all production data. This is the industry standard for a reason; it provides the "Goldilocks" zone of being able to perform maintenance (restarting nodes) without stopping your data ingestion. * Avoid `replication_num=2`: Unless you have a very specific 2-node cluster constraint and are willing to accept that any single-node maintenance window will block all data loading. Summary: If you are trying to save space on large tables, it is much better to use Column Compression (Zstd), Data Tiering (moving cold data to S3), or TTL/Partitioning rather than dropping to 2 replicas. References • administration/management/resource_management/Replica.md • knowledge/usage/best-practice.md • sql-reference/sql-statements/table_bucket_part_index/CREATE_TABLE.md References • administration/management/resource_management/Replica.md • faq/operation_maintenance_faq.md • sql-reference/sql-statements/table_bucket_part_index/CREATE_TABLE.md • administration/management/FE_configuration.md • knowledge/usage/best-practice.md
d
@Rocky now looking at materialised views to get us from raw_ingest to say, dim_users, how would you improve the below? Note that it's selecting any integration type/record type combos from raw_ingest that represent users on the target system. DROP MATERIALIZED VIEW IF EXISTS mv_users; CREATE MATERIALIZED VIEW mv_users DISTRIBUTED BY HASH(integration_id) REFRESH ASYNC PROPERTIES ("replication_num" = "1") AS /* ---- Bullhorn corporateUser --------------------------------------------- */ SELECT bh.integration_id, COALESCE(NULLIF(get_json_string(bh.payload, '$.id'), ''), bh.remote_id) AS user_id, CAST(FROM_UNIXTIME(NULLIF(get_json_int(bh.payload, '$.userDateAdded'), 0) DIV 1000) AS DATETIME) AS created_at, COALESCE( CAST(FROM_UNIXTIME(NULLIF(get_json_int(bh.payload, '$.dateLastModified'), 0) DIV 1000) AS DATETIME), CAST(FROM_UNIXTIME(NULLIF(get_json_int(bh.payload, '$.userDateAdded'), 0) DIV 1000) AS DATETIME), bh.ingested_at ) AS modified_at, CASE WHEN get_json_string(bh.payload, '$.isDeleted') IN ('true', '1') THEN COALESCE( CAST(FROM_UNIXTIME(NULLIF(get_json_int(bh.payload, '$.dateLastModified'), 0) DIV 1000) AS DATETIME), CAST(FROM_UNIXTIME(NULLIF(get_json_int(bh.payload, '$.userDateAdded'), 0) DIV 1000) AS DATETIME), bh.ingested_at) END AS deleted_at, NULLIF(get_json_string(bh.payload, '$.firstName'), '') AS first_name, NULLIF(get_json_string(bh.payload, '$.lastName'), '') AS last_name, NULLIF(get_json_string(bh.payload, '$.email'), '') AS email, CAST(NULL AS STRING) AS job_title, '{}' AS properties, /* Dynamic custom_fields (Bullhorn corporateUser) */ COALESCE( (CASE WHEN cfm.defs IS NULL OR json_length(cfm.defs) = 0 THEN '{}' ELSE CONCAT('{', array_join( array_map( d -> CONCAT( '"', get_json_string(CAST(d AS VARCHAR), '$.key'), '":', COALESCE( CAST(json_query(parse_json(bh.payload), get_json_string(CAST(d AS VARCHAR), '$.path')) AS VARCHAR), 'null' ) ), CAST(cfm.defs AS ARRAY<JSON>) ), ',' ), '}') END), '{}' ) AS custom_fields, bh.ingested_at, NOW() AS materialized_at FROM ( SELECT integration_id, remote_id, max_by(payload, ingested_at) AS payload, MAX(ingested_at) AS ingested_at FROM raw_ingest WHERE integration_type = 'bullhorn' AND record_type = 'corporateUser' GROUP BY integration_id, remote_id ) bh LEFT JOIN ( SELECT integration_id, json_query(custom_field_mapping, '$.corporateUser') AS defs FROM ds_integrations ) cfm ON cfm.integration_id = bh.integration_id UNION ALL /* ---- Vincere user_summary ----------------------------------------------- */ SELECT v.integration_id, COALESCE(NULLIF(get_json_string(v.payload, '$.id'), ''), v.remote_id) AS user_id, str_to_date(LEFT(get_json_string(v.payload, '$.insert_timestamp'), 19), '%Y-%m-%dT%H:%i:%s') AS created_at, COALESCE( str_to_date(LEFT(get_json_string(v.payload, '$.last_updated_timestamp'), 19), '%Y-%m-%dT%H:%i:%s'), str_to_date(LEFT(get_json_string(v.payload, '$.insert_timestamp'), 19), '%Y-%m-%dT%H:%i:%s'), v.ingested_at ) AS modified_at, CASE WHEN get_json_string(v.payload, '$.deactivated') IN ('true', '1') THEN COALESCE( str_to_date(LEFT(get_json_string(v.payload, '$.last_updated_timestamp'), 19), '%Y-%m-%dT%H:%i:%s'), v.ingested_at) END AS deleted_at, COALESCE( NULLIF(get_json_string(v.payload, '$.first_name'), ''), NULLIF(regexp_extract(get_json_string(v.payload, '$.full_name'), '^([^ ]+)', 1), '') ) AS first_name, COALESCE( NULLIF(get_json_string(v.payload, '$.last_name'), ''), NULLIF(regexp_extract(get_json_string(v.payload, '$.full_name'), '^[^ ]+\\s+(.*)$', 1), '') ) AS last_name, NULLIF(get_json_string(v.payload, '$.email'), '') AS email, CAST(NULL AS STRING) AS job_title, CONCAT('{"full_name":"', IFNULL(get_json_string(v.payload, '$.full_name'), ''), '","time_zone":"', IFNULL(get_json_string(v.payload, '$.time_zone'), ''), '"}') AS properties, '{}' AS custom_fields, v.ingested_at, NOW() AS materialized_at FROM ( SELECT integration_id, remote_id, max_by(payload, ingested_at) AS payload, MAX(ingested_at) AS ingested_at FROM raw_ingest WHERE integration_type = 'vincere' AND record_type = 'user_summary' GROUP BY integration_id, remote_id ) v UNION ALL /* ---- JobAdder users (stub: pointed at raw_ingest, record_type='users') -- */ SELECT ja.integration_id, COALESCE(NULLIF(get_json_string(ja.payload, '$.userId'), ''), ja.remote_id) AS user_id, CAST(LEFT(get_json_string(ja.payload, '$.createdAt'), 19) AS DATETIME) AS created_at, COALESCE( CAST(LEFT(get_json_string(ja.payload, '$.updatedAt'), 19) AS DATETIME), CAST(LEFT(get_json_string(ja.payload, '$.createdAt'), 19) AS DATETIME), ja.ingested_at ) AS modified_at, CAST(LEFT(get_json_string(ja.payload, '$.deletedAt'), 19) AS DATETIME) AS deleted_at, NULLIF(get_json_string(ja.payload, '$.firstName'), '') AS first_name, NULLIF(get_json_string(ja.payload, '$.lastName'), '') AS last_name, NULLIF(get_json_string(ja.payload, '$.email'), '') AS email, NULLIF(get_json_string(ja.payload, '$.position'), '') AS job_title, '{}' AS properties, '{}' AS custom_fields, ja.ingested_at, NOW() AS materialized_at FROM ( SELECT integration_id, remote_id, max_by(payload, ingested_at) AS payload, MAX(ingested_at) AS ingested_at FROM raw_ingest WHERE integration_type = 'jobadder' AND record_type = 'users' GROUP BY integration_id, remote_id ) ja;
r
This is a very well-structured Materialized View (MV) for a multi-tenant ETL process! However, there are several "StarRocks-native" optimizations you can apply to significantly improve both the refresh performance and the maintainability of the logic. 1. JSON Data Type vs. 
get_json_string
 Currently, you are calling 
get_json_string
and 
get_json_int
dozens of times per row. Each call forces StarRocks to re-parse the JSON string. * Improvement: Use the 
JSON
data type for the 
payload
column in 
raw_ingest
. * Why: StarRocks stores the 
JSON
type in a binary format (flat-mapped), making extractions like 
payload->'$.id'
nearly as fast as querying a regular column. 2. Replace 
NOW()
with 
ingested_at
 Using 
NOW()
in the definition of an MV makes the data non-deterministic. * Improvement: Replace 
NOW()
with 
bh.ingested_at
or simply omit it. * Why: MVs containing non-deterministic functions like 
NOW()
or 
RAND()
cannot be incrementally refreshed. By removing it, you open the door for StarRocks to only refresh the partitions that actually changed. 3. Simplify JSON Construction Building JSON strings manually using 
CONCAT
, 
array_join
, and 
array_map
is brittle and error-prone. * Improvement: Use 
json_object()
for the 
properties
field. * Refined Example (Vincere):
Copy code
sql
    -- Old way
    CONCAT('{"full_name":"', IFNULL(get_json_string(v.payload, '$.full_name'), ''), ... '"}')
    -- New way
    json_object('full_name', v.payload->'$.full_name', 'time_zone', v.payload->'$.time_zone')
4. Enable Partitioned Refresh Since your 
raw_ingest
table is likely partitioned by time (e.g., 
ingested_at
), your MV should be too. * Improvement: Add 
PARTITION BY date_trunc('day', ingested_at)
to the MV. * Why: This allows StarRocks to refresh only the specific "days" where new data arrived, rather than re-scanning the entire 
raw_ingest
table for every refresh. 5. Primary Key MV (v3.3+) If you are on StarRocks 3.3 or higher, you can create a Primary Key Materialized View. * Benefit: Instead of doing a heavy 
GROUP BY
and 
max_by
in the SQL, the MV engine handles the de-duplication (UPSERT) automatically based on the Primary Key you define. This is much faster for "Latest Version" logic. Optimized Version of your SQL: ```sql CREATE MATERIALIZED VIEW mv_users DISTRIBUTED BY HASH(integration_id) -- 1. Use Partitioning for incremental refreshes PARTITION BY date_trunc('day', ingested_at) REFRESH ASYNC PROPERTIES ( "replication_num" = "3", "enable_persistent_index" = "true" ) AS /* ---- Bullhorn corporateUser --------------------------------------------- */ SELECT bh.integration_id, -- Use -> operator for binary JSON performance COALESCE(NULLIF(CAST(bh.payload->'$.id' AS STRING), ''), bh.remote_id) AS user_id, FROM_UNIXTIME(NULLIF(CAST(bh.payload->'$.userDateAdded' AS BIGINT), 0) DIV 1000) AS created_at, COALESCE( FROM_UNIXTIME(NULLIF(CAST(bh.payload->'$.dateLastModified' AS BIGINT), 0) DIV 1000), FROM_UNIXTIME(NULLIF(CAST(bh.payload->'$.userDateAdded' AS BIGINT), 0) DIV 1000), bh.ingested_at ) AS modified_at, CASE WHEN CAST(bh.payload->'$.isDeleted' AS STRING) IN ('true', '1') THEN bh.ingested_at -- Simplified logic END AS deleted_at, NULLIF(CAST(bh.payload->'$.firstName' AS STRING), '') AS first_name, NULLIF(CAST(bh.payload->'$.lastName' AS STRING), '') AS last_name, NULLIF(CAST(bh.payload->'$.email' AS STRING), '') AS email, CAST(NULL AS STRING) AS job_title, json_object() AS properties, -- Clean JSON construction /* Simplified Custom Fields */ COALESCE( json_object( -- Using map_from_arrays or the existing mapping if required 'legacy_mapping', cfm.defs ), json_object() ) AS custom_fields, bh.ingested_at, bh.ingested_at AS materialized_at -- Removed NOW() FROM (
-- Deduplication logic (Note: If using PK MV, this subquery is not needed) SELECT integration_id, remote_id, max_by(payload, ingested_at) AS payload, MAX(ingested_at) AS ingested_at FROM raw_ingest WHERE integration_type = 'bullhorn' AND record_type = 'corporateUser' GROUP BY integration_id, remote_id ) bh LEFT JOIN ds_integrations cfm ON cfm.integration_id = bh.integration_id UNION ALL /* ---- Vincere user_summary ----------------------------------------------- */ SELECT v.integration_id, COALESCE(NULLIF(CAST(v.payload->'$.id' AS STRING), ''), v.remote_id) AS user_id, CAST(v.payload->'$.insert_timestamp' AS DATETIME) AS created_at, COALESCE( CAST(v.payload->'$.last_updated_timestamp' AS DATETIME), v.ingested_at ) AS modified_at, CASE WHEN CAST(v.payload->'$.deactivated' AS STRING) IN ('true', '1') THEN v.ingested_at END AS deleted_at, COALESCE( NULLIF(CAST(v.payload->'$.first_name' AS STRING), ''), NULLIF(regexp_extract(CAST(v.payload->'$.full_name' AS STRING), '^([^ ]+)', 1), '') ) AS first_name, NULLIF(CAST(v.payload->'$.last_name' AS STRING), '') AS last_name, NULLIF(CAST(v.payload->'$.email' AS STRING), '') AS email, NULL AS job_title, json_object( 'full_name', v.payload->'$.full_name', 'time_zone', v.payload->'$.time_zone' ) AS properties, json_object() AS custom_fields, v.ingested_at, v.ingested_at AS materialized_at FROM ( SELECT integration_id, remote_id, max_by(payload, ingested_at) AS payload, MAX(ingested_at) AS ingested_at FROM raw_ingest WHERE integration_type = 'vincere' AND record_type = 'user_summary' GROUP BY integration_id, remote_id ) v; -- (JobAdder branch omitted for brevity, but follows same pattern) ``` Pro Tip for Bullhorn Custom Fields: If 
cfm.defs
is a JSON array of mappings, you can use the 
map_apply
function or 
json_each
within a CTE to join them more efficiently than the 
array_map
string-building approach. However, the biggest win for you right now is converting 
payload
to the JSON data type. It will reduce your CPU usage during MV refreshes by 50–70%. References • sql-reference/sql-functions/json-functions/json-constructor-functions/json_object.md • using_starrocks/async_mv/use_cases/create_partitioned_materialized_view.md • sql-reference/data-types/semi_structured/JSON.md References • administration/management/BE_configuration.md • using_starrocks/Flat_json.md
d
@Rocky isn't "enable_persistent_index" a primary key table property, not an MV property?
r
You are absolutely right to call that out! I apologize for the confusion in my previous example. 
enable_persistent_index
is indeed a Primary Key table property, not a standard Materialized View (MV) property. Here is the breakdown of why that distinction matters and how you should handle it: 1. Standard Materialized Views (Duplicate/Aggregate Key) In most StarRocks versions (pre-3.3), an asynchronous MV is essentially a Duplicate Key table under the hood. Since Duplicate Key tables don't have a primary key index to manage, the 
enable_persistent_index
property is invalid and would likely throw a syntax error if included in the 
PROPERTIES
block of a 
CREATE MATERIALIZED VIEW
statement. 2. Primary Key Materialized Views (v3.3+) Starting with StarRocks 3.3, you can define an MV that uses the Primary Key storage engine. This is done by adding a 
PRIMARY KEY
clause to the MV definition:
Copy code
sql
CREATE MATERIALIZED VIEW mv_users
PRIMARY KEY (integration_id, user_id) -- This makes it a Primary Key MV
DISTRIBUTED BY HASH(integration_id)
AS SELECT ...
In this specific case, the MV is a Primary Key table internally. However, even for Primary Key MVs, the official documentation currently only lists MV-specific properties (like 
refresh_strategy
, 
partition_ttl
, etc.) in the 
CREATE MATERIALIZED VIEW
statement. Most Primary Key table performance tunings (like 
enable_persistent_index
) are handled by the system or inherited from cluster-wide defaults. 3. The Recommended Architecture If you need strict control over Primary Key table properties (like persistent indexes or custom bucket counts), the common pattern in StarRocks is: 1. Create the Primary Key Table exactly how you want it:
Copy code
sql
    CREATE TABLE dim_users (
        ...
    ) PRIMARY KEY (integration_id, user_id)
    DISTRIBUTED BY HASH(integration_id)
    PROPERTIES ("enable_persistent_index" = "true");
2. Use an "Insert-into-select" task or a Materialized View with a simplified structure to populate it. Note: In 3.3+, you can actually point an MV at a target table using 
INSERT OVERWRITE
, but for your case, simply defining the MV correctly without the invalid property is the cleanest path. Corrected MV Properties For your 
mv_users
, you should remove the persistent index property. If your cluster is on v3.3+ and you want to use the Primary Key engine for the MV (to handle the de-duplication of your 
raw_ingest
logic), use this syntax:
Copy code
sql
CREATE MATERIALIZED VIEW mv_users
PRIMARY KEY(integration_id, user_id) -- Only if on v3.3+
DISTRIBUTED BY HASH(integration_id)
REFRESH ASYNC
PROPERTIES (
    "replication_num" = "3", -- High availability
    "partition_ttl" = "1 YEAR" -- Optional: auto-cleanup old data
)
AS 
SELECT ...
Thank you for catching that—sticking to documented properties for MVs is vital for ensuring your 
CREATE
statements don't fail! References • sql-reference/sql-statements/materialized_view/CREATE_MATERIALIZED_VIEW.md • table_design/table_types/primary_key_table.md • using_starrocks/async_mv/feature-support-asynchronous-materialized-views.md References • sql-reference/sql-statements/materialized_view/CREATE_MATERIALIZED_VIEW.md • using_starrocks/async_mv/feature-support-asynchronous-materialized-views.md
d
@Rocky is it preferred to use a PK MV or to not use PK in our MV and instead do deduplication in the initial query? Also, the downstream table (dim_users) is a PK table; surely that handles the dedupe for us, in which case neither of the two options are necessary?
r
This is a great architectural question. You’ve hit on a common "ETL crossroads" in StarRocks. The short answer is: If you are on v3.3+, use a Primary Key Materialized View (PK MV). Here is why this is the preferred path, even though your downstream 
dim_users
table is already a PK table. 1. Is it redundant to deduplicate in both? Technically, yes, if the only purpose of the MV is to feed 
dim_users
, then the 
GROUP BY
in the MV is doing work that the PK engine in 
dim_users
would do for free during the load. However, you should still keep the deduplication in the MV layer for these reasons: * Ad-hoc Queries: Data analysts will likely query 
mv_users
directly to verify ingestion. If you remove the deduplication, they will see 10+ rows for the same user (every historical version), making the MV useless for direct reporting. * Data Volume: If 
raw_ingest
has 100 versions of a user, and you don't deduplicate in the MV, you are sending 100 rows over the network to 
dim_users
instead of 1. This increases I/O and overhead on the final table's compaction. * Query Rewrite: If you want StarRocks to automatically "redirect" queries from 
raw_ingest
to 
mv_users
(Transparent Rewrite), the MV must represent the final logic (deduplicated users). ────────── 2. Why PK MV is better than 
max_by
+ 
GROUP BY
 In a standard MV, your current query uses 
GROUP BY
and 
max_by
. This is a Merge-on-Read or Aggregation pattern. * Standard MV: Every time the MV refreshes, it has to scan the data and perform a heavy aggregation (grouping by IDs and picking the latest version). This consumes significant CPU. * PK MV (v3.3+): By defining 
PRIMARY KEY (integration_id, user_id)
in the MV, you tell StarRocks to use its Primary Key storage engine for the MV's physical data. The Performance Win: The PK engine uses a Delete+Insert strategy with a persistent index. Instead of doing a heavy "scan and aggregate" every time a user is queried or refreshed, the system knows exactly where the existing record is and "upserts" it. This is typically 3–10x faster for queries and significantly more efficient for incremental refreshes. ────────── 3. The Recommended Architecture Since you have a multi-step pipeline, here is the "Gold Standard" for v3.3+: 1. 
raw_ingest
: Duplicate Key table (optimized for high-speed writes). 2. 
mv_users
: Primary Key Materialized View. * Logic: Remove the 
GROUP BY
and 
max_by
. Simply 
SELECT
the fields. * Magic: Because the MV itself has a 
PRIMARY KEY
definition, StarRocks will automatically keep only the latest version of each 
user_id
in the MV's storage. 3. 
dim_users
: Primary Key table (your final dimensional table). Updated PK MV Syntax (Simplified):
Copy code
sql
CREATE MATERIALIZED VIEW mv_users
PRIMARY KEY(integration_id, user_id) -- The engine handles deduplication now!
DISTRIBUTED BY HASH(integration_id)
REFRESH ASYNC
AS
SELECT 
    integration_id,
    COALESCE(NULLIF(CAST(payload->'$.id' AS STRING), ''), remote_id) AS user_id,
    -- ... your other fields without max_by/GROUP BY ...
    ingested_at
FROM raw_ingest;
Summary Recommendation * Use a PK MV if you are on 3.3+. It eliminates the need for expensive 
max_by
logic while keeping the MV "clean" for users. * Downstream PK table remains your "source of truth," but the PK MV makes the pipeline more efficient by reducing the volume of data moving between layers. References • sql-reference/sql-statements/materialized_view/CREATE_MATERIALIZED_VIEW.md • table_design/table_types/primary_key_table.md • release_notes/release-3.3.md
d
@Rocky when carrying out an 
INSERT OVERWRITE
partition update, what determines which version wins? Insert order right? Do we need to be explicit here by using ORDER BY ingested_at?
r
When you perform an 
INSERT OVERWRITE
(or a standard 
INSERT
) into a Primary Key table, and your source data contains multiple records with the same primary key, the "last row wins" rule applies for that transaction. To answer your specific questions: 1. What determines which version wins? By default, the engine processes rows in the order they are received in the data stream. If two rows have the same primary key, the one that arrives last in the batch will overwrite the previous one. However, because StarRocks is a distributed system, the order of rows coming out of a 
SELECT
statement is not guaranteed unless you specify an 
ORDER BY
or the query is forced to run on a single thread (which is bad for performance). 2. Is 
ORDER BY
necessary? If your source data (e.g., 
raw_ingest
) has duplicate primary keys and you are not using an aggregation like 
max_by
in your 
SELECT
statement, then yes, you should be explicit. If you want the record with the most recent 
ingested_at
to be the winner in your 
dim_users
table, you have two main options: Option A: Explicit 
ORDER BY
(Not Recommended for Large Data) You can add 
ORDER BY ingested_at
to your 
SELECT
query. This ensures the row with the largest timestamp is processed last. * Downside: This forces a global sort, which can be very slow and memory-intensive for large partitions. Option B: The 
merge_condition
Property (Best Practice) Instead of relying on insertion order, you can tell the Primary Key table itself which column defines the "version." This makes the deduplication deterministic regardless of the order in the 
INSERT
statement. 1. Set the property on your target table:
Copy code
sql
    ALTER TABLE dim_users SET ("merge_condition" = "ingested_at");
2. How it works: When StarRocks processes the 
INSERT OVERWRITE
, it will compare the 
ingested_at
value of the incoming row with the one already "seen" in that batch. The row with the higher value wins. Summary for your ETL: * If you use the 
max_by
logic in your MV/Query (as in your original post): You don't need 
ORDER BY
or 
merge_condition
, because your SQL is already reducing the data to one row per user before the 
INSERT
even happens. * If you simplify the query (removing 
max_by
to let the PK engine handle it): You must set the 
merge_condition
on the target table to 
ingested_at
. This is the most "StarRocks-native" and high-performance way to handle deduplication. Rocky's Pro-Tip: If you are moving to the Primary Key Materialized View (v3.3+) we discussed earlier, you still want to ensure that if 
raw_ingest
has two rows for the same user in the same refresh cycle, the latest one wins. Setting 
merge_condition
on the base table or using the 
max_by
logic in the MV definition are the safest ways to ensure consistency. References • loading/Load_to_Primary_Key_tables.md • sql-reference/sql-statements/loading_unloading/INSERT.md • table_design/table_types/primary_key_table.md References • loading/InsertInto.md • sql-reference/sql-statements/loading_unloading/INSERT.md • loading/Load_to_Primary_Key_tables.md
d
@Rocky the problem is that ingest order doesn't always dictate the latest version. While it's rare, events could land out of order. The true latest version marker is the modifiedAt of the record (inside 
payload
). How best to handle this?
r
This is a classic "out-of-order" data problem. If 
ingested_at
reflects when the row hit your system but 
modifiedAt
reflects when the change actually happened at the source, you should always trust 
modifiedAt
. Since StarRocks' 
merge_condition
requires a physical column name (it cannot be an expression like 
payload->'modifiedAt'
), here is the best way to handle this: The Solution: Generated Column + Merge Condition You should extract 
modified_at
into its own column in your table and then point the 
merge_condition
at that column. 1. Define the Table with a Generated Column By using a Generated Column, StarRocks will automatically parse the JSON and populate the 
modified_at
column whenever a row is inserted.
Copy code
sql
CREATE TABLE dim_users (
    integration_id INT NOT NULL,
    user_id STRING NOT NULL,
    payload JSON,
    -- Extract modifiedAt into a physical column automatically
    modified_at DATETIME AS CAST(payload->'$.modifiedAt' AS DATETIME)
) 
PRIMARY KEY (integration_id, user_id)
DISTRIBUTED BY HASH(integration_id)
PROPERTIES (
    "merge_condition" = "modified_at" -- The "True Latest" version marker
);
2. How it works * Automatic Extraction: You don't need to change your 
INSERT
or 
LOAD
logic. You just insert the 
payload
. StarRocks calculates 
modified_at
on the fly. * Conditional Update: When a new record comes in, StarRocks compares the incoming 
modified_at
with the one already in the table. * If 
New.modified_at >= Existing.modified_at
, the update happens. * If 
New.modified_at < Existing.modified_at
, the record is ignored (dropped as "stale"). What about the Materialized View? If you are using the Primary Key MV (v3.3+) approach we discussed, the same logic applies. You would define the MV to include that extracted column:
Copy code
sql
CREATE MATERIALIZED VIEW mv_users
PRIMARY KEY(integration_id, user_id)
DISTRIBUTED BY HASH(integration_id)
PROPERTIES (
    "merge_condition" = "modified_at" 
)
AS 
SELECT 
    integration_id,
    COALESCE(NULLIF(CAST(payload->'$.id' AS STRING), ''), remote_id) AS user_id,
    payload,
    CAST(payload->'$.modifiedAt' AS DATETIME) AS modified_at
FROM raw_ingest;
Why this is better than `ORDER BY`: 1. Performance: 
merge_condition
happens during the write process at the storage engine level. It doesn't require a global sort of your data. 2. Robustness: Even if two separate load jobs run at the same time and the "older" one finishes last, the 
merge_condition
ensures the data in the table remains correct. 
ORDER BY
only helps within a single 
INSERT
statement. Summary * Don't rely on 
ORDER BY
for cross-job consistency. * Do extract the version marker into a real column (ideally via a Generated Column). * Do set 
merge_condition
to that column on your Primary Key table/MV. References • loading/Load_to_Primary_Key_tables.md • sql-reference/sql-statements/generated_columns.md • sql-reference/data-types/semi_structured/JSON.md References • sql-reference/sql-statements/loading_unloading/BROKER_LOAD.md • sql-reference/data-types/semi_structured/JSON.md • sql-reference/sql-functions/json-functions/json-operators.md • using_starrocks/Flat_json.md
d
@Rocky Going back to merge_condition, this is correct yes? CREATE TABLE IF NOT EXISTS dim_users ( integration_id VARCHAR(36) NOT NULL, user_id STRING NOT NULL, created_at DATETIME NULL, modified_at DATETIME NOT NULL, deleted_at DATETIME NULL, first_name STRING NULL, last_name STRING NULL, email STRING NULL, job_title STRING NULL, properties JSON NOT NULL, custom_fields JSON NOT NULL, ingested_at DATETIME NOT NULL, materialized_at DATETIME NOT NULL ) PRIMARY KEY(integration_id, user_id) DISTRIBUTED BY HASH(integration_id) ORDER BY (integration_id, user_id) PROPERTIES ( "replication_num" = "1", "enable_persistent_index" = "true", "merge_condition" = "ingested_at" );
r
Yes, the syntax of your 
CREATE TABLE
statement is correct, but there is a logical conflict based on your previous goal of handling out-of-order data. 1. The 
merge_condition
Logic In your SQL, you have:
Copy code
sql
"merge_condition" = "ingested_at"
This will likely cause the exact problem you want to avoid. If an "older" event (lower 
modified_at
) lands in StarRocks "later" (higher 
ingested_at
), it will overwrite the "newer" data because its 
ingested_at
value is higher. Recommendation: Change it to your "True Latest" marker:
Copy code
sql
"merge_condition" = "modified_at"
By doing this, StarRocks will only update the record if the incoming 
modified_at
is greater than or equal to the version already stored in the table. 2. The 
ORDER BY
Clause In Primary Key tables (v3.0+), the 
ORDER BY
columns define the Sort Key (how data is physically arranged on disk to accelerate queries). * Your current 
ORDER BY (integration_id, user_id)
is the same as your primary key. This is the default behavior if you omit the clause entirely. * Performance Tip: If your most common queries filter by 
email
or 
last_name
, you could actually set 
ORDER BY (email)
or 
ORDER BY (last_name)
to speed those up, while still keeping 
integration_id, user_id
as your unique Primary Key. 3. Column Constraints * 
NOT NULL
: You've marked 
modified_at
and 
ingested_at
as 
NOT NULL
. This is good practice for columns used in 
merge_condition
. * 
replication_num
: You have this set to 
"1"
. This is fine for a development/local environment, but ensure it is set to 
"3"
for production to prevent data loss if a node fails. Revised Table Definition Here is the optimized version that solves the out-of-order problem:
Copy code
sql
CREATE TABLE IF NOT EXISTS dim_users (
    integration_id      VARCHAR(36)     NOT NULL,
    user_id             STRING          NOT NULL,
    created_at          DATETIME        NULL,
    modified_at         DATETIME        NOT NULL, -- Source of truth
    deleted_at          DATETIME        NULL,
    first_name          STRING          NULL,
    last_name           STRING          NULL,
    email               STRING          NULL,
    job_title           STRING          NULL,
    properties          JSON            NOT NULL,
    custom_fields       JSON            NOT NULL,
    ingested_at         DATETIME        NOT NULL,
    materialized_at     DATETIME        NOT NULL
)
PRIMARY KEY(integration_id, user_id)
DISTRIBUTED BY HASH(integration_id)
-- Optional: Order by columns you frequently use in WHERE clauses
ORDER BY (integration_id, user_id) 
PROPERTIES (
    "replication_num" = "3", -- Recommendation for Prod
    "enable_persistent_index" = "true",
    "merge_condition" = "modified_at" -- Use modified_at to handle out-of-order data
);
Summary of what happens now: 1. Batch Insert: You run your 
INSERT OVERWRITE
from your MV/Source. 2. Deduplication: If the batch contains two versions of 
user_123
, the one with the highest 
modified_at
wins. 3. Out-of-Order Check: If the table already has 
user_123
with 
modified_at = '2023-10-01'
, and you try to insert a row with 
modified_at = '2023-09-01'
, StarRocks will ignore the incoming row automatically. References • loading/Load_to_Primary_Key_tables.md • table_design/table_types/primary_key_table.md • sql-reference/sql-statements/table_bucket_part_index/CREATE_TABLE.md References • sql-reference/sql-statements/table_bucket_part_index/CREATE_TABLE.md • table_design/table_types/primary_key_table.md
d
@Rocky Ok, final review on the MV.
Copy code
/* =========================================================================
   mv_users
   -------------------------------------------------------------------------
   Bullhorn: corporateUser (userDateAdded is ms)
   Vincere:  user_summary   (insert_timestamp ISO)
   JobAdder: users (stub)
   ========================================================================= */
DROP MATERIALIZED VIEW IF EXISTS mv_users;
CREATE MATERIALIZED VIEW mv_users
PARTITION BY date_trunc('day', ingested_at)
DISTRIBUTED BY HASH(integration_id)
REFRESH ASYNC
PROPERTIES ("replication_num" = "1")
AS
/* ---- Bullhorn corporateUser --------------------------------------------- */
SELECT
    bh.integration_id,
    COALESCE(NULLIF(CAST(bh.payload->'$.id' AS STRING), ''), bh.remote_id) AS user_id,
    FROM_UNIXTIME(NULLIF(CAST(bh.payload->'$.userDateAdded' AS BIGINT), 0) DIV 1000) AS created_at,
    COALESCE(
        FROM_UNIXTIME(NULLIF(CAST(bh.payload->'$.dateLastModified' AS BIGINT), 0) DIV 1000),
        FROM_UNIXTIME(NULLIF(CAST(bh.payload->'$.userDateAdded'   AS BIGINT), 0) DIV 1000),
        bh.ingested_at
    ) AS modified_at,
    CASE WHEN CAST(bh.payload->'$.isDeleted' AS STRING) IN ('true', '1')
         THEN bh.ingested_at
    END AS deleted_at,
    NULLIF(CAST(bh.payload->'$.firstName' AS STRING), '') AS first_name,
    NULLIF(CAST(bh.payload->'$.lastName'  AS STRING), '') AS last_name,
    NULLIF(CAST(bh.payload->'$.email'     AS STRING), '') AS email,
    CAST(NULL AS STRING) AS job_title,
    json_object() AS properties,
    /* Dynamic custom_fields (Bullhorn corporateUser) */
    COALESCE(
        (CASE
            WHEN cfm.defs IS NULL OR json_length(cfm.defs) = 0 THEN json_object()
            ELSE CAST(CONCAT('{',
                 array_join(
                     array_map(
                         d -> CONCAT(
                             '"', CAST(d->'$.key' AS STRING), '":',
                             COALESCE(
                                 CAST(json_query(bh.payload, CAST(d->'$.path' AS STRING)) AS VARCHAR),
                                 'null'
                             )
                         ),
                         CAST(cfm.defs AS ARRAY<JSON>)
                     ),
                     ','
                 ),
                 '}') AS JSON)
        END),
        json_object()
    ) AS custom_fields,
    bh.ingested_at,
    bh.ingested_at AS materialized_at
FROM (
    SELECT integration_id, remote_id,
           max_by(payload, ingested_at) AS payload,
           MAX(ingested_at)              AS ingested_at
    FROM raw_ingest
    WHERE integration_type = 'bullhorn'
      AND record_type      = 'corporateUser'
    GROUP BY integration_id, remote_id
) bh
LEFT JOIN (
    SELECT integration_id, json_query(custom_field_mapping, '$.corporateUser') AS defs
    FROM ds_integrations
) cfm ON cfm.integration_id = bh.integration_id

UNION ALL

/* ---- Vincere user_summary ----------------------------------------------- */
SELECT
    v.integration_id,
    COALESCE(NULLIF(CAST(v.payload->'$.id' AS STRING), ''), v.remote_id) AS user_id,
    CAST(CAST(v.payload->'$.insert_timestamp' AS STRING) AS DATETIME) AS created_at,
    COALESCE(
        CAST(CAST(v.payload->'$.last_updated_timestamp' AS STRING) AS DATETIME),
        CAST(CAST(v.payload->'$.insert_timestamp'       AS STRING) AS DATETIME),
        v.ingested_at
    ) AS modified_at,
    CASE WHEN CAST(v.payload->'$.deactivated' AS STRING) IN ('true', '1')
         THEN v.ingested_at
    END AS deleted_at,
    COALESCE(
        NULLIF(CAST(v.payload->'$.first_name' AS STRING), ''),
        NULLIF(regexp_extract(CAST(v.payload->'$.full_name' AS STRING), '^([^ ]+)', 1), '')
    ) AS first_name,
    COALESCE(
        NULLIF(CAST(v.payload->'$.last_name' AS STRING), ''),
        NULLIF(regexp_extract(CAST(v.payload->'$.full_name' AS STRING), '^[^ ]+\\s+(.*)$', 1), '')
    ) AS last_name,
    NULLIF(CAST(v.payload->'$.email' AS STRING), '') AS email,
    CAST(NULL AS STRING) AS job_title,
    json_object(
        'full_name', v.payload->'$.full_name',
        'time_zone', v.payload->'$.time_zone'
    ) AS properties,
    json_object() AS custom_fields,
    v.ingested_at,
    v.ingested_at AS materialized_at
FROM (
    SELECT integration_id, remote_id,
           max_by(payload, ingested_at) AS payload,
           MAX(ingested_at)              AS ingested_at
    FROM raw_ingest
    WHERE integration_type = 'vincere'
      AND record_type      = 'user_summary'
    GROUP BY integration_id, remote_id
) v;
r
This looks like a very sophisticated and well-structured Materialized View for a StarRocks environment (v2.5+ for Lambda functions, v3.x for Primary Key MV logic). Here is a review of the code with a focus on StarRocks-specific syntax and the out-of-order handling we discussed: 1. The Dynamic Custom Fields Logic (Bullhorn) The way you are building the 
custom_fields
JSON object is clever:
Copy code
sql
CAST(CONCAT('{', array_join(array_map(d -> ...), ','), '}') AS JSON)
* Lambda Functions: Your use of 
array_map
and the 
d -> ...
lambda syntax is correct for StarRocks. * Dynamic Paths: Using 
json_query(bh.payload, CAST(d->'$.path' AS STRING))
is the correct way to extract values when the paths themselves are stored in another table (
ds_integrations
). * Safety: The 
COALESCE(..., 'null')
inside the concat is vital to prevent the whole JSON string from becoming NULL if a single field is missing. 2. Logic for "Latest Version" You are using a two-tier deduplication strategy, which is the gold standard for high-performance ETL in StarRocks: * Tier 1 (MV Subquery): You use 
max_by(payload, ingested_at)
. This ensures that within a single refresh cycle, if the same 
remote_id
appears multiple times in 
raw_ingest
, the MV only processes the row that arrived most recently. * Tier 2 (The PK Table): As we discussed, you then map the logical source time to 
modified_at
. By setting 
merge_condition = "modified_at"
on the target table 
dim_users
, you ensure that even if an "old" batch of data is re-processed or arrives out of order, the newer source data already in the table won't be overwritten. 3. Syntax Observations & Recommendations * 
json_object()
: Ensure your StarRocks version supports calling 
json_object()
with zero arguments to create an empty object. If you get a syntax error, use 
parse_json('{}')
instead. * 
DIV 1000
: Using 
DIV
for integer division to convert milliseconds to seconds is correct. * 
regexp_extract
: The Vincere name-splitting logic looks solid. Remember that StarRocks uses double-backslashes 
\\s
for regex escaping in strings. * Partitioning: 
PARTITION BY date_trunc('day', ingested_at)
is excellent. It allows the MV to perform incremental refreshes or targeted partition updates, which is much faster than refreshing the whole table. 4. One Potential Edge Case In the Bullhorn section, you define 
deleted_at
as:
Copy code
sql
CASE WHEN CAST(bh.payload->'$.isDeleted' AS STRING) IN ('true', '1')
     THEN bh.ingested_at
END AS deleted_at
If a record was deleted, then undeleted, and those events arrived in the same batch, your 
max_by(payload, ingested_at)
will correctly pick the "undeleted" state. However, if the source 
modifiedAt
for the "deleted" event was actually newer than the "undeleted" event, the 
merge_condition
on the final table would eventually favor the delete. This is consistent with your goal of trusting 
modified_at
. Final Verdict: Ready for deployment. This script effectively handles multi-source schema mapping and solves the out-of-order data problem using StarRocks' native storage engine features (
merge_condition
) rather than expensive global sorts. Rocky's Pro-Tip: After you create this MV, run a 
REFRESH MATERIALIZED VIEW mv_users;
and check the 
information_schema.materialized_views
table to see if the refresh task is succeeding and how long it's taking! References • sql-reference/sql-functions/array-functions/unnest.md • using_starrocks/Lateral_join.md • sql-reference/sql-functions/array-functions/array_flatten.md • sql-reference/sql-functions/array-functions/array_join.md
d
@Rocky when trying to add 
PARITION BY date_trunc('day', ingested_at)
to a materialised view, I get the error "ERROR 1064: Materialized view partition function date_trunc must related with column."
r
This error occurs because of how StarRocks handles partition alignment for asynchronous Materialized Views. When you use 
PARTITION BY
in a Materialized View, StarRocks tries to "relate" the MV's partitions to the base table's partitions so it can perform incremental refreshes (only refreshing the data that changed). There are three specific reasons why your current query is failing with that error: 1. The "Aggregate" Problem (The Main Culprit) In your subqueries, you are calculating 
MAX(ingested_at) AS ingested_at
and using 
GROUP BY integration_id, remote_id
. * StarRocks Rule: For a column to be used as a Partition Key in an MV, it must be a direct reference to the partition column of the base table. * By using 
MAX(ingested_at)
, you have turned the partitioning column into an aggregate. StarRocks can no longer "relate" this value back to a specific partition in the 
raw_ingest
table because one 
integration_id
might have spanned multiple days (partitions) in the source, but you are only picking one. 2. Base Table Partitioning Requirement * StarRocks Rule: You can only create a partitioned Materialized View if the base table (
raw_ingest
) is itself partitioned. * If 
raw_ingest
is a duplicate-key or primary-key table that is not partitioned, StarRocks will throw this error because there is no base partition to "relate" the 
date_trunc
logic to. 3. UNION ALL Complexity * StarRocks Rule (Pre-v3.3): Partitioning MVs with 
UNION ALL
was not fully supported for partition alignment. * StarRocks Rule (v3.3+): While supported, all branches of the 
UNION
must provide a column that relates back to the base table's partition key. ────────── How to Fix This To make the MV partitioned, you must ensure 
ingested_at
is part of the "Identity" of the row in your grouping logic so the lineage is preserved. Option A: Include the day in your GROUP BY If you want the MV to be partitioned by day, you should group by the day as well. This allows StarRocks to map the base table's daily partitions to the MV's daily partitions.
Copy code
sql
CREATE MATERIALIZED VIEW mv_users
PARTITION BY date_trunc('day', ingested_at) -- This now "relates" to the column below
AS
SELECT 
    bh.integration_id,
    bh.user_id,
    -- ... other columns ...
    bh.ingested_at -- This is now a "clean" column from the subquery
FROM (
    SELECT 
        integration_id, 
        remote_id AS user_id,
        date_trunc('day', ingested_at) AS ingested_at, -- Group by the day
        max_by(payload, ingested_at) AS payload
    FROM raw_ingest
    WHERE integration_type = 'bullhorn'
    GROUP BY integration_id, remote_id, date_trunc('day', ingested_at)
) bh
UNION ALL
-- ... repeat same logic for Vincere branch ...
Option B: If you don't need incremental refresh If your 
raw_ingest
table is small or not partitioned, and you simply want the MV to be partitioned for query performance (and you're okay with a full refresh), you can often bypass this by wrapping the entire query in a outer SELECT:
Copy code
sql
CREATE MATERIALIZED VIEW mv_users
PARTITION BY date_trunc('day', ingested_at)
AS 
SELECT * FROM (
   /* Your entire existing UNION ALL query here */
) t;
Note: This might still fail if the base table isn't partitioned, as StarRocks strongly enforces the base-table-to-MV partition relationship for Async MVs. Summary Check: 1. Is 
raw_ingest
partitioned by 
ingested_at
? If not, you cannot use 
PARTITION BY
on the MV. 2. If it is, ensure you are not using 
MAX(ingested_at)
. Instead, include the column (or the truncated version) in your 
GROUP BY
list so the relationship to the base table's storage is maintained. References • using_starrocks/async_mv/use_cases/create_partitioned_materialized_view.md
d
@Rocky perhaps I'm getting confused here. All I want is to get data from raw_ingest into the dim/fct tables (such as dim_jobs). Should I be using a materialised view, or should I be doing something else? MVs feel like an unnecessary middle step.
r
You are absolutely right to question this. In many StarRocks architectures, using an asynchronous Materialized View (MV) just to populate a Primary Key (PK) table is indeed redundant and adds unnecessary complexity and storage. The Problem with the MV Approach As you've seen, Asynchronous MVs have strict "partition alignment" rules. They are designed more for query acceleration (transparently making 
SELECT
faster) than for ETL pipelines. When you use an MV to feed a PK table: 1. You store the data twice (once in the MV, once in the PK table). 2. You pay for the compute twice. 3. You run into the "partition function must relate with column" errors because MVs try to be too smart about partition dependencies. ────────── The Better Way: StarRocks Tasks (Internal Scheduler) If your goal is to get data from 
raw_ingest
into 
dim_users
(a Primary Key table), the standard "StarRocks-native" way is to use a Scheduled Task. A Task is simply an 
INSERT INTO ... SELECT
statement that StarRocks runs on a schedule. It gives you the full power of SQL without the strict rules of MVs. How to convert your logic to a Task: Instead of creating an MV, you just create a task that populates your 
dim_users
table:
Copy code
sql
CREATE TASK task_refresh_dim_users
SCHEDULE BY '5 MINUTE' -- Or '1 DAY', or CRON '0 0 * * *'
AS 
INSERT INTO dim_users (
    integration_id, user_id, created_at, modified_at, deleted_at, 
    first_name, last_name, email, job_title, properties, 
    custom_fields, ingested_at, materialized_at
)
SELECT 
    /* ... Put your full SQL logic from the MV here ... */
    /* ... including the Bullhorn and Vincere UNION ALL ... */
FROM raw_ingest ...
Why this is better for your case: 1. Direct Loading: You skip the middle-man. Data goes straight from 
raw_ingest
to 
dim_users
. 2. No Partition Errors: Tasks don't care about partition alignment. You can use 
MAX()
, 
GROUP BY
, and complex JSON logic freely. 3. Leverages PK Table Strengths: Since your 
dim_users
has 
PRIMARY KEY
and 
merge_condition
, you can just run this task frequently. StarRocks will handle the upserts and out-of-order logic automatically when the data lands in the table. 4. Flexible ETL: You can easily add a 
WHERE ingested_at > now() - interval 1 day
to your task to make it incremental and fast. When WOULD you use an MV? You only use an MV if you want StarRocks to automatically rewrite queries. _ _Example:* If users query 
raw_ingest
directly, but you want StarRocks to "secretly" redirect them to a pre-computed result to save time. * For Dimension and Fact modeling, creating dedicated PK tables and populating them via Tasks (or an external tool like dbt) is the recommended path. Recommendation 1. Keep your 
dim_users
table (the one with the Primary Key and 
merge_condition
). 2. Drop the Materialized View. 3. Create a Task to run your transformation logic. You can start the task manually to test it:
Copy code
sql
SUBMIT TASK task_refresh_dim_users;
-- Check progress
SELECT * FROM information_schema.task_runs WHERE task_name = 'task_refresh_dim_users';
References • sql-reference/sql-statements/loading_unloading/ETL/SUBMIT_TASK.md • loading/InsertInto.md • administration/management/task_management.md • using_starrocks/async_mv/use_cases/data_modeling_with_materialized_views.md References • loading/InsertInto.md • sql-reference/sql-statements/loading_unloading/INSERT.md
d
@Rocky what's the downside of using TASKS?
r
While Tasks are often the right choice for "heavy lifting" ETL (like your logic with complex JSON and grouping), they do have some distinct trade-offs compared to Materialized Views. Think of it this way: Tasks are an "imperative" ETL tool (you tell StarRocks how and when to move data), while MVs are a "declarative" modeling tool (you tell StarRocks what you want the data to look like, and it manages the rest). Here are the specific downsides of using Tasks: 1. Manual Incremental Logic (The "Watermark" Problem) With a Task, StarRocks doesn't know what data it has already processed. * MV: If set up correctly with partitions, an MV automatically tracks which source partitions have changed and only refreshes those. * Task: You must write your own 
WHERE
clause logic (e.g., 
WHERE ingested_at > now() - interval 1 hour
). If a task fails or your "lookback" window is too short, you might miss data. If it's too long, you’re re-processing data unnecessarily. 2. No Automatic Query Rewrite This is the biggest "killer feature" of MVs that Tasks lack. * MV: If a user (or a BI tool) writes a slow query against the 
raw_ingest
table, StarRocks' optimizer can say: _"Wait, I have the answer already calculated in 
mv_users
. I'll just get it from there instead."_ This happens transparently. * Task: Users must know to query 
dim_users
. If they query the raw table, they get the slow performance of the raw table. 3. Manual Dependency Management (DAGs) If your data pipeline has multiple steps (e.g., Raw → Refined → Fact Table): * MV: StarRocks understands the lineage. If you refresh the final Fact MV, it can automatically check if the upstream Refined MVs need refreshing first. * Task: You have to coordinate the timing yourself. If Task A (Refined) starts late and overlaps with Task B (Fact), Task B might process incomplete or old data. 4. Boilerplate and Maintenance * MV: You only maintain one object (the MV definition). The schema of the underlying storage is managed for you. * Task: You have to maintain the Table Schema (the 
CREATE TABLE
statement) and the Task Query separately. If you add a column to your logic, you have to 
ALTER TABLE
and then 
ALTER TASK
. 5. Freshness Monitoring * MV: You can check the 
LAST_REFRESH_TIME
and 
FRACTION_UPDATED
easily in system views to see exactly how "stale" your data is relative to the source. * Task: Monitoring is a bit more manual—you have to check 
information_schema.task_runs
to see if the last 
INSERT
succeeded, but that doesn't tell you if the data itself is actually up to date with the source. ────────── Which one should you choose? Feature | *Use TASKS if...* | *Use MATERIALIZED VIEWS if...* Logic Complexity | You need complex grouping, window functions, or multi-step logic. | Your logic is a relatively straightforward transformation or aggregation. Primary Key Table | You want to land data into a 
PRIMARY KEY
table (for 
merge_condition
). | You are okay with the MV's internal storage (Duplicate Key). Querying | You are building a formal "Star Schema" for BI tools to query directly. | You want to speed up existing queries against raw tables without changing SQL. Performance | You need maximum control over exactly how and when data moves. | You want StarRocks to handle partition-level incremental updates for you. My advice for your current project: Stick with the Task for populating 
dim_users
. Your logic (JSON parsing + Grouping + Union) is exactly what Tasks are built for, and the 
merge_condition
on your Primary Key table is a cleaner way to handle out-of-order data than the MV's partition alignment rules. References • using_starrocks/async_mv/use_cases/data_lake_query_acceleration_with_materialized_views.md
d
@Rocky with tasks, we will lose the benefits of partitioning though correct?
r
Actually, that is a common misconception! You do not lose the benefits of partitioning when using Tasks. Partitioning is a feature of the underlying table storage, not the Materialized View (MV). In fact, using a Task with a partitioned table is often easier to manage than an MV because it avoids the "related with column" errors you encountered. Here is how you keep the benefits: 1. Automatic Partition Creation (Expression Partitioning) In StarRocks v3.0+, you can use Expression Partitioning. When you create your 
dim_users
table, you define the partition logic once. When your Task runs an 
INSERT INTO
, StarRocks will automatically create new partitions based on the data.
Copy code
sql
CREATE TABLE dim_users (
    user_id INT,
    ingested_at DATETIME,
    ...
) 
PRIMARY KEY (user_id)
-- This is the "Magic" part:
PARTITION BY date_trunc('day', ingested_at) 
DISTRIBUTED BY HASH(user_id);
* The Benefit: You don't have to manually run 
ADD PARTITION
. If your Task inserts data for "2024-01-20", StarRocks creates that partition on the fly. 2. You Still Get "Partition Pruning" Even though a Task populated the data, the StarRocks optimizer sees that 
dim_users
is partitioned by day. _ If you run 
SELECT _ FROM dim_users WHERE ingested_at >= '2024-01-01'
, StarRocks will still skip all other partitions. The query performance is identical to an MV. 3. Automatic Data Lifecycle (TTL) One benefit of partitioning is dropping old data. You can set this up directly on the table using the 
partition_live_number
property. StarRocks will automatically drop the oldest partitions for you, regardless of whether a Task or MV put the data there.
Copy code
sql
ALTER TABLE dim_users SET ("partition_live_number" = "30"); -- Keep only last 30 days
4. How to handle "Incremental" loads in a Task This is the only part that is more "manual" than an MV. You need to tell the Task which data to move. A common pattern is to look back at the last hour/day:
Copy code
sql
CREATE TASK task_refresh_users 
SCHEDULE BY '1 HOUR' 
AS
INSERT INTO dim_users
SELECT * FROM raw_ingest 
WHERE ingested_at >= now() - INTERVAL 2 HOUR; -- "Lookback" window
*Since your 
dim_users
is a Primary Key table, any overlapping data from that 2-hour window will just be upserted (updated), so you don't get duplicates.* Summary Comparison Feature | Materialized View | Table + Task Partition Pruning | Yes | Yes Auto-create Partitions | Yes (via alignment) | Yes (via Expression Partitioning) Auto-cleanup (TTL) | Yes | Yes Incremental Logic | Automatic (strict rules) | Manual (via WHERE clause) Flexibility | Limited (must "relate" to base) | Unlimited (any SQL works) Bottom Line: Use a Task. You keep all the performance and storage benefits of partitioning, but you gain the freedom to use any SQL logic (like your complex JSON grouping) without the "partition function must relate with column" headache. References • table_design/data_distribution/expression_partitioning.md • table_design/data_distribution/feature-support-data-distribution.md
Open in Slack
PreviousNext
Powered by