Archive Data

Going into 2024, cost savings was one of the major goals for the storage team. Our AWS MySQL RDS storage footprint was rapidly growing, with petabytes of stored data. Moreover, some of our largest databases in our fleet were approaching the RDS maximum disk space of 64TB, which would render the databases inoperational. We noticed that our largest dataset in MySQL was our “cell history” and “action log” tables (which we will collectively call “archive data”), basically an audit log for the activity within a base. This is used to service our revision history features, and also often for internal debugging purposes. For some enterprise customers, we have committed to retain this data for up to 10 years. We wanted to stop storing this data in MySQL to save money, but we still needed to provide interactive level query latency and the same level of durability and availability guarantees as MySQL.

These are some characteristics of our archive data:

• The vast majority (petabytes, trillions of rows) is old data that is relatively infrequently accessed by application code.
• When it is accessed, the majority of the QPS comes from query patterns that are point selects or range queries used in pagination. The queries are also always filtered by a specific base.
• Old data is read-only, with the exception of when the data is hard deleted.
• The data is primary key’d by MySQL’s autoincr_id, meaning that it is essentially sorted in insertion order.

MySQL provides easy query access to this data, but is expensive and thus a poor fit for such cold traffic patterns. We formulated an idea to archive and migrate the data out of MySQL into AWS S3 (which is 10x cheaper byte for byte), partition the data by base into individual Apache Parquet files, and build a new query engine using Apache DataFusion to serve S3 Parquet files. Recent cell history and action log rows would continue to write to and serve from MySQL, making this a two-tier storage system.

Parquet

First of all, a quick overview of Parquet. It is a columnar file format designed for analytical workloads. Instead of storing data row-by-row like MySQL InnoDB, Parquet stores each column contiguously on disk. Furthermore, Parquet files are organized by row groups, each of which contains a horizontal partition of the dataset. A row group contains a chunk of rows for every column, and each column is stored as a separate column chunk.

The Parquet file format also contains metadata useful for query planning in an engine. File metadata provides offsets and size information useful for navigating the Parquet file and page header metadata is stored in-line with the page data, and contains useful statistics such as min/max values and bloom filters of the column chunks. Modern query engines that support querying Parquet are able to use the file format’s metadata to construct its execution plan and avoid scanning and filtering over many row groups entirely.

This is what a simplified visual representation of the type of data Parquet files have:

Parquet File
└── RowGroup[0]
│ ├── ColumnChunk[colA]
│ │ ├── ColumnIndex (min/max stats)
│ │ ├── BloomFilterIndex
│ │ └── Page data
│ └── ColumnChunk[colB]
…
└── File Metadata

In constructing our Parquet files, we did not significantly change the schema from its original MySQL schema. We kept them sorted by autoincr_id because most of our query patterns are point selects or range queries on autoincr_id (and some other additional conditions). This ordering allows our query engines to effectively consult metadata and selectively download the relevant byte range of data from S3. We also partitioned our Parquet files by base so that our queries (which are all per base) avoid scanning unrelated base data.

Lastly, we decided to register the S3 locations of these Parquet files in DynamoDB as our own additional layer of metadata. This allows our storage client code to easily register in the query engine which files it needs to query over. It was also important in supporting features like data residency and encrypting data by customer keys.

As a side node, thanks to the columnar file format, we were able to get significantly higher compression ratios — our final archive data set was also 10 times smaller than the original data set in MySQL. Coupled with S3 itself being 10 times cheaper byte for byte, we were actually able to build a system that was 100 times cheaper than MySQL in storage costs!

DataFusion

In early 2024, we began the project by first benchmarking a variety of query engines over Parquet files. All these engines speak SQL and are able to query Parquet files from S3.

• AWS Athena
• DuckDb
• Starrocks
• DataFusion

One of the first engines we tried was AWS Athena. However, Athena’s architecture is more suitable for general OLAP workloads, and we it didn’t meet our latency needs to provide an interactive, user-facing feature. The Athena API expects queries to be made with a StartQueryExecution call followed by a GetQueryExecution poll to retrieve the query result — as a result, we generally had seconds of latency. Also, as a managed service by AWS, there’s no isolation between separate bases, something we like to prioritize here at Airtable.

With DuckDB, we found that the query planner did not always effectively use projection pushdowns, a method to reduce the number of data scanned by first moving filters into the initial data scan. Some of our query patterns resulted in entire files being downloaded. Simple point select query on a single autoincr_id worked as expected and achieved subsecond latency, but we found it generally worse compared to DataFusion. In general, we believe the tool is more appropriate for ad-hoc analytics or if you just want to query Parquet fast. In fact, we leveraged DuckDB frequently for debugging purposes throughout our development process — it was extremely convenient to be able to run a CLI tool to quickly inspect contents of S3 Parquet files when working through validation issues.

Here’s a quick table of some benchmarked results on one select internal Airtable base. By no means is this representative of all query patterns for these various query engines. It is simply a small subset of the queries that we have.

