Polaris: Inside Microsoft's Distributed SQL Query Engine

What Is Polaris and Why It Matters#

Polaris is Microsoft's cloud-native distributed SQL query engine, conceived for Azure Synapse and now serving as the execution backbone of Microsoft Fabric's Warehouse and SQL analytics endpoint ^[S1]. Its creators set two explicit design goals at the outset: first, to unify traditional data-warehouse and big-data workloads on a single engine rather than forcing operators to maintain separate systems; second, to fully separate compute from all durable state so that the execution tier is stateless and therefore elastically scalable ^[S1].

For a practitioner this has concrete consequences. Every Fabric Warehouse and every SQL analytics endpoint — including the auto-provisioned endpoint on every Lakehouse — share one underlying distributed processing architecture, and the backend is serverless: compute capacity scales up and down autonomously to match workload demand rather than being provisioned by the user ^[S2]. There are no node-count sliders to configure, no MPP classes to pre-select, and no maintenance windows during which the cluster must be paused. The engine handles scale itself.

Understanding how Polaris works explains why Fabric Warehouse behaves the way it does under concurrency, what limits its latency floor, how it tolerates node failures without failing queries, and why certain patterns — broad fan-out queries, high concurrency, multi-petabyte scans — are exactly what it was designed for.

Core Concepts#

The Cell Abstraction: A Unified View of Any Dataset#

Polaris's most important architectural primitive is the data cell. Every dataset — whether it lives in ADLS Parquet files, Delta Lake tables, or a transactional store — is abstracted as a two-dimensional grid of cells indexed on two functions ^[S1]:

1. A user-defined partition function that supports aggressive partition pruning under range or equality predicates. Query predicates on the partition column eliminate whole slices of cells from the plan before any compute node touches them.
2. A system-chosen hash-distribution function that maps rows to many buckets so that cells — and therefore the computation over them — can be spread across arbitrarily many compute nodes ^[S1].

The cell abstraction makes the distributed query processor (DQP) storage-agnostic. The DQP operates only at cell granularity — it reasons about which cells are needed and how to redistribute them — while single-node extraction of the data inside a cell is delegated to a local execution engine. This means the same scale-out framework can query analytical stores and transactional stores alike without the DQP needing to understand their internal formats ^[S1].

Stateless Compute#

Polaris compute nodes carry no durable state. All persistent state — data files, metadata catalogs, and transaction logs — is externalized to remote storage and centralized highly-available services ^[S1]. The only things living on a compute node are caches: the SQL Server buffer pool in memory and a local SSD tier.

Caches can be rebuilt lazily from persisted data, so their loss never fails in-flight work.

Because caches can be rebuilt lazily from persisted remote data, their loss never fails in-flight work. This property is what enables two capabilities that would be impossible with stateful compute:

• Partial query restarts — if a node fails mid-query, its tasks can be redistributed and their cells re-fetched from storage and re-executed without restarting the whole query.
• Online topology changes — nodes can be added or removed while queries are running, because there is no state to migrate or redistribute ^[S1].

In Fabric's deployment this surfaces as a concrete guarantee: scaling the warehouse backend is an online operation — in-flight query processing continues uninterrupted while nodes are added — and when pressure subsides the topology scales back down and returns compute to the region ^[S2].

When a Query Arrives: From SQL to Distributed Tasks#

When a query arrives at the SQL frontend, the frontend optimizes it and hands the plan to the Distributed Query Processing (DQP) engine, which splits it into smaller units called tasks — the atoms of distributed execution — that run on backend compute nodes; results flow back through the frontend to the caller ^[S2].

A Fabric warehouse task reads files from OneLake, joins, groups, or orders data produced by other tasks, and for ingestion jobs writes results to the destination tables ^[S2]. Both the query path and the ingest path therefore run through the same task-based distributed engine — there is no separate bulk-load mechanism bypassing the DQP.

How It Works / Best Practices#

Design for the Distributed Path#

Rule: don't design latency-critical, single-row-lookup workloads against the Warehouse's distributed SQL path expecting uniform sub-second response. Why: node acquisition during scale-out typically takes a few seconds with no SLA on assignment, so the engine is not optimized for consistently sub-second latency on queries that require distributed processing ^[S2]. Example — route point-lookup traffic away from repeated ad-hoc distributed queries and toward a pre-aggregated serving table refreshed on a schedule:

-- Anti-pattern: high-frequency point lookups against the distributed engine,
-- each one potentially paying scale-out/scheduling latency
SELECT CustomerBalance FROM dbo.Accounts WHERE AccountId = @id;
-- ...called thousands of times per second from an app tier

-- Better: pre-aggregate on a schedule into a small serving table
CREATE TABLE dbo.AccountBalanceSnapshot AS
SELECT AccountId, CustomerBalance, SnapshotUtc = SYSUTCDATETIME()
FROM dbo.Accounts;

Exploit Partition Pruning#

Rule: align your most selective filter predicate with the table's partition column. Why: the cell model's user-defined partition function lets the optimizer skip whole cell slices before dispatching tasks, which cuts both I/O and the number of tasks scheduled ^[S1]. Example:

-- Anti-pattern: no partitioning column aligned with the dominant filter,
-- so every query over TransactionDate scans all cells
CREATE TABLE dbo.FactSales (
    TransactionId BIGINT,
    TransactionDate DATE,
    StoreId INT,
    SalesAmount DECIMAL(18,2)
);

-- Better: partition on the column most reporting queries filter by
CREATE TABLE dbo.FactSales (
    TransactionId BIGINT,
    TransactionDate DATE,
    StoreId INT,
    SalesAmount DECIMAL(18,2)
)
WITH (PARTITION (TransactionDate RANGE RIGHT FOR VALUES (...)));

Let the Engine Scale — Don't Build Your Own Retry Layer#

Rule: don't wrap every warehouse query in application-level node-failure retry logic. Why: the engine is fault tolerant at the task level — if a backend node becomes unhealthy, its operations are redistributed to healthy nodes for completion rather than failing the query ^[S2]. Example — a right-sized retry policy covers transient connection errors only, not compute-node health:

import pyodbc
from tenacity import retry, stop_after_attempt, retry_if_exception_type

@retry(stop=stop_after_attempt(3), retry_if_exception_type(pyodbc.OperationalError))
def run_query(conn_str, sql):
    with pyodbc.connect(conn_str) as conn:
        return conn.execute(sql).fetchall()

What Goes Wrong#

Sub-Second Latency Expectations#

The most common mismatch is expecting Fabric Warehouse to serve sub-second interactive queries. Because scale-out node acquisition takes a few seconds with no guaranteed SLA ^[S2], and because the DQP must schedule and dispatch tasks across the topology, distributed queries have an irreducible latency floor. Queries that can be served by a fully warm cache on a single node may be fast; queries requiring a fan-out will not be sub-second.

Ignoring Partition Design#

Without a well-chosen partition column the cell abstraction cannot prune cells, forcing the DQP to dispatch tasks over every cell in the dataset for any filter predicate ^[S1]. This wastes task slots in the workload graph, increases shuffle volume, and delays results. Partition column selection is the single highest-leverage schema decision for Polaris-backed tables.

Treating Scale-Down as a Cold-Start Risk#

During topology shrink, only the caches of removed nodes are lost ^[S1]. Data cells affinitized to surviving nodes remain warm. Practitioners who benchmark immediately after a scale-down event may see elevated latency on cells that were assigned to the removed nodes; this is not a system bug — it is expected warm-up behavior for newly re-affinitized cells.

Internals#

Architecture & design#

Polaris compute is stateless by design: durable state lives entirely in remote storage and centralized services, and only ephemeral caches live on compute nodes. Because caches can be lazily rebuilt, node loss never fails in-flight work — which is what makes partial query restarts and online topology changes possible ^[S1].

On top of that stateless compute layer sits the cell abstraction: every dataset is a grid of data cells keyed by a user partition function (for pruning) and a system hash-distribution function (for spreading computation) ^[S1]. Cells make the DQP storage-agnostic — it operates at cell granularity while a local single-node engine handles extraction inside a cell, so the same scale-out framework reads ADLS, Parquet, and Delta Lake alongside transactional stores ^[S1].

In Fabric, this architecture is exposed as one shared backend for both the Warehouse and the SQL analytics endpoint, running as a serverless service that scales compute autonomously rather than on fixed user-provisioned capacity ^[S2].

How it works internally#

Query compilation in Polaris is a two-phase process ^[S1]. In phase one, the SQL Server Cascades optimizer generates the logical search space in a structure called the MEMO, applying logical rewrites such as predicate pushdown, join reordering, and subquery unnesting, without yet reasoning about distribution. In phase two, a cost-based distributed planning pass enumerates physical distributed plans over the MEMO. Distribution properties serve a dual role: as correctness filters (an inner join, for example, requires hash-aligned inputs or a broadcast side), and as System-R-style 'interesting properties' that guide plan enumeration. The optimizer selects the plan that minimizes total data movement, since network shuffle is almost always the dominant cost in a distributed query ^[S1].

When a plan requires a dataset to change its distribution — for example, two tables must be hash-aligned before a join — Polaris inserts a data move enforcer operator, either hash re-distribution (shuffle) or broadcast. Enforcers are blocking: they must persist all output cells before any downstream consumer starts, and the operator subtree rooted at an enforcer defines the input/output boundary of a task ^[S1]. This blocking property lets SQL Server re-optimize with fresh statistics on intermediate results at each task boundary, since actual row counts and value distributions from completed tasks become available before the next task begins.

A Polaris task is the physical execution of an operator expression over one hash-distribution of its inputs, with three components: input cell collections, a task template (the compiled code for the expression), and an output cell collection. A full query plan becomes a DAG of task templates whose edges encode dataflow and precedence constraints ^[S1]. The DQP scheduler operates at task level and represents each query as this DAG, so tasks without mutual dependencies can execute simultaneously or out of order — giving both intra-query parallelism and cross-query concurrency ^[S2].

