arXiv CS open access

Distribution-Aware Distributed Database Testing (Extended Version)

· arxiv_cs

arXiv CS · Papers

Distributed database management systems (DDBMSs) introduce new challenges for assessing their reliability due to distribution-specific characteristics that affect query execution and optimization. Existing testing approaches, largely designed for centralized DBMSs, often fail to explore diverse dist

Conceptio Open Archive

Analyze & explore related research in Conceptio

Connect this record directly into your research workflow or AI agent. Explore citations across open repositories, query full text, and export structured proof bundles.

Explore in Search →Connect via MCP / API ↗

Distribution-Aware Distributed Database Testing (Extended Version) Zhou Zhou

Si Liu∗

East China Normal University

The Chinese University of Hong Kong, Shenzhen

Hengfeng Wei

Min Zhang∗

Hunan University

East China Normal University

arXiv:2609.18501v1 [cs.DB] 16 Sep 2026

Abstract Distributed database management systems (DDBMSs) introduce new challenges for assessing their reliability due to distributionspecific characteristics that affect query execution and optimization. Existing testing approaches, largely designed for centralized DBMSs, often fail to explore diverse distributed execution behaviors and suffer from low executability of generated test queries, thereby limiting their effectiveness in bug detection. We propose DAT (Distribution-Aware Testing), a novel automated approach for detecting query-processing bugs related to distribution strategies and distributed optimizations in DDBMSs, by systematically leveraging distribution-aware information throughout the testing pipeline. DAT builds on a set of techniques that capture diverse combinations of logical schemas and data distribution strategies, and performs guided query mutation to trigger a wide range of distributed query execution behaviors and optimizations, while improving query executability via historical feedback. We implement our approach in a tool, DistRanger, and evaluate it on four widely used production DDBMSs. It uncovers 31 previously unknown bugs, including 28 related to distributed query processing and optimization, and outperforms state-of-the-art testers. PVLDB Reference Format: Zhou Zhou, Si Liu, Hengfeng Wei, and Min Zhang. Distribution-Aware Distributed Database Testing (Extended Version). PVLDB, 1(1): XXX-XXX, 2026. doi:XX.XX/XXX.XX PVLDB Artifact Availability: The source code, data, and/or other artifacts have been made available at https://github.com/duerwuyi/DAT.

Introduction

Distributed database management systems (DDBMSs) are a key component of modern cloud computing infrastructure and are widely adopted to scale storage and computation beyond the limits of centralized DBMSs [22]. Unlike their centralized counterparts, where a schema fully determines data organization, DDBMSs additionally rely on distribution strategies [48], such as partitioning and ∗ Si Liu (liusi@cuhk.edu.cn) and Min Zhang (zhangmin@sei.ecnu.edu.cn) are the corre-

sponding authors. This work is licensed under the Creative Commons BY-NC-ND 4.0 International License. Visit https://creativecommons.org/licenses/by-nc-nd/4.0/ to view a copy of this license. For any use beyond those covered by this license, obtain permission by emailing info@vldb.org. Copyright is held by the owner/author(s). Publication rights licensed to the VLDB Endowment. Proceedings of the VLDB Endowment, Vol. 1, No. 1 ISSN 2150-8097. doi:XX.XX/XXX.XX

replication, to place data across physical nodes. This fundamental difference significantly impacts query processing. In particular, the execution of a SQL query in a DDBMS depends not only on its logical structure and semantics, but also on how the involved tables are distributed. As a result, the same query may exhibit vastly different execution behaviors, including data access patterns and communication costs, under different distribution strategies. To achieve efficient query processing under such complex execution settings, DDBMSs implement sophisticated distributed query optimizations, e.g., shard routing [40] and co-located joins [7], to reduce cross-node communication and localize computation whenever possible. However, these optimizations, as well as distributed query processing more broadly, are highly error-prone and can result in various types of bugs and issues in practice. These include logic bugs that produce incorrect query results (e.g., the one shown in Figure 1), crashes that lead to system failures, and timeouts that prevent queries from completing within a reasonable time. Challenges in Testing DDBMSs. Recent years have seen a surge of efforts on DBMS testing [8, 9, 30, 31, 33, 35–37, 49, 50, 52, 55, 65], which have found numerous bugs in real-world systems. However, existing testing approaches are unlikely to expose bugs like the one in Figure 1. For example, neither of the two representative state-ofthe-art testers, SQLancer [46, 52] and EET [31], identified this bug in a continuous 24-hour testing campaign (see also Section 5.3). The underlying reason is that these approaches are largely designed for testing centralized DBMSs and are agnostic to the distribution characteristics of DDBMSs. Consequently, they exhibit fundamental limitations in two key aspects when applied to DDBMSs, namely test diversity and query executability. First, test cases generated by existing approaches, though often complex (e.g., involving a range of SQL features and schemas), do not account for distribution-specific factors, such as distribution strategies and their interactions with query semantics. As a result, these approaches often fail to capture diverse execution behaviors unique to distributed settings. Moreover, their reliance on random testing, while effective for centralized DBMSs, often fails to trigger certain distributed query optimizations. This is because the distributed execution space is significantly larger and the conditions required to activate particular optimizations are rarely satisfied by chance. For instance, as shown in Figure 1, triggering the co-located join optimization bug in Vitess [62] requires specific structural patterns, i.e., two nested joins on a common sharding key sdkey (lines 12–13). This bug persisted for nearly three years across six major versions despite extensive prior testing efforts. Second, DDBMSs often reject SQL queries due to distributionspecific constraints or unsupported features [18, 57], as such queries

1 2

CREATE TABLE t0 (sdkey INT); CREATE TABLE t1 (sdkey INT);

3 4 5

/* VSchema ( specify dist . strategies per table ) */ {" sharded ": true , " tables " : { " t0 ": { " column_vindexes " : [ { " column " : " sdkey " , " name ": " i " } ] } , " t1 ": { " column_vindexes " : [ { " column " : " sdkey " , " name ": " i " } ] } } , " vindexes ": { " i " : { " type " : " xxhash " } } }

8 9

INSERT INTO t0 VALUES (NULL); INSERT INTO t1 VALUES (1);

10 11 12 13 14 15

Specifically, in addition to column-wise and row-wise table substitutions via 𝜌, it incorporates query transformations encoded in Δ, tailored to specific distributed query optimizations, such as shard routing and co-located joins. By applying sequences of mutations using 𝜇, DAT-generated queries can trigger a range of distributed execution behaviors and optimizations. To improve query executability, DAT incorporates a feedback mechanism during query mutation. By leveraging the history of previously executed and rejected queries in the DDBMS under test, it learns patterns of unsupported query structures under specific distribution strategies, which are encoded as feature paths, and guides subsequent generation toward diverse yet executable queries. This feedback loop, in turn, effectively mitigates the tension between test diversity and query executability. We implement DAT in a tool called DistRanger. It currently employs differential testing as the primary test oracle, i.e., identifying discrepancies between the execution results of the same query under centralized and distributed deployments of the same DBMS; any discrepancy indicates potential issues specific to the distributed layer. However, DAT can naturally incorporate other types of oracles, e.g., metamorphic ones [31, 45]. In addition, DAT can seamlessly integrate advances in query and schema generation into base query generation prior to mutation and SDM construction, respectively. Finally, the mutator 𝜇 is general and can be instantiated to incorporate additional distributed query optimizations beyond our current focus on shard routing and co-located joins. Together, these techniques establish DAT as a general framework for testing distributed query processing and optimization in DDBMSs. Sections 6 and 7 provide a detailed discussion of these aspects. Contributions. Overall, we make the following contributions. • Conceptually, we propose DAT, a novel approach for detecting query-processing bugs in DDBMSs that are related to distribution strategies and distributed optimizations. This problem presents fundamental challenges beyond centralized DBMS testing, which we address by leveraging distribution-aware information throughout the testing pipeline. • Technically, we develop a set of techniques that generate diverse test cases, including DDBMS instances and distributed queries, improve query executability, and mitigate the inherent tension between diversity and executability. • Practically, we realize DAT in an automated tester, DistRanger, and assess it on four production DDBMSs: Citus, Vitess, Apache ShardingSphere, and ClickHouse. It uncovers 31 previously unknown bugs (21 of which have been confirmed and 15 of which have been fixed), including 28 bugs whose manifestations or root causes are tied to distributed query processing and optimization. Compared to the state-of-the-art testers, DistRanger demonstrates superior effectiveness in bug detection.

SELECT * FROM (t0 AS r0 LEFT OUTER JOIN (t1 AS r1 JOIN t1 AS r2 ON (r1.sdkey = r2.sdkey)) ON (r0.sdkey = r1.sdkey)); --- expected result: one row returned --- actual result: none

Figure 1: A logic bug in Vitess found by our tool, along with the test query. Vitess incorrectly returns an empty result set, while the expected result contains one row. A JSON-based VSchema specifies distribution strategies per table (lines 4–7). sdkey denotes the common sharding key of the two tables.

may require non-trivial remote execution patterns and multi-round cross-node coordination (see also Section 2.3). As a result, a large fraction of queries generated by existing approaches, though valid in centralized DBMSs, fail to execute in distributed settings, which significantly limits overall testing efficacy from the outset. For example, we observe that, out of 100K queries produced by EET, 83% are executable on PostgreSQL, whereas only 25% are executable on its distributed deployment Citus [23]. Notably, these two aspects give rise to an inherent tension in DDBMS testing: increasing query complexity improves test diversity, but simultaneously undermines executability. This tension is largely absent in centralized DBMS testing and poses a fundamental challenge for testing DDBMSs. Our Approach. We introduce DAT (Distribution-Aware Testing), a novel automated testing approach for DDBMSs. It effectively discovers query-processing bugs related to distribution strategies and distributed optimizations that are unlikely to be uncovered by existing testing techniques. Our key insight is that distribution-aware information, i.e., data distribution strategies and their interaction with query semantics, can be systematically leveraged throughout the testing pipeline to address the aforementioned challenges. DAT begins by constructing a Schema-Distribution Matrix (SDM), which captures a wide range of combinations across diverse logical schemas and data distribution strategies. This provides a structured yet expressive exploration space for generating varied DDBMS instances for testing. In particular, this design offers two key properties: (i) column-wise, tables derived from the same schema can be interchanged to explore different distribution strategies without violating SQL syntactic validity; and (ii) row-wise, tables instantiated under aligned distribution strategies naturally satisfy the conditions required by certain distributed query optimizations. These properties enable effective test query generation over SDM, which DAT achieves through a generic mutator 𝜇 = (𝜌, Δ).

2 Background 2.1 Distribution Strategies In a centralized DBMS, a schema defines the logical organization of data as a set of relations (or tables). Yet, in a DDBMS, the schema alone does not determine data placement. The system must additionally define distribution strategies, which specify how data is 2

SELECT * FROM t2 WHERE t2.c0 = 140;

Table 1: Terminology for distribution strategies across four DDBMSs. Each row corresponds to a distribution strategy. Citus

ShardingSphere

Local [16] Reference [16] Distributed [16] Co-located [15]

Single [5] Broadcast [3] Sharding [4] Binding [2]

Vitess Unsharded [60] Reference [58] Sharded [59] Same Vindex [56]

ClickHouse MergeTree [20] N/A Distributed [19] N/A

sdkey

200

250

140

350

(a) All-shard scan

Node 1 t2_1

t8_1

Node 1 t2_1

t8_1

sdkey

200

250

200

250

140

t8_2

Node 2 t2_2 sdkey

sdkey

SELECT * FROM t2 WHERE t2.sdkey = 1; Node 2 t2_2

Node 1 t2_1