With Starrocks, we found solid performance results comparable to DataFusion, but the operational complexity involved with running a cluster full time in k8s to serve relatively low QPS queries for cold storage put us off. Like Athena, it also lacks strong isolation between bases.

After these investigations, we settled on DataFusion. It is an extensible query engine written in Rust, and we found it to be the best engine at using Parquet’s advanced features to implement queries efficiently. Its extensible nature also proved to be handy as we made query optimizations, which we will discuss more later. As an embedded library, we were also able to embed it into our worker processes that are already per base, which has a number of advantages:

• Low operational overhead: Since the engine is embedded, there is no additional service to manage in production. Local development, CI, etc. also did not require much additional setup.
• Strong isolation between bases: Again, since the engine is embedded, our existing architecture for per base processes provides us this guarantee. We did not need to introduce any new mechanism to prevent bases from contending with each other for CPU, RAM, or network bandwidth.
• Strong affinity with requests by base: This allows us to implement effective caching mechanisms, which we will discuss later.

Data Migration

After we settled on our choice of cost efficient storage and a query engine, we had to migrate the data out of MySQL. We prioritized a process to do a one-time migration of data out of MySQL into S3 to immediately start saving money.

In order to get a consistent view of the data for export, we chose to leverage AWS RDS’s snapshotting capabilities, which returns large Parquet files of the entire table. We also prototyped a system to run SQL directly on the databases and write Parquet files, but we did not productionize orchestrating such a system at scale. We preferred AWS’s snapshot capabilities because it runs on the backup database instance and does not incur additional load to our production systems.

However, these are snapshots of massive tables across a host of database shards. All our queries are filtered by specific bases, so we also decided that our Parquet files should be partitioned by base. In order to construct these partitioned serving Parquet files, we spun up a number of Flink jobs that parallelized over all our database shard snapshots, repartitioned these snapshots by base, and dumped them in some intermediate S3 directories. Then, we used AWS step functions to scan these S3 directories and enqueue the bases into AWS SQS. Lastly, from there, we ran some custom “compactor” code that merged these intermediate files together. In this compaction process, we merge-sorted the various files, deduplicated records, and made sure each individual final Parquet file did not exceed 1GB. This was an appropriate serving size we determined during initial benchmarking that had a good density of page groups per file.

Validation

It was an important priority for us to provide a seamless transition for our end users throughout this migration process, so before we launched this new system to serve live traffic, we had to do some validation first to ensure we didn’t corrupt data during the migration or introduce unexpected bugs.

We previously wrote about our bulk validation process to guard against data corruption in more depth in this blog post. TL;DR, we spun up a Starrocks cluster so that we could quickly run validation queries between the serving Parquet files and the RDS snapshots. Fortunately, we found zero cases of data corruption throughout this process — the repartitioner and compactor code had worked flawlessly.

However, this project introduced a lot of new storage client code. To give an idea of the new complexity here, we

• Wrote a query engine with DataFusion in Rust.
• Integrated the Rust query engine into our Node.js client code with napi-rs.
• Wrote new client code with logic to combine MySQL and S3 results, identify which S3 files to query, etc.
• Supported existing enterprise features such as encryption with customer provided keys, regional data residency, hard deletion, etc.

Bulk validation was a necessary test to ensure our data migration processes did not corrupt data, but it did not validate all the other client code and demonstrate that users see the exact same revision history in their bases before and after the migration.

Once we had the query engine built, we began to perform shadow validation on live traffic. Every request would read from MySQL like normal, but in the background, also issue the same request via the new query engine. We caught a number of simple implementation bugs, and also saw a variety of interesting issues, such as:

• Mismatched float precision between Javascript (our typical client) and Rust’s serde JSON library.
• An interesting case where it looks like we dropped entire database shards of data! But it turned out that DataFusion was unexpectedly doing a lexicographical instead of numerical sort.
• A crashing SIGABRT issue with async napi-rs and Node.js worker threads.
• Latency performance concerns.

In the end, we were able to resolve all these bugs prior to launching this to users and deleting the data in MySQL.

Performance Optimizations

During our staged rollout process, we discovered a number of performance issues around latency. They were caused by a variety of issues, ranging from inefficient query plans, bottlenecks in network requests to S3, or sparse filters resulting in more data downloaded than necessary. We’ll discuss some of the interesting improvements made here.

Caching

One particularly interesting performance optimization we made was around caching. As one can imagine building a query engine off S3, the bottlenecks are primarily in the network roundtrips to S3. DataFusion basically converts our SQL statement into a series of S3 GET statements. It fetches Parquet footer metadata, Parquet column chunk metadata, and then uses this information to decide which page groups and which column chunks of the Parquet files actually need to be fetched. We built a tiered caching system to reduce the number of S3 GET requests needed.

For the first layer, we were able to easily use DataFusion’s built-in cache infrastructure (CacheManagerConfig) to cache Parquet file metadata and S3 ListObjects calls.

