Apache Phoenix — SQL Layer on HBase with Server-Side Execution

The two halves of the engine

The thick client is the original deployment: your application loads the Phoenix JDBC driver, which is a full query engine. It parses SQL, resolves metadata from the SYSTEM.CATALOG table, builds a plan, decides how to split the scan for parallelism, dispatches the pieces to RegionServers, and merges what comes back. That means the client needs ZooKeeper and RegionServer connectivity and a nontrivial heap -- it is a cluster participant, not a thin connector.

The thin client exists because that is often unacceptable. The Phoenix Query Server is a standalone process running the thick client on your behalf and exposing it over HTTP using the Avatica protocol; applications connect with a small driver that has no cluster dependencies. This is the deployment for BI tools, non-Java languages, and anything outside the cluster's network boundary. The trade is an extra hop and a component to size and secure.

The server-side half is a set of coprocessors registered on Phoenix tables. Observers hook the scan path to perform aggregation and grouping in the region; another handles UPSERT SELECT and DELETE so that a rewrite-in-place never streams rows to the client and back; an index observer maintains secondary index tables as part of the write. Understanding which coprocessor does what is the difference between reading an EXPLAIN plan and guessing at it.

How a table maps onto HBase

A Phoenix table is an HBase table with a schema recorded in SYSTEM.CATALOG. The mapping is mechanical and worth knowing precisely, because every performance property follows from it.

The primary key columns become the HBase row key, serialised in declaration order and concatenated. Phoenix uses order-preserving byte encodings so that lexicographic byte ordering matches the declared type's natural ordering -- which is exactly what makes range predicates on the leading key columns translate into scan boundaries. Variable-length types in a composite key get a separator byte; fixed-width types do not need one.

Non-key columns become HBase columns in a column family, defaulting to family 0 when you do not name one. Since Phoenix 4.10 the default storage scheme encodes column qualifiers as small numbers rather than storing the column name in every cell, which is a large space saving on wide tables, and there is a single-cell storage scheme that packs a row's columns into one cell for further savings at the cost of per-column update granularity. A row with all columns null needs an empty key-value to exist at all, which Phoenix inserts automatically -- that is the marker that lets it distinguish 'row exists with null values' from 'row does not exist'.

Because the row key is the only sorted access path in HBase, the order of columns in your PRIMARY KEY clause is the single most consequential schema decision. Predicates on a prefix of the key are cheap; predicates on a trailing column alone are a full scan unless a skip scan or an index rescues them.

Row key design, salting and hotspots

HBase distributes by row key range, so a monotonically increasing leading key -- a timestamp, a sequence, an auto-increment id -- sends every write to one region and one RegionServer while the rest of the cluster idles. This is the classic HBase hotspot and Phoenix inherits it directly.

Phoenix's built-in answer is SALT_BUCKETS. Declaring SALT_BUCKETS = 16 prepends a single byte derived from a hash of the row key to every row, spreading writes evenly across sixteen key ranges and pre-splitting the table accordingly. Writes scale; point lookups still work, because Phoenix computes the same salt byte at query time. The cost is paid by range scans: a query over a contiguous range of the logical key now has to scan that range in every bucket and merge the results, so it fans out to sixteen scans instead of one. Salting is therefore right for write-heavy tables read by point lookup, and wrong for tables whose dominant access is a wide ordered range.

The alternative is to design a natural key whose leading component already has high cardinality and even distribution -- a tenant identifier, a device identifier, a hashed user id -- with the time component second. That preserves ordered range scans within a tenant, which is usually the query you actually run, and it avoids the salt fan-out entirely. Reserve salting for the case where no such natural leading column exists.

Query plans — full scan, range scan, skip scan, point lookup

EXPLAIN is the primary diagnostic and its vocabulary is small. A point lookup means the full primary key was supplied and Phoenix issues gets. A range scan means a prefix of the key was constrained, so the scan has real start and stop keys. A full scan means it did not, and the entire table will be read. The plan also reports how many chunks the scan was split into and whether aggregation is server-side.

The interesting case is the skip scan. Given a composite key of (region, device_type, event_time) and a query constraining region and event_time but leaving device_type open, a naive engine falls back to a full scan because the key prefix is broken. Phoenix instead uses a filter that seeks forward through the distinct values of the gap column, evaluating only the qualifying sub-ranges. On a key column with modest cardinality this converts a full scan into a handful of seeks, and it is one of the strongest reasons to put low-cardinality columns early in a composite key even when they are not always filtered on. It degrades as the gap column's cardinality rises -- with millions of distinct values the seeks stop being cheaper than reading through.

