Large-scale Incremental Processing Using Distributed Transactions and Notifications

✨ Summary

Overview

The paper introduces Percolator, a system for incrementally transforming extremely large datasets through many small, concurrent updates. Its motivating application was Google’s web-search index, where repeatedly running large MapReduce pipelines over the entire document repository created latency proportional to repository size and discarded work from previous runs. Percolator instead provides random access to a multi-petabyte repository, allowing processing costs to depend primarily on newly arrived or changed data. The system was deployed for Google’s search-indexing pipeline, later known as Caffeine. (research.google)

Core design

Percolator is built on Bigtable and GFS. It adds two major abstractions: distributed transactions and observers. Transactions provide cross-row and cross-table updates with ACID snapshot-isolation semantics. Observers are application-defined computations invoked when specified columns change; they form a chain in which one observer performs work and writes data that can trigger downstream observers.

The transaction protocol uses multiversion data, lock, and write metadata stored in Bigtable columns. A client obtains a strictly increasing start timestamp, buffers writes, and performs a two-phase commit. During prewrite, it checks for conflicting writes or locks, records the new data at the start timestamp, and installs locks. One lock is designated as the primary lock, while secondary locks refer to it. During commit, the client obtains a commit timestamp, replaces the primary lock with a write record, and then finalizes the secondary cells. The primary lock acts as the synchronization point for recovery: other clients can determine whether a failed transaction committed and either roll its locks back or roll them forward.

Reads use the transaction’s start timestamp to select a consistent snapshot. A reader waits when it encounters a visible conflicting lock, then follows a write record to the corresponding version of the data. The design deliberately uses snapshot isolation rather than serializability, which improves read efficiency but permits anomalies such as write skew. Percolator also avoids a centralized transaction manager and global deadlock detector, accepting higher conflict latency in exchange for scalability across thousands of machines.

Notification model

Observers are triggered through notification metadata. A notification column acts as a hint that an observed cell may require processing, while per-observer acknowledgment columns record the most recent successful execution. This supports message collapsing: multiple writes may result in a single observer execution, and at most one observer transaction commits for a particular observed-column change. The notification mechanism is intentionally separate from the triggering transaction, so it organizes computation but does not itself provide atomic maintenance of invariants.

Workers find pending notifications through distributed scans over a separate Bigtable locality group. Advisory locks reduce duplicate scanning, and the implementation uses randomized scanning with relocation when workers begin to cluster around the same table regions. The system also provides weaker, non-transactional notifications for high-contention cells; these avoid transactional conflicts but may cause multiple observer executions.

Evaluation and trade-offs

Compared with Google’s previous MapReduce-based indexing pipeline, the Percolator-based system processed the same daily crawl volume while reducing median document-processing latency by more than two orders of magnitude. The paper reports that the average age of documents appearing in search results fell by nearly 50 percent. In a synthetic clustering workload, Percolator processed newly arrived documents in roughly two seconds at a low crawl rate, whereas MapReduce required approximately twenty minutes to rescan the repository. Percolator’s advantage diminished at high update rates, where random lookups became more expensive than sequentially scanning the repository; in the reported configuration, Percolator saturated near a 40-percent-per-hour crawl rate.

The improvement required substantially more resources. The production incremental index used approximately twice the resources of the replaced system at the same crawl rate, and Percolator incurred significant RPC and metadata overhead relative to raw Bigtable. Microbenchmarks showed approximately fourfold write overhead for single-cell transactions, largely due to conflict checks and lock-management operations. A TPC-E-like benchmark demonstrated near-linear scaling from 11 to 15,000 CPU cores, but the authors estimated roughly thirty times more CPU per transaction than a contemporary single-machine commercial database. These costs were accepted because the system supported Internet-scale data, incremental updates, and resilience to machine failures rather than low-latency OLTP workloads.

Influence on subsequent systems and industry

The paper’s transaction design became a documented influence on TiKV and TiDB. TiKV explicitly describes its distributed transaction implementation as inspired by Percolator and adopts the primary/secondary-key model, timestamp oracle, multiversion data, locks, and write records, while replacing Bigtable’s storage layer with RocksDB and Raft. TiDB documentation likewise identifies Google’s Percolator transaction model as the basis for its distributed two-phase-commit design. (tikv.org)

The observer and notification ideas also influenced Apache Fluo, which was proposed as a distributed transaction and notification system for incrementally processing large datasets stored in Apache Accumulo. Its proposal explicitly identifies Percolator as the model for combining transactional updates with notification-triggered computation. (cwiki.apache.org)

Overall, the paper established a practical design point between batch processing and traditional databases: use a scalable distributed key-value store, add client-coordinated transactions and timestamp-based versioning, and organize computation around changes rather than repeated full-repository scans. Its documented influence is especially clear in distributed transactional key-value systems and incremental data-processing frameworks.