Next, we cached the rest of the Parquet page header metadata. Together with the built-in file metadata cache, we effectively reduced the constant need to roundtrip to S3 during the query planning process. With pushdown filtering, DataFusion only had to consult the cached metadata to get a good idea of which page group byte ranges it needed to scan. Being an extensible query engine, DataFusion makes it simple to write your own cache implementation. Typically, DataFusion provides a default parquet reader interface (which basically implements functions like get_metadata, get_bytes, get_byte_ranges), but it also allows you to substitute your own implementation instead. So, we wrote an implementation that cached metadata results in-memory. It was also straightforward to add observability and other instrumentation here, and we were able to confirm that we typically have a 99%+ cache hit ratio here. This was as expected because of how our DataFusion engine is embedded in a per-base process and how Parquet files are similarly partitioned by base — the request affinity is as good as it gets here.

Lastly, we also implemented an on-disk cache that preemptively downloads the Parquet files for a base. Similar to the previous cache layer, we were able to implement this by substituting the default provided S3 ObjectStore implementation with a custom implementation that downloaded to and read from the local disk. Caching entire Parquet files on disk is more work and cost than just metadata though — fortunately, this was only needed for an extremely small number of bases with poorer performance due to large amounts of data and pathological query patterns.

Overall, we found DataFusion’s extensible nature to be easy to work with, flexible, and the ideal tool to build high performance query engines.

Custom Indexes

We previously highlighted that most of our queries are point selects and range queries on autoincr_id, but we also often have additional conditions that filter rows far more efficiently than just autoincr_id. Some examples include queries that

• filter by the action type, e.g. looking for actions that updated Airtable column names.
• filter by the row being updated
• omit updates from our sync feature

Some of these additional filters resulted in matching on a tiny fraction of rows, which makes fetching and scanning entire Parquet files wasteful. For example, imagine filtering out sync updates, but having a base that was predominantly made of sync actions — such naive queries would naturally be slow.

No database system with varying query patterns is complete without secondary indexes. In order to make these queries more efficient, we built an indexing system leveraging DataFusion that scans through Parquet files and writes indexes as new Parquet files. Our DataFusion and surrounding client code is aware of these index files, and queries them first in order to generate a more efficient query on the original Parquet files. It was easy to build this ad-hoc indexing system mainly because of how our data is read-only. We never have to worry about our data changing and also having to update indexes in sync, which more sophisticated database systems typically have to do.

Bloom Filters

As we previously mentioned, most of our query patterns are similar to point selects or range queries on an autoincr_id, the “primary key” of how our Parquet files are laid out. However, we do have some significantly lower QPS queries that were point selects on a different, unique identifier. This unique identifier was randomly distributed, so the min/max statistics on Parquet were useless — we’d end up having to fetch and scan every single page group and then apply the filter afterwards. We could address this with the same custom index strategy as above, but we found it simpler to just rely on Parquet’s bloom filter metadata, which DataFusion understands.

Bloom filters are a space efficient probabilistic data structure that tests whether an element is a member of the set, with false positives being possible but false negatives impossible. In this case, each column chunk metadata for the identifier contained the bloom filter. Because false negatives are impossible, they can be used to filter out some page groups that definitely don’t need to be fetched.

Conclusion

Overall, we were able to move petabytes of data out of MySQL, build a system that was 100x cheaper in storage costs, and save millions of dollars per year, all while maintaining interactive query latencies for our users.

For future work, we have plans to make this system incrementally archive data out of MySQL. We prioritized a manual batch archiving process to migrate petabytes of data out of MySQL and start saving money immediately, but there’s still a lot of engineering work to be done to make this a fully automatic system with less operational burden. We envision setting up a CDC system like Flink to handle this, but there are going to be a lot more unsolved and interesting problems around how we handle compacting Parquet files together, rebuilding indexes, managing the operational side of things, etc. Also, our initial implementation targeted our largest dataset, but there are other log-like tables we could onboard to this system as well.

If this type of work optimizing database query engines and working with petabytes of data sounds exciting to you, apply to Airtable! We’re hiring at https://airtable.com/careers.

Thanks to all past and present Airtablets on storage and across the organization who contributed to this project: Nathan Chou, Aiden Dou, Riley Hockett, Matthew Jin, Daniel Kozlowski, Brian Larson, Keunwoo Lee, Mike Milkin, Gavin Towey, Andrew Wang, Xiaobing Xia, Alex Yao, Brian Zhang, Kun Zhou

Apache, Apache Parquet, and Apache DataFusion are trademarks of The Apache Software Foundation.

How we reduced archive storage costs by 100x and saved millions was originally published in The Airtable Engineering Blog on Medium, where people are continuing the conversation by highlighting and responding to this story.

Overview

At Airtable, we store our application or “base-scoped” data on a number of sharded MySQL instances in Amazon’s Relational Database Service (RDS). Each Airtable base is associated with a single one of these sharded instances, and as the base and the data in the base changes, we store some append-only data associated with the history of the base. This data powers features such as undo and revision history, allowing us to display record revisions as far back as 3 years ago for our enterprise customers. As we have grown as a business, these append-only tables have become increasingly large and their content now represents close to half of all the data that we store at Airtable. Additionally, much of this data is very infrequently accessed, but is stored in the same storage layer as all of our base-scoped data for every day product use making it expensive.

