The author previously presented this topic at the VeloDB Webinar. This article expands on that presentation by providing more detailed comparative information, including test data and in-depth technical explanations: https://www.youtube.com/embed/qnxX-FOd8Wc?si=TcEF_w-XhqgQyP4A

In the past year, there's an increasing number of users looking to use Apache Doris as an alternative to Elasticsearch, so I'd like to provide an in-depth comparison of the two to serve as a reference for users.

Apache Doris is a real-time data warehouse commonly used for observability, cyber security analysis, online reports, customer profiles, data lakehouse and more. Elasticsearch is more like a search engine, but it is also widely used for data analytics, so there's an overlap in their use cases. The comparison in this post will focus on the real-time analytics capabilities of Apache Doris and Elasticsearch from a user-oriented perspective:

1. Open source
2. System architecture
3. Real-time writes
4. Real-time storage
5. Real-time queries

Open source​

The license decides the level of openness of an open source product, and further decides whether users will be trapped in a vendor lock-in situation.

Apache Doris is operated under the Apache Software Foundation and it is governed by Apache License 2.0. This is a highly liberal open source license. Anyone can freely use, modify, and distribute Apache Doris in open source or commercial projects.

Elasticsearch has undergone several changes in licenses. It was subject to Apache License 2.0 at the very beginning. Then in 2021, it switched to the Elastic License and SSPL, mostly because some cloud service providers were providing Elasticsearch as a managed service. The Elastic company tried to protect its business interests and moved to the Elastic License in order to restrict certain commercial use. And then in 2024, Elasticsearch announced that "Elasticsearch Is Open Source. Again!" by adding the AGPL License. This license has less restrictions on cloud providers.

The difference in licenses reflects the differing ways in which the two open-source projects are managed and operated. Apache Doris is operated under the Apache Software Foundation and adheres to "the Apache Way" and maintains vendor neutrality. It will always be under Apache License and maintain a high level of openness. Elasticsearch is owned and run by the Elastic company, so it is free to change its license based on the changing needs of the company.

System architecture​

The system architecture of Apache Doris and Elasticsearch is relevant to how users can deploy them and the software/hardware prerequisites that must be met.

Apache Doris supports various deployment models, especially after the release of verion 3.0. It can be deployed on-premise the traditional way, which means compute and storage are integrated within the same hardware. It can also be deployed with compute and storage decoupled, providing higher flexibility and elasticity.

Apache Doris enables the isolation of computing workloads, which makes it well-suited for multi-tenancy. In addition to decoupling compute and storage, it also provides tiered storage so you can choose different storage medium for your hot and cold data.

For workload isolation, Apache Doris provides the Compute Group and Workload Group mechanism.

Compute Group is a mechanism for physical isolation between different workloads in a compute-storage decoupled architecture. One or more BE nodes can form a Compute Group.

Workload Group is an in-process mechanism for isolating workloads. It achieves resource isolation by finely partitioning or limiting resources (CPU, IO, Memory) within the BE process, which is implemented by Linux CGroup and offers strict hard isolation.

Elasticsearch supports on-premise and cloud deployment. It does not support compute-storage decoupling. Elasticsearch implements workload isolation by Thread Groups, which provides limited soft resource isolation.

Real-time writes​

The next step after deploying the system is to write data into it. Apache Doris and Elasticsearch are very different in data ingestion.

Write capabilities​

Normally there are two ways to write real-time data into a data system. One is push-based, and the other is pull-based. A push-based method means users actively push the data into the database system, such as via HTTP. A pull-based method, for example, is where database system pulls data from data source such as Kafka message queue.

Elasticsearch supports push-based ingestion, but it requires Logstash to perform pull-based data ingestion, making it less convenient.

As for Apache Doris, it supports both push-based method (HTTP Stream Load) and pull-based method (Routine Load from Kafka, Broker Load from object storage and HDFS). In addition, output plugins for Logstash and Beats are available to enable seamless data ingestion from Logstash or Beats into Doris.

In addition, Doris provides a special write transaction mechanism. By setting a label for a batch of data through the Load API, attempting to re-load a label that has been successfully load before will result in an error, thereby achieving data deduplication. This mechanism ensures that data is written without loss or duplication without relying on the uniqueness of primary keys at the storage layer. Additionally, having a unique label for each batch of data offers better performance compared to having a unique primary key for each individual record.

Write performance​

Apache Doris, as a real-time data warehouse, supports real-time data writes and updates. One of the standout features of Doris is its incredibly high write throughput. This is because the Doris community has put a lot of effort in optimizing it for high-throughput writes such as vertorized writting and single replica indexing.

On the other hand, the low write throughput of Elasticsearch is a well-known pain point. This is due to Elasticsearch's needs to generate complex inverted index on multiple replicas, causing a lot of overheads.

We compare the write performance of Doris and Elasticsearch using Elasticsearch's official benchmark httplogs, under the same hardware resources and storage schema (including field types and inverted indexes). The test environment and results are as follows:

Under the premise of creating the same inverted index, Apache Doris delivers a much higher write throughput than Elasticsearch. This is due to some key advancements in performance optimization, including:

1. Doris is implemented in C++, which is more efficient than Elasticsearch's Java implementation.
2. The vectorized execution engine of Doris fully utilizes CPU SIMD instructions to accelerate both data writing and inverted index building. Its columnar storage also facilitates vectorized execution.
3. The inverted index of Doris is simplified for real-time analytics scenarios, eliminating unnecessary storage structures such as forward indexes and norms.

Real-time storage​

Storage format and cost​

As a real-time data warehouse, Doris adheres to the typical relational database model, with a storage hierarchy that includes catalog, database, and table levels. A table consists of one or more columns, each with its own data type, and indexes can be created on the columns.

By default, Doris stores tables in a columnar format, meaning that data for a single column is stored in contiguous physical files. This not only achieves a high compression ratio but also ensures high query efficiency in analysis scenarios where only certain columns are accessed, as only the required data is read and processed. Additionally, Doris supports an optional row storage format to accelerate detailed point queries.

In Elasticsearch, data is stored in Lucene using a document model. In this model, an Elasticsearch index is equivalent to a database table, the index mapping corresponds to the database table schema, and fields within the index mapping are akin to columns in a database, each with its own type and index type.

By default, Elasticsearch uses row-based storage (the _source field), where each field has an associated inverted index created for it, but it also supports optional columnar storage (docvalue). Generally, row-based storage is essential in Elasticsearch because it lays the foundation for raw detailed data queries.

Storage space consumption, or more straightforward, storage cost, is a big concern for users. This is another pain point of Elasticsearch - it creates huge storage costs.

Apache Doris community have made a lot of optimizations and successfully reduced the storage cost significantly. We have put a lot of work to simplify inverted index, which is supported in Doris since version 2.0, and it now takes up much less space. Doris supports the ZSTD algorithm, which is much more efficient than GZIP and LZ4, and it can reach a data compression ratio of 5:1 ~ 10:1. Since the compression ratio of Elasticsearch is usually about 1.5:1, Doris is 3~5 times more efficient than Elasticsearch in data compression.

As demonstrated in the httplogs test results table above, 32GB of raw data occupies 3.2GB in Doris, whereas Elasticsearch defaults to 19.4GB. Even with the latest logsdb optimization mode enabled, Elasticsearch still consumes 11.7GB. Compared to Elasticsearch, Doris reduces storage space by 83% (73% with logsdb enabled).

Table model​

Apache Doris provides three data models. As for Elasticsearch, I put it as "one and two half" models because it only provides limited support for certain data storage models. I will get to them one by one.

• The first is the Duplicate model, which means it allows data duplication. This is suitable for storing logs. Both Doris and Elasticsearch support this model.

• The second is the primary key model. It works like OLTP databases, which means the data will be deduplicated by key. The primary key model in Doris provides high performance and many user-friendly features. Just like many databases, you can define one or multiple fields as the primary and unique key.

  • However, in Elasticsearch, you can only use a special field _id as the unique identifier for a document (a row in database). Unlike database primary key, there are many limitations for it.

  • The _id field can not be used for aggregation or sorting

  • The _id field can not be larger than 512 bytes

  • The _id field can not be multiple fields, so you will need to combine the multiple fields into one before you can use them as the primary key, but the length of the combined _id can still not exceed 512 bytes.

• The third is the aggregation model, which means data will be aggregated or rolled up.

In its early stages, Elasticsearch provided the rollup feature through the commercial X-Pack plugin, allowing users to create rollup jobs that configured dimensions, metric fields, and aggregation intervals for rollup indexes based on a base index. However, X-Pack rollup had several limitations:

1. Data Asynchrony: Rollup jobs ran in the background, meaning the data in the rollup index was not synchronized in real time with the base index.
2. Specialized Query Requirement: Queries targeting rolled-up data required dedicated rollup queries, and users had to manually specify the rollup index in their queries.

Perhaps these reasons explain why Elasticsearch has deprecated the use of rollup starting in version 8.11.0 and now recommends downsampling as an alternative. Downsampling eliminates the need to specify a separate index and simplifies querying, but it also comes with its own constraints:

1. Time-Series Exclusive: Downsampling is only applicable to time-series data, relying on time as the primary dimension. It cannot be used for other data types, such as business reporting data.
2. Index Replacement: A downsampled index replaces the original index, meaning aggregated data and raw data cannot coexist.
3. Read-Only: Downsampling can only be performed on the original index after it transitions to a read-only state. Data actively being written (e.g., real-time ingestion) cannot undergo downsampling.

As a real-time data warehouse excelling in aggregation and analytics, Doris has supported aggregation capabilities since its early use cases in online reporting and analytics. It offers two flexible mechanisms:

1. Aggregation Table Model:
   • Data is directly aggregated during ingestion, eliminating the storage of raw data.
   • Only aggregated results are stored, drastically reducing storage costs.
2. Aggregated Materialized Views/ Rollup:
   • Users can create aggregated materialized views on a base table. Data is written to both the base table and the materialized view simultaneously, ensuring atomic and synchronized updates.
   • Queries target the base table, and the query optimizer automatically rewrites queries to leverage the materialized view for accelerated performance.

This design ensures real-time aggregation while maintaining query flexibility and efficiency in Doris.

Flexible schema​

In most cases, especially for online business scenarios, users often need to update or modify the data schema. Elasticsearch supports flexible schema, but it only allows users to dynamically add columns.

When you write JSON data into Elasticsearch, if the data contains a new field which is not in index mapping, Elasticsearch will create a new field in the schema through dynamic mapping, so you don't need to change the schema beforehand, like you have to do in traditional databases. However, in Elasticsearch, users cannot directly delete a field of the schema. For that, you will need a reindex operation that reads and writes the entire index in Elasticsearch.

In addition, Elasticsearch does not allow adding index for an already existed field. Imagine that you have 100 fields in a schema, and you have created inverted index for 10 of them, and then some days later, you want to add inverted index for another 5 fields, but no, you can't do that in Elasticsearch. Likewise, deleting index for a field is not allowed either.

As a result, to avoid such troubles, most Elasticsearch users will just create inverted index for all of their fields, but that causes much slower data writing and much larger storage usage.

Also, Elasticsearch does not allow you to modify your index names or your field names. To sum up, if you need to make any changes to your data other than adding a field, you will need to re-index everything. That's a highly resource-intensive and time-consuming operation.

What about Apache Doris? Doris supports all these schema changes. You can add or delete fields and indexes as you want, and you can change the name of any table or field easily. Doris can finish these changes within just a few milliseconds. Such flexibility will save you a lot of time and resource, especially when you have changing business needs.

Real-time queries​

We break down the real-time querying of Elasticsearch and Apache Doris into three dimensions: usability, query capabilities, and performance.

Usability​

One key aspect of usability is reflected in the user interface.

The Doris SQL interface is designed to maintain protocol and syntax compatibility with MySQL. It even does not have its own client or JDBC driver, allowing the use of MySQL’s client and JDBC driver directly. This is a big convenience for many users familiar with MySQL. In fact, this choice was made early in the original design of Doris to make it more user-friendly. Since MySQL is the most widely used open-source OLTP database, Doris is designed to complement MySQL, creating the best practice for combining OLTP (MySQL) with OLAP (Doris). This allows engineers and data analysts to work seamlessly within a single interface and set of syntax, covering both transactional and analytical workloads.

On the other hand, Elasticsearch has its own Domain Specific Language (DSL), which is originally designed for searching. This DSL is part of the Elasticsearch ecosystem, it often requires engineers to invest a lot of time and effort to fully understand and use the DSL effectively.

Let's take a look at one example:

Imagine we want to search for a particular keyword within a particular time range, and then group and order the data by time interval to visualize it as a trend chart.

In Doris, this can be done with only 7 lines of SQL. However, in Elasticsearch, the same process requires 30 lines.

And it's not just about the code complexity. I believe many of you may find this relatable: The first attempt to learn Elasticsearch's DSL is quite challenging. And even after gaining familiarity with it, crafting a query often requires frequent reference to documentation and examples. Due to its inherent complexity, writing DSL queries from scratch remains a challenging task. In contrast, SQL is simple, clear, and highly structured. Most engineers can easily write a query like this without much effort.

Query capabilities​

Elasticsearch is good at searching. As its name suggests, it was built for search. But beyond search and aggregation queries, it doesn't support complex analytical queries, such as multi-table JOIN.

Apache Doris is a general-purpose data warehouse. It provides strong support for complex analytics, including multi-table JOIN, UDF, subquery, window function, logic view, materialized view, and data lakehouse. And Doris has been improving its searching capabilities since version 2.0. We have introduced inverted index and full-text search, so it is getting increasingly competitive in this area.

Search​

While this article focuses on real-time analytics scenarios, we have chosen not to overlook Elasticsearch’s core strength: its search capabilities.

Elasticsearch, renowned for its search performance, leverages the Apache Lucene library under the hood. It provides two primary indexing structures:

• Inverted indexes for text fields.
• BKD-Tree indexes for numeric, date, and geospatial data.

Elasticsearch supports the following search paradigms:

1. Exact Matching: equality or range queries for numeric, date, or non-tokenized string fields (keyword), powered by BKD-Tree indexes (numeric/date) or non-tokenized inverted indexes (exact term matches).
2. Full-Text Search: keyword, phrase, or multi-term matching. During ingestion, text is split into tokens using a configurable tokenizer. Queries match tokens, phrases, or combinations (e.g., "quick brown fox"). Elasticsearch supports some advanced full-text search features, such as relevance scoring, auto-completion, spell-checking, and search suggestions.
3. Vector Search based on ANN index.

