Sharding, Replication, and Distributed Execution in Apache Doris

Course README · Open Lab 9 · Open Quiz 9

Module goal

See how Doris spreads one table over several BEs and keeps it available: how Tablets and their replicas are placed, what happens when a BE joins, an FE fails, or a BE stops during a write, and which rows a distributed Join moves.

An ecommerce event service has outgrown the single-FE, single-BE sandbox of the earlier modules. Operators need to add BE capacity without changing the table's shards, new client sessions must be able to avoid a failed FE, writes must continue through a supported replica failure, and a recurring event–user report should avoid unnecessary network movement. Three problems stay separate throughout: giving clients a healthy FE entry point, placing and recovering Tablet replicas across BEs, and moving intermediate rows only where a query requires it.

Learning objectives

After completing this module, you will be able to:

1. Explain how Partitions, Buckets, Tablets, and replicas relate, and count them for a table.
2. Explain why adding a BE moves existing replicas without changing the table's Partitions or Buckets.
3. Separate FE metadata election from the client connection that must be reopened after its FE fails.
4. Trace a query from Tablet selection to the BE scan tasks that a Query Profile records.
5. Apply the default majority-write rule, and tell a live BE from a current replica after an outage.
6. Explain how Bucket Shuffle and Colocate Join reduce the data that a distributed Join moves.

Module structure

9.1 Identify the Roles in a Distributed Doris Request

The event service is moving beyond one FE and one BE. Before changing the topology, the team must identify which component accepts the client session, which component plans the query, and which components store and scan data. A client submitting SQL does not connect directly to every BE. It opens a MySQL protocol session to an FE query port, or to a proxy that selects an FE for that session.

The client reaches one FE session; the FE plans and coordinates work that selected BEs execute.

Frontend responsibilities

An FE maintains and uses cluster metadata, parses SQL, creates a distributed query plan, selects execution locations, dispatches plan fragments, and coordinates the result. In a high-availability FE deployment:

• Follower FEs participate in metadata replication and election;
• one Follower is the current Master;
• other healthy Follower FEs remain part of the voting group; and
• an Observer can replicate metadata for read scaling but does not participate in election.

The Master role can move after failure. The hostname that an application used does not move with it.

Backend responsibilities

A BE stores Tablet replicas and executes assigned operators. A scan Fragment Instance on a BE reads selected local Tablet replicas, and later operators can aggregate, join, sort, or exchange intermediate data.

Adding more BEs can provide:

• more storage locations for Tablet replicas;
• more possible scan and execution locations; and
• more aggregate CPU, memory, and I/O capacity in a real multi-host cluster.

It does not automatically add Buckets to an existing table, and it does not guarantee that every query uses every BE.

Proxy and client responsibilities

A TCP proxy or multi-address driver chooses an FE for a new session. It does not create Doris plans, store Doris metadata, or execute BE operators. An application still owns retry policy for a connection or request that fails.

Keep these health questions separate:

Each row answers only its own question; none of them answers the other five.

The responsibilities are now separate. The data layer begins with two BEs and two replicas; the first change adds a third BE without redefining the event table's Partition or Bucket layout.

9.2 Relate Partitions, Buckets, Tablets, and Replicas

Module 4 introduced Partition and Bucket design. A multi-BE cluster makes the physical consequences visible.

For the base index of a table with a fixed Bucket count per Partition:

Adding a BE creates another placement target; it does not change the logical shard definition.

For example, a two-Partition table with six Buckets per Partition has twelve logical Tablets in its base index. With two replicas, Doris maintains two physical copies of each logical Tablet.

Partition A                       Partition B
  Bucket 0 -> Tablet A0             Bucket 0 -> Tablet B0
  Bucket 1 -> Tablet A1             Bucket 1 -> Tablet B1
  ...                               ...

Tablet A0 replicas: copy on one BE + copy on another BE

SQL reads the logical rows represented by Tablet A0 once. Replication does not double the business answer.

When every BE stores every Tablet

If the number of available placement targets equals the replication count, every logical Tablet needs one replica on every BE. With two BEs and two replicas, both BEs cover the complete set of logical Tablets.

This does not mean Doris stores one indivisible copy of the table on each BE. The storage unit is still a Tablet replica, and each Tablet has its own ID, version, size, and health state.

When BEs store different Tablet subsets

After adding a third BE while keeping two replicas per Tablet, Doris has more placement targets than copies. The Tablet scheduler can move some existing replicas to the new BE while keeping the configured replica count:

before expansion
    Tablet T1 -> BE A, BE B
    Tablet T2 -> BE A, BE B

after balancing
    Tablet T1 -> BE A, BE C
    Tablet T2 -> BE A, BE B

The exact balanced distribution depends on scheduler state, disk usage, placement constraints, and cluster load. A new BE reporting Alive=true means that it joined the available topology. It does not prove that existing replicas have already moved.

Expansion does not reshard the table

Adding a BE can change replica placement. It does not automatically change:

• Partition boundaries;
• the table's distribution column;
• the number of Buckets in an existing Partition;
• logical Tablet IDs; or
• SQL-visible rows.

More placement targets and more logical shards are different changes. A table with too few Buckets can still limit scan parallelism after adding BEs.

Read placement metadata with the correct scope

Do not add the capacity value reported by every BE container in a shared Docker filesystem as if each value represented an independent disk. Likewise, Tablet metadata RowCount is not a substitute for an exact SQL count.

Use a bounded wait when observing balancing. A fixed sleep can finish before the scheduler acts or waste time after balancing is complete. If a small table does not visibly move within the observation window, report that limit instead of forcing a manual move and calling it automatic balancing.

See Elastic Expansion and Data Replica Management.

Expansion changes where replicas can live. The application, however, is still connected directly to one FE. Stopping that FE tests the next boundary: whether Doris metadata election also preserves the client's existing network session.

9.3 Separate FE Election from Client Load Balancing

Suppose three Follower FEs form the voting group and the client directly connects to the current Master. If that FE stops, the remaining majority can elect another Master and preserve FE metadata service availability.

The existing MySQL TCP session still terminates:

Election can preserve the FE service, but the application must create a new session through a healthy entry point.

The application must establish a new session through another address or a stable entry point.

Choose an entry mechanism

Lab 9 uses HAProxy because the Notebook client is Python rather than JDBC and the experiment needs visible TCP health and connection counters. This is an environment choice, not a Doris requirement or a universal recommendation.

Understand what HAProxy counts

A proxied MySQL session has two TCP legs:

Python client ---> HAProxy listening port ---> selected FE query port

The Lab's statistics use these fields:

Round-robin distributes newly created sessions, not individual statements inside one persistent session. A connection pool can therefore make statement counts look uneven even when new sessions were distributed as configured.

Use distinct evidence for distinct behavior

A controlled FE failure should establish four facts separately:

1. The directly connected FE was identified before it was stopped.
2. The old connection failed rather than migrating invisibly.
3. The surviving FEs retained a voting majority and one became Master.
4. A newly created proxy session avoided the failed target and completed a business query.

This supports “new connections can be re-established around one failed FE.” It does not support “in-flight queries migrate” or “the system has zero interruption.” One HAProxy container is also a single proxy failure point.

See FE Load Balancing.

After the failed FE returns, the service again has a complete FE voting group, a stable client entry, and replicas on all three BEs. The team can now follow one regional aggregation from the FE plan to the actual BE scan tasks.

9.4 Follow a Query from SQL to BE Scan Tasks

Replica placement shows where Doris could read a Tablet. It does not show which copy one query actually used.

Consider a regional aggregation over an event table:

SELECT region,
       COUNT(*) AS event_count,
       SUM(revenue) AS total_revenue
FROM event_detail
GROUP BY region
ORDER BY region;

The simplified distributed path is:

1. connected FE parses and optimizes the SQL
2. predicates prune Partitions and determine required Tablets
3. FE selects one healthy replica for each required Tablet
4. physical plan is split into Plan Fragments at Exchange boundaries
5. Fragment Instances are assigned to selected BEs
6. each BE builds PipelineTasks and scans its local selected Tablets
7. local partial results travel through Exchanges
8. an upper/root Fragment merges results and returns them to the client

Distinguish the execution units

One BE can run several PipelineTasks. Counting PipelineTask blocks does not tell you how many replicas exist or how many BEs participated. Local pipeline parallelism is different from distributed node parallelism.

Read the plan from the scan upward

A cropped plan can contain a structure such as:

PLAN FRAGMENT: scan and partial aggregation
  TABLE: event_detail(base index)
  partitions = selected/total
  tablets = selected/total
  numNodes = planned node count

EXCHANGE

PLAN FRAGMENT: merge or final aggregation

This establishes the planned Tablet scope, Fragment boundaries, and data distribution. It does not establish the actual host or actual scan count for one execution.

Read the DetailProfile as runtime evidence

The Profile can then show:

Fragment
  Pipeline(host=<BE address>)
    PipelineTask
      OLAP_SCAN_OPERATOR
        TabletIds: [...]
        ScanRows: ...

For each BE host, read the Tablet IDs and ScanRows inside the matching scan operator. If a BE stores an eligible replica but does not appear in this scan scope, it did not perform this table scan in that execution.

The FE normally selects one replica per required logical Tablet. It does not scan every replica and then deduplicate the business rows. Replica selection can vary between executions as node health, scheduling, and placement change.

Do not hard-code:

• container IP addresses;
• Tablet IDs;
• exact per-BE scan rows; or
• that every available BE must participate.

Several BE hosts in the scan Profile establish that the work was distributed; 9.7 states what a single Docker host cannot test.

For the plan hierarchy, see Product Concepts and EXPLAIN.

The regional aggregation shows how healthy replicas support distributed reads. The next operational question is whether the service can also publish a new event while one BE is unavailable, and how the returning copy becomes current after the write.

9.5 Commit Through a Majority and Recover the Lagging Replica

Replication affects read availability, write commit rules, storage cost, and failure tolerance. Those properties depend on the configured replica count.

By default, Doris considers a load successful when more than half of the replicas are written successfully:

This is why Lab 9 does not use its two-replica analytics table to promise a successful write during one BE outage. It creates an isolated three-replica event table after all three BEs are healthy.

The min_load_replica_num feature can change the required number according to the documented table and global configuration precedence. Lowering the threshold trades some immediate replica coverage for load availability. Do not assume the default majority rule when an environment overrides it.

Follow the commit and recovery states

Two of three replicas can commit the write; the returning BE is alive before its replica is necessarily current.

The exact numbers do not matter. Compare versions relative to each other. During the outage, FE metadata can retain the stopped replica's last known version. That does not mean the Notebook read files from a stopped process.

Heartbeat recovery is not data recovery

Two milestones must be checked separately:

1. SHOW BACKENDS reports the returning BE as alive.
2. Every target Tablet replica reports a complete current version.

The second can happen so quickly that an immediate sample already shows equal versions. If so, report that the transition was not captured; do not invent an older value. A bounded replica-health wait should finish only when the intended replicas are healthy and current.

Use stop/start or another process-level interruption that preserves the BE's data when teaching version catch-up. Removing the container and storage would be a different replacement-replica experiment.

Replica convergence is not backup. A logically wrong write can be replicated successfully to every BE. Module 10 separates online copies from recycle-bin recovery and independent backup.

See Load High Availability and Data Replica Management.

Once all three replicas are current again, temporary recovery state no longer confounds the next plan comparison. The team can use the stable layout to ask how much event–user Join input must move for a recurring report.

9.6 Compare Bucket Shuffle and Colocate Join Data Movement

The restored cluster repeatedly joins event measures to user segments on user_id. The logical Inner Join and its business answer stay fixed. The decision is how much Join input should move across the distributed execution.

The logical Join stays the same; the storage and placement contract determines which input moves.

Reuse the Event-Side Layout with Bucket Shuffle

Module 6 separated logical Join semantics, physical Hash Join, and Join data distribution. Here the business query is fixed so the distribution path is the only intended change.

The report asks for event count and revenue by user segment:

1. select one event time Partition;
2. aggregate event rows to one intermediate row per user_id;
3. join each user to a new, returning, or loyal segment; and
4. aggregate the joined measures by user_segment.

The event-side intermediate relation has this shape:

(user_id, event_count, total_revenue)

Because the event table is Hash distributed by user_id, the per-user aggregation can preserve an existing user_id Bucket layout. The first user dimension deliberately uses RANDOM distribution:

event-side state for user 1001 -> event Bucket selected by hash(user_id)
user row for 1001             -> arbitrary RANDOM user Tablet

The matching rows may not initially be at the same Join execution location.

Route only the user side

Bucket Shuffle keeps the compatible event-side layout. The Join-side Exchange routes each user row according to the event Bucket mapping:

RANDOM user row
    (1001, returning)
           |
           | hash and route with retained event Bucket mapping
           v
Fragment Instance responsible for that event Bucket
    event state (1001, count, revenue)
    user row    (1001, returning)
           |
           v
        local Hash Join

The user side moves; the retained event-side intermediate rows do not need a general repartition. A Bucket number is a logical position in the retained layout, not a permanent BE address. The FE still selects actual Tablet replicas and execution hosts for this query.

The Lab filters one physical event Partition because the documented direct physical-table Bucket Shuffle conditions require a usable single-Partition left-side layout. The broader Doris definition also allows an intermediate operator result to be the retained bucketed relation when its distribution property is preserved.