The Project

The Live Shard Data Archive (LSDA) project allowed us to shrink the disc volumes of the bulk of our RDS instances by taking this infrequently accessed, append-only data and storing it in a cheaper storage solution, S3. Once it was stored in that cheaper solution, we were able to drop the old data, and rebuild the RDS instances to reclaim the space.

Moving this data to S3 required three major steps. First, we had to archive and transform the data from RDS into S3 such that it could be accessed by our codebase in an efficient and consistent way. Second, we had to validate that this archived data matched the existing source data in RDS, and that our process of archiving did not cause any inconsistencies between the two datasets. Finally, we had to make application code changes to serve this data from S3. After these steps, we were able to truncate the data from RDS that we were serving from S3, allowing us to shrink the allocated storage space of these instances, saving a substantial amount of our overall RDS bill. This blog post will focus on the first phase of the second step of that process, data validation.

Archiving and Transformation

The data archival from RDS into its final shape in S3 was a three-step process. First, we exported snapshots of our databases into S3 as parquets. This is a built-in RDS feature. To optimize query latency of the archive, we then repartitioned the snapshot exports by our customers. This was a challenging process due to the amount of data (>1PB) and the number of files involved (>10M). We used Apache Flink to incrementally ingest these files and repartitioned them into per customer partitions. Finally, we ran a highly-concurrent Kubernetes rewriter job that sorted the archive for each customer, and added the proper index and bloom filters to the rewritten parquets to speed up the most common query patterns.

Validation overview and considered approaches

Validating the data required us to do a row by row comparison from the archive to our source data and make sure that for every row, these values were equal. Naively, an easy way to do this would be to read a row from our archive, find that row in RDS, and confirm they are the same. However, we were dealing with almost 1PB of data and close to 2 trillion rows. Additionally, our RDS instances in production serve customer traffic, so saddling them with these additional requests was not really an option, especially at the volume we would require to validate our entire archive. As a result, we decided to use the original, unmodified RDS export as our source data for this validation project. This data was stored in S3, and while that alleviated the problem of querying serving instances, it would simply be too slow to go row by row and validate this data. Ultimately, we decided that if we had all of the data in some relational database, we could just join the two tables together, and find any discrepancies in the data that way.

Leveraging StarRocks for Data Validation — Airtable Data Infrastructure Team

For the data validation project, the Data Infrastructure team helped the Storage team in selecting the best tool to complete the validation work efficiently.

Why Use StarRocks to Address the Data Validation Problem?

The core of the data validation problem involves performing a large number of join operations between two massive datasets, each containing nearly a trillion rows of data. The primary challenge was executing these computationally intensive join operations efficiently.

After thorough investigation, StarRocks was chosen due to its exceptional join performance. It can handle these operations with affordable computational costs, whereas other query engines struggle significantly with the same workload.

To address the problem, we decided to load raw Parquet files from S3 into local tables in StarRocks. By leveraging StarRocks’ colocation mechanism, we could efficiently perform the join operations required for data validation.

StarRocks Architecture

The diagram above illustrates the StarRocks architecture, which can access the following data sources:

1. Data Lakes on S3: Includes Hudi Lake, Delta Lake, Iceberg Lake, and Paimon Lake.
2. Native Format on S3: StarRocks allows the creation of tables that persist data directly in S3 using its native format.
3. Raw Parquet, JSON, or CSV Files on S3: Queries can be executed directly on raw Parquet, JSON, or CSV files stored in S3.

In our specific scenario, we loaded raw Parquet files from S3 storage into StarRocks’ local tables to perform data validation, as highlighted above.

Ingestion Optimization: Enhances data ingestion performance in StarRocks

We had to load nearly 1 trillion rows of data from raw Parquet files into StarRocks local tables. The dataset consisted of hundreds of millions of small Parquet files. Without proper optimization and parallelization, ingesting the entire dataset would have taken several months.

To accelerate the ingestion throughput, we implemented the following optimizations:

1. Reduce the Number of Replicas (from 3 to 1):
   Since this is a one-time validation task, maintaining high availability for production is unnecessary. Reducing the number of replicas significantly decreases the total data volume to be ingested.
2. Increase Internal Ingestion Parallelism:
   As the validation process involves ingestion first, followed by join-based validation, ingestion performance does not affect serving scenarios. We increased parallelism by tuning the following parameters:
   - pipeline_dop
   - pipeline_sink_dop
3. Increase the Number of Buckets per Partition:
   Given the large data volume, we ensured that each bucket contained no more than 5GB of data. Increasing the number of buckets per partition significantly improves ingestion throughput. Although this may cause compaction to lag behind, it is not a concern in our specific scenario.

These optimizations collectively help to efficiently handle the massive data ingestion process required for our validation workload.

Ingestion

Export