Polaris deliberately makes tasks maximal units of work: wherever hash-aligned joins or other consecutive operations do not require an enforcer between them, they are fused into a single task rather than split into multiple smaller ones ^[S1]. Each fused task template is then re-encoded back into T-SQL that runs natively in a SQL Server instance on the compute node — combining big-data-style scale-out coordination with SQL Server's mature, scale-up columnar execution for single-node work within each task ^[S1].

Execution is orchestrated as a hierarchical composition of finite state machines over the query DAG, task templates, and individual tasks, with simple states like Ready/Failed/Success and composite states like Run/Blocked. Transient failures are retried at task granularity, and only escalate to the parent state machine if not retriable; all transitions are logged for reproducibility and resumable execution ^[S1]. In Fabric's deployment this translates to a published guarantee: if a backend node becomes unhealthy, the operations running on it are redistributed to healthy nodes for completion rather than failing the query ^[S2].

Scheduling coordinates across all active queries using a global workload graph that unions the task DAGs of every active query. Each task carries a multi-dimensional resource demand vector — CPU and memory are stretchable, local temp disk is rigid — and compute nodes are modeled as multi-dimensional resource bins. Supported policies include FIFO, sorting by resource demand, and sorting by proximity to the DAG root to free shared resources sooner ^[S1]. In Fabric's deployment, incoming tasks default to FIFO scheduling, but when idle capacity exists the scheduler switches to a best-fit placement approach to improve concurrency ^[S2].

Finally, task placement is affinitized: the same data cells are consistently assigned to the same compute nodes, preserving SQL Server buffer-pool and local SSD caches across topology changes. Shrinking the topology loses only the caches of the removed nodes; growing it preserves all existing caches while only the new nodes warm up ^[S1].

Performance characteristics#

The VLDB 2020 paper documents two scale experiments that characterize the system's performance envelope ^[S1]. First, all 22 TPC-H queries were executed against a 1-petabyte dataset on a pool of 420 execution nodes; the most join-intensive queries, Q9 and Q21, completed in under two hours. This experiment validates the cell-based distribution and the workload-graph scheduler at extreme dataset size. Second, 5,000 concurrent TPC-DS sessions ran on just 10 compute nodes, completing roughly 550,000 instantiated tasks in total, with up to about 9,000 tasks packed concurrently across those 10 nodes at peak ^[S1]. This second experiment demonstrates that the DAG-level scheduler can extract dense concurrency from a small node pool — relevant to Fabric's serverless model, where node count may be low during scale-down periods.

On the elasticity side, Fabric's documentation confirms the qualitative behavior without publishing throughput numbers: scaling is an online operation, in-flight processing continues while nodes are added, and the topology scales back down — returning compute to the region — once pressure subsides ^[S2]. New-node acquisition typically takes a few seconds with no SLA on assignment, which directly explains why the engine doesn't target consistently sub-second latency on distributed queries ^[S2].

Worked Example: 1 TB Daily Fact-Table Load with Concurrent Reporting#

Inference: The following scenario assembles verified Polaris mechanics into a plausible end-to-end pattern. Each numbered step is grounded in the verified claims cited.

Scenario: A team loads 1 TB of daily transaction data from ADLS Delta files into a Fabric Warehouse fact table, while 200 concurrent reporting users query the table throughout the day.

Design decisions grounded in Polaris mechanics:

1. Partition the fact table on transaction date — the most selective reporting predicate. The cell abstraction's partition pruning eliminates cells outside the queried date range before any tasks are dispatched ^[S1]. On a 1 TB table partitioned daily over two years, a single-day query touches roughly 0.14% of cells.

2. Choose a distribution key aligned to the most common join — for example, the product ID if most reporting queries join the fact table to a product dimension. This makes the join hash-aligned with zero enforcer overhead for those queries; the optimizer will not inject a shuffle ^[S1].

3. Submit the daily ingest as a single batch operation rather than row-by-row or small-batch inserts. Polaris routes both ingest and query through the same task-based DQP ^[S2], so large batches fully exploit parallelism while many small writes produce many small tasks that compete for scheduler slots.

4. Trust the autoscaler during the ingest surge. When 200 concurrent sessions plus the ingest job create resource pressure, the scheduler triggers scale-out ^[S2]. In-flight reporting queries continue running during the topology expansion ^[S2]; users see no interruption.

5. Skip application-level retry logic for node failures. The state machine hierarchy retries failed tasks at task granularity and escalates only non-retriable failures ^{[S1] [S2]}. A query that encounters a node failure is recovered by the engine, not by the application.

Monitoring signals:

• High task concurrency in the workload graph indicates the scheduler is packing work efficiently ^[S1].
• If reporting queries show unexpectedly high latency after a scale-down, check whether their data cells were affinitized to removed nodes — the warm-up period for re-affinitized cells is normal ^[S1].
• The 5,000-session / 10-node benchmark result from the VLDB paper ^[S1] provides an existence proof that dense concurrency is achievable: 200 concurrent sessions on a larger node pool is well within the engine's design envelope.