Doris also supports a rich set of indexes to accelerate queries, including prefix-sorted indexes, ZoneMap indexes, BloomFilter indexes and N-Gram BloomFilter indexes.

Starting from version 2.0, Doris has added inverted indexes and BKD-Tree indexes to support exact matching and full-text search. Vector search is currently under development and is expected to be available within the next six months.

There are some differences between the indexes of Apache Doris and Elasticsearch:

1. Diverse indexing strategies: Doris does not rely solely on inverted indexes for query acceleration. It offers different indexes for different scenarios. ZoneMap, BloomFilter, and N-Gram BloomFilter indexes are skip indexes that accelerate analytical queries by skipping non-matching data blocks based on WHERE conditions. The N-Gram BloomFilter index is specifically optimized for LIKE-style fuzzy string matching. Inverted index and Prefix-sorted indexes are point query indexes that speed up point queries by locating which rows satisfy the WHERE conditions through the index and directly read those rows.
2. Performance centric design: Doris prioritizes speed and efficiency over advanced search features. To achieve faster index building and query performance (see the Query Performance section in the following), it omits certain complex functionalities of inverted indexes at present, such as relevance scoring, auto-completion, spell-checking, and search suggestions. Since there are more critical in document or web search scenarios, which are not Doris’ primary focus on real-time analytics.

For exact matching and common full-text search requirements, such as term, range, phrase, multi-field matching, Doris is fully capable and reliable in most use cases. For complex search needs (e.g., relevance scoring, auto-completion, spell-checking, or search suggestions), Elasticsearch remains the more suitable choice.

Aggregation​

Both Elasticsearch and Doris support a wide range of aggregation operators, including:

• Basic aggregations: MIN, MAX, COUNT, SUM, AVG.
• Advanced aggregations: PERCENTILE (quantiles), HISTOGRAM (bucketed distributions).
• Aggregation modes: Global aggregation, key-based grouping (e.g., GROUP BY).

However, their approaches to aggregation analysis differ significantly in the following aspects:

1. Query syntax and usability: Doris uses standard SQL syntax (GROUP BY clauses + aggregation functions), making it intuitive for analysts and developers familiar with SQL. Elasticsearch relies on a custom Domain-Specific Language (DSL) with concepts like metric agg, bucket agg, and pipeline agg. Complexity increases with nested or multi-layered aggregations, requiring steeper learning curves.

2. Result accuracy: Elasticsearch outputs inaccurate approximate result for many aggregation types by default. Aggregations (e.g., terms agg) are executed per shard, with each shard returning only its top results (e.g., top 10 buckets). These partial results are merged globally, leading to potentially inaccurate final outputs. As a serious OLAP database, Doris ensures exact results by processing full datasets without approximations.

3. Query Performance: Doris demonstrates significant performance advantages over Elasticsearch in aggregation-heavy OLAP workloads like those on ClickBench (see more in the Query performance section bellow).

JOIN​

Elasticsearch does not support JOIN, making it unable to execute common benchmarks like TPC-H or TPC-DS. Since JOINs are critical in data analysis, Elasticsearch provides some complex workarounds with significant limitations:

1. Parent-child relationships and has_child / has_parent queries: Simulate JOINs by establishing parent-child relationships within a single index. Child documents store the parent document’s ID. has_child queries first find matching child documents, then retrieve their parent documents via the stored parent IDs. has_parent queries reverse this logic. It's complex and Elasticsearch explicitly warns against equating this approach to database-style JOINs.

   We don’t recommend using multiple levels of relations to replicate a relational model. Each level of relation adds an overhead at query time in terms of memory and computation. For better search performance, denormalize your data instead.

2. Terms lookup: Similar to an IN-subquery by fetching a value list from one index and using it in a terms query on another. It's only suitable for large-table-to-small-table JOINs (e.g., filtering a large dataset using a small reference list), but performs very poorly for large-table-to-large-table JOINs due to scalability issues.

Doris provides comprehensive support for JOIN operations, including:

• INNER JOIN
• CROSS JOIN
• LEFT / RIGHT / FULL OUTER JOIN
• LEFT / RIGHT SEMI JOIN
• LEFT / RIGHT ANTI JOIN

Further more, Doris’ query optimizer adaptively selects optimal execution plans for JOIN operations based on data characteristics and statistics, including:

• Small-large table JOIN:
  • Broadcast JOIN: The smaller table is broadcast to all nodes for local joins.
• Large-large table JOIN:
  • Bucket Shuffle JOIN: Used when the left table’s bucketing distribution aligns with the JOIN key.
  • Colocate JOIN: Applied when both the left and right tables share identical bucketing distributions with the JOIN keys.
  • Runtime Filter: Reduces data scanned in the left table by pushing down predicates from the right table using runtime-generated filters.

This intelligent optimization ensures efficient JOIN execution across diverse data scales and distributions, including TPC-H and TPC-DS benchmarks.

Lakehouse​

Data warehouses addresses the need for fast data analysis, while data lakes are good at data storage and management. The integration of them, known as "lakehouse", is to facilitate the seamless integration and free flow of data between the data lake and data warehouse. It enables users to leverage the analytic capabilities of the data warehouse while harnessing the data management power of the data lake.

Apache Doris can work as a data lakehouse with its Multi Catalog feature. It can access databases and data lakes including Apache Hive, Apache Iceberg, Apache Hudi, Apache Paimon, LakeSoul, Elasticsearch, MySQL, Oracle, and SQLServer. It also supports Apache Ranger for privilege management.

Elasticsearch does not support querying external data, and of course, it does not support lakehouse either.

Query performance​

Apache Doris has been extensively optimized for multiple scenarios, so it can deliver high performance in many use cases. For example, Doris can achieve tens of thousands of QPS for high-concurrency point queries, and it delivers industry-leading performance in aggregation analysis on a global scale.

Elasticsearch is good at point queries (retrieving just a small amount of data). However, it might struggle with complex analytical workloads.

Elasticsearch httplogs and the Microsoft Azure logsbench are benchmarks for log storage and search. Both tests show that Doris is about 34 times faster than Elasticsearch in data writing, but only uses 1/61/4 of the storage space that Elasticsearch uses. Then for data queries, Doris is more than 2 times faster than Elasticsearch.

ClickBench is a benchmark created and maintained by the ClickHouse team to evaluate the performance of analytical databases. It results show that Apache Doris delivers out-of-the-box query performance 21 times faster than Elasticsearch and 6 times faster than a fine-tuned Elasticsearch.

Comparison Summary​

In summary, Doris follows Apache License 2.0, which is a highly open license to ensure that Doris remains truly open and will continue to uphold such openness in the future.

Secondly, Doris supports both compute-storage coupled and decoupled mode, while Elasticsearch only supports the former, which enable more powerful elasticity and resource isolation.

Thirdly, Doris provides high performance in data ingestion. It is usually 3~5 times faster than Elasticsearch.

In terms of storage, compared to Elasticsearch, Doris can reduce storage costs by 70% or more. It is also 10 times faster than Elasticsearch in data updates.

As for data queries, Doris outperforms Elasticsearch by a lot. It also provides analytic capabilities that Elasticsearch does have, including multi-table JOIN, materialized views, and data lakehouse.

Use cases​

Many users have replaced Elasticsearch with Apache Doris in their production environments and received exciting results. I will introduce some user stories from the fields of observability, cyber security, and real-time business analysis.

Observability​

User A: a world-famous short video provider

• Daily incremental data: 800 billion rows (500 TB)
• Average write throughput: 10 million row/s (60GB/s)
• Peak write throughput: 30 million row/s (90GB/s)

Apache Doris supports logging and tracing data storage for this tech giant and meets the data import performance requirements for nearly all use cases within the company.

User B: NetEase - one of the World's Highest-Yielding Game Companies

• Replacing Elasticsearch with Apache Doris for log analysis: reducing storage consumption by 2/3 and achieving 10X query performance
• Replacing InfluxDB with Apache Doris for time-series data analysis: saving 50% of server resources and reducing storage consumption by 67%

User C: an observability platform provider

Apache Doris offers a special data type, VARIANT, to handle semi-structured data in log and tracing, reducing costs by 70% and delivering 2~3 times faster full-text search performance compared to the Elasticsearch-based solution.

Cyber security​

User A: QAX - a listed company & leading cyber security leader

The Doris-based security logging solution uses 40% less storage space, delivers 2X faster write performance, and supports full-text search, aggregation analysis, and multi-table JOIN capabilities in one database system.

User B: a payment platform with nearly 600 million registered users

As a unified security data storage solution, Apache Doris delivers 4X faster write speeds, 3X better query performance, and saves 50% storage space compared to the previous architecture with diverse tech stacks using Elasticsearch, Hive and Presto.

User C: a leading cyber security solution provider

Compared to Elasticsearch, Doris delivers 2 times faster write speeds, 4 times better query performance, and a 4 times data compression ratio.

Business analysis​

User A: a world-leading live e-commerce company

The user previously relied on Elasticsearch to handle online queries for their live stream detail pages, but they faced big challenges in cost and concurrency. After migrating to Apache Doris, they achieve:

• 3 times faster real-time writes: 30w/s -> 100w/s
• 4 times higher query concurrency: 500QPS -> 2000QPS

User B: Tencent Music Entertainment

Previously, TME content library used both Elasticsearch and ClickHouse to meet their needs for data searching and analysis, but managing two separate systems was complex and inefficient. With Doris, they are able to unify the two systems into one single platform to support both data searching and analysis. The new architecture delivers 4X faster write speeds, reduces storage costs by 80%, and supports complex analytical operations.

User C: a web browser provider with 600 million users

After migrating to Apache Doris for a unified solution for log storage and report analysis, the company doubled its aggregation analysis efficiency, reduced storage consumption by 60%, and cut SQL development time by half.

Taking Apache Doris to the next level​

For the Apache Doris community developers, the path to making Doris good enough to replace Elasticsearch wasn't easy. In 2022, we started adding inverted index capabilities to Doris. At that time, this decision was met with skepticism. Many viewed inverted indexes as a feature exclusive to Elasticsearch, a domain few in the industry dared to venture into. Nevertheless, we went with it, and today we can confidently say that we have succeeded.

In 2022, we developed this feature from the ground up, and after a year of dedicated effort, we open-sourced it. Initially, we had only one user, QAX, who was willing to test and adopt the feature. We are deeply grateful to them for their early support during this pivotal stage.

By 2023, the value of Doris with inverted indexes became increasingly evident, leading to broader adoption by about 10 companies.

The growth momentum has continued, and as of 2024, we are experiencing rapid expansion, with over 100 companies now leveraging Doris to replace Elasticsearch.

Looking ahead, I am very much looking forward to what 2025 will bring. This progress, advancing from the ground up to such significant milestones, has been made possible by the incredible support from the Doris community users and developers. We encourage everyone to join the Apache Doris Slack community and join the dedicated channel #elasticsearch-to-doris, where you can receive technical assistance, stay updated with the latest news about Doris, and engage with more Doris developers and users.

More on Apache Doris:

• Apache Doris for log and time series data analysis in NetEase, why not Elasticsearch and InfluxDB?
• From Elasticsearch to Apache Doris: upgrading an observability platform
• Log Analysis: How to Digest 15 Billion Logs Per Day and Keep Big Queries Within 1 Second
• Replacing Apache Hive, Elasticsearch and PostgreSQL with Apache Doris
• Log Analysis: Elasticsearch VS Apache Doris
• A Deep Dive into Inverted Index in Apache Doris
• VARIANT in Apache Doris 2.1.0: a new data type 8 times faster than JSON for semi-structured data analysis

Connect with me on Linkedin

Apache Doris on GitHub

Apache Doris Website

In the age of data-driven decision-making, the exponential growth of data volume and the ever-evolving demands for analytics pose great challenges. Data streams in from diverse sources (such as application logs, network interactions, and mobile devices), spanning structured, semi-structured, and unstructured formats. This diversity places pressure on storage and analytical systems. Meanwhile, the surge in demand for real-time analytics and exploratory queries requires systems to deliver millisecond-level responsiveness while achieving optimal cost efficiency and elastic scalability.

Apache Doris emerged in the era of integrated storage and computation, built on a classic Shared Nothing architecture. In this design, storage and computation are co-located on Backend (BE) nodes, leveraging an MPP (Massively Parallel Processing) distributed computing model. This architecture delivers key advantages, including high availability, simplified deployment, seamless horizontal scalability, and exceptional real-time analytical performance.

For real-time analytics and small-scale data processing, Apache Doris stands out with predictable, stable low-latency performance, making it an indispensable solution. However, when scaling to large-scale data processing, it encounters certain challenges, primarily in:

• Relatively high costs & low elasticity: Balancing storage and compute resources remains a big challenge. Storage capacity must be sufficient to accommodate all data, while compute resources need to handle query workloads efficiently. However, dynamically scaling clusters is often time-consuming, prompting enterprises to over-provision resources to ensure stability. This approach simplifies operations but leads to resource waste and increased costs.
• Limited workload isolation: Since Apache Doris 2.0, Workload Groups provide soft isolation, while Resource Groups offer a degree of hard isolation. However, neither mechanism ensures complete physical isolation, which can impact performance in multi-tenant or resource-intensive environments.
• Operational complexity: Managing an OLAP system with built-in distributed storage requires not only overseeing compute nodes but also ensuring efficient storage administration. Storage management is inherently complex, and misconfigurations or improper operations can lead to data loss, making maintenance highly demanding.

Even so, in the absence of stable and large-scale storage support, a integrated storage-compute architecture remains the optimal choice.

As cloud infrastructure matures, enterprises increasingly seek deeper Apache Doris integration with public clouds, private clouds, and Kubernetes (K8s) container platforms to unlock greater elasticity and flexibility. Public clouds offer mature object storage with on-demand compute resources, eliminating the need for pre-allocated space, while private clouds leverage technologies like K8s and MinIO to build scalable resource platforms. This evolution in cloud infrastructure has also accelerated Apache Doris’ transition to a storage-compute decoupled architecture, enabling lower costs, high elasticity, and enhanced workload isolation.

Apache Doris Compute-Storage Decoupled Mode​

Since version 3.0, Apache Doris has supported both the compute-storage decoupled mode and the compute-storage coupled mode.

01 Compute-Storage Decoupled​

In the compute-storage decoupled mode, Apache Doris adopts a three-tier architecture consisting of three layers: shared storage, compute groups, and meta data service:

Shared Storage Layer