Once StarRocks was set up, we needed to get all of the data ingested from both our RDS export and the transformed archive which we planned on serving from. We decided that it was not cost and time efficient to store all ~1PB in StarRocks, so we decided to just hash all of the non-key columns in the table. We ingested two tables as a part of this process, but our examples will focus primarily on just one, _actionLog.

Initial solution: Simple table with hashed non-key columns

We started with a table schema for each of our tables which looked like this:

CREATE TABLE `_rdsExportActionLog` (
`id` bigint(20) NOT NULL COMMENT "",
`application` varchar(65533) NOT NULL COMMENT "",
`hash_value` varchar(65533) NULL COMMENT ""
) ENGINE=OLAP
PRIMARY KEY(`id`, `application`)
DISTRIBUTED BY HASH(`id`, `application`)
ORDER BY(`application`)
PROPERTIES (
"replication_num" = "1",
"colocate_with" = "action_log_group",
"in_memory" = "false",
"enable_persistent_index" = "true",
"replicated_storage" = "true",
"compression" = "ZSTD"
);

And we started to load the data using insert statements like this

INSERT INTO \`exportActionLog\`
WITH LABEL ${label}
(id, application, hash_value)
SELECT id, application, XX_HASH3_64(CONCAT_WS(',',
<columns>)) as hash_value
FROM FILES(
"path" = "s3://${bucket}/${folder}*.parquet",
"format" = "parquet",
);

Data distribution and Loading Bottlenecks

However, we found that this loading operation took a really long time, on the order of almost 1 day to load two of our shards. We found that as we added more data and the table got bigger, our ingestion rate slowed further. What was also odd is that if we increased the local parallelization of this ingestion, i.e. ingested multiple shards at once, we didn’t see almost any performance boost, and when we set this value to be more than five, we saw a lot of this:

JobId: 14094
Label: insert_16604b11–7f2d-11ef-888c-46341e0f370e
State: LOADING
Progress: ETL:100%; LOAD:99%
Type: INSERT

You can see here that our load value is 99%, but a number of these large loads would just get stuck at this value and not be able to progress past this state quickly despite getting to 99% quickly. Per the StarRocks documentation, “When all data is loaded into StarRocks, 99% is returned for the LOAD parameter. Then, loaded data starts taking effect in StarRocks. After the data takes effect, 100% is returned for the LOAD parameter.” Evidently we were experiencing some bottlenecks on the data taking effect with our initial solution.

Improvement 1: Increase bucket count

We were distributing our data by id and application, but we had yet to specify the number of buckets to distribute this data into (more info on StarRocks data distribution). Our hypothesis was that as we stored more and more data, these buckets got increasingly larger and more cumbersome which led us to the slow down that we were seeing. We consulted the StarRocks team who suggested that for our data volume, we should look at specifying somewhere on the order of 7200 buckets for the smaller of the two tables, so we changed our schema to look like this:

CREATE TABLE `exportActionLog` (
`id` bigint(20) NOT NULL COMMENT "",
`application` varchar(65533) NOT NULL COMMENT "",
`hash_value` varchar(65533) NULL COMMENT ""
) ENGINE=OLAP
PRIMARY KEY(`id`, `application`)
DISTRIBUTED BY HASH(`id`, `application`) BUCKETS 7200
ORDER BY (`application`)
PROPERTIES (
"replication_num" = "1",
"colocate_with" = "action_log_group",
"in_memory" = "false",
"enable_persistent_index" = "true",
"replicated_storage" = "true",
"compression" = "ZSTD"
);

However, while we were able to load this data much more quickly than we were previously, we ran into this memory issue:

message: 'primary key memory usage exceeds the limit. 
tablet_id: 10367, consumption: 126428346066, limit: 125241246351. 
Memory stats of top five tablets: 53331(73M)53763(73M)53715(73M)53667(73M)53619(73M): 

Improvement 2: Partition the table by shard ID

We realized that it would make sense to just partition the table by the shardId and try to load it that way. This would allow us to specify a number of buckets for each partition, and they would be stored in a more efficient manner. Using some rough math, we found that:

actionLog => 10TB (Hashed) => 10 * 1024 / 148 shards = 69GB per shard => 
34 buckets to host it => add some buffer, 64 buckets per partition

In total: 64 buckets per partition * 148 shards = 9472 buckets

We figured that using this distribution could also allow us to validate shard by shard which would help to not overwhelm the memory of the cluster. In the end we created this table and adjusted our loading statement to pull the shard ID from the S3 file path.

CREATE TABLE `exportActionLog` (
`id` bigint(20) NOT NULL COMMENT "",
`application` varchar(65533) NOT NULL COMMENT "",
`shard` int(11) NOT NULL COMMENT "",
`hash_value` varchar(65533) NULL COMMENT ""
) ENGINE=OLAP
PRIMARY KEY(`id`, `application`, `shard`)
PARTITION BY (`shard`)
DISTRIBUTED BY HASH(`id`, `application`) BUCKETS 64
ORDER BY(`application`)
PROPERTIES (
"replication_num" = "1",
"colocate_with" = "action_log_group_partition_by_shard",
"in_memory" = "false",
"enable_persistent_index" = "true",
"replicated_storage" = "true",
"compression" = "ZSTD"
);

This drastically sped up our ingestion time, LSDA data was successfully loaded into StarRocks from S3 in under 10 hours. The average throughput was approximately 2 billion rows per minute.

Archive

Now that the full RDS export, our source of truth data for validation, had been ingested into StarRocks, we needed to ingest the archive we were going to serve the data from and run our validation process across the two datasets. Unfortunately, the archive data was not stored in the same format as the export data which presented an additional set of challenges. Drawing from our experience with the export ingestion, we decided to create the tables in an identical way, with both the same number of buckets and the same partitioning strategy by shardId. This also allowed us to have the export and the archive in the same colocation group, such that we could use StarRocks’ colocation join.

CREATE TABLE `_rdsArchiveActionLog` (
`autoincr_id` bigint(20) NOT NULL COMMENT "",
`applicationId` varchar(65533) NOT NULL COMMENT "",
`shardId` int(11) NOT NULL COMMENT "",
`hash_value` varchar(65533) NULL COMMENT ""
) ENGINE=OLAP
PRIMARY KEY(`autoincr_id`, `applicationId`, `shardId`)
PARTITION BY (`shardId`)
DISTRIBUTED BY HASH(`autoincr_id`, `applicationId`) BUCKETS 64
ORDER BY(`applicationId`)
PROPERTIES (
"replication_num" = "1",
"colocate_with" = "action_log_group_partition_by_shard",
"in_memory" = "false",
"enable_persistent_index" = "true",
"replicated_storage" = "true",
"compression" = "ZSTD"
);

Discrepancies in directory structure between RDS export and archive

However, a major difference between our export and our archive was the way in which they were stored in S3. For our export, it was stored in large directories, each corresponding to an individual shard and partition in StarRocks. This made our insert queries to StarRocks quite simple; we could just wildcard all of the parquet files across these directories, and from the directory we were able to get the shard ID which we could then insert into the row and use for partitioning.

For the archive, our data was stored by application to make serving from S3 in the application simpler. This meant that we had over 6 million small directories corresponding to each application that did not necessarily correspond to a given shard, some applications can be stored across multiple shards. We also had not stored the source shard information in S3, so unlike with our export, we had no ability to easily get our shardId from our single call to S3. Instead, we had stored yet to be validated file metadata in DynamoDb. Given this information, we created a Global Secondary Index in DynamoDb with a sort key as the shard ID, and then queried all of the files for that shardId to insert those.

Grouping inserts in StarRocks using Union

Additionally, StarRocks only lets you specify a single path for each insert statement, meaning that each file would need its own insert statement. This was in stark contrast to our export which only had about ~160 folders and corresponding insert statements. We now had more than 6 million insert statements for our archive. To solve this problem, we figured we could just try to heavily parallelize the process. We thought that we could just run multiple shards at once and run multiple processes per shard. Note, this is all through Node and TS, so the idea of multithreading for this is not really true, but we could run multiple processes which would run if others were blocked on I/O. We tried to run a single shard with 10 threads, but hit this problem:

message: 'Failed to load data into tablet 14775287, 
because of too many versions, current/limit: 1006/1000. 
You can reduce the loading job concurrency, or increase loading data batch size. 
If you are loading data with Routine Load, you can increase 
FE configs routine_load_task_consume_second and max_routine_load_batch_size

Essentially, because multiple processes were running insertions at the same time, we were creating too many versions of the table for StarRocks to compact based on the compaction frequency. Additionally, you can no longer increase the version limit in StarRocks beyond 1000.

Given this, we tried to pursue a strategy which would let us reduce the number of insert queries to try to hand off more of the work to StarRocks (at this point it was running at low CPU and memory and the process was taking a while/failing). To do this, given that we can only specify a single path per select statement, we just created a for loop to generate an insert statement like this

INSERT INTO <table>
SELECT * FROM FILE_1
UNION ALL
SELECT * FROM FILE_2
……

With this strategy, we were actually able to run this for 100 applications at once, speeding this up significantly. This allowed us to load all ~1 trillion rows on the archive side from 6 millions applications in just about 3 days.

Acknowledgements

Thank you to Daniel Kozlowski, Kun Zhou, Matthew Jin and Xiaobing Xia for all of their contributions on this project.

Live Shard Data Archive: Export and Ingestion to StarRocks for Validation was originally published in The Airtable Engineering Blog on Medium, where people are continuing the conversation by highlighting and responding to this story.

Imagine a marketing team asking, “Can you find past campaigns similar to this one?”, a product management team asking “can you find engineers whose expertise matches this project?”, or in my case “can you find past issues (called “escalations” internally) similar to this one?”. Embedding powered systems can allow teams to quickly identify insights that would normally take hours of manual labor.

Thinking About Building An Embedding Powered System:

Embeddings are at the heart of semantic search. Embeddings are vectors, an array of numbers, numerical representations of data, that capture meaning in a way that machines can understand. Concretely, an embedding can tell you that “cat” is similar to “tabby”, but not similar to “car”.

Building a system around embeddings isn’t trivial. There are multiple dimensions to consider:

• (Focus of this post) Embedding Lifecycle Management
  - Triggering — As data changes embeddings need to be (re)generated. We need to durably track the state of generation.
  - Generation — Generating the embeddings via API calls (over the network, or in-process)
  - Persistence — Where and how are the embedding stored
  - Deletion — Deleting old data when it’s no longer needed
  - Consistency — How do we maintain consistency between our vector data and primary database?
  - Migrations — This has several flavors e.g. Changes to the DB schema, changes to the DB engine (e.g. LanceDB vs Opensearch), Changes to the AI embedding model, or changes to the underlying storage layer’s encryption
  - Disaster Recovery — What happens if we experience data loss?-
• Data Preparation
  - Corpus Choice — What data will we embed?
  - Configuration — Embedding model choice, chunking strategy, other knobs
• Query And Access
  - Querying — Via semantic search or direct access
  - Indexing — Data structures for efficient semantic similarity queries
  - Permissions — Making semantic search respect our permissions model
• Operational Concerns
  - Security — Embeddings are sensitive customer data and must be treated accordingly
  - Cost — How do we keep the cost of generation, persistence, and querying manageable?
  - Partition Management — If you have many collections, how do you handle scaling the number of partitions?

Given all that, we needed a system that was not just functional but robust and adaptable.

A High-Level Overview of Airtable’s Architecture

Airtable offers an app building platform built on top of a custom, in-memory database backed by MySql.

Let’s introduce two useful terms we’ll be using

1. MemApp — Our in memory database, manages all reads and writes for a base.
   - The data is ultimately persisted in MySql
2. Base — A particular instance of MemApp

A critical detail: MemApp is a single-writer database. All writes occur serially, which becomes important when we talk about maintaining consistency.

Building A Data Model

Fully Or Eventually Consistent?

Our first fundamental design question was this: Should embeddings be stored within MemApp to ensure full consistency with the data? While appealing in theory, this approach had two major drawbacks:

1. Cost: Memory usage is a significant factor in Airtable’s expenses. Storing embeddings in MemApp would be prohibitively expensive because embeddings are often 10x the size of the underlying data.

2. Performance: Achieving strong consistency would require generating embeddings within transactions. This process would be too slow for bulk updates and would limit us to less capable, in-house models rather than leveraging top-tier providers like OpenAI.

The solution? Embrace eventual consistency, and store and process embeddings outside of MemApp. This decision led us to design a data flow where embeddings are generated asynchronously and stored in a separate vector database — with their state tracked within MemApp.

Data Modeling

We anticipated that AI models, embedding providers, and storage engines would evolve. To future-proof our system, we introduced an abstraction within MemApp called an embedding config. This allows developers to map data from MemApp to a table in a vector database, without worrying about the underlying complexities.

An embedding config has the following primary components:

1. data subscription — A declarative description of the data a user has told us they want to embed.
2. embedding strategy — How the data is to be embedded
3. storage configuration — Where the data will be stored.
4. triggering configuration — Once data is out of date, when do we re-generate it?

We’ll be using the terms data subscription and embedding config repeatedly in this post.

Because the system is eventually consistent, we knew we needed to track the state of embedding for every piece of data that a data subscription needs to embed, which leads us to…

Tracking Consistency:

Given that we would be eventually consistent, the latest version of data persisted to our vector database will always lag behind the data in MemApp. We need to be able to determine which data has been embedded and what may be out of date (for filtering stale results or fixing up the vector database). To accomplish this it helps have an ordering of data versions. Ordered data versions also lets us handle out-of-order write operations.

Since MemApp is a serializable database, its transaction number is perfect for this — a BigInt that increments with each write. For each piece of data we embed we create an embedding state:

type EmbeddingState = {
  // last recorded transaction that caused a write to the vector database
  // this may be lower than the actual number in between when a write to
  // the vector database occurs and when the update to MemApp occurs
  // (those writes to MemApp may also be lost due to system failures)
  lastPersistedTransaction: number | null,
  // always correct and up to date, we update this transactionally when
  // data in the data subscription is updated
  lastUpdatedTransaction: number
}

Being more concrete, we expect updates over the lifetime of an embedding state to look like this:

// new data to be embedded enters the data subscription
{
  lastPersistedTransaction: null,
  lastUpdatedTransaction: 2,
}
// once the data is embedded, we mark it as up to date
{
  lastPersistedTransaction: 2,
  lastUpdatedTransaction: 2,
}
// data is edited, we mark it as stale so it can be re-embedded
{
  lastPersistedTransaction: 2,
  lastUpdatedTransaction: 5,
}

The Simple Life Of An Embedding

Here’s a rough sketch of our system when data changes and needs to be re-embedded

Let’s walk through the flow in practice

Initialization

Upon creating a new embedding config, MemApp automatically provisions a new vector database table and generates embedding states for each relevant data chunk.

Detection

When data changes, we update the embedding state’s lastUpdatedTransaction to reflect the current transaction.

Triggering

Tasks are created to generate the embeddings for each config that saw data in its data subscription updated. Currently we do this within the transaction that the detection occurs.

Generation

Our embedding service processes the task(s), and generates the embeddings. We have ample retry logic in case this fails, there’s also a separate blog post here on how to prevent poorly behaving bases from consuming our global rate limits from AI providers or otherwise tanking quality of service.

Persistence

Once embeddings are generated we store them in our vector database. We prevent out-of-order writes by making the insertion conditional on the transaction number for the write being greater than the transaction number stored in the vector DB.

If the update has been outpaced we silently exit the flow since no more work is necessary.

We then confirm with MemApp that the write has happened, updating the lastPersistedTransaction. Once again we handle potential out of order writes by only ever increasing the value.

Deletion

Deleting individual embedding states will delete the data in the embedding store. Again we use conditional deletions to handle out of order writes.

Deleting an entire embedding config triggers automatic cleanup of all data: we delete the associated embedding states and the vector database table. This reduces storage costs and lets us meet various data retention guarantees. This actually creates a very nice property to the system — the real meat of this post…

Operations In Practice — Migrations And Failures

Systems fail and requirements change. We needed a strategy that handled:

• Vector database Failures
  - Database Corruption — the database exists, but is in a bad state
  - Catastrophic Data Loss — the database is deleted or unavailable
• Vector Database Migrations
  - Schema Migrations — storing new metadata
  - Database Engine Migration — e.g. LanceDB to Milvus or vice versa
  - Data Residency Migration — e.g. US to EU
  - KMS migration — when a customer using our KMS feature rotates their encryption key
• Permission Changes
  - AI Provider Changes — Airtable supports multiple AI providers. What if a user forbids use of OpenAI and only allows models served via Bedrock
  Note: Embeddings can/should not be compared across models. So updates require re-embedding everything.
• MemApp Migrations
  - Deprecating AI Models — We need to update the AI model to one that is still supported
  - Updating the Embedding Strategy — If we change how we embed data, we re-embed everything to keep comparisons apples-to-apples
• MemApp Operations
  - Updating Data Subscriptions — This changes the data we’re embedding, and so usually requires completely re-embedding the data
  - Cloning a base — Airtable allows you to clone a database. We need separate vector stores to keep the data model sensible over time.
  - Base Snapshot Restoration — Airtable has multi-year old snapshots that can serialized in a database
• MemApp Vector Database Synchronization
  - MemApp corruption — If we don’t create embedding states for some of our data
  - Detection Change Bugs — what if we don’t propagate changes to the vector store because we fail to detect them?
  - Task Queue Failures — What happens if our task queue violates its “at least once delivery” of updates?

You can imagine many ways to handle these cases. One pattern I found myself coming up with over and over again was:

1. Deleting the old embedding config (cleaning up old data if it exists)
2. Creating a new new embedding config in the old ones place (possibly with new settings)

We call this process resetting the embedding config.

I realized we could handle all of these cases if any time the system detects that embeddings are, or about to become, invalid, we reset the embedding config.

Moving a base from the US to the EU is illustrative.

When a base moves from the US to the EU sensitive data related to the base must move to the EU, with nothing in the US remaining. Since embeddings are sensitive they must be moved as well. To handle this we delete every embedding config in the base, our system then automatically deletes all existing embedding data as part of the normal cleanup process. We then re-create the embedding configs with a configuration to store the data in the EU. Voila! All the embedding data for the base has now been migrated.

You can repeat this exercise for every single case listed above. This approach had a few tradeoffs we had to consider:

Cost

Regenerating embeddings incurs expenses, but these are manageable compared to storage and indexing costs from incremental updates.

Downtime

Resetting embeddings causes temporary unavailability of semantic search features. Given the rarity of resets and the speed of re-embedding (p99.9 under 2 minutes), this was acceptable.

Handling this kind of downtime is already necessary at the product level since users can trigger large scale generations in the product very easily (e.g. cloning a base).

Potential Runaway Resets

We had to implement safeguards to prevent continuous reset loops, which could lead to spiraling costs and prolonged downtime. We used a combination of metrics, alerts, rate limiting, and idempotent requests.

Conclusion:

Our embedding system isn’t perfect, but it’s resilient and adaptable. By accepting eventual consistency and designing for graceful handling of failures and migrations, we’ve built a foundation that supports powerful semantic search capabilities and are continuing to iterate on this system as we find new ways to leverage AI to increase the power of the Airtable platform.

Some other interesting topics we totally glossed over but may touch on in the future:

1. Failures in the embedding generation
2. Downtime for MemApp
3. Managing global AI rate limits
4. Choice of vector indices
5. How we apply filters and permissions

Building a Resilient Embedding System for Semantic Search at Airtable was originally published in The Airtable Engineering Blog on Medium, where people are continuing the conversation by highlighting and responding to this story.