The other line to read is whether aggregation is marked as happening server-side. CLIENT in front of an aggregate step means rows are being shipped to the driver and combined there, which is a plan you almost never want on a large table and usually indicates the grouping does not line up with the key or that an index was not selected.

Server-side execution — what the coprocessors actually do

Aggregation is the headline. For SELECT device_type, COUNT(*), AVG(latency) FROM events GROUP BY device_type, the region-level coprocessor scans locally, maintains partial aggregates per group, and returns one small result per region. The client merges partials. Network transfer is proportional to the number of groups times the number of regions rather than to the number of rows, which is the difference between a query that finishes and one that does not.

The same principle covers UPSERT SELECT and DELETE. When the source and target of an UPSERT SELECT are the same table and the operation is not transactional, Phoenix can run the entire rewrite inside the regions -- rows are read, transformed and written without leaving the RegionServer. A large DELETE with a predicate behaves the same way, emitting tombstones locally. This is why bulk maintenance in Phoenix is dramatically faster than the equivalent loop in application code.

The consequence to respect: coprocessors run inside the RegionServer JVM, with its heap and its lifecycle. A query that groups by a very high-cardinality column builds large partial-aggregate maps in that heap. A mismatched Phoenix and HBase version, or a half-completed upgrade, puts incompatible coprocessor classes on the classpath. Both failure modes take out RegionServers rather than merely failing a query, which is the central operational reason Phoenix upgrades are planned rather than rolled out casually.

Secondary indexes — global versus local

HBase has one sorted access path. Phoenix adds secondary indexes by maintaining additional sorted structures whose row key begins with the indexed columns, and it keeps them current through the index coprocessor on the write path rather than asking the application to do dual writes.

A global index is a separate HBase table keyed by (indexed columns, primary key of the base row). Reads are excellent: the index is itself range-scannable and a lookup goes straight to the qualifying keys. Writes are the cost -- every base-table write becomes writes to the base table and to each affected index, landing on different regions and probably different servers. Global indexes suit read-heavy tables and are the default choice.

A local index stores index entries in the same region as the data they index, in a dedicated column family. Writes stay local, which is far cheaper and keeps the index consistent with the row without cross-server coordination -- the right trade for write-heavy tables. Reads pay instead: because the index is partitioned by the base table's key rather than by the indexed value, a lookup must consult every region and merge. Local indexes scale well on write and poorly on region count.

Both kinds support covered columns via INCLUDE. Without them, an index lookup that needs a non-indexed column must go back to the base table row by row, and Phoenix's optimiser frequently decides that double hop is worse than a scan and silently ignores the index. Including the handful of columns the query projects turns the index into a covering one and is usually what makes the plan flip. Functional indexes -- indexing an expression such as UPPER(name) -- work the same way and are matched only when the query uses the identical expression.

What happens when an index write fails

This is the part of Phoenix that surprises people, and it is a correctness question rather than a performance one. A base-table write and its global-index writes are not a distributed transaction. If the index write fails -- a RegionServer is down, a region is splitting -- Phoenix must choose between blocking writes and letting the index drift.

The behaviour is controlled by an index failure policy. The default posture is to protect correctness: the index is marked as needing rebuild and, depending on configuration and version, either the write is rejected or the index is disabled and excluded from query planning until it is rebuilt. An automatic rebuild task then replays the missing entries from the base table. What you must not assume is that a silently degraded index is still being consulted -- queries continue to return correct results because the optimiser stops using a disabled index, but they get slower without any application-visible error.

Operationally this means two things. Monitor index state in SYSTEM.CATALOG rather than assuming indexes are healthy, and know the rebuild command -- ALTER INDEX ... REBUILD -- and roughly how long a rebuild of your largest index takes, because that is your recovery time. Newer Phoenix versions add a consistency-checking and repair tool that compares an index against its base table and reports divergence; running it periodically on critical indexes is cheap insurance.

Statistics and parallelism

Phoenix parallelises a scan by splitting it into chunks and running them across a client-side thread pool. How it chooses the split points determines whether a query uses the cluster or one thread of it. Without statistics, the natural split boundary is the region, so a table with four regions gets four parallel scans no matter how large it is.