physically partitioned for scalability or replicated for availability across nodes [48]. These strategies are typically defined per table. In practice, DDBMSs often use different terminology for conceptually similar distribution strategies; Table 1 summarizes the terminology across four representative systems. To avoid ambiguity, we adopt the terminology of Citus [23] in this paper. Local tables store all data on a single node, allowing queries over them to be executed without cross-node communication. Reference tables store a full copy of the table on each node, either via explicit replication (e.g., in Vitess) or router-level broadcast (e.g., in ShardingSphere). This enables local access during distributed query execution. Distributed tables are horizontally partitioned across nodes based on a sharding key and sharding method. Querying such tables requires accessing multiple nodes and merging partial results. Co-located tables are distributed tables with aligned partitioning schemes, such that rows sharing the same sharding key reside on the same node. This enables efficient joins on the sharding key, commonly referred to as co-located joins [7]. Different DDBMSs often specify these distribution strategies using different languages, e.g., SQL or SQL-like DDL (Data Definition Language) statements in Citus, ShardingSphere, and ClickHouse, and a JSON-based schema in Vitess (Figure 1). Note that DDBMSs may impose restrictions on certain strategies (e.g., in Citus, local tables exist only on the coordinator node), and some systems provide additional system-specific variants [44]. A table may also simultaneously exhibit both partitioning and replication (e.g., replicated shards); yet, such cases can still be expressed using these strategies.

2.2

SELECT * FROM t2 JOIN t8 ON t2.sdkey = t8.sdkey;

SELECT * FROM t2 JOIN t8 ON t2.c0 = t8.c6

Node 2 t2_2

Node 1 t2_1

sdkey

200

250 350

140

(b) Shard routing scan

Node 2 t2_2 sdkey

t8_2

sdkey

250

200

250

200

350

140

350

140

(d) Co-located join

Figure 2: An illustration of the two distributed query optimizations: shard routing (a vs. b) and co-located joins (c vs. d). In each case, both tables are sharded by sdkey. SELECT * FROM t2 LEFT JOIN t11 ON t2.c3 = t11.c2 WHERE (t2.sdkey = 1) AND EXISTS( SELECT 1 FROM t4 WHERE t4.val > t2.val); Node 1 t2: shard1

LEFT JOIN

t11

correlated via EXISTS

Node 2 t2: shard2

t4: shard1

t4: shard2

t11: replia1

t11: replia2

Node 1 t2: shard1

Node 2 t2: shard2

t4: shard1

t4: shard2

t11: replia1

t11: replia2

Supported

Unsupported

ERROR: complex joins are only supported when all distributed tables are co-located and joined on their distribution columns.

Figure 3: An example of a SQL query rejected by Citus (v13.0).

Example 2.1. In Figure 2(a) and (b), the table t2 is sharded by t2.sdkey mod 2. As a result, querying t2 often requires an all-shard scan accessing both nodes, e.g., by SELECT ∗ FROM t2 WHERE t2.c0 = 140 (Figure 2(a)). However, when the WHERE clause includes the predicate t2.sdkey = 1, the DDBMS can leverage the sharding strategy of t2 to route the query to Node 1 only, as shown in Figure 2(b).

Distributed Query Processing and Optimization

Co-located Joins. A join combines rows from two tables based on a join condition. In a DDBMS, joining two distributed tables may require accessing rows across multiple shards when matching rows cannot be found locally. Such joins are referred to as cross-shard joins and can result in significant communication overhead. However, if the tables are co-located, i.e., rows with the same sharding key are placed on the same nodes, a join on the sharding key can be executed locally without accessing remote nodes. This optimization is known as co-located joins.

In a DDBMS, data is often partitioned across multiple nodes (or shards), and a query may need to access one or multiple shards depending on the data it touches. Hence, executing a query involves not only computing query results but also coordinating data access and communication across nodes. Distributed query optimization aims to minimize such communication overhead by exploiting data distribution and choosing efficient execution strategies. Below, we introduce two representative distributed query optimizations, shard routing and co-located joins, which are widely supported by existing DDBMSs [7, 14, 61]. Shard Routing. Since a distributed table is partitioned into multiple shards, scanning the entire table may require accessing all shards, which is referred to as all-shard access. However, if a query predicate identifies the shard containing the requested data, the DDBMS can avoid unnecessary accesses. This optimization is known as shard routing (or shard pruning) [7, 40].

Example 2.2. In Figure 2(c) and (d), t2 and t8 are co-located on sdkey, meaning that rows with the same sdkey reside on the same node. In Figure 2(d), the join condition t2.sdkey = t8.sdkey allows each shard of t2 to join only with the corresponding local shard of t8. In contrast, in Figure 2(c), the join is on c0 and c6 rather than the sharding key. As a result, each shard of t2 may need to join with all shards of t8, potentially incurring high communication overhead. 3

Base Queries

Schema-Distribution Matrix Distribution Schema 1 Schema 2 Strategy Local on the t0 t6 same node Local on t1 t7 different nodes Co-located t2 t8 (setting 1) Co-located t3 t9 (setting 2) Distributed

t10

Reference

t11

① DDBMS Instance

Generation

SELECT * FROM t0 LEFT JOIN t11 ON t0.c3 = t11.c2 Q� WHERE EXISTS(SELECT 1 FROM t4 WHERE t4.val > t0.val);

mutate Q� Q�

SELECT * FROM t2 LEFT JOIN t11 ON t2.c3 = t11.c2 Q� WHERE (t2.sdkey = 1) AND EXISTS(SELECT 1 FROM t4 WHERE t4.val > t2.val);

Yes, execute Q�

Historical Paths

No, mutate again

Accepted? No, update

Extract

Feature Paths

② SDM-Guided Query Mutation

Executed? Yes

Match

Mutated Query Candidates

Centralized Deploy.

Distributed Deploy.

occ: 90 200 100 suc: 50 195 8

③ History-Guided Feedback

Update

Result Comparison = ≠

④ Bug Checking

Figure 4: Workflow of our DAT approach.

2.3

Unsupported SQL Features in DDBMSs

foreign keys), and populating them with random data. It then generates a set of distribution strategies, each with multiple instances depending on how data is placed. Together, the schemas and distribution strategies form the schema–distribution matrix (SDM), which provides a structured yet expressive space for generating diverse DDBMS instances and guiding query mutation in Step 2 . Figure 4 shows an example SDM containing two logical schemas and six distribution strategies, including randomly distributed tables, reference tables, two types of co-location settings, and tables localized on a single node or across different nodes. Overall, this yields 12 distributed table instances.

Although distributed query optimization improves execution efficiency, DDBMSs often conservatively reject complex SQL queries at planning time and return errors indicating unsupported features [18, 57]. This is because processing such queries may introduce non-trivial remote execution patterns, and some executions may require multiple rounds of coordination across nodes, which fall outside the system’s design assumptions. Note that the set of unsupported SQL features may vary across DDBMSs and even across different versions of the same system. Nevertheless, these limitations are commonly rooted in mismatches between query semantics and data distribution constraints. The following illustrates a SQL query that is not executable in Citus (v13.0) and serves as a running example throughout the paper.

2 SDM-Guided Query Mutation. Given a DDBMS instance, DAT generates a base query and applies mutations to diversify it, aiming to trigger a range of distributed execution behaviors and optimizations. In particular, these mutations are guided by the SDM, e.g., by replacing a table with another one under a different distribution strategy to explicitly trigger the corresponding optimizations. Figure 4 shows an example of triggering the shard-routing optimization. Specifically, we select a table from the base query 𝑄 0 , namely the local table t0, on which shard routing cannot be directly applied. We then replace t0 with another table from the same column of the SDM, namely t2, and insert a shard-routing predicate t2.sdkey = 1 into the WHERE clause. This mutation increases the likelihood of triggering shard routing in the DDBMS under test. Additional mutation types are described in Section 4.2.

Example 2.3. As shown in Figure 3, t2 and t4 are two distributed tables, while t11 is a reference table. The query involves a shardrouting predicate t2.sdkey = 1 that restricts access to a shard of t2, as well as two types of interactions between tables. The first arises from t2 LEFT JOIN t11, which is supported by Citus. The second is an EXISTS-based correlation between t4 and t2. Despite shard routing on t2, the correlation with t4 still induces a cross-shard semi-join, which is not supported. As a result, the planner fails to produce a query plan for the entire query, which is ultimately rejected by Citus with an error. Query executability poses a major obstacle to DDBMS testing: a high rejection rate of generated queries prevents effective exploration of system behavior, thereby limiting bug detection.

3 History-Guided Feedback. Mutated queries may be rejected by a DDBMS due to unsupported features (Section 2.3), which in turn reduces overall testing efficacy. To address this, DAT incorporates a feedback mechanism that interacts with Step 2 , leveraging the history of previously executed and rejected queries by the DDBMS to guide generation toward executable queries. Specifically, this mechanism extracts structural features from a mutated query, e.g., join types and correlations, and encodes them as feature paths (intuitively, how tables are related through SQL operations). These paths are then matched against historical paths, which also maintain executability statistics (e.g., their frequency of occurrence and successful execution counts). As shown in Figure 4, the mechanism extracts structural features from a mutated query 𝑄 1 , such as the correlation between t2 and t4 and the presence of the t2 LEFT JOIN t11 construct, and encodes them as two feature paths. DAT rejects this candidate because one

Overview of DAT