Read the plan evidence locally around the Join

The expected evidence has this shape:

VHASH JOIN
  join op: INNER JOIN(BUCKET_SHUFFLE)
  equal join conjunct: user_id = user_id
  input VEXCHANGE for the moved side

event scan: one selected time Partition
user scan: RANDOM-distributed table

An Exchange above the Join may still be required to merge final GROUP BY user_segment results or order the output. “Bucket Shuffle moves one Join input” does not mean “the entire query has no Exchange.” Read the Exchange in relation to the Hash Join and its two input branches.

The [shuffle] hint in the Lab prevents the small user dimension from taking a reasonable Broadcast path, making the Shuffle comparison observable. The hint does not prove Bucket Shuffle. The actual INNER JOIN(BUCKET_SHUFFLE) plan label is the evidence.

See Doris Joins and Join Optimization.

Pre-position Both Inputs for Colocate Join

Bucket Shuffle reuses one layout for one query. If two large tables repeatedly join on a stable Key, the table design can make the relationship durable.

A Colocation Group gives participating tables a compatible Bucket sequence and replica placement. For the event–user relationship:

BE A: event Bucket 0 replicas + user Bucket 0 replicas
BE B: event Bucket 1 replicas + user Bucket 1 replicas
BE C: event Bucket 2 replicas + user Bucket 2 replicas

Rows with the same user_id map to corresponding Buckets, and the replicas of those Buckets are placed on compatible BEs. The Hash Join can run locally at those locations.

Matching DDL is necessary but insufficient

For physical tables in the same Colocation Group, confirm:

• the tables belong to the same group;
• distribution columns have compatible types and counts;
• Bucket counts match;
• replica counts are compatible;
• the Join equality includes the Bucket Key; and
• group status reports IsStable=true.

Two tables that both say DISTRIBUTED BY HASH(user_id) are not necessarily colocated. Without a shared stable placement contract, corresponding Buckets can reside on different BEs.

Compare the same business answer

The Bucket Shuffle and Colocate queries must keep the same:

• time range and selected event Partition;
• event-side per-user aggregation;
• Join Key and logical Join type;
• user-segment data;
• output grouping and measures; and
• complete expected result.

Only the user table changes: its layout becomes HASH(user_id) in the Colocation Group, and its key becomes a Unique Key on user_id. Every user already has one row, so the key does not change the Join's input. The Colocate plan should report:

INNER JOIN(COLOCATE)

There should be no VEXCHANGE between that Hash Join and either input scan. An Exchange above the Join can still serve final aggregation or result merge.

Colocate is not automatically faster for every workload. A small Bucket count can limit parallelism, a Colocation Group constrains placement and balancing, and maintaining the relationship has an operating cost. Use it when repeated Join movement is a demonstrated bottleneck and the participating table designs can remain compatible.

See Colocation Join.

The two plans answer the same business question but use different placement contracts, and the VEXCHANGE next to each Join shows which input moves. The team must now state precisely what every topology, plan, and runtime observation supports.

9.7 State the Availability and Performance Boundary

Module 9 combines several successful observations, but each supports a limited conclusion.

Shared-host limitations

All Module 9 containers share:

• one Docker Engine;
• one physical host and memory pool;
• one physical failure domain;
• host networking and storage infrastructure; and
• one HAProxy container.

Stopping one process is useful for observing Doris state transitions. It does not test a host outage, rack or availability-zone isolation, production network throughput, independent disks, or proxy high availability.

Do not compare single-host elapsed times to rank distributed strategies. Use the Lab to identify selected plans, data movement, execution locations, and replica states. Measure production performance only with representative data, concurrency, hosts, and failure domains.

Preserve the intended failure experiment

Container commands also define the evidence boundary:

• stop/start preserves a container and its writable data for a return-and- catch-up experiment;
• removing a container or its storage creates a replacement-data experiment; and
• docker compose down can delete the isolated environment when data is not mounted externally.

The Lab therefore uses stop/start and never treats a destructive teardown as a temporary node failure.

9.8 Move from Replica Health to Data Lifecycle

The event service now has a healthy FE voting group, a reusable client entry, healthy online replicas placed across the available BEs, and understood Join distribution choices. Those properties keep current data available and place current query work. They do not decide how long historical events should remain online or how to undo a valid but incorrect operation that has reached every replica.

Keep the next set of responsibilities separate:

Module 10 therefore starts with a different business question: as the event history keeps growing, how should the team control storage cost, remove whole time ranges from the online table, handle late data, and recover from an eligible mistaken deletion?

Lab 9: Observe Sharding, Replication, and Distributed Execution

Open Lab 9 — Observe Sharding, Replication, and Distributed Execution after completing Lab 8, and plan about 70 minutes, a little more on a slow network. It uses an isolated Module 9 Compose project rather than the single-node doris-rta container from Modules 1–8.

Check the environment before you start:

• The Lab runs three FEs, three BEs, and an HAProxy container on one Docker host. Docker needs at least 4 CPUs, 8 GB of memory, and 20 GiB of free disk space. The first Lab cell checks these values before it starts any Module 9 container and stops with an explanation if one is too low. In the tested run, the seven containers used about 6.4 GiB of memory at their peak.
• Stop the single-node sandbox first with docker stop doris-rta. The Lab checks for it and stops instead of stopping it for you.
• With the Doris image already downloaded, the first start takes about one minute, including the HAProxy image download. On a new machine, downloading the Doris image takes longer.

The first complete run imports a read-only, bounded seven-hour course sample when no reusable local baseline exists. Its 12,062 rows are spread over all 40 course files, so Doris reads about 59 MB of them: on a slow network this load can take more than half as long as the full load in Lab 1. Later runs can preserve only the 12,062 committed logical event rows as a temporary local source, then rebuild the controlled physical event table. Other Module 9 business objects are rebuilt by the sections that own them. This avoids repeated S3 reads while ensuring the two-BE placement experiment starts from the intended DDL.

Read Lab 9 evidence in four layers:

1. SQL answers establish the visible business data.
2. FE and BE metadata establish membership, placement, and replica versions.
3. EXPLAIN establishes planned Fragments, scans, Join strategy, and Exchanges.
4. DetailProfile establishes the BE hosts, Tablet IDs, and scan rows for one execution.

The Stop and Restart controls preserve the isolated containers and verify the persisted event baseline, user dimension, and outage purchase. They are not a backup test. Do not run down while you still need the cluster.

If the Module 9 cluster reaches a state that the Lab cannot recover from, for example after a container was deleted or the FEs lost their majority, reset it. From the course directory, run:

docker compose --file level3/module09-sharding-replication/compose.multinode.yml down

Then rerun the Lab from Section 1.1. The command deletes only the Module 9 containers, the data stored in them, and their network; the single-node doris-rta container is not affected. The next run imports the seven-hour sample again. Use the same command to remove the cluster when you finish the course.

The Lab ends with a practice: count the Tablets and replicas of two tables before you look at them. Predict the result first, then write the code in the empty cell; the collapsed reference solution after it also runs with Run All and leaves the Lab's tables as they were.

Summary

• A table has one logical Tablet per Bucket in each Partition, and replication stores copies of every Tablet: two Partitions of six Buckets with two replicas are 12 Tablets and 24 replicas, and SQL still returns each row once.
• Adding a BE adds a placement target: the Tablet scheduler moves some existing replicas onto it and keeps the replica count, while the Partitions, the Buckets, and the logical Tablets stay as they are.
• When an FE fails, the surviving FE majority can elect a Master, but the client's session to the failed FE ends; a new session through a proxy or a multi-address driver reaches a healthy FE, and HAProxy's counters describe TCP sessions, not Doris queries.
• The FE selects one healthy replica of each Tablet that a query needs and sends Fragment Instances to the BEs; the plan shows the planned Tablet scope, and only the Profile shows which BEs scanned which Tablets.
• By default a load commits when more than half of the replicas succeed, so three replicas tolerate one stopped BE and two do not; a returning BE is alive before its replicas reach the current version, and replication is not a backup.
• Bucket Shuffle keeps one compatible bucketed input in place and routes the other input to it; Colocate places both inputs' matching Buckets together in advance, so the Join needs no Exchange for its inputs.

Quiz 9: Sharding, Replication, and Distributed Execution in Apache Doris

Open Quiz 9: Sharding, Replication, and Distributed Execution in Apache Doris. It has six single-choice questions and needs no running environment.

Next module: Module 10: Controlling Storage Cost and Data Lifecycle in Apache Doris.

References

• Architecture and execution: Product Concepts and EXPLAIN
• FE entry points: FE Load Balancing
• BE expansion and replicas: Elastic Expansion and Data Replica Management
• Write availability: Load High Availability
• Join movement: Doris Joins, Join Optimization, and Colocation Join
• Lab environment: All-in-One Image