UPDATE STATISTICS collects guideposts -- key positions at roughly fixed byte intervals within each region, controlled by the guidepost width setting. With guideposts present, a single region can be split into many parallel chunks, and the optimiser gains size estimates it uses to choose between an index and a base-table scan and to decide which side of a join to broadcast. The practical effect on a large table is frequently a several-fold latency improvement for scan-heavy queries, and it is one of the most commonly skipped steps in a Phoenix deployment.

Guidepost width is a real tuning knob rather than a formality. Too wide and you get little extra parallelism; too narrow and you generate an enormous number of small chunks plus a large statistics table, and the client spends its time coordinating. Statistics also go stale after heavy write activity, so refreshing them belongs in the same maintenance window as major compactions.

Joins, subqueries, and the limits of the SQL surface

Phoenix supports joins, and the honest framing is that it supports them adequately for operational queries and poorly for analytics. The default strategy is a hash join: one side is read into memory and broadcast to the servers scanning the other side. That is fast when the broadcast side is small, and it fails outright -- or spills into misery -- when it is not. The USE_SORT_MERGE_JOIN hint switches to a sort-merge strategy for large-to-large joins, which is slower but survives.

Because join strategy is chosen from statistics and hints rather than from a sophisticated cost model, join-heavy workloads on Phoenix need explicit attention: check the plan, force the strategy when the optimiser gets it wrong, and keep the smaller relation genuinely small. The general guidance that holds across deployments is to treat Phoenix as an operational SQL layer -- point lookups, ranged reads, aggregates over bounded key ranges, upserts -- and to route multi-table analytical work to an engine designed for it, reading the same HBase tables through Spark or a query engine that has a real optimiser.

Two other surface details matter in day-to-day use. Phoenix has UPSERT rather than separate INSERT and UPDATE, which mirrors HBase's put semantics: there is no read-modify-write and no primary-key violation to catch, so an accidental re-run overwrites rather than erroring. And ON DUPLICATE KEY exists for atomic read-modify-write on a single row, implemented server-side with a row lock -- the right tool for counters, and one that quietly serialises concurrent writers to the same key.

Views, multi-tenancy and bulk loading

Views in Phoenix are more structural than in a relational database. A view can be defined over an existing HBase table to give it a Phoenix schema without rewriting data, which is the standard migration path for tables that predate Phoenix. Views can also add columns to a shared base table, so many tenants or many entity types coexist in one physical table with different logical schemas.

That mechanism underpins multi-tenancy. A table declared MULTI_TENANT reserves the leading primary key column as a tenant identifier; a connection opened with a TenantId property sees only that tenant's rows and may create tenant-specific views with tenant-specific columns. Isolation is enforced by the driver rewriting every query with the tenant prefix, which makes it cheap and consistent -- and makes it critical that application code cannot obtain an untenanted connection.

Bulk loading bypasses the write path entirely. The CSV bulk-load tool runs a MapReduce job that generates HFiles in the table's own key order and hands them to HBase, skipping the write-ahead log, the memstore and the flush cycle. For initial loads and large periodic imports this is an order of magnitude faster than upserting through JDBC and avoids the compaction storm that a large upsert batch provokes. It maintains indexes as part of the job, which is why the tool must be used rather than a hand-rolled HFile writer.

Operational realities before you commit

Version coupling is tight. A Phoenix release targets specific HBase minor versions, and the client, the server coprocessors and the system tables must agree. Upgrades involve a system-catalog migration that runs on first connection from a new client, so a rolling upgrade with mixed client versions is a hazard, not a convenience. Plan Phoenix upgrades as cluster events.

The system catalog is a shared dependency. All metadata lives in SYSTEM.CATALOG; historically it was constrained to a single region, and while newer versions relax that, heavy DDL and very large numbers of views still concentrate load there. Thousands of tenant views is a known stress pattern -- test it at your intended scale rather than at ten.

Coprocessors are in your blast radius. Everything Phoenix pushes down runs in RegionServer JVMs. Size heaps with that in mind, watch garbage-collection pauses after enabling Phoenix on a busy cluster, and treat a Phoenix query that destabilises a RegionServer as an expected class of incident with a runbook rather than as a surprise.

Know the fit. Phoenix is the right answer when you already run HBase, need low-latency operational access by key or key range, and want JDBC, secondary indexes and a schema rather than hand-written scan code. It is the wrong answer as a general-purpose analytical warehouse, and it is a poor reason to adopt HBase in the first place if you do not otherwise need it.