This section provides an overview of DAT (Distribution-Aware Testing), an automated approach for detecting query-processing bugs in DDBMSs that are related to distribution strategies and distributed optimizations. DAT is unique in its ability to generate diverse yet executable queries that can effectively exercise a wide range of distributed execution behaviors and optimizations, thereby uncovering issues specific to the distributed layer of DDBMSs. Figure 4 shows DAT’s workflow, consisting of four main steps. 1 DDBMS Instance Generation. DAT starts by generating a set of logical schemas, including tables and their relationships (e.g., via 4

Schema s�

CREATE TABLE t0 ( c0 INT4 PRIMARY KEY, c1 DOUBLE, c2 TEXT, c3 TIMESTAMP );

...

Distribution Strategy × Co-located ×

× Distributed

shard_count = 2 method = hash sharding_key = c0

Reference ×

Distribution Table built Table built Strategy from s� from s�

node_id = rand

shard_count = 5 method = range sharding_key = c1

information required to define a table in a centralized setting, including its attributes, data types, and integrity constraints. Next, each schema, instantiated with random data, is combined with a distribution strategy to form a distributed table instance. For example, 𝑠 1 is paired with six different distribution strategies to generate the table instances t0 through t5. How are these distribution strategies generated? By default, we randomly configure their parameters, such as the number of shards, the sharding method, and the sharding key. This yields a distributed table instance in the “Distributed” row of the SDM, e.g., t4. While random generation is generally effective for exploring the space of distribution configurations, it may be insufficient for deliberately triggering certain distributed query execution logic and optimizations, which often require multiple distributed tables to satisfy alignment conditions. For example, co-located joins require participating tables to share the same sharding parameters. To this end, DAT introduces constraints on distribution parameters.

Schema-Distribution Matrix

node_id = 1

shard_count = rand × sharding_key = rand method = rand

s�

CREATE TABLE t6 ( ... FOREIGN KEY (c4) REFERENCES t0(c0) );

Parameters & Constraints

Local

Local on node 1

Local on diff. nodes

Co-located (setting 1)

Co-located (setting 2)

Distributed

t10

Reference

t11

replicas = 3

...

Figure 5: An illustration of how an SDM is constructed.

path matches a frequently rejected pattern in the historical record (in red ), despite another having a high success rate (in green ). As mutation proceeds iteratively, a subsequent candidate 𝑄 3 that passes this check is submitted to the DDBMS for execution.

Example 4.1. Consider the “Co-located” case, specifically the dashed box, in Figure 5. DAT defines a parameter setting for colocated tables by constraining the shard count (2), the sharding method (hash), and the sharding key (c0). It then instantiates multiple tables under this setting. As a result, table instances in the same row “Co-located (setting 1)”, e.g., t2 and t8, are naturally aligned and thus co-located, as they share the same sharding parameter setting, even though they are instantiated from different schemas.

4 Bug Checking. DAT executes the mutated query against the DDBMS under test to detect bugs. As shown in Figure 4, it employs differential testing to identify discrepancies between the execution results of the same query (e.g., 𝑄 3 ) under centralized and distributed deployments of the same system; any mismatch indicates a potential query-processing bug related to the distributed layer. Note that, regardless of whether a query is successfully executed, this information is fed back to Step 3 to guide future mutations. The above steps illustrate one iteration of our testing process, which can be repeated. In subsequent iterations, we can either continue from Step 2 with a new base query or restart from Step 1 to explore a new DDBMS instance with different SDMs. Step 3 persists throughout the entire testing process, recording information about all previously executed or attempted queries. DAT leverages recent advances [30, 53] in centralized DBMS testing to produce syntactically valid SQL queries and diverse schemas; these are not part of our contributions. The next section presents DAT’s core techniques tailored for testing DDBMSs.

When generating an SDM, although the types of distribution strategies (Table 1) are relatively limited compared to schemas, we consider a larger number of their instantiations to enable diverse distributed execution behaviors and optimizations. The exact number |D| is determined by the testing goal and the alignment conditions required to trigger specific optimizations, and may vary across DDBMSs due to differences in their supported strategies. Example 4.2. As shown in Figure 5, DAT introduces two parameter constraints for local tables, and similarly for co-located tables. This results in two distinct rows in each case. For example, local tables may be placed on a fixed node (both t0 and t6 on Node 1) or on different nodes (for t1 and t7). Compared to “Co-located (setting 1)”, t3 and t9 are co-located tables instantiated under a different parameter setting (five shards, range sharding on the key c1).

4 Core Techniques in DAT 4.1 DDBMS Instance Generation

Note that in an SDM, tables share the same schema columnwise. Replacing one with the other preserves SQL syntactic validity while potentially inducing different distributed execution behaviors. Tables share the same distribution strategy row-wise, enabling preconditions required by certain optimizations (e.g., co-located joins). Overall, the SDM provides a strong basis not only for generating diverse DDBMS instances, but also for guiding query mutation to explore a broad range of distributed behaviors, as we will see next.

When generating a database instance for testing, unlike in the centralized setting where only the logical schema is considered, the distributed setting must additionally account for how tables are partitioned or replicated across physical nodes. To this end, we organize table instances in a DDBMS using the Schema-Distribution Matrix (SDM), denoted by M. Let S be a set of logical schemas and D be a set of distribution strategies under consideration, with |S| = 𝑛 and |D | = 𝑚. Then M is an 𝑚 ×𝑛 matrix indexed by D × S, where each entry M [𝑑, 𝑠] denotes the table instance obtained by instantiating schema 𝑠 ∈ S under distribution strategy 𝑑 ∈ D and populating it with randomly generated data. Figure 5 illustrates the construction of the SDM shown in Figure 4, which consists of two schemas and six distribution strategies. First, we generate a set of schemas, e.g., 𝑠 1 , which specifies all

4.2

SDM-Guided Query Mutation

With the SDM in place, the next step is to generate test SQL queries. We start from base queries that are either generated by existing tools [30, 53] or observed to be executable during testing. For the former, we initialize them to access only local tables on the same node, i.e., from the “Local on the same node” row of the SDM. 5

SELECT * FROM t0 LEFT JOIN t11 ON join_cond WHERE pred;

❶ Constructing AST

❷ Extracting Feature Paths

SELECT

Base Query FROM

WHERE

❸ Normalizing

dist. table

LEFT JOIN

t11

...

ref. table

SELECT

SELECT * FROM t2 LEFT JOIN t11 ON join_cond WHERE pred;

SELECT * FROM t2 LEFT JOIN t11 ON join_cond WHERE t2.sdkey = 1 AND pred;

SELECT * FROM t2 LEFT JOIN t8 ON t2.sdkey = t8.sdkey WHERE pred;

휇 = ({푡0 → 푡2}, ∅)

휇 = ({푡0 → 푡2}, ∆��)

휇 = ({푡0 → 푡2, 푡11 → 푡8}, ∆�� )

Strategy Replacement

Predicate Injection

LEFT JOIN

AND

t11

EXISTS

SELECT

FROM

WHERE

FROM

dist. table

❹ Annotating t2.sdkey

Join Alignment

shard routing predicate

Figure 6: Examples of three types of query mutations, along with corresponding 𝜌 and Δ specifications.

t4.val

t2.val correlation

Such queries are unlikely to be rejected because this setting closely resembles a centralized DBMS. Given a base query, DAT proceeds by applying a series of mutations to it, aimed at triggering diverse distributed query executions and optimizations. Underlying this process is a generic mutator 𝜇 = (𝜌, Δ), where 𝜌 maps each table instance M [𝑑, 𝑠] in the base query to another (or the same) table instance M [𝑑 ′, 𝑠 ′ ] in the SDM, and Δ denotes a set of additional (optional) transformations to the base query tailored to specific distributed query optimizations. In the following, we present three instances of 𝜇, with corresponding examples shown in Figure 6. The first serves as a general mutation for exploring diverse distributed behaviors, while the remaining two target specific distributed query optimizations. We assume that the base query is executable in the DDBMS under test.

... SELECT

❺ Matching

dist. table SRP

LEFT JOIN

ref. table

...

ref. table

SELECT COR

occ: 200 suc: 195

... SELECT COR

FROM

dist. table

dist. table occ: 100 suc: 8

Figure 7: An illustration of the history-guided feedback mechanism using the SQL query in Figure 3. Example 4.4. In Figure 6 (middle branch), after 𝜌 replaces t0 with t2, ΔPI inserts the predicate t2.sdkey = 1 into the WHERE clause, making the sharding key of t2 explicitly appear in the filter condition. When processing this query, the optimizer can infer that only the shards containing rows with t2.sdkey = 1 are relevant. Join Alignment (JA). This mutation targets co-located joins, a query optimization that executes joins locally between distributed tables without cross-shard communication. ΔJA applies only when all table instances participating in a JOIN are already co-located after applying 𝜌 (based on the SDM). It then rewrites the join conditions so that joins between the participating tables are explicitly expressed as equalities on their common sharding keys.

Strategy Replacement (SR). In this mutation, 𝜌 replaces tables in a query with other tables from the same column of the SDM (i.e., 𝑠 = 𝑠 ′ ) but with different distribution strategies (i.e., 𝑑 ≠ 𝑑 ′ ). Since the new tables share the same schema as the original ones, the mutated query remains syntactically valid. Oftentimes, changing the distribution strategy of even a single table can already alter distributed query execution behavior. For example, 𝜌 can be used to trigger different join optimizations across table types, e.g., between a reference table and a distributed table. Note that SR does not involve additional transformations to the base query, i.e., ΔSR = ∅.

Example 4.5. In Figure 6 (right branch), after applying 𝜌, t0 and t11 are replaced with t2 and t8, respectively. ΔJA then rewrites the ON predicate as t2.sdkey = t8.sdkey, where t2 and t8 are colocated on the sharding key as determined during SDM construction. Consequently, the join is transformed into a form that is likely to be executed as a co-located join.

4.3

Example 4.3. In Figure 6 (left branch), 𝜌 replaces t0 with t2. Before the mutation, t0 and (a copy of) the reference table t11 can be joined locally on the same node. After the mutation, the join structure is preserved, but its execution semantics change, as t0 is replaced by the distributed table t2 (recall from Section 2.1 that colocated tables are a special case of distributed tables). Consequently, the join may no longer be executed locally on the same node.

History-Guided Feedback

While the mutation diversifies generated queries, many may be rejected by the DDBMS due to unsupported SQL features (Section 2.3), reducing overall testing efficacy. To address this, we introduce a feedback mechanism that guides the generation toward executable queries by leveraging historical successful and failed cases. Figure 7 illustrates our history-guided feedback mechanism using the SQL query in Figure 3. It begins by parsing the abstract syntax tree (AST) [1] of a mutated query candidate, which provides a structured representation of the query’s syntactic constructs. Formally, the AST of a SQL query 𝑄 is defined as 𝐴(𝑄) = (𝑉 , 𝐸), where 𝑉 is the set of AST nodes representing syntactic constructs in 𝑄, including statement constructs (e.g., SELECT), table-reference constructs (e.g., t2), relational operator constructs (e.g., LEFT JOIN), and expression constructs (e.g., the AND predicate); 𝐸 ⊆ 𝑉 ×𝑉 is the set of directed edges capturing parent–child relationships between nodes.

Predicate Injection (PI). This mutation is designed to trigger shard routing, a distributed query optimization that limits query execution to a subset of relevant shards rather than all shards. In addition to introducing at least one distributed table (via 𝜌 guided by the SDM), this mutation applies ΔPI which inserts a predicate into the WHERE clause of the base query. In particular, the predicate must be defined on the sharding key of the distributed table and inserted into the filter condition associated with that table. 6

The AST corresponding to the query in Figure 3 is shown in Step ❶, which can be constructed using existing approaches [30, 46]. We next describe the remaining four core steps, which determine whether a mutated query candidate qualifies as a test query.

affect execution behavior (e.g., by reducing cross-shard communication), but is not reflected in the feature path itself. To address this issue, we augment feature paths with node-level annotations to capture execution semantics beyond structural information, thereby enabling more accurate matching. In our current design, we consider two annotation families most relevant to distributed execution behaviors in DDBMSs: optimization annotations and correlation annotations.

❷ Extracting Feature Paths. We aim to avoid generating mutated queries that are similar to those previously rejected. A key challenge, therefore, is how to represent such similarity. Our insight is that if a base query is executable but a mutated query is not, the cause of rejection likely stems from unsupported features introduced by the replaced tables, as the mutation preserves the schema but alters their distribution strategies, along with the associated transformations (as defined in 𝜇). To capture this, we define feature paths. Intuitively, a feature path captures the structural relationship between two table nodes in the AST, encoding the sequence of relational operators and predicates that connect them, and thus characterizes how their distribution strategies jointly affect the executability of the query. Feature paths are extracted by starting from each mutated table node and traversing the AST, viewed as an undirected graph, to other table nodes. All intermediate nodes along such paths correspond to SQL constructs. As the AST is a tree, the path between any two table nodes is unique. Note that paths between unchanged tables are not extracted, as they are assumed to remain executable given that they already appear in the original executable query. As shown in Figure 7 (❷), two feature paths are extracted from the AST by traversing the blue and red paths, respectively, both starting from the mutated table t2.

Optimization Annotations. These annotations capture distributed optimizations associated with tables or operators in a query, such as shard routing and co-located joins, which are frequently triggered by our mutations. For example, for shard routing, we record routing predicates and annotate the corresponding table node with “SRP”. For example, the starting node of each path in Figure 7(❹) is annotated to indicate that a shard-routing predicate is applied to t2. Similarly, for co-located joins, normalizing participating tables into co-located tables is insufficient, as multiple JOINs may appear along a path. Therefore, we annotate the specific JOIN nodes that perform co-located joins with “CLJ”. Correlation Annotations. A correlation occurs when a subquery references a table in an outer query block. For example, t2 is referenced in an EXISTS subquery (via the predicate t4.val > t2.val), creating a dependency between the outer query and the subquery. Such dependencies may introduce additional execution and communication overhead in DDBMSs, e.g., when subquery evaluation depends on values from the outer query; moreover, they may require complex distributed semi-join or anti-join execution across shards, potentially causing an otherwise executable query to be rejected. Accordingly, correlation annotations capture such crossblock dependencies by annotating the corresponding nodes with “COR”, e.g., the SELECT node in Figure 7(❹), so that correlated paths can be distinguished during matching. Note that, following Step ❸, feature path equivalence is then redefined by incorporating these annotations. ❺ Matching against Historical Paths. DAT maintains a path set 𝑆𝐻 that stores feature paths extracted from all previously executed test queries, together with their execution statistics. Each element in 𝑆𝐻 is represented as a tuple (𝑝, occ, suc), where 𝑝 is a feature path obtained after normalization and annotation, occ is its occurrence count, and suc is its success count. Two paths are considered a match if they are equivalent, independent of their associated statistics. Let 𝑄 be a mutated query candidate, and let P (𝑄) denote its set of (normalized and annotated) feature paths. For each 𝑝 ∈ P (𝑄), if there exists an entry 𝑠 = (𝑝 ′, occ, suc) ∈ 𝑆𝐻 such that 𝑝 matches 𝑝 ′ ,

❸ Normalization. Queries that exhibit similar distributed execution behaviors may nonetheless produce different feature paths, because the above representation retains concrete table identities. For example, the execution of t2 LEFT JOIN t11 may be similar to that of t9 LEFT JOIN t5 in the DDBMS, as the involved tables share the same distribution strategies (i.e., a distributed table joined with a reference table; see Figure 5), yet are treated as distinct feature paths due to their different names. This introduces unnecessary variability, which inflates the feature path space, leads to a lower matching rate with historical patterns in Step ❺, and ultimately weakens the effectiveness of the feedback guidance. To address this issue, we normalize feature paths based on the effective distribution roles of table references, rather than their concrete names. The effective role of a table reference is the distribution strategy it exhibits in the given query context, as determined by its SDM configuration and the surrounding SQL structure. For example, t2 behaves as a distributed table, as shown in Figure 7 (❸), despite originating from a co-located table in Figure 5, as it does not participate in any co-located join. After normalization, two feature paths are considered equivalent if they share the same sequence of non-table SQL constructs and their endpoint table references have the same effective distribution roles. This yields a canonical representation, enabling matching across queries with similar distributed execution behaviors.

suc(𝑠) ≤𝜃 occ(𝑠) then 𝑄 is filtered out by DAT; otherwise, 𝑄 is submitted to the DDBMS for execution. Here, 𝑘 is the minimum occurrence threshold and 𝜃 is the success-rate threshold. In practice, we choose a sufficiently large 𝑘 (e.g., 10) to ensure that decisions are based on reliable historical evidence. Moreover, some unsupported features may not always lead to query rejection, e.g., due to optimizer pruning. Therefore, a path is considered unsupported only when its empirical success rate is consistently below 𝜃 . In our experiments, we set 𝜃 = 10%, and the rationale is provided in Section 5.4.2. occ(𝑠) ≥ 𝑘

❹ Annotation. Another limitation of original feature paths is that they fail to capture execution-relevant information outside the path. For example, a shard-routing predicate (t2.sdkey = 1) may restrict access to t2 to only a subset of its shards during distributed execution, rather than all shards. Such information may significantly 7

and

Table 2: Summary of the tested DDBMSs. DDBMS

Architecture

Base DBMS

GitHub Stars

LoC

Initial Release

Tested Version

Citus ShardingSphere Vitess ClickHouse

extension middleware cloud-native cloud-native

PostgreSQL PostgreSQL MySQL –

12.7K 20.8K 21.2K 49K

350K 790K 210K 1.1M

2012 2016 2010 2012

v12.5 w/ PostgreSQL 16.2; v13.2 w/ PostgreSQL 17.2; v14.0 w/ PostgreSQL 18.2 v5.5.2 w/ PostgreSQL 17.0 v24.0.0 w/ MySQL 8.4.8 v25.12.4.35

Table 3: Summary of the bugs detected by DistRanger, with distribution-related bugs shown in parentheses.

As shown in Figure 7(❺), two feature paths are matched in the history. One has a high empirical success rate (97.5%), while the other has a very low success rate (8%) due to the unsupported feature shown in Figure 3. Therefore, this candidate query is filtered out by DAT prior to its execution in the DDBMS.

DDBMS

Evaluation

We implement our DAT approach as a tool called DistRanger. It builds on the codebase of DynSQL [30], which provides a basis for generating syntactically complex and valid SQL queries that can be executed on centralized DBMSs,1 as well as diverse database schemas. The overall codebase consists of 17 KLOC in C++, including 1.6 KLOC for the query mutator and the feedback mechanism, 1.9 KLOC for DDBMS instance generation (i.e., SDMs) and bug checking, and 5.7 KLOC for supporting multiple DDBMSs. We conduct an extensive evaluation of DistRanger and compare it against state-of-the-art testers for DDBMS bug detection. We aim to address the following questions:

#detected

#confirmed

#fixed

Citus ShardingSphere Vitess ClickHouse

8 (8) 9 (7) 12 (12) 2 (1)

7 (7) 8 (6) 5 (5) 1 (0)

7 (7) 7 (5) 1 (1) 0 (0)

Total

31 (28)

21 (18)

15 (13)

Q1. Can DistRanger uncover previously unknown bugs in production DDBMSs (Section 5.2)? Q2. Does DistRanger outperform the state-of-the-art in generating high-quality queries and identifying bugs (Section 5.3)? Q3. How does each of our key design choices contribute to DistRanger’s effectiveness gains (Section 5.4)? Q4. What are the resource overheads of DistRanger (Section 5.5)?

testing, we construct, for each distributed deployment, a corresponding centralized deployment on a single node. In particular, for Citus, ShardingSphere, and Vitess, the centralized deployment corresponds to their underlying base DBMSs (see Table 2). For each SDM, we randomly generate 4–9 schemas and 4–6 distribution strategies of different types. Each generated table contains 40–120 rows chosen uniformly at random. We regenerate the SDM after every 1K generated queries, and each query is mutated for up to 10 times. While larger parameter settings (e.g., more schemas or mutation times) may expose additional bugs, the above settings already suffice to uncover many bugs in practice, as we will see. All experiments were conducted on a 64-bit Ubuntu 22.04 machine with an AMD Ryzen 5975WX (32 cores) and 256 GB RAM. Each data point is averaged over five runs.

5.1

5.2

Experimental Setup

We evaluate recent releases of four widely used production DDBMSs of different kinds. Citus [23] extends PostgreSQL into a distributed database, enabling horizontal scaling across multiple nodes. Apache ShardingSphere [34] operates as a database middleware layer (e.g., on top of PostgreSQL), aiming to maximize the computing capabilities of existing databases. Vitess [62] is a cloud-native distributed database system built around MySQL, supporting virtually unlimited scalability through generalized sharding. ClickHouse [21] is a standalone, columnar DBMS for online analytical processing (OLAP), enabling real-time analytical queries over large datasets. Table 2 summarizes their key characteristics, including architectures, popularity, codebase sizes, and tested versions. We deploy each DDBMS cluster using both Docker Compose and Kubernetes. For Citus and ShardingSphere, each cluster consists of one coordinator node and five data nodes. For Vitess, we employ five shard nodes and two primary nodes. For ClickHouse, we deploy two database nodes and three keeper nodes. For differential

Uncovering New Bugs

Overall, we reported 31 previously unknown bugs across the four DDBMSs, as summarized in Table 3. A complete list of these bugs, including the corresponding DDBMS, detailed description, bug type, status, and involved SQL features, is provided in the Appendix [66]. Among them, 28 are related to distribution-specific mechanisms across the query processing pipeline, including rewriting, planning, optimization, and execution, as well as distribution strategies. This demonstrates DistRanger’s effectiveness in triggering distributionspecific bugs beyond those observed in centralized DBMSs. The remaining three bugs include one timestamp format inconsistency and two SQL parser issues. Notably, some of these bugs remained undetected for years, e.g., half of the bugs in Citus went undetected for over four years. We present two additional representative bugs in the Appendix [66], complementing the one shown in Section 1. As of this writing, 21 bugs have been confirmed by the developers, with 15 already fixed in recent releases. Our test oracles are highly unlikely to produce false positives, as they rely on clear signals such as crashes, timeouts, errors, unexpected exceptions, and discrepancies in the results of the same query under centralized and distributed executions of the same system. We provide the

1 These queries involve common SQL constructs, e.g., DQL and DML, and advanced

features, including subqueries, aggregation functions, and window functions. 8

DDBMS

DistRanger!MT

DistRanger!FB

total

DistRanger dist.

total

SQLancer dist.

total

EET dist.

total

dist.

total

dist.

Citus ClickHouse ShardingSphere Vitess

3 2 4 7

3 1 2 7

0 0 – –

0 1 2 2

0 1 1 2

3 2 2 6

3 1 0 6

1 2 3 5

1 1 1 5

Total

# distinct feature paths

DistRanger

2 1

3 0

Time (hours)

DistRanger 16 (13)

6 (5) 0 (0)

12 16 Time (hours)

DistRanger!MT Vitess

·105

EET

Figure 8: Cumulative bugs detected in the four DDBMSs over 24 hours. Distribution-related bugs are shown in parentheses.

EET

·104

SQLancer

16 12 8 4 0 1

ShardingSphere

Citus

·104

Cumulative bugs

Table 4: Bugs detected over 24 hours of testing. SQLancer’s current implementation only supports testing Citus and ClickHouse.

Time (hours)

·103

ClickHouse

Time (hours)

Figure 9: Comparison of query diversity, measured as the cumulative number of distinct feature paths extracted from the executable queries on each DDBMS over 24 hours. Table 5: Comparison of query executability rates over 24 hrs.

distribution of detected bugs by type in the Appendix [66], where half of them are logic bugs.

5.3

DDBMS Citus ClickHouse ShardingSphere Vitess

Comparison with State-of-the-Art

We compare DistRanger against two representative state-of-the-art DBMS testing tools, namely SQLancer [46, 52] and EET [31]. We select SQLancer2 as a strong baseline for end-to-end database testing due to its success in finding numerous bugs in production DBMSs over the years. Its current implementation supports testing Citus and ClickHouse, for which Ternary Logic Partitioning (TLP) [45] is used as the test oracle. TLP partitions a query into three derived queries whose results are combined and compared with the original query result; any mismatch indicates a bug. EET is a recent testing tool that supports diverse queries by incorporating rich SQL features (e.g., subqueries and various join types). Given a query, EET rewrites selected expressions into semantically equivalent but structurally more complex forms, whose results are expected to be identical. Since EET also builds on DynSQL, its query generator is largely comparable to that of DistRanger, enabling a fair comparison of query executability and diversity in DDBMSs. Moreover, the shared codebase allows us to readily equip EET with our SDM to generate DDBMS instances, which enables comparisons between DistRanger and EET across all target DDBMSs.

DistRanger

SQLancer

EET

DistRanger!FB

87.6% 88.7% 72.0% 84.6%

38.9% 33.0% – –

25.3% 85.1% 3.1% 13.8%

47.0% 84.5% 22.6% 55.3%

In addition, we analyze how the bugs detected by each tool accumulates over time, as shown in Figure 8. Compared with SQLancer and EET, DistRanger provides two key advantages: it discovers the largest number of bugs overall and continues to uncover new bugs throughout the testing period. This demonstrates that DistRanger can detect bugs more effectively over time. These performance gaps arise from fundamental design differences. Both SQLancer and EET primarily target centralized settings, which limits the quality of their generated queries in terms of (i) executability, preventing a large fraction of them from being executed in DDBMSs, and (ii) diversity, hindering their ability to trigger diverse system behaviors. A detailed analysis is provided below. 5.3.2 Test Query Executability. Executability is crucial, as only executable queries can be used to exercise system behaviors and expose potential bugs. We measure the executability rate of queries generated by the three tools across the four DDBMSs. As shown in Table 5, DistRanger significantly outperforms EET (by up to 22.2×), as well as SQLancer (by up to 1.7× on the two systems it supports). This demonstrates the effectiveness of our feedback mechanism in guiding the generator toward queries accepted by DDBMSs. Note that DistRanger’s executability rate varies across DDBMSs. This is primarily due to differences in unsupported features across systems, as well as the limited time (24 hours) available for our feedback mechanism to learn and guide generation. Moreover, achieving

5.3.1 Bug Detection. To examine how effective each tool is in bug detection, we conducted a 24-hour empirical analysis. The results are shown in Table 4. Overall, DistRanger identifies a total of 16 bugs, including 13 related to distributed query processing and optimization in DDBMSs,3 significantly outperforming SQLancer and EET. Notably, all bugs detected by EET are also found by DistRanger. 2 Specifically, we use the latest version of SQLancer as of April 27, 2026 [51]. 3 Note that Table 3 reports a larger number of bugs, as our full testing campaign

spanned several months. 9

EET

DistRanger

250 7.05

10 6 4 2 0

2.49 1

DistRanger

DistRanger!MT

355.55 397.67

450 300

Mult. factor (×)

359.5 Mult. factor (×)

Mult. factor (×)

SQLancer 500

204.8 90.26

10 4 2

45.7 1.76 1

4.79 1

0 R us S

Cit

us C Cit

use

ck Cli

J J J R re SR re C itess S itess C ouse SR us C Cit Sphe Sphe V V g g ckH i n n l i i C ard hard Sh S

DistRanger

266.67

225.66

198.83 97.52 5.94

3.8 1

2.76 1

J J J R R us S itus C here SR here C itess S itess C use SR o C V V gSp ingSp ckH i n l i C ard hard Sh S

s SR

u Cit

300 200 100 10 4 2 0

Cit

Figure 10: Frequency of triggering distributed query optimizations, shard routing (SR) and co-located joins (CJ), across four DDBMSs. For ClickHouse, we evaluate only SR, as it does not support CJ. Results are normalized, and the y-axis reports multiplicative factors relative to the baseline. a 100% executability rate is inherently challenging. One reason is that queries containing unsupported features may still be accepted by the DDBMS due to optimizer branch pruning, which can mask invalid substructures. We defer a detailed analysis to Section 5.4.2.

Occurrences

SQLancer

5.3.3 Test Query Diversity. The diversity of generated queries is another key factor for effective DDBMS testing. A tester that repeatedly generates structurally similar queries may exercise only a limited portion of the distributed query processing logic, even if they are executable. Therefore, we evaluate whether DistRanger can continuously explore diverse queries over time, compared to EET (as both tools share a common base query generator). As described in Section 4.3, a feature path summarizes table-totable structural relationships together with their effective distribution strategies. Accordingly, the number of unique feature paths serves as a proxy for the diversity of distributed execution patterns covered by a tool. We measure this using the cumulative number of distinct feature paths extracted from executable queries generated during 24 hours on each DDBMS. As shown in Figure 9, DistRanger consistently covers more unique feature paths than EET across all the DDBMSs. This demonstrates that DistRanger enables broader exploration of distributed query execution behaviors.

EET

DistRanger!MT

·103

10 7.5 5 2.5 0

a Loc

in nce out ere rd r a h S

Ref

4 3 2 1 0

ard

-sh

All

ate

lo c

Co-

DistRanger

·103

d nce har ss-s Cro

ere Ref

Figure 11: Distributions of shard-access patterns (left), including local, reference-table, and distributed-table access, and distributed-join patterns (right), including co-located, reference-table, and cross-shard joins, exercised by 10K executable queries in Citus. The distributed-table access patterns include shard routing and all-shard access. tool exercise different distributed execution behaviors, we additionally measure various shard-access and distributed-join patterns (Section 2.2). This provides complementary insights into DDBMS behaviors beyond the trigger counts reported earlier for shard routing and co-located joins. Due to space limitations, we present the results for Citus in Figure 11 and defer the results for the remaining DDBMSs to the Appendix [66].4 All results align with our expectations: for the two target distributed optimizations, namely shard routing and colocated joins, significantly more queries generated by DistRanger exercise the corresponding patterns compared with SQLancer and EET. Additionally, due to the randomness in query generation, DistRanger also exercises patterns beyond the two target optimizations, and each query may exercise zero or multiple patterns.

5.3.4 Triggering Specific Query Optimizations. Another advantage of DAT is its generality: it can be instantiated to target specific distributed query optimizations, such as shard routing and co-located joins (Section 4.2). Accordingly, we compare how frequently three tools trigger these two optimizations. For each tool, we generate 10K executable queries per run across five runs under randomly generated DDBMS instances. We instrument key entry points in the optimization logic to record when each optimization is triggered (see the Appendix [66] for details); thus, trigger counts approximate its execution frequency. Since SQLancer does not support generating distributed tables for ClickHouse and thus cannot trigger the shard-routing logic, we randomly distribute the generated tables across the ClickHouse cluster to enable a meaningful comparison. As shown in Figure 10 (left and middle), DistRanger triggers the two distributed query optimizations more frequently than both SQLancer and EET, achieving improvements of up to several hundred times. This is expected, as our mutation strategies, i.e., ΔPI and ΔJA , are explicitly designed to trigger these optimizations, while query generation in SQLancer and EET is largely agnostic to them.

5.4

A Closer Look At DistRanger

We now analyze how each of DistRanger’s key design choices contributes to its effectiveness over state-of-the-art DBMS testers. 5.4.1 Ablation Analysis. We examine two variants: (i) DistRanger without query mutation (i.e., using random query generation instead), denoted as DistRanger!MT , to assess its effect on triggering diverse distributed query execution and optimizations; and (ii) DistRanger without the feedback mechanism, denoted as DistRanger!FB , to evaluate its impact on the executability of generated queries.

5.3.5 Coverage of Distributed Execution Patterns. To gain a more comprehensive understanding of how queries generated by each

4 For Citus, we analyze query plans, which expose accessed tables, shards, and whether

joins are executed in a co-located manner or across shards. 10

30 20 10 0

FP rate (%)

QE rate (%)

90 80 70 60 50

memory usage and runtime overhead of DistRanger using a 24hour testing campaign on Vitess as a case study. First, the primary source of DistRanger’s memory usage is storing the feature path set, as shown in Figure 13b. In comparison, the overhead of storing the SDM and query ASTs is negligible. This also explains why DistRanger consumes more memory than SQLancer and EET, as shown in Figure 13a. Nevertheless, this additional memory overhead is a worthwhile trade-off, since the feature path set underlies our feedback mechanism, which improves query executability and thus enhances bug-finding effectiveness, as we have demonstrated. Note that although DistRanger incurs higher memory usage, its overall memory consumption during 24-hour testing remains practical on modern machines. Moreover, Figure 13c shows the runtime breakdown of DistRanger. The majority of runtime is spent on executing queries and other components, including the logging and checkpointing mechanisms used for bug probing. In contrast, maintaining the feature path set accounts for only 0.37% of the total runtime overhead, indicating that this cost is negligible.

QE FP

0 30

𝜃 (%)

Figure 12: Query executability (QE) and false-positive (FP) rates under different values of 𝜃 . Overall, both variants perform significantly worse on their respective metrics, as expected. In particular, as shown in Table 5, DistRanger!FB exhibits a lower query executability rate than DistRanger across the four DDBMSs, e.g., with up to a 2.2× reduction for ShardingSphere. Moreover, DistRanger!MT explores fewer distinct feature paths than DistRanger within the same time budget, as shown in Figure 9, with up to a 6× reduction for ShardingSphere. In addition, we evaluate the effectiveness of our mutation strategies in targeting distributed query optimizations. Specifically, we compare how frequently DistRanger and DistRanger!MT trigger the two optimizations, namely shard routing and co-located joins. As shown in Figure 10 (right) and Figure 11 (with additional results in the Appendix [66]), DistRanger!MT triggers both optimizations much less frequently than DistRanger. For instance, the gap in Figure 10 reaches approximately 190–250× on ShardingSphere. Finally, we examine how the above degradations impact bug detection capability. As shown in Table 4, both variants find noticeably fewer bugs than DistRanger. This is expected: without the feedback mechanism, fewer queries can be successfully executed by the DDBMSs in the first place; without the query mutation, the probability of triggering diverse distributed execution behaviors is reduced, thereby lowering the likelihood of exposing bugs.

(D)DBMS Testing. Recent years have witnessed a surge of efforts on testing DBMSs, including SQLancer [46, 52] and EET [31]. However, bugs arising from query processing and optimization in distributed settings remain largely underexplored. Addressing this gap is the focus of our approach DAT. DistSQL [25] is a recent differential testing framework that targets logic bugs in DDBMSs. It compares the results of the same query executed on a DBMS configured in centralized and distributed modes. Yet, its code is not publicly available as of this writing. Given its complexity (approximately 28 KLOC), a reimplementation would require substantial assumptions and undermine a fair experimental comparison. Hence, we provide below a qualitative comparison. Overall, DAT differs from DistSQL in three key aspects:

5.4.2 Threshold Selection and Trade-offs. In our approach, the feedback mechanism rejects a feature path, and thus any query containing it, when its historical query executability (QE) rate falls below 𝜃 (Section 4.3). Figure 12 shows our analysis of 40K queries (10K per DDBMS). As 𝜃 increases, the overall QE rate improves and eventually stabilizes, while the false-positive (FP) rate continues to increase. The FP rate is defined as the fraction of executable queries that are rejected by the feedback filter. In black-box testing, false positives are difficult to avoid. Consider a simple example: a predicate such as WHERE TRUE OR right always evaluates to true. The optimizer may prune the right branch as a simplification, even if it contains unsupported features. As a result, the query can be executed by the DDBMS. When 𝜃 is too small (e.g., 0%–5%), branch pruning can delay learning, as the feedback mechanism may require many counterexamples before recognizing that a query is actually not executable, resulting in a limited QE rate. When 𝜃 is too large (e.g., ≥ 20%), executable feature paths may be misclassified as unsupported by our feedback mechanism, leading to higher FP rates and overly conservative mutations. Hence, in practice, we use a moderate 𝜃 (e.g., 10%) to balance this trade-off.

5.5

Related Work

(1) While DAT currently uses the same oracle for identifying bugs, it is designed as a general testing framework. Additional test oracles can be naturally incorporated to enable the detection of a broader range of issues (see below and Section 7). In contrast, DistSQL tightly couples its logic bug detection with differential testing, relying on query plan discrepancies across deployment modes to generate queries that are more likely to trigger bugs. (2) DAT emphasizes generating executable test queries, whereas DistSQL does not. Therefore, we conjecture that a non-trivial fraction of DistSQL-generated queries may be rejected by DDBMSs, thereby reducing overall testing efficacy. (3) DAT enables targeted generation of queries that trigger specific distributed optimizations via SDM-guided mutation (i.e., 𝜇), whereas DistSQL lacks such targeted exploration, potentially missing bugs related to these optimizations. Nonetheless, DistSQL can expose certain classes of bugs not yet covered by DistRanger. First, its query generator supports SQL features not emphasized in our implementation (e.g., window functions with PARTITION BY), enabling it to generate relevant bug-triggering queries. Second, it explores bugs related to system configurations, such as enabling optional components or changing parameters.

Resource Overhead Analysis

Analyzing the resource costs of a testing tool provides valuable insights into its practical trade-offs. Therefore, we measure the 11

Tool

Memory

SQLancer EET DistRanger

269.06 MB 12.19 MB 641.15 MB

Component

Memory

SDM Query ASTs Feature paths Others

0.84 MB 0.03 MB 615.98 MB 24.30 MB

DDBMS gen. Query exec.

Other runtime overhead RHS components

58.37% 6.59%

Query gen. Mutation

0.91%

Maintain path set

0.43%

34.13% (a) total memory usage comparison

(b) memory breakdown of DistRanger

0.11% 0.37%

Figure 13: Memory usage and runtime breakdown during 24-hour testing of Vitess. Other memory usage and runtime overhead are mainly attributed to the logging and checkpointing mechanisms used for bug probing. These gaps reflect implementation choices rather than fundamental limitations of DAT. To demonstrate its extensibility, we performed a preliminary extension of DistRanger by incorporating ShardingSphere’s configuration options into the DDBMS instance generation stage. With this extension, we uncovered a bug in ShardingSphere, which has been confirmed by the developers. This bug is triggered when two configurations, i.e., encryption and auto keygeneration, are enabled. Compared with the setting where both options are disabled, the same query produces different results. There is also a substantial body of research on testing distributed systems. While some involves distributed databases, its focus is largely orthogonal, e.g., on synchronization failures [24] and recovery bugs [26]. In contrast, DAT focuses on issues arising from query processing. Exploring synergies between DAT and these techniques, e.g., fault injection [29, 47], is interesting future work, as it may, e.g., help trigger more diverse distributed execution behaviors.

DAT is designed as a general testing framework that currently focuses on test generation and already integrates the oracles in (i) and (iii) within DistRanger. Extending it to incorporate additional oracles, particularly those in (ii), is feasible, either directly or with minor adaptations. For instance, incorporating SQLancer’s TLP or EET’s semantically equivalent query transformations incurs no additional cost: for a query and its derived query, we simply obtain an additional pair of execution results in the distributed setting; as a result, we compare the equivalence of four query results in total.

Discussion and Conclusion

We have presented DAT, a novel automated testing approach tailored to DDBMSs, together with its implementation, DistRanger. Using DistRanger, we have uncovered many previously unknown bugs in production DDBMSs and demonstrated its superior effectiveness in bug detection compared to the state-of-the-art. This work complements prior efforts on centralized settings by addressing issues that arise specifically in the distribution layers of DDBMSs. Next, we discuss the limitations of this work and outline additional directions for future work beyond those covered in Section 6.

From a technical perspective, our work is closely related to two lines of DBMS testing research, i.e., test generation and test oracles, which have thus far primarily focused on centralized settings. Test Generation. A large body of prior work [8, 30, 35–37, 42, 53, 55] focuses on generating diverse test cases, including complex SQL queries and varied database schemas and configurations. DAT builds upon these advances, e.g., DynSQL [30]. Among these approaches, mutation-based techniques are often used to increase the diversity of queries, while feedback mechanisms are employed to guide test generation, aiming to improve test case quality (e.g., trigger a wider range of system behaviors). However, queries that are syntactically valid and accepted by centralized DBMSs are not necessarily executable in DDBMSs due to distribution-specific constraints. Paradoxically, increasing query complexity, which has proven effective in triggering diverse system behaviors in centralized settings, can raise the likelihood of rejection in distributed settings. DAT addresses this dilemma by mutating queries based on the SDM and leveraging historical executability results to guide the mutation process.

Limitations. While testing can find bugs, it does not guarantee their absence. DAT shares this inherent limitation. Formally verifying DDBMSs could overcome this issue, but it is challenging [11, 27]. Moreover, DistRanger may produce duplicate bug reports, as different test cases can trigger bugs with the same root cause. We currently perform manual deduplication, while its automation remains a challenge in system testing [13, 63]. Finally, DAT relies on explicit specifications to define how data is distributed. For systems with distribution strategies that are dynamically determined or not directly controllable, we could extract the realized data distribution from system-level metadata (e.g., region-level placement information in TiDB [43]) and incorporate it into the testing process. Extension to Other Distributed Query Optimizations. We have demonstrated DAT’s effectiveness in triggering distributed query optimizations via shard routing and co-located joins. Other optimizations can be incorporated into DistRanger by defining Δ rules, along with the mapping 𝜌, that transform the query to satisfy optimization-specific structural preconditions. For example, testing aggregation pushdown [6, 17] (e.g., GROUP BY on the sharding key) can be enabled by replacing tables with sharded counterparts, rewriting the query to enforce the required aggregate and GROUP BY structure, and aligning grouping keys with sharding keys.

Test Oracles. An orthogonal line of work is dedicated to the design of test oracles, which determine whether a test output matches the expected result. Representative oracles include (i) differential testing [42, 53], which compares outputs of identical queries executed under different configurations; (ii) metamorphic testing [9, 31, 33, 45, 49, 50, 65], which verifies expected relations among the outputs of related inputs; and (iii) crash and exception detection [30], achievable by simply inspecting system outputs. 12

Discovering Other Types of Bugs. DAT can serve as a basis for detecting other types of bugs, e.g., inefficient distributed query processing. A key challenge lies in the lack of effective test oracles, i.e., determining the expected execution time of a distributed query. Moreover, despite recent advances in black-box checking of isolation levels [12, 28, 32, 38, 41, 54, 64], e.g., serializability and weaker levels [10, 39], they are largely limited to key-value stores. DAT can be extended to relational DDBMSs by wrapping mutated SQL queries as transactions and exploiting their interactions with diverse distribution strategies to explore concurrency behaviors.

[16] Citus Data. Accessed: April 12, 2026. Concepts: Table Types. https://docs. citusdata.com/en/stable/get_started/concepts.html#table-types. [17] Citus Data. Accessed: April 12, 2026. Querying Distributed Tables (SQL): Aggregate Functions. https://docs.citusdata.com/en/latest/develop/reference_sql.html. [18] Citus Data. Accessed: April 12, 2026. SQL Support and Workarounds. https: //docs.citusdata.com/en/stable/develop/reference_workarounds.html. [19] ClickHouse. Accessed: April 12, 2026. Distributed table engine. https://clickhouse. com/docs/engines/table-engines/special/distributed. [20] ClickHouse. Accessed: April 12, 2026. MergeTree table engine. https://clickhouse. com/docs/engines/table-engines/mergetree-family/mergetree. [21] ClickHouse. Accessed: Feb. 13, 2026. Fast Open-Source OLAP DBMS. https: //clickhouse.com/. [22] George Coulouris, Jean Dollimore, Tim Kindberg, and Gordon Blair. 2011. Distributed Systems: Concepts and Design. Addison-Wesley Publishing Company. [23] Umur Cubukcu, Ozgun Erdogan, Sumedh Pathak, Sudhakar Sannakkayala, and Marco Slot. 2021. Citus: Distributed PostgreSQL for Data-Intensive Applications. In Proceedings of the 2021 International Conference on Management of Data (SIGMOD ’21). ACM, 2490–2502. https://doi.org/10.1145/3448016.3457551 [24] Cundi Fang, Jie Liang, Zhiyong Wu, Jingzhou Fu, Zhouyang Jia, Chun Huang, Yu Jiang, and Shanshan Li. 2025. DepState: Detecting Synchronization Failure Bugs in Distributed Database Management Systems. Proc. ACM Softw. Eng. 2, ISSTA, Article ISSTA088 (June 2025), 22 pages. https://doi.org/10.1145/3728965 [25] Zi-Xuan Fu, Jia-Ju Bai, Hong-Bo Feng, and Kang Chen. 2026. One DBMS, Two Modes, and a Bunch of Bugs: Catching Logic Bugs in Distributed DBMSs via Differential Testing. Proc. ACM Manag. Data 4, 1, Article 59 (April 2026), 24 pages. https://doi.org/10.1145/3786673 [26] Yu Gao, Wensheng Dou, Dong Wang, Wenhan Feng, Jun Wei, Hua Zhong, and Tao Huang. 2023. Coverage Guided Fault Injection for Cloud Systems. In Proceedings of the 45th International Conference on Software Engineering (ICSE ’23). IEEE Press, 2211–2223. https://doi.org/10.1109/ICSE48619.2023.00186 [27] Heidi Howard, Markus A. Kuppe, Edward Ashton, Amaury Chamayou, and Natacha Crooks. 2025. Smart Casual Verification of the Confidential Consortium Framework. In 22nd USENIX Symposium on Networked Systems Design and Implementation (NSDI 25). USENIX Association, 259–276. https: //www.usenix.org/conference/nsdi25/presentation/howard [28] Kaile Huang, Si Liu, Zhenge Chen, Hengfeng Wei, David Basin, Haixiang Li, and Anqun Pan. 2023. Efficient Black-Box Checking of Snapshot Isolation in Databases. Proc. VLDB Endow. 16, 6 (Feb. 2023), 1264–1276. https://doi.org/10. 14778/3583140.3583145 [29] Jepsen. Accessed: April 1, 2026. Distributed Systems Safety Research. http: //jepsen.io/. [30] Zu-Ming Jiang, Jia-Ju Bai, and Zhendong Su. 2023. DynSQL: Stateful Fuzzing for Database Management Systems with Complex and Valid SQL Query Generation. In 32nd USENIX Security Symposium (USENIX Security 23). USENIX Association, 4949–4965. https://www.usenix.org/conference/usenixsecurity23/presentation/ jiang-zu-ming [31] Zu-Ming Jiang and Zhendong Su. 2024. Detecting Logic Bugs in Database Engines via Equivalent Expression Transformation. In 18th USENIX Symposium on Operating Systems Design and Implementation (OSDI 24). USENIX Association, 821–835. https://www.usenix.org/conference/osdi24/presentation/jiang [32] Kyle Kingsbury and Peter Alvaro. 2020. Elle: inferring isolation anomalies from experimental observations. Proc. VLDB Endow. 14, 3 (Nov. 2020), 268–280. https://doi.org/10.14778/3430915.3430918 [33] Jinhui Lai, Chi Zhang, Bingyan Li, Chenglin Liang, Jie Liang, Zhiyong Wu, Jingzhou Fu, Yu Jiang, and Zichen Xu. 2025. SRS: Detecting Logic Bugs of Join Implementation in DBMSs via Set Relation Synthesis. Proc. ACM Manag. Data 3, 6, Article 363 (Dec. 2025), 24 pages. https://doi.org/10.1145/3769828 [34] Ruiyuan Li, Liang Zhang, Juan Pan, Junwen Liu, Peng Wang, Nianjun Sun, Shanmin Wang, Chao Chen, Fuqiang Gu, and Songtao Guo. 2022. Apache ShardingSphere: A Holistic and Pluggable Platform for Data Sharding. In 2022 IEEE 38th International Conference on Data Engineering (ICDE). 2468–2480. https: //doi.org/10.1109/ICDE53745.2022.00230 [35] Jie Liang, Yaoguang Chen, Zhiyong Wu, Jingzhou Fu, Mingzhe Wang, Yu Jiang, Xiangdong Huang, Ting Chen, Jiashui Wang, and Jiajia Li. 2023. SequenceOriented DBMS Fuzzing. In 2023 IEEE 39th International Conference on Data Engineering (ICDE). 668–681. https://doi.org/10.1109/ICDE55515.2023.00057 [36] Jie Liang, Zhiyong Wu, Jingzhou Fu, Mingzhe Wang, Chengnian Sun, and Yu Jiang. 2024. Mozi: Discovering DBMS Bugs via Configuration-Based Equivalent Transformation. In Proceedings of the IEEE/ACM 46th International Conference on Software Engineering (ICSE ’24). ACM, Article 135, 12 pages. https://doi.org/10. 1145/3597503.3639112 [37] Yu Liang, Song Liu, and Hong Hu. 2022. Detecting Logical Bugs of DBMS with Coverage-based Guidance. In 31st USENIX Security Symposium (USENIX Security 22). USENIX Association, 4309–4326. https://www.usenix.org/conference/ usenixsecurity22/presentation/liang [38] Si Liu, Long Gu, Hengfeng Wei, and David Basin. 2024. Plume: Efficient and Complete Black-Box Checking of Weak Isolation Levels. Proc. ACM Program.

ACKNOWLEDGMENTS We appreciate the anonymous reviewers for their valuable feedback. This work was partially supported by Shanghai Trusted Industry Internet Software Collaborative Innovation Center. Si Liu was supported by the University Development Fund (UDF01004607). Hengfeng Wei was supported by the NSFC (62472214) and the Fundamental Research Funds for the Central Universities.

References [1] Alfred V. Aho, Monica S. Lam, Ravi Sethi, and Jeffrey D. Ullman. 2006. Compilers: Principles, Techniques, and Tools (2nd Edition). Addison-Wesley. [2] Apache ShardingSphere. Accessed: April 2026. Alter Sharding Table Reference Rule. https://shardingsphere.apache.org/document/current/en/usermanual/shardingsphere-proxy/distsql/syntax/rdl/rule-definition/sharding/ alter-sharding-table-reference-rule/. Create Broadcast Ta[3] Apache ShardingSphere. Accessed: April 2026. ble Rule. https://shardingsphere.apache.org/document/current/en/usermanual/shardingsphere-proxy/distsql/syntax/rdl/rule-definition/broadcasttable/create-broadcast-table-rule/. [4] Apache ShardingSphere. Accessed: April 2026. Create Sharding Table Rule. https://shardingsphere.apache.org/document/current/en/user-manual/ shardingsphere-proxy/distsql/syntax/rdl/rule-definition/sharding/createsharding-table-rule/. [5] Apache ShardingSphere. Accessed: April 2026. Load Single Table. https://shardingsphere.apache.org/document/current/en/usermanual/shardingsphere-proxy/distsql/syntax/rdl/rule-definition/singletable/load-single-table/. [6] Apache ShardingSphere. Accessed: April 2026. Merger Engine: Group-by Merger. https://shardingsphere.apache.org/document/current/en/reference/ sharding/merge. [7] Apache ShardingSphere. Accessed: April 2026. Route Engine. https:// shardingsphere.apache.org/document/current/en/reference/sharding/route/. [8] Jinsheng Ba and Manuel Rigger. 2023. Testing Database Engines via Query Plan Guidance. In Proceedings of the 45th International Conference on Software Engineering (ICSE ’23). IEEE Press, 2060–2071. https://doi.org/10.1109/ICSE48619. 2023.00174 [9] Jinsheng Ba and Manuel Rigger. 2024. Keep It Simple: Testing Databases via Differential Query Plans. Proc. ACM Manag. Data 2, 3, Article 188 (May 2024), 26 pages. https://doi.org/10.1145/3654991 [10] Hal Berenson, Phil Bernstein, Jim Gray, Jim Melton, Elizabeth O’Neil, and Patrick O’Neil. 1995. A critique of ANSI SQL isolation levels. In Proceedings of the 1995 ACM SIGMOD International Conference on Management of Data (SIGMOD ’95). ACM, 1–10. https://doi.org/10.1145/223784.223785 [11] James Bornholt, Rajeev Joshi, Vytautas Astrauskas, Brendan Cully, Bernhard Kragl, Seth Markle, Kyle Sauri, Drew Schleit, Grant Slatton, Serdar Tasiran, Jacob Van Geffen, and Andrew Warfield. 2021. Using Lightweight Formal Methods to Validate a Key-Value Storage Node in Amazon S3. In Proceedings of the ACM SIGOPS 28th Symposium on Operating Systems Principles (SOSP ’21). ACM, 836–850. https://doi.org/10.1145/3477132.3483540 [12] Zhiheng Cai, Si Liu, Hengfeng Wei, Yuxing Chen, and Anqun Pan. 2025. Fast Verification of Strong Database Isolation. Proc. VLDB Endow. 19, 4 (Dec. 2025), 563–575. https://doi.org/10.14778/3785297.3785300 [13] Yang Chen, Alex Groce, Chaoqiang Zhang, Weng-Keen Wong, Xiaoli Fern, Eric Eide, and John Regehr. 2013. Taming compiler fuzzers. In Proceedings of the 34th ACM SIGPLAN Conference on Programming Language Design and Implementation (PLDI ’13). ACM, 197–208. https://doi.org/10.1145/2491956.2462173 [14] Citus Data. Accessed: April 12, 2026. Citus Technical Documentation. https: //github.com/citusdata/citus/blob/main/src/backend/distributed/README.md. [15] Citus Data. Accessed: April 12, 2026. Concepts: Co-Location. https://docs. citusdata.com/en/stable/get_started/concepts.html#co-location. 13

1 2 3 4 5 6 7 8 9 10

CREATE TABLE t1 (vkey INT4); CREATE TABLE t2 (vkey INT4); INSERT INTO t2 (vkey) values (5); SELECT create_distributed_table('t1', 'vkey'); SELECT create_reference_table('t2');

[53] SQLsmith. Accessed: Nov. 2025. SQLsmith. https://github.com/anse1/sqlsmith. [54] Cheng Tan, Changgeng Zhao, Shuai Mu, and Michael Walfish. 2020. Cobra: Making Transactional Key-Value Stores Verifiably Serializable. In 14th USENIX Symposium on Operating Systems Design and Implementation (OSDI 20). USENIX Association, 63–80. https://www.usenix.org/conference/osdi20/presentation/tan [55] Xiu Tang, Sai Wu, Dongxiang Zhang, Feifei Li, and Gang Chen. 2023. Detecting Logic Bugs of Join Optimizations in DBMS. Proc. ACM Manag. Data 1, 1, Article 55 (May 2023), 26 pages. https://doi.org/10.1145/3588909 [56] Vitess. Accessed: April 12, 2026. Lookup as Primary Vindex. https://vitess.io/ docs/24.0/user-guides/vschema-guide/lookup-as-primary/. [57] Vitess. Accessed: April 12, 2026. MySQL Compatibility: SQL Support. https: //vitess.io/docs/24.0/reference/compatibility/mysql-compatibility/#sql-support. [58] Vitess. Accessed: April 12, 2026. Reference Tables. https://vitess.io/docs/24.0/ reference/vreplication/reference_tables/. [59] Vitess. Accessed: April 12, 2026. Sharded Keyspace. https://vitess.io/docs/24.0/ user-guides/vschema-guide/sharded/. [60] Vitess. Accessed: April 12, 2026. Unsharded Keyspace. https://vitess.io/docs/24. 0/user-guides/vschema-guide/unsharded/. [61] Vitess. Accessed: April 2, 2026. Query Plans Classification. https://vitess.io/docs/ 24.0/reference/query-serving/metrics. [62] Vitess. Accessed: Feb. 13, 2026. Scalable. Reliable. MySQL-compatible. Cloudnative. https://vitess.io/. [63] Qianqian Wang, Chris Parnin, and Alessandro Orso. 2015. Evaluating the usefulness of IR-based fault localization techniques. In Proceedings of the 2015 International Symposium on Software Testing and Analysis (ISSTA 2015). ACM, 1–11. https://doi.org/10.1145/2771783.2771797 [64] Hengfeng Wei, Jiang Xiao, Na Yang, Si Liu, Zijing Yin, Yuxing Chen, and Anqun Pan. 2025. Boosting End-to-End Database Isolation Checking via MiniTransactions. In 2025 IEEE 41st International Conference on Data Engineering (ICDE). 3998–4010. https://doi.org/10.1109/ICDE65448.2025.00298 [65] Chi Zhang and Manuel Rigger. 2025. Constant Optimization Driven Database System Testing. Proc. ACM Manag. Data 3, 1, Article 24 (Feb. 2025), 24 pages. https://doi.org/10.1145/3709674 [66] Zhou Zhou, Si Liu, Hengfeng Wei, and Min Zhang. 2026. Distribution-Aware Distributed Database Testing (Extended Version). Technical Report. https://github. com/duerwuyi/DAT/blob/main/tech-rpt.pdf.

--- t1's DS --- t2's DS

SELECT * FROM (t2 FULL OUTER JOIN t1 ON (t2.vkey = t1.vkey)) WHERE NOT((85) IN (SELECT 1 FROM t2)); --- expected result: 1 row returned --- actual result: 32 rows returned

Figure 14: A logic bug found in Citus: FULL OUTER JOIN with a reference table yields 32 duplicate rows. 1 2 3 4 5 6 7 8 9 10 11 12 13

SET DEFAULT SINGLE TABLE STORAGE UNIT = ds_0; CREATE TABLE t0 (c2 TEXT,c6 TEXT); SET DEFAULT SINGLE TABLE STORAGE UNIT = ds_1; CREATE TABLE t1 (vkey INT4); CREATE BROADCAST TABLE RULE t2; CREATE TABLE t2 (vkey INT4);

--- t0's DS --- t1's DS --- t2's DS

SELECT * FROM t0 AS ref_0 WHERE (ref_0.c2) IN (SELECT ref_0.c6 AS c_0 FROM (t2 AS ref_1 FULL OUTER JOIN t1 AS ref_3 ON (ref_1.vkey = ref_3.vkey))); --- expected: the execution succeeds with correct result --- actual result: the request times out after 30s

Figure 15: A bug found in ShardingSphere: a correlated subquery triggers per-row rescans, exhausting the connection pool and causing timeouts.

Lang. 8, OOPSLA2, Article 302 (Oct. 2024), 29 pages. https://doi.org/10.1145/ 3689742 [39] Si Liu, Luca Multazzu, Hengfeng Wei, and David A. Basin. 2024. NOC-NOC: Towards Performance-optimal Distributed Transactions. Proc. ACM Manag. Data 2, 1, Article 9 (March 2024), 25 pages. https://doi.org/10.1145/3639264 [40] Microsoft Learn. Accessed: April 3, 2026. Citus for PostgreSQL: Performance Tuning. https://learn.microsoft.com/en-us/postgresql/citus/performance-tuning? view=citus-14. Section: Scale-out performance. [41] Lasse Møldrup and Andreas Pavlogiannis. 2025. AWDIT: An Optimal Weak Database Isolation Tester. Proc. ACM Program. Lang. 9, PLDI, Article 209 (June 2025), 25 pages. https://doi.org/10.1145/3742465 [42] PingCAP. Accessed: April 2026. Go-Randgen. https://github.com/pingcap/gorandgen. [43] PingCAP. Accessed: July 2026. TiDB Documentation: Placement Rules. https: //docs.pingcap.com/tidb/stable/configure-placement-rules/ [44] PingCAP. Accessed: March 11, 2026. TiDB Documentation: TiFlash Overview. https://docs.pingcap.com/tidb/stable/tiflash-overview/ [45] Manuel Rigger and Zhendong Su. 2020. Finding bugs in database systems via query partitioning. Proc. ACM Program. Lang. 4, OOPSLA, Article 211 (Nov. 2020), 30 pages. https://doi.org/10.1145/3428279 [46] Manuel Rigger and Zhendong Su. 2020. Testing Database Engines via Pivoted Query Synthesis. In 14th USENIX Symposium on Operating Systems Design and Implementation (OSDI 20). USENIX Association, 667–682. https://www.usenix. org/conference/osdi20/presentation/rigger [47] AWS Cloud Services. Accessed: April 2026. AWS Fault Injection Service. https: //aws.amazon.com/fis/. [48] Avi Silberschatz, Henry F. Korth, and S. Sudarshan. 2020. Database System Concepts, Seventh Edition. McGraw-Hill Book Company. [49] Jiansen Song, Wensheng Dou, Yu Gao, Ziyu Cui, Yingying Zheng, Dong Wang, Wei Wang, Jun Wei, and Tao Huang. 2024. Detecting Metadata-Related Logic Bugs in Database Systems via Raw Database Construction. Proc. VLDB Endow. 17, 8 (April 2024), 1884–1897. https://doi.org/10.14778/3659437.3659445 [50] Jiansen Song, Wensheng Dou, Yingying Zheng, Yu Gao, Ziyu Cui, Wei Wang, and Jun Wei. 2025. Detecting Schema-Related Logic Bugs in Relational DBMSs via Equivalent Database Construction. Proc. VLDB Endow. 18, 7 (March 2025), 2281–2294. https://doi.org/10.14778/3734839.3734861 [51] SQLancer. Accessed: April 27, 2026. Commit d38ddcb. https://github.com/ sqlancer/sqlancer/tree/d38ddcb. [52] SQLancer. Accessed: Nov. 2025. SQLancer. https://github.com/sqlancer/sqlancer.

More Experimental Results

This section provides additional details to complement the experimental results in Section 5. It consists of five parts: (1) a complete bug list, including detailed descriptions, bug types, status, and SQL features involved in bug-triggering queries, as shown in Table 8, which corresponds to Table 3; (2) two additional representative bugs found in Citus (Figure 14) and ShardingSphere (Figure 15), respectively, which complement the one shown in Section 1; (3) the distribution of detected bugs by type, as shown in Table 6; (4) the entry points instrumented in the codebase of each DDBMS for measuring optimization triggering, as shown in Table 7; (5) the coverages of shard-access (Figure 16) and distributed-join (Figure 17) patterns for ShardingSphere, Vitess, and ClickHouse. Below, we provide further details on (2), (4), and (5).

A.1

Representative Bugs

Example A.1. Figure 14 shows a SQL query that triggers a logic bug in Citus. The query performs a FULL OUTER JOIN between a distributed table t1, sharded by vkey (32 shards by default), and a reference table t2. Executing the query should return a single row. Instead, Citus returns 32 identical rows, one per shard, indicating that the full outer join is executed independently on each shard, with the coordinator simply concatenating the results rather than properly aggregating them. This bug stems from a flaw in the distributed planner that prevents Citus from recognizing certain joins, e.g., FULL OUTER JOIN with pseudo-constant restrictions. As a result, the outer join is not 14

Table 6: Distribution of the detected bugs by type. Bug Type

Citus

ShardingSphere

Vitess

ClickHouse

Total (31)

Logic Crash Error Exception Timeout

5 3 0 0 0

4 0 3 1 1

6 2 3 0 1

1 0 0 1 0

16 5 6 2 2

For Citus, we instrument its distributed planner. Specifically, we use shard_pruning.c to measure shard-routing behavior because this file implements the shard routing (or pruning) logic that determines whether only a subset of shards needs to be accessed. For co-located joins, we instrument query_colocation_checker.c, which checks whether distributed tables satisfy the co-location requirements needed for co-located join execution. For ShardingSphere, we instrument the Java routing layer and record whether execution reaches the corresponding route-engine logic inside the Java virtual machine. We use routeByShardingConditions to measure shard routing, since this component routes a query based on extracted sharding conditions. For co-located joins, we use routeByShardingConditionsWithCondition, which additionally handles routing decisions involving co-located joins. For Vitess, we leverage its built-in query planning and routing components. We instrument RoutingParameters.equal to find when Vitess recognizes shard routing. For co-located joins, we instrument joinMerging.mergeShardedRouting, which is invoked when Vitess attempts to perform co-located joins. For ClickHouse, we evaluate optimize_skip_unused_shards by comparing query execution with this option enabled and disabled. A query is counted as triggering shard routing when enabling this option changes the internal execution path or query plan, indicating that ClickHouse can skip irrelevant shards by shard routing. ClickHouse does not expose the co-located join optimization.

Table 7: Instrumented entry points for measuring optimization triggering. Each entry point corresponds to a specific function or source file serving as a measurement target. DDBMS

Opt.

Instrumented entry point

Citus Citus ShardingSphere ShardingSphere Vitess Vitess ClickHouse

SR CJ SR CJ SR CJ SR

shard_pruning.c query_colocation_checker.c routeByShardingConditions routeByShardingConditionsWithCondition RoutingParameters.equal joinMerging.mergeShardedRouting optimize_skip_unused_shards

considered during planning, leading to an incorrect distributed execution plan. Citus treats this as a critical logic bug requiring user attention. In the current release, it explicitly raises an error instead of silently producing incorrect results. Example A.2. Figure 15 illustrates a query that exposes a timeout issue in ShardingSphere caused by connection pool exhaustion. The federation layer in ShardingSphere relies on SQL rewriting to retrieve data across data nodes, but it mishandles the correlated subquery. At runtime, for each outer row of ref_0, it rescans all shards of ref_1 and ref_3 without reusing intermediate results. Moreover, each scan acquires a new backend connection, quickly exhausting the connection pool and resulting in timeouts. This bug is severe, as it can render the system unavailable: once the connection pool is saturated, subsequent requests may be delayed or fail while waiting for available connections. Moreover, it can be exploited by an attacker to launch denial-of-service attacks.

A.2

A.3

Coverage of Distributed Execution Patterns

In addition to Citus, we measure various shard-access and distributedjoin patterns for ShardingSphere, Vitess, and ClickHouse. For ClickHouse and Citus, we analyze query plans, which expose accessed tables, shards (for distributed tables), and whether joins are executed in a co-located or cross-shard manner. For ShardingSphere and Vitess, which do not expose physical plans in the same form, we use PREVIEW and VEXPLAIN to inspect rewritten or routed SQL queries and infer shard accesses and join patterns. The experimental results are shown in Figure 16 and Figure 17. They are consistent with the observations for Citus: for the two target distributed optimizations, i.e., shard routing and co-located joins, significantly more queries generated by DistRanger exercise the corresponding patterns compared with SQLancer, EET, and DistRanger!MT . Additionally, due to the randomness in query generation, DistRanger also exercises patterns beyond the two target optimizations, and each query may exercise zero or more patterns.

Instrumented Entry Points

To measure how frequently generated queries trigger distributed query optimizations, we instrument entry points for the distributed query planning or routing logic in the codebase. Table 7 summarizes the instrumentation targets for the four DDBMSs. Since different systems expose shard routing and co-located joins through different internal modules, the instrumentation is system-specific.

SQLancer

EET

Pattern Occurrences

ShardingSphere

DistRanger!MT ClickHouse

9,500 7,500

Vitess

7,500

6,500

5,000

4,000

2,500

2,000

5,000 2,500 0 al Loc

g tin nce rou ere rd Ref a h S

DistRanger

0 -s All

d har

a Loc

t rou rd Sha

ing

0 l

ard -sh All

a Loc

ce tin ren rou rd Sha

e Ref

g All

a -sh

Figure 16: Distribution of different shard-access patterns (local, reference-table, and distributed-table access) exercised by 10K executable queries. The distributed-table access patterns include shard routing and all-shard access. ClickHouse does not support reference tables. SQLancer

EET

DistRanger!MT

DistRanger

Pattern Occurrences

ShardingSphere

Vitess

3,700 3,000

2,000

1,000

0 C

d cate o-lo

0 ce

en efer

d cate o-lo

d har ss-s Cro

eren

Ref

d har ss-s Cro

Figure 17: Distribution of different distributed-join patterns (co-located, reference, and cross-shard joins) exercised by 10K executable queries. ClickHouse is omitted because it does not support co-located joins.

Table 8: A complete list of bugs detected by DistRanger. DDBMS

Description

Type

Status

Features Involved

Citus

logic

fixed

join

Citus

logic

fixed

DML, subquery

Citus

logic

fixed

DML

Citus

logic

fixed

subquery

5 6

Citus Citus

crash crash

investigating DQL fixed DML, subquery

Citus

logic

fixed

subquery

Citus

crash

fixed

subquery, join

ShardingSphere

logic

investigating metadata

ShardingSphere

logic

fixed

window

ShardingSphere

A FULL OUTER JOIN between a distributed table and a reference table produce duplicate result due to no access to set_join_pathlist_hook. An UPDATE statement with distributed table produce wrong result due to wrong subplan of WHERE filter. Subquery with reference table returns wrong result due to wrong single-shard routing logic. Adding WHERE TRUE in subquery flips result due to wrong distributed query planning. Server crashes when re-adding a node during reset. A Delete statement crashes due to an assertion failure in distributed query rewriting. Subquery with LIMIT returns wrong result due to wrong distributed subplan. Segmentation fault occurs on LEFT JOIN to a distributed table and correlated subqueries. Wrong result of information_schema.tables when sharding a table. Incorrect result window aggregate on sharded table due to wrong aggregation logic. Correlated subquery in HAVING cannot reference outer query column in SQL parser.

error

fixed

subquery, group by Continued on next page

Table 8: A complete list of bugs detected by DistRanger (Continued) DDBMS

Description

Type

Status

Features Involved

ShardingSphere

exception

fixed

13 14

ShardingSphere ShardingSphere

logic timeout

fixed confirmed

/ subquery, join

ShardingSphere

error

fixed

subquery, join

ShardingSphere

error

fixed

join, subquery

ShardingSphere

logic

fixed

distinct

18 19

ClickHouse ClickHouse

exception logic

confirmed subquery investigating join

Vitess

error

investigating shard

Vitess

error

confirmed

Vitess

logic

investigating subquery, window

Vitess

crash

investigating subquery, join

Vitess

timeout

investigating subquery, join

25 26

Vitess Vitess

Exception when SELECT version() due to wrong function verify and routing. Timestamp output adds trailing ".0" in SQL format. Correlated subquery triggers timeout due to wrong distributed query rewriting. Correlated IN-subquery fails in a non-join subquery due to wrong distributed planing logic. Co-located join inside scalar subquery fails due to wrong distributed query rewriting. Result of SELECT DISTINCT duplicates due to wrong distributed query rewriting. Nested EXISTS with deep correlated reference fails. Result of GLOBAL RIGHT OUTER JOIN on distributed tables duplicates. ApplyVSchema accepts empty auto_increment object and later INSERT fails. Cannot insert into pinned table, got no primary vindex due to wrong distributed query routing. Window function without single-shard restriction returns wrong result with uncorrelated scalar subquery. VTGate panic when WHERE contains LTRIM on sharded tables. VTGate planning appears to hang on queries with many uncorrelated scalar subqueries. subquery is incorrectly dropped on sharded keyspace. Unknown column in WHERE clause of scalar subquery.

logic error

Vitess

logic

investigating subquery investigating subquery, aggregation fixed subquery

Vitess

crash

confirmed

subquery

Vitess

logic

confirmed

subquery

Vitess

logic

investigating join

Vitess

logic

confirmed

Planner drops constant-true OR branch and returns wrong result on sharded keyspace. VTGate panic in planner when comparing two scalar subqueries. Reference table SELECT incorrectly rewritten to WHERE 1!=1 with nested EXISTS. Reference table OUTER JOIN on preserved side duplicates rows across shards. Co-located join rewrite breaks LEFT OUTER JOIN semantics due to wrong distributed query rewriting.

shard

join

Distribution-Aware Distributed Database Testing (Extended Version)

Analyze & explore related research in Conceptio

Related documents