Data is persisted in the shared storage layer, allowing compute nodes to access and share data seamlessly. This design enhances compute node flexibility and reduces operational overhead. Leveraging mature and reliable shared storage results in ultra-low storage costs and high data reliability. Whether using public cloud object storage or enterprise-managed shared storage, this approach greatly reduces the maintenance complexity of Apache Doris.

Compute Groups

The compute layer consists of multiple compute groups responsible for executing query plans. Each query is executed within a single compute group, ensuring isolation and scalability. Compute nodes are stateless, utilizing local disks as high-speed caches to accelerate queries while sharing the same data and metadata services. Each compute group operates independently, supporting on-demand scaling, and local caches remain isolated to ensure workload separation and performance consistency.

Meta Data Service

The meta data layer manages system meta data, including databases, tables, schemas, rowset meta data, and transaction information, with support for horizontal scaling. Future iterations of Apache Doris’ compute-storage decoupled mode will introduce stateless Frontend (FE) nodes, where memory consumption is decoupled from cluster size. This evolution will eliminate memory bottlenecks, allowing FE nodes to operate with minimal memory requirements.

02 Architecture Design​

Traditional compute-storage decoupling approaches typically store both data and meta data in shared storage while centralizing transaction management on a single FE node. However, this design introduces several challenges:

• Write performance bottlenecks: The two-phase commit protocol, driven by FE Master, incurs high latency and low throughput.
• Small file proliferation: Frequent meta data writes generate excessive small files, leading to system instability and inflated storage costs.
• Scalability constraints: Since FE nodes manage meta data in memory, an increasing number of Tablets amplifies memory pressure, eventually causing write bottlenecks.
• Data deletion risks: Relying on delta computation with timeout-based mechanisms for deletion introduces challenges in synchronizing writes and deletions. As a result, there is a risk of unintended data loss due to misalignment between ongoing writes and scheduled deletions.

Compared to traditional approaches, Apache Doris effectively addresses these challenges through a shared meta data service:

• Real-time ingestion: The meta data service provides a globally consistent view, enabling low-latency, high-throughput writes. Benchmarks show that the Apache Doris compute-storage decoupled mode achieves 100X higher performance than other solutions at 50 concurrent writes and 11X higher performance at 500 concurrent writes.
• Optimized small file management: Data is written to shared storage, while meta data is handled by the meta data service. This effectively reduces small file overheads. Tests indicate that the Apache Doris compute-storage decoupled mode generates only 1/2 the number of write files compared to other industry solutions.
• Enhanced scalability: In future versions of Apache Doris, FE metadata will be moved to the meta data service to eliminate cluster scaling limitations and ensure seamless expansion.
• Reliable data deletion: Doris employs a forward deletion mechanism based on a globally consistent view. This ensures mutual exclusion between writes and deletions, thus eliminating the risk of accidental data loss.

03 What makes it stand out​

The Apache Doris compute-storage decoupled architecture provide values for users in three aspects: cost efficiency, elasticity, and workload isolation.

Firstly, it brings a 90% cost reduction compared to the compute-storage coupled mode.

• Pay-as-you-go: Unlike traditional coupled architectures, there’s no need to pre-provision compute and storage resources. Storage costs scale with actual usage, while compute resources can be dynamically adjusted based on demand.
• Single-replica storage: Instead of maintaining three replicas in costly block storage, data is stored as a single replica in low-cost object storage, with hot data cached in block storage for performance. This dramatically reduces storage footprint and hardware costs. For example, S3 costs only 25% to 50% of AWS EBS.
• Lower resource consumption: In compute-storage decoupled mode, compaction operations only process a single data replica, thus largely reducing resource usage compared to multi-replica environments.

Secondly, with a stateless compute node design, Doris enables on-demand resource scaling to meet fluctuating workloads efficiently.

• Elastic auto-scaling: Compute resources can be dynamically scaled to accommodate traffic spikes or workload variations. When demand increases, Doris can rapidly scale out compute nodes; when demand drops, resources scale down automatically, avoiding unnecessary costs.
• Fine-grained compute resource allocation: Doris allows compute nodes to be strategically assigned to specific compute groups based on workload requirements. For example, high-performance nodes handle complex queries and high-concurrency workloads, while standard nodes manage lightweight queries and infrequent requests.

Thirdly, Doris provides efficient resource scheduling and workload isolation mechanisms.

• Cross-business isolation: Different business units can be assigned dedicated compute groups with physical isolation, so workloads operate on dedicated resources without interference.
• Offline workload isolation: Large-scale batch processing tasks can be segregated into dedicated compute groups, so users can leverage low-cost resources for offline data processing without impacting real-time business performance.
• Read-write isolation: Doris allows dedicated compute groups for read and write operations to ensure consistent query response times even under high write loads.

Benchmarking and comparison​

To provide a clear evaluation of the compute-storage decoupled architecture of Apache Doris, we conducted a series of benchmark tests across multiple dimensions, including data ingestion, query performance, and resource cost efficiency.

01 Ingestion performance​

High-concurrency ingestion

We compared Apache Doris' coupled and decoupled modes with other mainstream solutions under the same compute resources. The tests measured real-time ingestion performance under two levels of concurrency:

• 50 concurrent writes: Ingesting 250 files, each containing 20,000 rows.
• 500 concurrent writes: Ingesting 10,000 files, each containing 500 rows.

Test results:

• At 50 concurrent writes, Doris' compute-storage decoupled mode performed on par with the coupled mode while achieving 100X the write performance of other industry compute-storage decoupled solutions.
• At 500 concurrent writes, Doris' decoupled mode experienced a slight performance drop compared to the coupled mode, yet still maintained an 11X advantage over other compute-storage decoupled architectures.

Batch data ingestion

To evaluate batch data ingestion efficiency, we conducted tests using TPC-H 1TB and TPC-DS 1TB datasets, comparing the compute-storage coupled and decoupled modes of Apache Doris. Data was loaded using S3 Load. Under default configurations, multiple tables were ingested sequentially, and the total ingestion time was measured for comparison.

Hardware configuration:

• Cluster size: 4 compute instances (1 FE, 3 BE)
• CPU: 48 cores per instance
• Memory: 192GB per instance
• Network Bandwidth: 21 Gbps
• Storage: Enhanced SSD

As is shown, even when using a single replica in both architectures, the compute-storage decoupled mode outperforms the coupled mode in batch data ingestion by 20.05% and 27.98% in the two benchmarks, respectively. (In real-world deployments, the coupled mode typically adopts a three-replica strategy. This further amplifies the write performance gains of the decoupled mode.)

02 Query Performance​

In the compute-storage decoupled mode, Apache Doris leverages a multi-tier caching mechanism to accelerate queries. It improves overall query efficiency by speeding up data access and minimizing reliance on shared storage. The cache hierarchy includes:

• Doris Page Cache: In-memory caching of decompressed data.
• Linux Page Cache: In-memory caching of compressed data.
• Local Disk Cache: Persistent caching of compressed data.

Hardware configuration:

• Cluster size: 4 compute instances (1 FE, 3 BE)
• CPU: 48 cores per instance
• Memory: 192GB per instance
• Network Bandwidth: 21 Gbps
• Storage: Enhanced SSD

We conducted performance benchmarking under different caching scenarios in both compute-storage coupled and decoupled modes. Using the TPC-DS 1TB dataset, the test results are as follows:

• Full cache hit: We execute the query twice and measure the runtime of the second execution, ensuring that all data is preloaded into the cache. Query performance in compute-storage decoupled mode matches that of the coupled architecture with no performance degradation.
• Partial cache hit (This scenario best reflects real-world usage.): Before the test begins, all caches are cleared, and we measure the runtime of the first execution while data is gradually loaded into the cache. Compared to the coupled architecture, query performance remains nearly identical, with an overall performance overhead of about 10%.
• No cache hit: All caches are cleared before each SQL execution, ensuring that every query runs without cached data. Compared to the coupled architecture, query performance sees an approximate 35% degradation.

03 Resource Cost​

Operational cost for online workloads

Taking a real-world enterprise workload as an example, we compare the cost differences between compute-storage coupled and decoupled modes in Apache Doris.

• Compute-storage coupled mode: The dataset in Doris has a size of 100TB per replica, resulting in a total of 300TB with three replicas. To prevent frequent scaling operations from impacting business stability, disk usage is maintained at about 50%. Thus, the monthly resource cost amounts to $36,962.7 (as detailed below).

• Compute-storage decoupled mode: With the same data scale, adopting the compute-storage decoupled model only requires storing a single replica in object storage, while hot data is cached on local disks. As shown below, the monthly resource cost is reduced to $22,212.65, achieving a 40% cost savings.

Historical data cost

For example, with 200TB of historical data, the resource utilization under both the compute-storage coupled and decoupled modes is shown below. The coupled model incurs a monthly cost of $48,851.10, whereas the decoupled model reduces the cost to just $4,502.40—cutting expenses by over 90%.

What's next​

Powered by compute-storage decoupling, Apache Doris excels in real-time analytics, lakehouse analytics, observability and log storage & analysis. Looking ahead, Apache Doris will continue to enhance its capabilities in this mode. We will introduce new features such as snapshots, time travel, and Cross-Cluster Replication (CCR) support, and achieve stateless FE to further improve system stability and usability.

If you're interested in Apache Doris' compute-storage decoupled mode and its future development, we invite you to join the #compute-storage-decoupled channel in the Apache Doris Slack community, where you can connect with core developers and users. We look forward to your thoughts and contributions!

Join us live on March 27 for more insights into the Apache Doris compute-storage decoupled mode!

To handle large datasets, distributed databases introduce strategies like partitioning and bucketing. Data is divided into smaller units based on specific rules and distributed across different nodes, so databases can perform parallel processing for higher performance and data management flexibility.

Like in many databases, Apache Doris shards data into partitions, and then a partition is further divided into buckets. Partitions are typically defined by time or other continuous values. This allows query engines to quickly locate the target data during queries by pruning irrelevant data ranges.

Bucketing, on the other hand, distributes data based on the hash values of one or more columns, which prevents data skew.

Prior to version 2.1.0, there are two way you can create data partitions in Apache Doris:

• Manual Partition: Users specify the partitions in the table creation statement, or modify them through DDL statements afterwards.

• Dynamic Partition: The system automatically maintains partitions within a pre-defined range based on the data ingestion time.

In Apache Doris 2.1.0, we have introduced Auto Partition. It supports partitioning data by RANGE or by LIST and further enhances flexibility on top of automatic partitioning.

Evolution of partitioning strategies in Doris​

In the design of data distribution, we focus more on partition planning, because the choice of partition columns and partition intervals heavily depends on the actual data distribution patterns, and a good partition design can largely improve the query and storage efficiency of the table.

In Doris, the data table is divided into partitions and then buckets in a hierarchical manner. The data within the same bucket then forms a data tablet, which is the minimum physical storage unit in Doris for data replication, inter-cluster data scheduling, and load balancing.

Manual Partition​

Doris allows users to manually create data partitions by RANGE and by LIST.

For time-stamped data like logs and transaction records, users typically create partitions based on the time dimension. Here's an example of the CREATE TABLE statement:

CREATE TABLE IF NOT EXISTS example_range_tbl
(
    `user_id` LARGEINT NOT NULL COMMENT "User ID",
    `date` DATE NOT NULL COMMENT "Data import date",
    `timestamp` DATETIME NOT NULL COMMENT "Data import timestamp",
    `city` VARCHAR(20) COMMENT "Location of user",
    `age` SMALLINT COMMENT "Age of user",
    `sex` TINYINT COMMENT "Sex of user",
    `last_visit_date` DATETIME REPLACE DEFAULT "1970-01-01 00:00:00" COMMENT "Last visit date of user",
    `cost` BIGINT SUM DEFAULT "0" COMMENT "User consumption",
    `max_dwell_time` INT MAX DEFAULT "0" COMMENT "Maximum dwell time of user",
    `min_dwell_time` INT MIN DEFAULT "99999" COMMENT "Minimum dwell time of user"
)
ENGINE=OLAP
AGGREGATE KEY(`user_id`, `date`, `timestamp`, `city`, `age`, `sex`)
PARTITION BY RANGE(`date`)
(
    PARTITION `p201701` VALUES LESS THAN ("2017-02-01"),
    PARTITION `p201702` VALUES LESS THAN ("2017-03-01"),
    PARTITION `p201703` VALUES LESS THAN ("2017-04-01"),
    PARTITION `p2018` VALUES [("2018-01-01"), ("2019-01-01"))
)
DISTRIBUTED BY HASH(`user_id`) BUCKETS 16
PROPERTIES
(
    "replication_num" = "1"
);

The table is partitioned by the data import date date, and 4 partitions have been pre-created. Within each partition, the data is further divided into 16 buckets based on the hash value of the user_id.

With this partitioning and bucketing design, when querying data from 2018 onwards, the system only need to scan the p2018 partition. This is what the query SQL looks like:

mysql> desc select count() from example_range_tbl where date >= '20180101';
+--------------------------------------------------------------------------------------+
| Explain String(Nereids Planner)                                                      |
+--------------------------------------------------------------------------------------+
| PLAN FRAGMENT 0                                                                      |
|   OUTPUT EXPRS:                                                                      |
|     count(*)[#11]                                                                    |
|   PARTITION: UNPARTITIONED                                                           |
|                                                                                      |
|    ......                                                                            |
|                                                                                      |
|   0:VOlapScanNode(193)                                                               |
|      TABLE: test.example_range_tbl(example_range_tbl), PREAGGREGATION: OFF.          |
|      PREDICATES: (date[#1] >= '2018-01-01')                                          |
|      partitions=1/4 (p2018), tablets=16/16, tabletList=561490,561492,561494 ...      |
|      cardinality=0, avgRowSize=0.0, numNodes=1                                       |
|      pushAggOp=NONE                                                                  |
|                                                                                      |
+--------------------------------------------------------------------------------------+

If the data is distributed unevenly across partitions, the hash-based bucketing mechanism can further divide the data based on the user_id. This helps to avoid load imbalance on some machines during querying and storage.

However, in real-world business scenarios, one cluster may have tens of thousands of tables, which means it is impossible to manage them manually.

CREATE TABLE `DAILY_TRADE_VALUE`
(
    `TRADE_DATE`              datev2 NOT NULL COMMENT 'Trade date',
    `TRADE_ID`                varchar(40) NOT NULL COMMENT 'Trade ID',
    ......
)
UNIQUE KEY(`TRADE_DATE`, `TRADE_ID`)
PARTITION BY RANGE(`TRADE_DATE`)
(
    PARTITION p_200001 VALUES [('2000-01-01'), ('2000-02-01')),
    PARTITION p_200002 VALUES [('2000-02-01'), ('2000-03-01')),
    PARTITION p_200003 VALUES [('2000-03-01'), ('2000-04-01')),
    PARTITION p_200004 VALUES [('2000-04-01'), ('2000-05-01')),
    PARTITION p_200005 VALUES [('2000-05-01'), ('2000-06-01')),
    PARTITION p_200006 VALUES [('2000-06-01'), ('2000-07-01')),
    PARTITION p_200007 VALUES [('2000-07-01'), ('2000-08-01')),
    PARTITION p_200008 VALUES [('2000-08-01'), ('2000-09-01')),
    PARTITION p_200009 VALUES [('2000-09-01'), ('2000-10-01')),
    PARTITION p_200010 VALUES [('2000-10-01'), ('2000-11-01')),
    PARTITION p_200011 VALUES [('2000-11-01'), ('2000-12-01')),
    PARTITION p_200012 VALUES [('2000-12-01'), ('2001-01-01')),
    PARTITION p_200101 VALUES [('2001-01-01'), ('2001-02-01')),
    ......
)
DISTRIBUTED BY HASH(`TRADE_DATE`) BUCKETS 10
PROPERTIES (
  ......
);

In the above example, data is partitioned on a monthly basis. This requires the database administrator (DBA) to manually add a new partition each month and maintain table schema regularly. Imagine the case of real-time data processing, where you might need to create partitions daily or even hourly, manually doing this is no long a choice. That's why we introduced Dynamic Partition.

Dynamic Partition​

By Dynamic Partition, Doris automatically creates and reclaims data partitions as long as the user specifies the partition unit, the number of historical partitions, and the number of future partitions. This functionality relies on a fixed thread on the Doris Frontend. It continuously polls and checks for new partitions to be created or old partitions to be reclaimed, and updates the partition schema of the table.

This is an example CREATE TABLE statement for a table which is partitioned by day. The start and end parameters are set to -7 and 3, respectively, meaning that data partitions for the next 3 days will be pre-created and the historical partitions that are older than 7 days will be reclaimed.

CREATE TABLE `DAILY_TRADE_VALUE`
(
    `TRADE_DATE`              datev2 NOT NULL COMMENT 'Trade date',
    `TRADE_ID`                varchar(40) NOT NULL COMMENT 'Trade ID',
    ......
)
UNIQUE KEY(`TRADE_DATE`, `TRADE_ID`)
PARTITION BY RANGE(`TRADE_DATE`) ()
DISTRIBUTED BY HASH(`TRADE_DATE`) BUCKETS 10
PROPERTIES (
    "dynamic_partition.enable" = "true",
    "dynamic_partition.time_unit" = "DAY",
    "dynamic_partition.start" = "-7",
    "dynamic_partition.end" = "3",
    "dynamic_partition.prefix" = "p",
    "dynamic_partition.buckets" = "10"
);

Over time, the table will always maintain partitions within the range of [current date - 7, current date + 3]. Dynamic Partition is particularly useful for real-time data ingestion scenarios, such as when the ODS (Operational Data Store) layer directly receives data from external sources like Kafka.

The start and end parameters define a fixed range for the partitions, allowing the user to manage the partitions only within this range. However, if the user needs to include more historical data, they would have to dial up the start value, and that could lead to unnecessary metadata overhead in the cluster.

Therefore, when applying Dynamic Partition, there is a trade-off between the convenience and efficiency of metadata management.

Developers' words​

As the complexity of business adds up, Dynamic Partition becomes inadequate because:

• It only supports partitioning by RANGE but not by LIST.

• It can only be applied to the current real-world timestamps.

• It only supports a single continuous partition range, and cannot accommodate partitions outside of that range.

Given these functional limitations, we started to plan a new partitioning mechanism that can both automate partition management and simplify data table maintenance.

We figured out that the ideal partitioning implementation should:

• Save the need for manually creating partitions after table creation;

• Be able to accommodate all ingested data in corresponding partitions.

The former stands for automation and the latter for flexibility. The essence of realizing them both is associating partition creation with the actual data.

Then we started to think about: What if we hold off the creation of partitions until the data is ingested, rather than doing it during table creation or through regular polling. Instead of pre-constructing the partition distribution, we can define the "data-to-partition" mapping rules, so the partitions are created after data arrives.

Compared to Manual Partition, this whole process would be fully automated, eliminating the need for human maintenance. Compared to Dynamic Partition, it avoids having partitions that are not used, or partitions that are needed but not present.

Auto Partition​

With Apache Doris 2.1.0, we bring the above plan into fruition. During data ingestion, Doris creates data partitions based on the configured rules. The Doris Backend nodes that are responsible for data processing and distribution will attempt to find the appropriate partition for each row of data in the DataSink operator of the execution plan. It no longer filters out data that does not fit into any existing partition or reports an error for such a situation, but automatically generates partitions for all ingested data.

Auto Partition by RANGE​

Auto Partition by RANGE provides an optimized partitioning solution based on the time dimension. It is more flexible than Dynamic Partition in terms of parameter configuration. The syntax for it is as follows:

AUTO PARTITION BY RANGE (FUNC_CALL_EXPR)
()
FUNC_CALL_EXPR ::= DATE_TRUNC ( <partition_column>, '<interval>' )

The <partition_column> above is the partition column (i.e., the column that the partitioning is based on). <interval>specifies the partition unit, which is the desired width of each partition.

For example, if the partition column is k0 and you want to partition by month, the partition statement would be AUTO PARTITION BY RANGE (DATE_TRUNC(k0, 'month')). For all the imported data, the system will call DATE_TRUNC(k0, 'month') to calculate the left endpoint of the partition, and then the right endpoint by adding one interval.

Now, we can apply Auto Partition to the DAILY_TRADE_VALUE table introduced in the previous section on Dynamic Partition.

CREATE TABLE DAILY_TRADE_VALUE
(
    `TRADE_DATE`    DATEV2 NOT NULL COMMENT 'Trade Date',
    `TRADE_ID`      VARCHAR(40) NOT NULL COMMENT 'Trade ID',
    ......
)
AUTO PARTITION BY RANGE (DATE_TRUNC(`TRADE_DATE`, 'month'))
()
DISTRIBUTED BY HASH(`TRADE_DATE`) BUCKETS 10
PROPERTIES
(
    ......
);

After importing some data, these are the partitions we get:

mysql> show partitions from DAILY_TRADE_VALUE;
Empty set (0.10 sec)

mysql> insert into DAILY_TRADE_VALUE values ('2015-01-01', 1), ('2020-01-01', 2), ('2024-03-05', 10000), ('2024-03-06', 10001);
Query OK, 4 rows affected (0.24 sec)
{'label':'label_2a7353a3f991400e_ae731988fa2bc568', 'status':'VISIBLE', 'txnId':'85097'}

mysql> show partitions from DAILY_TRADE_VALUE;
+-------------+-----------------+----------------+---------------------+--------+--------------+--------------------------------------------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+
| PartitionId | PartitionName   | VisibleVersion | VisibleVersionTime  | State  | PartitionKey | Range                                                                          | DistributionKey | Buckets | ReplicationNum | StorageMedium | CooldownTime        | RemoteStoragePolicy | LastConsistencyCheckTime | DataSize | IsInMemory | ReplicaAllocation       | IsMutable | SyncWithBaseTables | UnsyncTables |
+-------------+-----------------+----------------+---------------------+--------+--------------+--------------------------------------------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+
| 588395      | p20150101000000 | 2              | 2024-06-01 19:02:40 | NORMAL | TRADE_DATE   | [types: [DATEV2]; keys: [2015-01-01]; ..types: [DATEV2]; keys: [2015-02-01]; ) | TRADE_DATE      | 10      | 1              | HDD           | 9999-12-31 23:59:59 |                     | NULL                     | 0.000    | false      | tag.location.default: 1 | true      | true               | NULL         |
| 588437      | p20200101000000 | 2              | 2024-06-01 19:02:40 | NORMAL | TRADE_DATE   | [types: [DATEV2]; keys: [2020-01-01]; ..types: [DATEV2]; keys: [2020-02-01]; ) | TRADE_DATE      | 10      | 1              | HDD           | 9999-12-31 23:59:59 |                     | NULL                     | 0.000    | false      | tag.location.default: 1 | true      | true               | NULL         |
| 588416      | p20240301000000 | 2              | 2024-06-01 19:02:40 | NORMAL | TRADE_DATE   | [types: [DATEV2]; keys: [2024-03-01]; ..types: [DATEV2]; keys: [2024-04-01]; ) | TRADE_DATE      | 10      | 1              | HDD           | 9999-12-31 23:59:59 |                     | NULL                     | 0.000    | false      | tag.location.default: 1 | true      | true               | NULL         |
+-------------+-----------------+----------------+---------------------+--------+--------------+--------------------------------------------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+
3 rows in set (0.09 sec)

As is shown, partitions are automatically created for the imported data, and it doesn't create partitions that are beyond the range of the existing data.

Auto Partition by LIST​

Auto Partition by LIST is to shard data based on non-time-based dimensions, such as region and department. It fills that gap for Dynamic Partition, which does not support data partitioning by LIST.

Auto Partition by RANGE provides an optimized partitioning solution based on the time dimension. It is more flexible than Dynamic Partition in terms of parameter configuration. The syntax for it is as follows:

AUTO PARTITION BY LIST (`partition_col`)
()

This is an example of Auto Partition by LIST using city as the partition column:

mysql> CREATE TABLE `str_table` (
    ->     `city` VARCHAR NOT NULL,
    ->     ......
    -> )
    -> DUPLICATE KEY(`city`)
    -> AUTO PARTITION BY LIST (`city`)
    -> ()
    -> DISTRIBUTED BY HASH(`city`) BUCKETS 10
    -> PROPERTIES (
    ->     ......
    -> );
Query OK, 0 rows affected (0.09 sec)

mysql> insert into str_table values ("Denver"), ("Boston"), ("Los_Angeles");
Query OK, 3 rows affected (0.25 sec)

mysql> show partitions from str_table;
+-------------+-----------------+----------------+---------------------+--------+--------------+-------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+
| PartitionId | PartitionName   | VisibleVersion | VisibleVersionTime  | State  | PartitionKey | Range                                     | DistributionKey | Buckets | ReplicationNum | StorageMedium | CooldownTime        | RemoteStoragePolicy | LastConsistencyCheckTime | DataSize | IsInMemory | ReplicaAllocation       | IsMutable | SyncWithBaseTables | UnsyncTables |
+-------------+-----------------+----------------+---------------------+--------+--------------+-------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+
| 589685      | pDenver7        | 2              | 2024-06-01 20:12:37 | NORMAL | city         | [types: [VARCHAR]; keys: [Denver]; ]      | city            | 10      | 1              | HDD           | 9999-12-31 23:59:59 |                     | NULL                     | 0.000    | false      | tag.location.default: 1 | true      | true               | NULL         |
| 589643      | pLos5fAngeles11 | 2              | 2024-06-01 20:12:37 | NORMAL | city         | [types: [VARCHAR]; keys: [Los_Angeles]; ] | city            | 10      | 1              | HDD           | 9999-12-31 23:59:59 |                     | NULL                     | 0.000    | false      | tag.location.default: 1 | true      | true               | NULL         |
| 589664      | pBoston8        | 2              | 2024-06-01 20:12:37 | NORMAL | city         | [types: [VARCHAR]; keys: [Boston]; ]      | city            | 10      | 1              | HDD           | 9999-12-31 23:59:59 |                     | NULL                     | 0.000    | false      | tag.location.default: 1 | true      | true               | NULL         |
+-------------+-----------------+----------------+---------------------+--------+--------------+-------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+
3 rows in set (0.10 sec)

After inserting data for the cities of Denver, Boston, and Los Angeles, the system automatically created corresponding partitions based on the city names. Previously, this type of custom partitioning could only be achieved through manual DDL statements. This is how Auto Partition by LIST simplifies database maintenance.

Tips & notes​

Manually adjust historical partitions

For tables that receive both real-time data and occasional historical updates, since Auto Partition does not automatically reclaim historical partitions, we recommend two options:

• Use Auto Partition, which will automatically create partitions for the occasional historical data updates.

• Use Auto Partition and manually create a LESS THAN partition to accommodate the historical updates. This allows for a clearer separation of historical and real-time data, and makes data management easier.

mysql> CREATE TABLE DAILY_TRADE_VALUE
    -> (
    ->     `TRADE_DATE`    DATEV2 NOT NULL COMMENT 'Trade Date',
    ->     `TRADE_ID`      VARCHAR(40) NOT NULL COMMENT 'Trade ID'
    -> )
    -> AUTO PARTITION BY RANGE (DATE_TRUNC(`TRADE_DATE`, 'DAY'))
    -> (
    ->     PARTITION `pHistory` VALUES LESS THAN ("2024-01-01")
    -> )
    -> DISTRIBUTED BY HASH(`TRADE_DATE`) BUCKETS 10
    -> PROPERTIES
    -> (
    ->     "replication_num" = "1"
    -> );
Query OK, 0 rows affected (0.11 sec)

mysql> insert into DAILY_TRADE_VALUE values ('2015-01-01', 1), ('2020-01-01', 2), ('2024-03-05', 10000), ('2024-03-06', 10001);
Query OK, 4 rows affected (0.25 sec)
{'label':'label_96dc3d20c6974f4a_946bc1a674d24733', 'status':'VISIBLE', 'txnId':'85092'}

mysql> show partitions from DAILY_TRADE_VALUE;
+-------------+-----------------+----------------+---------------------+--------+--------------+--------------------------------------------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+
| PartitionId | PartitionName   | VisibleVersion | VisibleVersionTime  | State  | PartitionKey | Range                                                                          | DistributionKey | Buckets | ReplicationNum | StorageMedium | CooldownTime        | RemoteStoragePolicy | LastConsistencyCheckTime | DataSize | IsInMemory | ReplicaAllocation       | IsMutable | SyncWithBaseTables | UnsyncTables |
+-------------+-----------------+----------------+---------------------+--------+--------------+--------------------------------------------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+
| 577871      | pHistory        | 2              | 2024-06-01 08:53:49 | NORMAL | TRADE_DATE   | [types: [DATEV2]; keys: [0000-01-01]; ..types: [DATEV2]; keys: [2024-01-01]; ) | TRADE_DATE      | 10      | 1              | HDD           | 9999-12-31 23:59:59 |                     | NULL                     | 0.000    | false      | tag.location.default: 1 | true      | true               | NULL         |
| 577940      | p20240305000000 | 2              | 2024-06-01 08:53:49 | NORMAL | TRADE_DATE   | [types: [DATEV2]; keys: [2024-03-05]; ..types: [DATEV2]; keys: [2024-03-06]; ) | TRADE_DATE      | 10      | 1              | HDD           | 9999-12-31 23:59:59 |                     | NULL                     | 0.000    | false      | tag.location.default: 1 | true      | true               | NULL         |
| 577919      | p20240306000000 | 2              | 2024-06-01 08:53:49 | NORMAL | TRADE_DATE   | [types: [DATEV2]; keys: [2024-03-06]; ..types: [DATEV2]; keys: [2024-03-07]; ) | TRADE_DATE      | 10      | 1              | HDD           | 9999-12-31 23:59:59 |                     | NULL                     | 0.000    | false      | tag.location.default: 1 | true      | true               | NULL         |
+-------------+-----------------+----------------+---------------------+--------+--------------+--------------------------------------------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+
3 rows in set (0.10 sec)

NULL partition

With Auto Partition by LIST, Doris supports storing NULL values in NULL partitions. For example:

mysql> CREATE TABLE list_nullable
    -> (
    ->     `str` varchar NULL
    -> )
    -> AUTO PARTITION BY LIST (`str`)
    -> ()
    -> DISTRIBUTED BY HASH(`str`) BUCKETS auto
    -> PROPERTIES
    -> (
    ->     "replication_num" = "1"
    -> );
Query OK, 0 rows affected (0.10 sec)

mysql> insert into list_nullable values ('123'), (''), (NULL);
Query OK, 3 rows affected (0.24 sec)
{'label':'label_f5489769c2f04f0d_bfb65510f9737fff', 'status':'VISIBLE', 'txnId':'85089'}

mysql> show partitions from list_nullable;
+-------------+---------------+----------------+---------------------+--------+--------------+------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+
| PartitionId | PartitionName | VisibleVersion | VisibleVersionTime  | State  | PartitionKey | Range                              | DistributionKey | Buckets | ReplicationNum | StorageMedium | CooldownTime        | RemoteStoragePolicy | LastConsistencyCheckTime | DataSize | IsInMemory | ReplicaAllocation       | IsMutable | SyncWithBaseTables | UnsyncTables |
+-------------+---------------+----------------+---------------------+--------+--------------+------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+
| 577297      | pX            | 2              | 2024-06-01 08:19:21 | NORMAL | str          | [types: [VARCHAR]; keys: [NULL]; ] | str             | 10      | 1              | HDD           | 9999-12-31 23:59:59 |                     | NULL                     | 0.000    | false      | tag.location.default: 1 | true      | true               | NULL         |
| 577276      | p0            | 2              | 2024-06-01 08:19:21 | NORMAL | str          | [types: [VARCHAR]; keys: []; ]     | str             | 10      | 1              | HDD           | 9999-12-31 23:59:59 |                     | NULL                     | 0.000    | false      | tag.location.default: 1 | true      | true               | NULL         |
| 577255      | p1233         | 2              | 2024-06-01 08:19:21 | NORMAL | str          | [types: [VARCHAR]; keys: [123]; ]  | str             | 10      | 1              | HDD           | 9999-12-31 23:59:59 |                     | NULL                     | 0.000    | false      | tag.location.default: 1 | true      | true               | NULL         |
+-------------+---------------+----------------+---------------------+--------+--------------+------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+
3 rows in set (0.11 sec)

However, Auto Partition by RANGE does not support NULL partitions, because the NULL values will be stored in the smallest LESS THAN partition, and it is impossible to reliably determine the appropriate range for it. If Auto Partition were to create a NULL partition with a range of (-INFINITY, MIN_VALUE), there would be a risk of this partition being inadvertently deleted in production, as the MIN_VALUE boundary may not accurately represent the intended business logic.

Summary​

Auto Partition covers most of the use cases of Dynamic Partition, while introducing the benefit of upfront partition rule definition. Once the rules are defined, the bulk of partition creation work is automatically handled by Doris instead of a DBA.

Before utilizing Auto Partition, it's important to understand the relevant limitations:

1. Auto Partition by LIST supports partitioning based on multiple columns, but each automatically created partition only contains one single value, and the partition name cannot exceed 50 characters in length. Note that the partition names follow specific naming conventions, which have particular implications for metadata management. That means not all of the 50-character space is at the user's disposal.

2. Auto Partition by RANGE only supports a single partition column, which must be of type DATE or DATETIME.

3. Auto Partition by LIST supports NULLABLE partition column and inserting NULL values. Auto Partition by RANGE does not support NULLABLE partition column.

4. It is not recommended to use Auto Partition in conjunction with Dynamic Partition after Apache Doris 2.1.3.

Performance comparison​

The main functional differences between Auto Partition and Dynamic Partition lie in partition creation and deletion, supported partition types, and their impact on import performance.

Dynamic Partition uses fixed threads to periodically create and reclaim partitions. It only supports partitioning by RANGE. In contrast, Auto Partition supports both partitioning by RANGE and by LIST. It automatically creates partitions on-demand based on specific rules during data ingestion, providing a higher level of automation and flexibility.

Dynamic Partition does not slow down data ingestion speed, while Auto Partition causes certain time overheads because it firstly checks for existing partitions and then creates new ones on demand. We will present the performance test results.

Auto Partition: ingestion workflow​

This part is about how data ingestion is implemented with the Auto Partition mechanism, and we use Stream Load as an example. When Doris initiates a data import, one of the Doris Backend nodes takes on the role of the Coordinator. It is responsible for the initial data processing work and then dispatching the data to the appropriate BE nodes, known as the Executors, for execution.

In the final Datasink Node of the Coordinator's execution pipeline, the data needs to be routed to the correct partitions, buckets, and Doris Backend node locations before it can be successfully transmitted and stored.

To enable this data transfer, the Coordinator and Executor nodes establish a communication channels:

• The sending end is called the Node Channel.

• The receiving end is called the Tablets Channel.

This is how Auto Partition comes into play during the process of determining the correct partitions for the data:

Previously, without Auto Partition, when a table does not have the required partition, the behavior in Doris is for the BE nodes to accumulate errors until a DATA_QUALITY_ERROR is reported. Now, with Auto Partition enabled, a request will be initiated to the Doris Frontend to create the necessary partition on-the-fly. After the partition creation transaction is completed, the Doris Frontend responds to the Coordinator, which then opens the corresponding communication channels (Node Channel and Tablets Channel) to continue the data ingestion process. This is a seamless experience for users.

In a real-world cluster environment, the time spent by the Coordinator waiting for the Doris Frontend to complete partition creation can incur large overheads. This is due to the inherent latency of Thrift RPC calls, as well as lock contention on the Frontend under high load conditions.

To improve the data ingestion efficiency in Auto Partition, Doris has implemented batching to largely reduce the number of RPC calls made to the FE. This brings a notable performance enhancement for data write operations.

Note that when the FE Master completes the partition creation transaction, the new partition becomes immediately visible. However, if the import process ultimately fails or is canceled, the created partitions are not automatically reclaimed.

Auto Partition performance​

We tested the performance and stability of Auto Partition in Doris, covering different use cases:

Case 1: 1 Frontend + 3 Backend; 6 randomly generated datasets, each having 100 million rows and 2,000 partitions; ingested the 6 datasets concurrently into 6 tables

• Objective: Evaluate the performance of Auto Partition under high pressure and check for any performance degradation.

• Results: Auto Partition brings an average performance loss less than 5%, with all import transactions running stably.

Case 2: 1 Frontend + 3 Backend; ingesting 100 rows per second from Flink by Routine Load; testing with 1, 10, and 20 concurrent transactions (tables), respectively

• Objective: Identify any potential or data backlog issues that could arise with Auto Partition under different concurrency levels.

• Results: With or without Auto Partition enabled, the data ingestion was successful without any backpressure issues across all the concurrency levels tested, even at 20 concurrent transactions when the CPU utilization reached close to 100%.

To conclude the results of these tests, the impact of enabling Auto Partition on data ingestion performance is minimal.

Conclusion and future plans​

Auto Partition has simplified DDL and partition management since Apache Doris 2.1.0. It is useful in large-scale data processing and makes it easy for users to migrate from other database systems to Apache Doris.

Moreover, we are committed to expanding the capabilities of Auto Partition to support more complex data types.

Plans for Auto Partition by RANGE:

• Support numeric values;

• Allowing users to specify the left and right boundaries of the partition range.

Plans for Auto Partition by LIST:

• Allow merging multiple values into the same partition based on specific rules.

Join Apache Doris open-source community for more information and further guidance.

What makes a modern database system? The three key modules are query optimizer, execution engine, and storage engine. Among them, the role of execution engine to the DBMS is like the chef to a restaurant. This article focuses on the execution engine of the Apache Doris data warehouse, explaining the secret to its high performance.

To illustrate the role of the execution engine, let's follow the execution process of an SQL statement:

• Upon receiving an SQL query, the query optimizer performs syntax/lexical analysis and generates the optimal execution plan based on the cost model and optimization rules.

• The execution engine then schedules the plan to the nodes, which operate on data in the underlying storage engine and then return the query results.

The execution engine performs operations like data reading, filtering, sorting, and aggregation. The efficiency of these steps determines query performance and resource utilization. That's why different execution models bring distinction in query efficiency.

Volcano Model​

The Volcano Model (originally known as the Iterator Model) predominates in analytical databases, followed by the Materialization Model and Vectorized Model. In a Volcano Model, each operation is abstracted as an operator, so the entire SQL query is an operator tree. During query execution, the tree is traversed top-down by calling the next() interface, and data is pulled and processed from the bottom up. This is called a pull-based execution model.

The Volcano Model is flexible, scalable, and easy to implement and optimize. It underpins Apache Doris before version 2.1.0. When a user initiates an SQL query, Doris parses the query, generates a distributed execution plan, and dispatches tasks to the nodes for execution. Each individual task is an instance. Take a simple query as an example:

select age, sex from employees where age > 30

In an instance, data flows between operators are propelled by the next() method. If the next() method of an operator is called, it will first call the next() of its child operator, obtain data from it, and then process the data to produce output.

next() is a synchronous method. In other words, the current operator will be blocked if its child operator does not provide data for it. In this case, the next() method of the root operator needs to be called in a loop until all data is processed, which is when the instance finishes its computation.

Such execution mechanism faces a few bottlenecks in single-node, multi-core use cases:

• Thread blocking: In a fixed-size thread pool, if an instance occupies a thread and it is blocked, that will easily cause a deadlock when there are a large number of instances requesting execution simultaneously. This is especially the case when the current instance is dependent on other instances. Additionally, if a node is running more instances than the number of CPU cores it has, the system scheduling mechanism will be heavily relied upon and a huge context switching overhead can be produced. In a colocation scenario, this will lead to an even larger thread switching overhead.

• CPU contention: The threads might compete for CPU resources so queries of different sizes and between different tenants might interfere with each other.

• Underutilization of the multi-core computing capabilities: Execution concurrency relies heavily on data distribution. Specifically, the number of instances running on a node is limited by the number of data buckets on that node. In this case, it's important to set an appropriate number of buckets. If you shard the data into too many buckets, that will become a burden for the system and bring unnecessary overheads; if the buckets are too few, you will not be able to utilize your CPU cores to the fullest. However, in a production environment, it is not always easy to estimate the proper number of buckets you need, thus performance loss.

Pipeline Execution Engine​

Based on the known issues of Volcano Model, we've replaced it with the Pipeline Execution Engine since Apache Doris 2.0.0.

As the name suggests, the Pipeline Execution Engine breaks down the execution plan into pipeline tasks, and schedules these pipeline tasks into a thread pool in a time-sharing manner. If a pipeline task is blocked, it will be put on hold to release the thread it is occupying. Meanwhile, it supports various scheduling strategies, meaning that you can allocate CPU resources to different queries and tenants more flexibly.

Additionally, the Pipeline Execution Engine pools together data within data buckets, so the number of running instances is no longer capped by the number of buckets. This not only enhances Apache Doris' utilization of multi-core systems, but also improves system performance and stability by avoiding frequent thread creation and deletion.

Example​

This is the execution plan of a join query. It includes two instances:

As illustrated, the Probe operation can only be executed after the hash table is built, while the Build operation is reliant on the computation results of the Exchange operator. Each of the two instances is divided into two pipeline tasks as such. Then these tasks will be scheduled in the "ready" queue of the thread pool. Following the specified strategies, the threads obtain the tasks to process. In a pipeline task, after one data block is finished, if the relevant data is ready and its runtime stays within the maximum allowed duration, the thread will continue to compute the next data block.

Design & implementation​

Avoid thread blocking

As is mentioned earlier, the Volcano Model is faced with a few bottlenecks:

1. If too many threads are blocked, the thread pool will be saturated and unable to respond to subsequent queries.

2. Thread scheduling is entirely managed by the operating system, without any user-level control or customization.

How does Pipeline Execution Engine avoid such issues?

1. We fix the size of the thread pool to match the CPU core count. Then we split all operators that are prone to blocking into pipeline tasks. For example, we use individual threads for disk I/O operations and RPC operations.

2. We design a user-space polling scheduler. It continuously checks the state of all executable pipeline tasks and assigns executable tasks to threads. With this in place, the operating system doesn't have to frequently switch threads, thus less overheads. It also allows customized scheduling strategies, such as assigning priorities to tasks.

Parallelization

Before version 2.0, Apache Doris requires users to set a concurrency parameter for the execution engine (parallel_fragment_exec_instance_num), which does not dynamically change based on the workloads. Therefore, it is a burden for users to figure out an appropriate concurrency level that leads to optimal performance.

What's the industry's solution to this?

Presto's idea is to shuffle the data into a reasonable number of partitions during execution, which requires minimal concurrency control from users. On the other hand, DuckDB introduces an extra synchronization mechanism instead of shuffling. We decide to follow Presto's track of Presto because the DuckDB solution inevitably involves the use of locks, which works against our purpose of avoiding blocking.

Unlike Presto, Apache Doris doesn't need an extra Local Exchange mechanism to shards the data into an appropriate number of partitions. With its massively parallel processing (MPP) architecture, Doris already does so during shuffling. (In Presto's case, it re-partitions the data via Local Exchange for higher execution concurrency. For example, in hash aggregation, Doris further shards the data based on the aggregation key in order to fully utilize the CPU cores. Also, this can downsize the hash table that each execution thread has to build.)

Based on the MPP architecture, we only need two improvements before we achieve what we want in Doris:

• Increase the concurrency level during shuffling. For this, we only need to have the frontend (FE) perceive the backend (BE) environment and then set a reasonable number of partitions.

• Implement concurrent execution after data reading by the scan layer. To do this, we need a logical restructuring of the scan layer to decouple the threads from the number of data tablets. This is a pooling process. We pool the data read by scanner threads, so it can be fetched by multiple pipeline tasks directly.

PipelineX​

Introduced in Apache Doris 2.0.0, the pipeline execution engine has been improving query performance and stability under hybrid workload scenarios (queries of different sizes and from different tenants). In version 2.1.0, we've tackled the known issues and upgraded this from an experimental feature to a robust and reliable solution, which is what we call PipelineX.

PipelineX has provided answers to the following issues that used to challenge the Pipeline Execution Engine:

• Limited execution concurrency

• High execution overhead

• High scheduling overhead

• Poor readability of operator profile

Execution concurrency​

The Pipeline Execution Engine remains under the restriction of the static concurrency parameter at FE and the tablet count at the storage layer, making itself unable to capitalize on the full computing resources. Plus, it is easily affected by data skew.

For example, suppose that Table A contains 100 million rows but it has only 1 tablet, which means it is not sharded enough, let's see what can happen when you perform an aggregation query on it:

 SELECT COUNT(*) FROM A GROUP BY A.COL_1;

During query execution, the query plan is divided into two fragments. Each fragment, consisting of multiple operators, is dispatched by frontend (FE) to backend (BE). The BE starts threads to execute the fragments concurrently.

Now, let's focus on Fragment 0 for further elaboration. Because there is only one tablet, Fragment 0 can only be executed by one thread. That means aggregation of 100 million rows by one single thread. If you have 16 CPU cores, ideally, the system can allocate 8 threads to execute Fragment 0. In this case, there is a concurrency disparity of 8 to 1. This is how the number of tablets restricts execution concurrency and also why we introduce the idea of Local Shuffle mechanism to remove that restriction in Apache Doris 2.1.0. So this is how it works in PipelineX:

• The threads execute their own pipeline tasks, but the pipeline tasks only maintain their runtime state (known as Local State), while the information that shared across all pipeline tasks (known as Global State) is managed by one pipeline object.

• On a single BE, the Local Shuffle mechanism is responsible for data distribution and data balancing across pipeline tasks.

Apart from decoupling execution concurrency from tablet count, Local Shuffle can avoid performance loss due to data skew. Again, we will explain with the foregoing example.

This time, we shard Table A into two tablets instead of one, but the data is not evenly distributed. Tablet 1 and Tablet 3 hold 10 million and 90 million rows, respectively. The Pipeline Execution Engine and PipelineX Execution Engine respond differently to such data skew:

• Pipeline Execution Engine: Thread 1 and Thread 2 executes Fragment 1 concurrently. The latter takes 9 times as long as Thread 1 because of the different data sizes they deal with.

• PipelineX Execution Engine: With Local Shuffle, data is distributed evenly to the two threads, so they take almost equal time to finish.

Execution overhead​

Under the Pipeline Execution Engine, because the expressions of different instances are individual, each instance is initialized individually. However, since the initialization parameters of instances share a lot in common, we can reuse the shared states to reduce execution overheads. This is what PipelineX does: it initializes the Global State at a time, and the Local State sequentially.

Scheduling overhead​

In the Pipeline Execution Engine, the blocked tasks are put into a blocked queue, where a dedicated thread takes polls and moves the executable tasks over to the runnable queue. This dedicated scheduling thread consumes a CPU core, and incurs overheads that can be particularly noticeable on systems with limited computing resources.

As a better solution, PipelineX encapsulates the blocking conditions as dependencies, and the task status (blocked or runnable) will be triggered to change by event notifications. Specifically, when RPC data arrives, the relevant task will be considered as ready by the ExchangeSourceOperator and then moved to the runnable queue.

That means PipelineX implements event-driven scheduling. A query execution plan can be depicted as a DAG, where the pipeline tasks are abstracted as nodes and the dependencies as edges. Whether a pipeline task gets executed depends on whether all its associated dependencies have satisfied the requisite conditions.

For simplicity of illustration, the above DAG only shows the dependencies between the upstream and downstream pipeline tasks. In fact, all blocking conditions are abstracted as dependencies. The complete execution workflow of a pipeline task is as follows:

In event-driven execution, a pipeline task will only be executed after all its dependencies satisfy the conditions; otherwise, it will be added to the blocked queue. When an external event arrives, all blocked tasks will be re-evaluated to see if they're runnable.

The event-driven design of PipelineX eliminates the need for a polling thread and thus the consequential performance loss under high cluster loads. Moreover, the encapsulation of dependencies enables a more flexible scheduling framework, making it easier to spill data to disks.

Operator profile​

PipelineX has reorganized the metrics in the operator profiles, adding new ones and obsoleting irrelevant ones. Besides, with the dependencies encapsulated, we monitor how long the dependencies take to get ready by the metric WaitForDependency, so the profile can provide a clear picture of the time spent in each step. These are two examples:

• Scan Operator: The total execution time of OLAP_SCAN_OPERATOR is 457.750ms, including that spent in data reading by the scanner (436.883ms) and that in actual execution.

  OLAP_SCAN_OPERATOR  (id=4.  table  name  =  Z03_DI_MID):
      -  ExecTime:  457.750ms
      -  WaitForDependency[OLAP_SCAN_OPERATOR_DEPENDENCY]Time:  436.883ms
  

• Exchange Source Operator: The execution time of EXCHANGE_OPERATOR is 86.691us. The time spent waiting for data from upstream is 409.256us.

  EXCHANGE_OPERATOR  (id=3):
      -  ExecTime:  86.691us
      -  WaitForDependencyTime:  0ns
          -  WaitForData0:  409.256us
  

What's next​

From the Volcano Model to the Pipeline Execution Engine, Apache Doris 2.0.0 has overcome the deadlocks under high cluster load and greatly increased CPU utilization. Now, from the Pipeline Execution Engine to PipelineX, Apache Doris 2.1.0 is more production-friendly as it has ironed out the kinks in concurrency, overheads, and operator profile.

What's next in our roadmap is to support spilling data to disk in PipelineX to further improve query speed and system reliability. We also plan to advance further in terms of automation, such as self-adaptive concurrency and auto execution plan optimization, accompanied by NUMA technologies to harvest better performance from hardware resources.

If you want to talk to the amazing Doris developers who lead these changes, you are more than welcome to join the Apache Doris community.

Job scheduling is an important part of data management as it enables regular data updates and cleanups. In a data platform, it is often undertaken by workflow orchestration tools like Apache Airflow and Apache Dolphinscheduler. However, adding another component to the data architecture also means investing extra resources for management and maintenance. That's why Apache Doris 2.1.0 introduces a built-in Job Scheduler. It is strategically more tailored to Apache Doris, and brings higher scheduling flexibility and architectural simplicity.

The Doris Job Scheduler triggers the pre-defined operations at specific time points or intervals, thus allowing for efficient and reliable task automation. Its key capabilities include:

• Efficiency: It adopts the TimeWheel algorithm to ensure that the triggering of tasks is precise to the second.

• Flexibility: It supports both one-time jobs and regular jobs. For the latter, users can define the start/end time, and intervals of minutes, hours, days, or weeks.

• Execution thread pool and processing queue: It is supported by a Disruptor-based single-producer, multi-consumer model to avoid task execution overload.

• Traceability: It keeps track of the latest task execution records (configurable), which are queryable by a simple command.

• Availability: Like Apache Doris itself, the Doris Job Scheduler is easily recoverable and highly available.

Syntax & examples​

Syntax description​

A valid job statement consists of the following elements:

• CREATE JOB: Specifies the job name as a unique identifier.

• The ON SCHEDULE clause: Specifies the type, trigger time, and frequency of the job.

  • AT timestamp: This is used to specify a one-time job. AT CURRENT_TIMESTAMP means that the job will run immediately upon creation.

  • EVERY: This is used to specify a regular job. You can define the execution frequency of the job. The interval can be measured in weeks, days, hours, and minutes.

    • The EVERY clause supports an optional STARTS clause with a timestamp to define the start time of the recurring schedule. CURRENT_TIMESTAMP can be used. It also supports an optional ENDS clause to specify the end time for the job.

• The DO clause defines the action to be performed when the job is executed. At this time, the only supported operation is INSERT.

  CREATE
      JOB
      job_name
      ON SCHEDULE schedule
      [COMMENT 'string']
      DO execute_sql;
  
  schedule: {
      AT timestamp 
    | EVERY interval
      [STARTS timestamp ]
      [ENDS timestamp ]
  }
  
  interval:
      quantity { WEEK |DAY | HOUR | MINUTE
              }
              
  

  Example:

  CREATE JOB my_job ON SCHEDULE EVERY 1 MINUTE DO INSERT INTO db1.tbl1 SELECT * FROM db2.tbl2;
  

  The above statement creates a job named my_job, which is to load data from db2.tbl2 to db1.tbl1 every minute.

More examples​

Create a one-time job: Load data from db2.tbl2 to db1.tbl1 at 2025-01-01 00:00:00.

CREATE JOB my_job ON SCHEDULE AT '2025-01-01 00:00:00' DO INSERT INTO db1.tbl1 SELECT * FROM db2.tbl2;

Create a regular job without specifying the end time: Load data from db2.tbl2 to db1.tbl1 once a day starting from 2025-01-01 00:00:00.

CREATE JOB my_job ON SCHEDULE EVERY 1 DAY STARTS '2025-01-01 00:00:00' DO INSERT INTO db1.tbl1 SELECT * FROM db2.tbl2 WHERE  create_time >=  days_add(now(),-1);

Create a regular job within a specified period: Load data from db2.tbl2 to db1.tbl1 once a day, beginning at 2025-01-01 00:00:00 and finishing at 2026-01-01 00:10:00.

CREATE JOB my_job ON SCHEDULER EVERY 1 DAY STARTS '2025-01-01 00:00:00' ENDS '2026-01-01 00:10:00' DO INSERT INTO db1.tbl1 SELECT * FROM db2.tbl2 create_time >=  days_add(now(),-1);

Asynchronous execution: Because jobs are executed in an asynchronous manner in Doris. Tasks that require asynchronous execution, such as insert into select, can be implemented by a job.

For example, to asynchronously execute data loading from db2.tbl2 to db1.tbl1, simply create a one-time job for it and schedule it at current_timestamp.

CREATE JOB my_job ON SCHEDULE AT current_timestamp DO INSERT INTO db1.tbl1 SELECT * FROM db2.tbl2;

Auto data synchronization​

The combination of the Job Scheduler and the Multi-Catalog feature of Apache Doris is an efficient way to implement regular data synchronization across data sources.

This is useful in many cases, such as for an e-commerce user who regularly needs to load business data from MySQL to Doris for analysis.

Example: To filter consumers by total consumption amount, last visit time, sex, and city in the table below, and import the query results to Doris regularly.

Step 1: Create a table in Doris

CREATE TABLE IF NOT EXISTS user_activity
(
    `user_id` LARGEINT NOT NULL COMMENT "User ID",
    `date` DATE NOT NULL COMMENT "Time of data import",
    `city` VARCHAR(20) COMMENT "User city",
    `age` SMALLINT COMMENT "User age",
    `sex` TINYINT COMMENT "User sex",
    `last_visit_date` DATETIME REPLACE DEFAULT "1970-01-01 00:00:00" COMMENT "Time of user's last visit",
    `cost` BIGINT SUM DEFAULT "0" COMMENT "User's total consumption amount",
    `max_dwell_time` INT MAX DEFAULT "0" COMMENT "Maximum dwell time of user",
    `min_dwell_time` INT MIN DEFAULT "99999" COMMENT "Minimum dwell time of user"
)
AGGREGATE KEY(`user_id`, `date`, `city`, `age`, `sex`)
DISTRIBUTED BY HASH(`user_id`) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 1"
);

Step 2: Create a catalog in Doris to map to the data in MySQL

CREATE CATALOG activity PROPERTIES (
    "type"="jdbc",
    "user"="root",
    "jdbc_url" = "jdbc:mysql://127.0.0.1:9734/user?useSSL=false",
    "driver_url" = "mysql-connector-java-5.1.49.jar",
    "driver_class" = "com.mysql.jdbc.Driver"
);

Step 3: Ingest data from MySQL to Doris. Leverage the catalog mechanism and the Insert Into method for full data ingestion. (We recommend that such operations be executed during low-traffic hours to minimize potential service disruptions.)

• One-time job: Schedule a one-time full-scale data loading that starts at 2024-8-10 03:00:00.

  CREATE JOB one_time_load_job
  ON SCHEDULE 
  AT '2024-8-10 03:00:00'
  DO
  INSERT INTO user_activity FROM SELECT * FROM activity.user.activity 
  
  

• Regular job: Create a regular job to update data periodically.

  CREATE JOB schedule_load
  ON SCHEDULE EVERY 1 DAY
  DO
  INSERT INTO user_activity FROM SELECT * FROM activity.user.activity where create_time >=  days_add(now(),-1)
  

Technical design & implementation​

Efficient scheduling often comes at the cost of significant resource consumption, and high-precision scheduling is even more resource-intensive. To implement job scheduling, some people rely on the built-in scheduling capabilities of Java, while others employ job scheduling libraries. But what if we want higher precision and lower memory usage than these solutions can reach? For that, the Doris makers combine the TimingWheel algorithm with the Disruptor framework to achieve second-level job scheduling.

To implement the TimingWheel algorithm, we leverage the HashedWheelTimer in Netty. The Job Manager puts tasks every 10 minutes (by default) in the TimeWheel for scheduling. In order to ensure efficient task triggering and avoid high resource usage, we adopt a Disruptor-based single-producer, multi-consumer model. The TimeWheel only triggers tasks but does not execute jobs directly. Tasks that need to be triggered upon expiration will be put into a Dispatch thread and distributed to an appropriate execution thread pool. Tasks that need to be executed immediately will be directly submitted to the corresponding execution thread pool.

This is how we improve processing efficiency by reducing unnecessary traversal: For one-time tasks, their definition will be removed after execution. For recurring tasks, the system events in the TimeWheel will periodically fetch the next round of execution tasks. This helps to avoid the accumulation of tasks in a single bucket.

In addition, for transactional tasks, the Job Scheduler can ensure data consistency and integrity by the transaction association and transaction callback mechanisms.

Applicable scenarios​

The Doris Job Scheduler is a Swiss Army Knife. It is not only useful in ETL and data lake analytics as we mentioned, but also critical for the implementation of asynchronous materialized views. An asynchronous materialized view is a pre-computed result set. Unlike normal materialized views, it can be built on multiple tables. Thus, as you can imagine, changes in any of the source tables will lead to the need for updates in the asynchronous materialized view. That's why we apply the job scheduling mechanism for periodic data refreshing in asynchronous materialized views, which is low-maintenance and also ensures data consistency.

Where are we going with the Doris Job Scheduler? The Apache Doris developer community is looking at:

• Displaying the distribution of tasks executed in different time slots on the WebUI.

• DAG jobs. This will allow data warehouse task orchestration within Apache Doris, which will unlock many possibilities when it is combined with the Multi-Catalog feature.

• Support for more operations such as UPDATE and DELETE.

This is an in-depth introduction to the workload isolation capabilities of Apache Doris. But first of all, why and when do you need workload isolation? If you relate to any of the following situations, read on and you will end up with a solution:

• You have different business departments or tenants sharing the same cluster and you want to prevent the interference of workloads among them.

• You have query tasks of varying priority levels and you want to give priority to your critical tasks (such as real-time data analytics and online transactions) in terms of resources and execution.

• You need workload isolation but also want high cost-effectiveness and resource utilization rates.

Apache Doris supports workload isolation based on Resource Tag and Workload Group. Resource Tag isolates the CPU and memory resources for different workloads at the level of backend nodes, while the Workload Group mechanism can further divide the resources within a backend node for higher resource utilization.

Resource isolation based on Resource Tag​

Let's begin with the architecture of Apache Doris. Doris has two types of nodes: frontends (FEs) and backends (BEs). FE nodes store metadata, manage clusters, process user requests, and parse query plans, while BE nodes are responsible for computation and data storage. Thus, BE nodes are the major resource consumers.

The main idea of a Resource Tag-based isolation solution is to divide computing resources into groups by assigning tags to BE nodes in a cluster, where BE nodes of the same tag constitute a Resource Group. A Resource Group can be deemed as a unit for data storage and computation. For data ingested into Doris, the system will write data replicas into different Resource Groups according to the configurations. Queries will also be assigned to their corresponding Resource Groups for execution.

For example, if you want to separate read and write workloads in a 3-BE cluster, you can follow these steps:

1. Assign Resource Tags to BE nodes: Bind 2 BEs to the "Read" tag and 1 BE to the "Write" tag.

2. Assign Resource Tags to data replicas: Assuming that Table 1 has 3 replicas, bind 2 of them to the "Read" tag and 1 to the "Write" tag. Data written into Replica 3 will be synchronized to Replica 1 and Replica 2 and the data synchronization process consumes few resources of BE 1 and BE2.

3. Assign workload groups to Resource Tags: Queries that include the "Read" tag in their SQLs will be automatically routed to the nodes tagged with "Read" (in this case, BE 1 and BE 2). For data writing tasks, you also need to assign them with the "Write" tag, so they can be routed to the corresponding node (BE 3). In this way, there will be no resource contention between read and write workloads except the data synchronization overheads from replica 3 to replicate 1 and 2.

Resource Tag also enables multi-tenancy in Apache Doris. For example, computing and storage resources tagged with "User A" are for User A only, while those tagged with "User B" are exclusive to User B. This is how Doris implements multi-tenant resource isolation with Resource Tags at the BE side.

Dividing the BE nodes into groups ensures a high level of isolation:

• CPU, memory, and I/O of different tenants are physically isolated.

• One tenant will never be affected by the failures (such as process crashes) of another tenant.

But it has a few downsides:

• In read-write separation, when the data writing stops, the BE nodes tagged with "Write" become idle. This reduces overall cluster utilization.

• Under multi-tenancy, if you want to further isolate different workloads of the same tenant by assigning separate BE nodes to each of them, you will need to endure significant costs and low resource utilization.

• The number of tenants is tied to the number of data replicas. So if you have 5 tenants, you will need 5 data replicas. That's huge storage redundancy.

To improve on this,we provide a workload isolation solution based on Workload Group in Apache Doris 2.0.0, and enhanced it in Apache Doris 2.1.0

Workload isolation based on Workload Group​

The Workload Group-based solution realizes a more granular division of resources. It further divides CPU and memory resources within processes on BE nodes, meaning that the queries in one BE node can be isolated from each other to some extent. This avoids resource competition within BE processes and optimizes resource utilization.

Users can relate queries to Workload Groups, and thus limit the percentage of CPU and memory resources that a query can use. Under high cluster loads, Doris can automatically kill the most resource-consuming queries in a Workload Group. Under low cluster loads, Doris can allow multiple Workload Groups to share idle resources.

Doris supports both CPU soft limit and CPU hard limit. The soft limit allows Workload Groups to break the limit and utilize idle resources, enabling more efficient utilization. The hard limit is a hard guarantee of stable performance because it prevents the mutual impact of Workload Groups.

(CPU soft limit and CPU hard limit are contradictory to each other. You can choose between them based on your own use case.)

Its differences from the Resource Tag-based solution include:

• Workload Groups are formed within processes. Multiple Workload Groups compete for resources within the same BE node.

• The consideration of data replica distribution is out of the picture because Workload Group is only a way of resource management.

CPU soft limit​

CPU soft limit is implemented by the cpu_share parameter, which is similar to weights conceptually. Workload Groups with higher cpu_share will be allocated more CPU time during a time slot.

For example, if Group A is configured with a cpu_share of 1, and Group B, 9. In a time slot of 10 seconds, when both Group A and Group B are fully loaded, Group A and Group B will be able to consume 1s and 9s of CPU time, respectively.

What happens in real-world cases is that, not all workloads in the cluster run at full capacity. Under the soft limit, if Group B has low or zero workload, then Group A will be able to use all 10s of CPU time, thus increasing the overall CPU utilization in the cluster.

A soft limit brings flexibility and a higher resource utilization rate. On the flip side, it might cause performance fluctuations.

CPU hard limit​

CPU hard limit in Apache Doris 2.1.0 is designed for users who require stable performance. In simple terms, the CPU hard limit defines that a Workload Group cannot use more CPU resources than its limit whether there are idle CPU resources or not.

This is how it works:

Suppose that Group A is set with cpu_hard_limit=10% and Group B with cpu_hard_limit=90%. If both Group A and Group B run at full load, Group A and Group B will respectively use 10% and 90% of the overall CPU time. The difference lies in when the workload of Group B decreases. In such cases, regardless of how high the query load of Group A is, it should not use more than the 10% CPU resources allocated to it.

As opposed to soft limit, a hard limit guarantees stable system performance at the cost of flexibility and the possibility of a higher resource utilization rate.

Memory resource limit​

The memory of a BE node comprises the following parts:

• Reserved memory for the operating system.

• Memory consumed by non-queries, which is not considered in the Workload Group's memory statistics.

• Memory consumed by queries, including data writing. This can be tracked and controlled by Workload Group.

The memory_limit parameter defines the maximum (%) memory available to a Workload Group within the BE process. It also affects the priority of Resource Groups.

Under initial status, a high-priority Resource Group will be allocated more memory. By setting enable_memory_overcommit, you can allow Resource Groups to occupy more memory than the limits when there is idle space. When memory is tight, Doris will cancel tasks to reclaim the memory resources that they commit. In this case, the system will retain memory resources for high-priority resource groups as much as possible.

Query queue​

It happens that the cluster is undertaking more loads than it can handle. In this case, submitting new query requests will not only be fruitless but also interruptive to the queries in progress.

To improve on this, Apache Doris provides the query queue mechanism. Users can put a limit on the number of queries that can run concurrently in the cluster. A query will be rejected when the query queue is full or after a waiting timeout, thus ensuring system stability under high loads.

The query queue mechanism involves three parameters: max_concurrency, max_queue_size, and queue_timeout.

Tests​

To demonstrate the effectiveness of the CPU soft limit and hard limit, we did a few tests.

• Environment: single machine, 16 cores, 64GB

• Deployment: 1 FE + 1 BE

• Dataset: ClickBench, TPC-H

• Load testing tool: Apache JMeter

CPU soft limit test​

Start two clients and continuously submit queries (ClickBench Q23) with and without using Workload Groups, respectively. Note that Page Cache should be disabled to prevent it from affecting the test results.

Comparing the throughputs of the two clients in both tests, it can be concluded that:

• Without configuring Workload Groups, the two clients consume the CPU resources on an equal basis.

• Configuring Workload Groups and setting the cpu_share to 2:1, the throughput ratio of the two clients is 2:1. With a higher cpu_share, Client 1 is provided with a higher portion of CPU resources, and it delivers a higher throughput.

CPU hard limit test​

Start a client, set cpu_hard_limit=50% for the Workload Group, and execute ClickBench Q23 for 5 minutes under a concurrency level of 1, 2, and 4, respectively.

As the query concurrency increases, the CPU utilization rate remains at around 800%, meaning that 8 cores are used. On a 16-core machine, that's 50% utilization, which is as expected. In addition, since CPU hard limits are imposed, the increase in TP99 latency as concurrency rises is also an expected outcome.

Test in simulated production environment​

In real-world usage, users are particularly concerned about query latency rather than just query throughput, since latency is more easily perceptible in user experience. That's why we decided to validate the effectiveness of Workload Group in a simulated production environment.

We picked out a SQL set consisting of queries that should be finished within 1s (ClickBench Q15, Q17, Q23 and TPC-H Q3, Q7, Q19), including single-table aggregations and join queries. The size of the TPC-H dataset is 100GB.

Similarly, we conduct tests with and without configuring Workload Groups.

As the results show:

• Without Workload Group (comparing Test 1 & 2): When dialing up the concurrency of Client 2, both clients experience a 2~3-time increase in query latency.

• Configuring Workload Group (comparing Test 3 & 4): As the concurrency of Client 2 goes up, the performance fluctuation in Client 1 is much smaller, which is proof of how it is effectively protected by workload isolation.

Recommendations & plans​

The Resource Tag-based solution is a thorough workload isolation plan. The Workload Group-based solution realizes a better balance between resource isolation and utilization, and it is complemented by the query queue mechanism for stability guarantee.

So which one to choose for your use case? Here is our recommendation:

• Resource Tag: for use cases where different business lines of departments share the same cluster, so the resources and data are physically isolated for different tenants.

• Workload Group: for use cases where one cluster undertakes various query workloads for flexible resource allocation.

In future releases, we will keep improving user experience of the Workload Group and query queue features:

• Freeing up memory space by canceling queries is a brutal method. We plan to implement that by disk spilling, which will bring higher stability in query performance.

• Since memory consumed by non-queries in the BE is not included in Workload Group's memory statistics, users might observe a disparity between the BE process memory usage and Workload Group memory usage. We will address this issue to avoid confusion.

• In the query queue mechanism, cluster load is controlled by setting the maximum query concurrency. We plan to enable dynamic maximum query concurrency based on resource availability at the BE. This is to create backpressure on the client side and thus improve the availability of Doris when clients keep submitting high loads.

• The main idea of Resource Tag is to group the BE nodes, while that of Workload Group is to further divide the resources of a single BE node. For users to grasp these ideas, they need to learn about the concept of BE nodes in Doris first. However, from an operational perspective, users only need to understand the resource consumption percentage of each of their workloads and what priority they should have when cluster load is saturated. Thus, we will try and figure out a way to flatten the learning curve for users, such as keeping the concept of BE nodes in the black box.

For further assistance on workload isolation in Apache Doris, join the Apache Doris community.

Apache Doris is an all-in-one data platform that is capable of real-time reporting, ad-hoc queries, data lakehousing, log management and analysis, and batch data processing. As more and more companies have been replacing their component-heavy data architecture with Apache Doris, there is an increasing need for a more convenient data migration solution. That's why the Doris SQL Convertor is made.

Most database systems run their own SQL dialects. Thus, migration between systems often entails modifications of SQL syntaxes. Since SQLs work closely with a company's business logic, in many cases, users have to modify their business logic, too. To reduce the transition pain for users, Apache Doris 2.1 provides the Doris SQL Convertor. It supports the SQL syntaxes of Presto, Trino, Hive, ClickHouse, and PostgreSQL. With it, users can execute queries with their old SQL syntaxes directly in Doris or batch convert their existing SQL statements on the visual interface.

Doris SQL Convertor​

The Doris SQL Convertor requires zero rewriting of SQL. Simply set sql_dialect = "trino" in the session variable, then you can execute queries in Doris using Trino SQLs.

The SQL compatibility of it has been proven by extensive tests. For example, a user tested the Doris SQL Convertor with over 30,000 SQL queries from their production environment. Turned out that the Convertor successfully converted 99.6% of the Trino SQLs and 98% of the ClickHouse SQLs.

Currently, Presto, Trino, Hive, ClickHouse, and PostgreSQL dialects are supported. We are working to add Teradata, SQL Server, and Snowflake to the list, and consistently increase the compatibility level of each SQL dialect.

Installation & usage​

SQL conversion service​

1. Download Doris SQL Convertor

2. On any frontend (FE) node, start the service using the following command.

• The SQL conversion service is stateless and can be started or stopped at any time.

• port=5001 in the command specifies the service port. (You can use any available port.)

• It is advisable to start a service individually for each FE node.

nohup ./doris-sql-convertor-1.0.1-bin-x86 run --host=0.0.0.0 --port=5001 &

3. Start a Doris cluster (Use Doris 2.1.0 or newer).

4. Set the URL for SQL conversion service in Doris. 127.0.0.1:5001 in the command represents the IP and port number of the node where the service is deployed.

MySQL> set global sql_converter_service_url = "http://127.0.0.1:5001/api/v1/convert"

After deployment, you can execute SQL directly in the command line. You can start the service by set sql_dialect = XXX. The following examples are based on ClickHouse SQL dialects.

• Presto

mysql> set sql_dialect=presto;                                                                                                                                                                                                             
Query OK, 0 rows affected (0.00 sec) 

mysql> SELECT cast(start_time as varchar(20)) as col1,                                                                                                                                                                                     
            array_distinct(arr_int) as col2,                                                                                                                                                                                             
            FILTER(arr_str, x -> x LIKE '%World%') as col3,                                                                                                                                                                              
            to_date(value,'%Y-%m-%d') as col4,                                                                                                                                                                                           
            YEAR(start_time) as col5,                                                                                                                                                                                                    
            date_add('month', 1, start_time) as col6,                                                                                                                                                                                    
            REGEXP_EXTRACT_ALL(value, '-.') as col7,                                                                                                                                                                                     
            JSON_EXTRACT('{"id": "33"}', '$.id')as col8,                                                                                                                                                                                 
            element_at(arr_int, 1) as col9,                                                                                                                                                                                              
            date_trunc('day',start_time) as col10                                                                                                                                                                                        
         FROM test_sqlconvert                                                                                                                                                                                                            
         where date_trunc('day',start_time)= DATE'2024-05-20'                                                                                                                                                                            
     order by id;                                                                                                                                                                                                                        
+---------------------+-----------+-----------+------------+------+---------------------+-------------+------+------+---------------------+                                                                                                
| col1                | col2      | col3      | col4       | col5 | col6                | col7        | col8 | col9 | col10               |                                                                                                
+---------------------+-----------+-----------+------------+------+---------------------+-------------+------+------+---------------------+                                                                                                
| 2024-05-20 13:14:52 | [1, 2, 3] | ["World"] | 2024-01-14 | 2024 | 2024-06-20 13:14:52 | ['-0','-1'] | "33" |    1 | 2024-05-20 00:00:00 |                                                                                                
+---------------------+-----------+-----------+------------+------+---------------------+-------------+------+------+---------------------+                                                                                                
1 row in set (0.03 sec)    

• ClickHouse

mysql> set sql_dialect=clickhouse;                                                                                                                                             
Query OK, 0 rows affected (0.00 sec)                                                                                                                                           
                                                                                                                                                                               
mysql> select  toString(start_time) as col1,                                                                                                                                   
             arrayCompact(arr_int) as col2,                                                                                                                                  
             arrayFilter(x -> x like '%World%',arr_str)as col3,                                                                                                              
             toDate(value) as col4,                                                                                                                                          
             toYear(start_time)as col5,                                                                                                                                      
             addMonths(start_time, 1)as col6,                                                                                                                                
             extractAll(value, '-.')as col7,                                                                                                                                 
             JSONExtractString('{"id": "33"}' , 'id')as col8,                                                                                                                
             arrayElement(arr_int, 1) as col9,                                                                                                                               
             date_trunc('day',start_time) as col10                                                                                                                           
          FROM test_sqlconvert                                                                                                                                               
          where date_trunc('day',start_time)= '2024-05-20 00:00:00'                                                                                                          
     order by id;                                                                                                                                                   
+---------------------+-----------+-----------+------------+------+---------------------+-------------+------+------+---------------------+                                    
| col1                | col2      | col3      | col4       | col5 | col6                | col7        | col8 | col9 | col10               |                                    
+---------------------+-----------+-----------+------------+------+---------------------+-------------+------+------+---------------------+                                    
| 2024-05-20 13:14:52 | [1, 2, 3] | ["World"] | 2024-01-14 | 2024 | 2024-06-20 13:14:52 | ['-0','-1'] | "33" |    1 | 2024-05-20 00:00:00 |                                    
+---------------------+-----------+-----------+------------+------+---------------------+-------------+------+------+---------------------+                                    
1 row in set (0.02 sec)

Visual interface​

For large-scale conversion, it is recommended to use the visual interface, on which you can batch upload the files for dialect conversion.

Follow these steps to deploy the visual conversion interface:

1. Environment: Docker, Docker-Compose

2. Get Doris-SQL-Convertor Docker image

3. Create a network for the image

docker network create app_network

4. Decompress the package

tar xzvf doris-sql-convertor-1.0.1.tar.gz

cd doris-sql-convertor

5. Edit the environment variables

FLASK_APP=server/app.py
FLASK_DEBUG=1
API_HOST=http://doris-sql-convertor-api:5000

# DOCKER TAG
API_TAG=latest
WEB_TAG=latest

6. Start it up

sh start.sh

After deployment, you can access the service by ip:8080 via your local browser. 8080 is the default port. You can change the mapping port. On the visual interface, you can select the source dialect type and target dialect type, and then click "Convert".

Join the Apache Doris community to seek guidance from the Doris makers or provide your feedback!

For years, JDBC and ODBC have been commonly adopted norms for database interaction. Now, as we gaze upon the vast expanse of the data realm, the rise of data science and data lake analytics brings bigger and bigger datasets. Correspondingly, we need faster and faster data reading and transmission, so we start to look for better answers than JDBC and ODBC. Thus, we include Arrow Flight SQL protocol into Apache Doris 2.1, which provides tens-fold speedups for data transfer.

High-speed data transfer based on Arrow Flight SQL​

As a column-oriented data warehouse, Apache Doris arranges its query results in the form of data Blocks in a columnar format. Before version 2.1, the Blocks must be serialized into bytes in row-oriented formats before they can be transferred to a target client via a MySQL client or JDBC/ODBC driver. Moreover, if the target client is a columnar database or a column-oriented data science component like Pandas, the data should then be de-serialized. The serialization-deserialization process is a speed bump for data transmission.

Apache Doris 2.1 has a data transmission channel built on Arrow Flight SQL. (Apache Arrow is a software development platform designed for high data movement efficiency across systems and languages, and the Arrow format aims for high-performance, lossless data exchange.) It allows high-speed, large-scale data reading from Doris via SQL in various mainstream programming languages. For target clients that also support the Arrow format, the whole process will be free of serialization/deserialization, thus no performance loss. Another upside is, Arrow Flight can make full use of multi-node and multi-core architecture and implement parallel data transfer, which is another enabler of high data throughput.

For example, if a Python client reads data from Apache Doris, Doris will first convert the column-oriented Blocks to Arrow RecordBatch. Then in the Python client, Arrow RecordBatch will be converted to Pandas DataFrame. Both conversions are fast because the Doris Blocks, Arrow RecordBatch, and Pandas DataFrame are all column-oriented.

In addition, Arrow Flight SQL provides a general JDBC driver to facilitate seamless communication between databases that supports the Arrow Flight SQL protocol. This unlocks the the potential of Doris to be connected to a wider ecosystem and to be used in more cases.

Performance test​

The "tens-fold speedups" conclusion is based on our benchmark tests. We tried reading data from Doris using PyMySQL, Pandas, and Arrow Flight SQL, and jotted down the durations, respectively. The test data is the ClickBench dataset.

Results on various data types are as follows:

As shown, Arrow Flight SQL outperforms PyMySQL and Pandas in all data types by a factor ranging from 20 to several hundreds.

Usage​

With support for Arrow Flight SQL, Apache Doris can leverage the Python ADBC Driver for fast data reading. I will showcase a few frequently executed database operations using the Python ADBC Driver (version 3.9 or later), including DDL, DML, session variable setting, and show statements.

01 Install library​

The relevant library is already published on PyPI. It can be installed simply as follows:

pip install adbc_driver_manager
pip install adbc_driver_flightsql

Import the following module/library to interact with the installed library:

import adbc_driver_manager
import adbc_driver_flightsql.dbapi as flight_sql

>>> print(adbc_driver_manager.__version__)
1.1.0
>>> print(adbc_driver_flightsql.__version__)
1.1.0

02 Connect to Doris​

Create a client for interacting with the Doris Arrow Flight SQL service. Prerequisites include: Doris frontend (FE) host, Arrow Flight port, and login username/password.

Configure parameters for Doris frontend (FE) and backend (BE):

• In fe/conf/fe.conf, set arrow_flight_sql_port to an available port, such as 8070.

• In be/conf/be.conf, set arrow_flight_sql_port to an available port, such as 8050.

Note: The arrow_flight_sql_port port number configured in fe.conf and be.conf is different

After modifying the configuration and restarting the cluster, searching for Arrow Flight SQL service is started in the fe/log/fe.log file indicates that the Arrow Flight Server of FE has been successfully started; searching for Arrow Flight Service bind to host in the be/log/be.INFO file indicates that the Arrow Flight Server of BE has been successfully started.

Suppose that the Arrow Flight SQL services for the Doris instance will run on ports 8070 and 8050 for FE and BE respectively, and the Doris username/password is "user" and "pass", the connection process would be:

conn = flight_sql.connect(uri="grpc://{FE_HOST}:{fe.conf:arrow_flight_sql_port}", db_kwargs={
            adbc_driver_manager.DatabaseOptions.USERNAME.value: "user",
            adbc_driver_manager.DatabaseOptions.PASSWORD.value: "pass",
        })
cursor = conn.cursor()

Once the connection is established, you can interact with Doris using SQL statements through the returned cursor object. This allows you to perform various operations such as table creation, metadata retrieval, data import, and query execution.

03 Create table and retrieve metadata​

Pass the query to the cursor.execute() function, which creates tables and retrieves metadata.

cursor.execute("DROP DATABASE IF EXISTS arrow_flight_sql FORCE;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("create database arrow_flight_sql;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("show databases;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("use arrow_flight_sql;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("""CREATE TABLE arrow_flight_sql_test
    (
         k0 INT,
         k1 DOUBLE,
         K2 varchar(32) NULL DEFAULT "" COMMENT "",
         k3 DECIMAL(27,9) DEFAULT "0",
         k4 BIGINT NULL DEFAULT '10',
         k5 DATE,
    )
    DISTRIBUTED BY HASH(k5) BUCKETS 5
    PROPERTIES("replication_num" = "1");""")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("show create table arrow_flight_sql_test;")
print(cursor.fetchallarrow().to_pandas())

If the returned StatusResult is 0, that means the query is executed successfully. (Such design is to ensure compatibility with JDBC.)

  StatusResult
0            0

  StatusResult
0            0

                   Database
0         __internal_schema
1          arrow_flight_sql
..                      ...
507             udf_auth_db

[508 rows x 1 columns]

  StatusResult
0            0

  StatusResult
0            0
                   Table                                       Create Table
0  arrow_flight_sql_test  CREATE TABLE `arrow_flight_sql_test` (\n  `k0`...

04 Ingest data​

Execute an INSERT INTO statement to load test data into the table created:

cursor.execute("""INSERT INTO arrow_flight_sql_test VALUES
        ('0', 0.1, "ID", 0.0001, 9999999999, '2023-10-21'),
        ('1', 0.20, "ID_1", 1.00000001, 0, '2023-10-21'),
        ('2', 3.4, "ID_1", 3.1, 123456, '2023-10-22'),
        ('3', 4, "ID", 4, 4, '2023-10-22'),
        ('4', 122345.54321, "ID", 122345.54321, 5, '2023-10-22');""")
print(cursor.fetchallarrow().to_pandas())

If you see the following returned result, the data ingestion is successful.

  StatusResult
0            0

If the data size to ingest is huge, you can apply the Stream Load method using pydoris.

05 Execute queries​

Perform queries on the above table, such as aggregation, sorting, and session variable setting.

cursor.execute("select * from arrow_flight_sql_test order by k0;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("set exec_mem_limit=2000;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("show variables like \"%exec_mem_limit%\";")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("select k5, sum(k1), count(1), avg(k3) from arrow_flight_sql_test group by k5;")
print(cursor.fetchallarrow().to_pandas())

The results are as follows:

   k0            k1    K2                k3          k4          k5
0   0       0.10000    ID       0.000100000  9999999999  2023-10-21
1   1       0.20000  ID_1       1.000000010           0  2023-10-21
2   2       3.40000  ID_1       3.100000000      123456  2023-10-22
3   3       4.00000    ID       4.000000000           4  2023-10-22
4   4  122345.54321    ID  122345.543210000           5  2023-10-22

[5 rows x 6 columns]

  StatusResult
0            0

    Variable_name Value Default_Value Changed
0  exec_mem_limit  2000    2147483648       1

           k5  Nullable(Float64)_1  Int64_2 Nullable(Decimal(38, 9))_3
0  2023-10-22         122352.94321        3            40784.214403333
1  2023-10-21              0.30000        2                0.500050005

[2 rows x 5 columns]

06 Complete code​

# Doris Arrow Flight SQL Test

# step 1, library is released on PyPI and can be easily installed.
# pip install adbc_driver_manager
# pip install adbc_driver_flightsql
import adbc_driver_manager
import adbc_driver_flightsql.dbapi as flight_sql

# step 2, create a client that interacts with the Doris Arrow Flight SQL service.
# Modify arrow_flight_sql_port in fe/conf/fe.conf to an available port, such as 8070.
# Modify arrow_flight_sql_port in be/conf/be.conf to an available port, such as 8050.
conn = flight_sql.connect(uri="grpc://{FE_HOST}:{fe.conf:arrow_flight_sql_port}", db_kwargs={
            adbc_driver_manager.DatabaseOptions.USERNAME.value: "root",
            adbc_driver_manager.DatabaseOptions.PASSWORD.value: "",
        })
cursor = conn.cursor()

# interacting with Doris via SQL using Cursor
def execute(sql):
    print("\n### execute query: ###\n " + sql)
    cursor.execute(sql)
    print("### result: ###")
    print(cursor.fetchallarrow().to_pandas())

# step3, execute DDL statements, create database/table, show stmt.
execute("DROP DATABASE IF EXISTS arrow_flight_sql FORCE;")
execute("show databases;")
execute("create database arrow_flight_sql;")
execute("show databases;")
execute("use arrow_flight_sql;")
execute("""CREATE TABLE arrow_flight_sql_test
    (
         k0 INT,
         k1 DOUBLE,
         K2 varchar(32) NULL DEFAULT "" COMMENT "",
         k3 DECIMAL(27,9) DEFAULT "0",
         k4 BIGINT NULL DEFAULT '10',
         k5 DATE,
    )
    DISTRIBUTED BY HASH(k5) BUCKETS 5
    PROPERTIES("replication_num" = "1");""")
execute("show create table arrow_flight_sql_test;")


# step4, insert into
execute("""INSERT INTO arrow_flight_sql_test VALUES
        ('0', 0.1, "ID", 0.0001, 9999999999, '2023-10-21'),
        ('1', 0.20, "ID_1", 1.00000001, 0, '2023-10-21'),
        ('2', 3.4, "ID_1", 3.1, 123456, '2023-10-22'),
        ('3', 4, "ID", 4, 4, '2023-10-22'),
        ('4', 122345.54321, "ID", 122345.54321, 5, '2023-10-22');""")


# step5, execute queries, aggregation, sort, set session variable
execute("select * from arrow_flight_sql_test order by k0;")
execute("set exec_mem_limit=2000;")
execute("show variables like \"%exec_mem_limit%\";")
execute("select k5, sum(k1), count(1), avg(k3) from arrow_flight_sql_test group by k5;")

# step6, close cursor 
cursor.close()