Spark SQL — ANSI SQL over Files and Warehouses

Why it matters

Spark SQL is not a convenience wrapper bolted onto the DataFrame API for people who dislike Scala. It is a peer entry point: a SQL string and a chain of DataFrame calls arrive at the same logical plan and are executed by the same machinery, so choosing between them is an ergonomic and governance decision rather than a performance one. That equivalence is the single most useful thing to internalise, because it means a team can hand analysts a SQL endpoint without splitting the platform into a fast path and a slow path.

The other half of Spark SQL is the part that has nothing to do with query text. It is a catalog - a namespace of tables, views, functions and databases that outlives any one job, that other engines can read, and that turns a directory of Parquet files into an object with a name, a schema and a lifecycle. Most of the operational surprises in a Spark SQL deployment come from that half: where the metadata actually lives, which catalog a three-part name resolved to, whether a view was session-scoped or persistent, and what the engine will and will not enforce on your behalf.

This article covers the SQL surface and the catalog. The plan rewriting behind it - the four Catalyst phases, rule batches, pushdown and cost-based decisions - is developed in the Spark SQL optimizer article, and plan output in reading explain plans.

Two front doors, one query plan

spark.sql("...") takes a string and hands back a DataFrame. That return type is the whole story: the SQL front door does not lead to a separate engine, it leads into the same object graph the API builds. You can start in SQL and finish in the API, or the reverse, in one expression.

orders = spark.sql("SELECT id, customer_id, amount FROM sales.orders WHERE amount > 0")
big = orders.filter("amount > 1000").groupBy("customer_id").count()   # API on a SQL result

big.createOrReplaceTempView("big_customers")
spark.sql("SELECT * FROM big_customers ORDER BY count DESC LIMIT 20")  # SQL on an API result

Because both routes converge before any optimisation happens, the folklore that "SQL is slower than the DataFrame API" or the reverse is simply false for equivalent queries. What does differ is everything around the query. SQL text is a string, so it can be stored in a file, diffed in review, generated by a BI tool, or sent over a wire to a server that has never seen your code. API calls are values, so they can be composed by functions, unit tested without a parser, and refactored by an IDE. The failure modes differ too: a typo in a SQL string is caught at analysis time when you submit it, while a typo in a column name inside a builder chain is caught at the same moment, but a malformed builder chain fails to compile in Scala and never reaches Spark at all.

One practical caution about SQL as strings: build predicates with named parameters where your version supports them rather than interpolating values into the text. String interpolation of user input is the same injection hazard here that it is against any database, and it also defeats plan reuse.

What ANSI mode actually changes

Spark has two arithmetic and casting personalities, selected by spark.sql.ansi.enabled. Which one your cluster starts in has changed across major releases, so check it for your version instead of assuming - the important thing is knowing what each mode does, because the difference is between a wrong answer and a failed job.

With ANSI mode disabled, Spark follows the permissive convention it inherited from the Hadoop SQL world: an invalid cast returns NULL rather than raising, integral arithmetic that exceeds the type's range wraps silently, and division by zero yields NULL. CAST('twelve' AS INT) is NULL. An INT multiplication that overflows produces a negative number with no warning anywhere in the logs. This is the behaviour that produces the classic incident where a revenue total is quietly wrong for a quarter and nothing in the pipeline ever failed.

With ANSI mode enabled, those same operations raise a runtime error instead. A failed cast throws, overflow throws, division by zero throws. The query dies, loudly, with a message naming the expression. That is what a conventional relational database would have done all along, and it is the correct default for anything whose output someone will make a decision from.

-- ANSI disabled: both rows come back, one of them silently wrong
SELECT CAST(sku AS INT) FROM raw_events;      -- 'A-1099' -> NULL
SELECT 2147483647 + 1;                        -- wraps to -2147483648

-- ANSI enabled: the same statements raise instead of returning a value

Turning ANSI on is not free. It will surface every latent data quality problem in your pipeline at once, and it will do so as job failures rather than as null columns, which is why the switch is best flipped on a backfill of historical data before it is flipped in production. Budget for a round of fixes, not a config change.

Keeping null semantics under ANSI - the try functions

The objection to strict mode is legitimate: sometimes a NULL genuinely is the right answer. A dirty ingest column where ten rows in a million are unparseable should not take down a job that processes the other 999,990 correctly. Spark answers this with an explicit family of try_ functions that keep the permissive semantics at the call site instead of globally.

SELECT
  try_cast(sku AS INT)            AS sku_num,     -- NULL on a bad value
  try_divide(revenue, units)      AS unit_price,  -- NULL on units = 0
  try_add(counter, delta)         AS counter,     -- NULL on overflow
  try_element_at(tags, 5)         AS fifth_tag    -- NULL on out of range
FROM raw_events;

The design point is that permissiveness becomes a visible decision in the query text. A reviewer reading try_cast knows the author accepted nulls for unparseable input; a reviewer reading CAST under ANSI knows the author expects it never to fail. Under the permissive global mode, both authors wrote CAST and neither expressed an intent, which is exactly why the mistake was invisible.

A separate and frequently confused setting governs the write path rather than the read path: spark.sql.storeAssignmentPolicy controls which implicit casts are allowed when a value is stored into a table column by an INSERT. Its LEGACY value permits any cast, including lossy ones; ANSI permits reasonable widening but rejects nonsense such as string into int at analysis time; STRICT rejects anything lossy at all. It is independent of spark.sql.ansi.enabled, so a cluster can be strict about inserts and permissive about expressions, or the reverse. Check both when you are tracking down why a value changed shape on the way into a table.

Three-part names and the catalog you plug in

A table reference in Spark SQL has up to three parts: catalog.namespace.table. The leading part is not decorative. The analyser hands the first segment to a catalog manager, which looks for a registered implementation under that name and asks it to load the rest of the identifier. Anything that does not begin with a registered catalog name falls through to the built-in session catalog, which is registered under the reserved name spark_catalog.

Registering another catalog is a configuration line. Each one lives in its own namespace, with its own connection settings, and queries can span them in a single statement:

# register a second catalog under the name "lake"
spark.sql.catalog.lake                = org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.lake.type           = hive
spark.sql.catalog.lake.uri            = thrift://metastore-host:9083

# make unqualified names resolve there instead of spark_catalog
spark.sql.defaultCatalog              = lake

SELECT o.id, c.segment
FROM lake.sales.orders o
JOIN spark_catalog.reference.customers c ON o.customer_id = c.id;

The more consequential move is replacing spark_catalog itself. Table formats ship a session catalog implementation that you register under that reserved name, which lets it intercept resolution for ordinary two-part names while delegating anything it does not own back to the default behaviour. That is how CREATE TABLE ... USING delta and MERGE INTO start working on plain, unqualified table names without every query being rewritten to carry a catalog prefix.

From code, spark.catalog exposes the same namespace programmatically - listDatabases(), listTables(), listColumns(), tableExists(), currentDatabase() - which is what metadata tooling and schema-drift checks should use instead of parsing SHOW TABLES output. The connector-side contract that a catalog implementation has to satisfy is covered in the DataSource V2 article.

What the Hive metastore actually gives you

"Spark SQL uses the Hive metastore" is true and routinely misread. Spark uses the metastore as a metadata service: it asks for a table's schema, its partition list, its storage location, its format and its properties. It does not use Hive's query planner, Hive's execution engine, or - for the common formats - Hive's readers. The plan is Spark's, the tasks are Spark's, and for Parquet and ORC tables Spark converts the Hive table definition into its own native, vectorised data source rather than going through the Hive SerDe path. That conversion is governed by spark.sql.hive.convertMetastoreParquet and spark.sql.hive.convertMetastoreOrc, and it is the reason a Hive-defined table is fast in Spark and occasionally reads a corner-case type differently than Hive would.

Whether Spark talks to a metastore at all is decided by spark.sql.catalogImplementation, which is either hive or in-memory. The in-memory catalog is real and useful: databases, tables and views exist for the life of the application and then vanish. The hive setting is what enableHiveSupport() turns on.

spark.sql.catalogImplementation       = hive
spark.sql.warehouse.dir               = s3://my-lake/warehouse
spark.hadoop.hive.metastore.uris      = thrift://metastore-host:9083

# talk to a metastore whose version differs from the bundled client
spark.sql.hive.metastore.version      = 3.1.3
spark.sql.hive.metastore.jars         = path
spark.sql.hive.metastore.jars.path    = /opt/hms-client/*.jar

The failure everyone meets once: with Hive support enabled and no metastore URI configured, Spark starts an embedded Derby metastore in a metastore_db directory under the working directory. It works, it is single-process, and the second concurrent driver fails with a message about another instance having already booted the database. Two symptoms follow from the same cause - tables that "disappear" between runs because each run started in a different directory, and startup failures on a shared edge node. Neither is a bug; it is the default doing exactly what a single-user embedded database does. Point hive.metastore.uris at a real metastore before anything shared touches it. The internals of the metastore service itself are covered in the Hive metastore article.

Managed tables, external tables, and querying a path directly

Spark recognises three ways to name data, and the difference between them is entirely about who owns the files.

A managed table is created without a LOCATION. Spark chooses the directory under spark.sql.warehouse.dir and considers the data its own, which means DROP TABLE deletes the files. An external table is created with an explicit LOCATION. Spark records where the data is but does not claim it, so DROP TABLE removes the catalog entry and leaves every byte in place. On a shared lake where several engines and several teams write to the same prefixes, external is almost always the right choice, and the one-word difference between the two statements is the difference between a recoverable mistake and a restore from backup.

-- managed: Spark owns the directory, DROP deletes the data
CREATE TABLE sales.orders (id BIGINT, customer_id BIGINT, amount DECIMAL(12,2))
USING parquet
PARTITIONED BY (dt DATE);

-- external: you own the directory, DROP removes only the metadata
CREATE TABLE sales.orders_ext (id BIGINT, amount DECIMAL(12,2))
USING parquet
PARTITIONED BY (dt DATE)
LOCATION 's3://my-lake/raw/orders/';

-- partitions written out of band are invisible until the catalog is told
ALTER TABLE sales.orders_ext RECOVER PARTITIONS;

That last statement is worth remembering. For a partitioned table, the catalog holds the partition list, and a file dropped into a new dt= directory by an external process does not exist as far as the planner is concerned until the partition is registered. Queries return no error - they return fewer rows, which is worse.

The third route skips the catalog entirely. SELECT * FROM parquet.`s3://my-lake/raw/orders/dt=2026-08-01/` reads a path as if it were a table, with the schema taken from the files. It is excellent for ad hoc inspection and a poor basis for anything scheduled: there is no schema of record, no partition metadata, no place to record ownership, and nothing stopping a writer from changing the layout underneath a query that assumed it.

Views - temporary, global temporary, and persistent

All three kinds of view store a query, not a result. At analysis time the view's definition is inlined into the plan of whatever referenced it, so a view costs nothing until it is queried and never goes stale. What separates the three is scope and lifetime, and mixing them up is a common cause of a job that works in a notebook and fails on a schedule.

A temporary view is registered outside the catalog and takes precedence during resolution, so a temp view named orders shadows a catalog table named orders for that session. That is convenient for testing and dangerous in a long notebook, where a stale temp view silently replaces the real table for every subsequent cell.

Global temporary views must be referenced through their database: SELECT * FROM global_temp.daily_totals, never FROM daily_totals. They exist so that two sessions inside the same application - two JDBC connections to one Thrift server, for instance - can share a definition, and they die with the application no matter how it ends.

Persistent views are metastore objects with a real consequence: their definition is stored as SQL text, so it can only reference objects that will still exist when someone else resolves it. Spark rejects a persistent view built on a temporary view for exactly this reason. Spark also records the view's output column names and types at creation, so if an underlying table later loses a column the view fails at resolution with a message naming the mismatch rather than returning a differently-shaped result.

The Thrift server - pointing a BI tool at Spark

The Spark Thrift server is a HiveServer2-compatible endpoint started by sbin/start-thriftserver.sh, listening on port 10000 by default. Any client that speaks the HiveServer2 protocol - beeline, a JDBC driver, a BI tool's Hive connector - can attach and issue SQL, and what runs behind it is ordinary Spark SQL.

./sbin/start-thriftserver.sh \
  --master yarn --deploy-mode client \
  --hiveconf hive.server2.thrift.port=10000 \
  --conf spark.scheduler.mode=FAIR \
  --conf spark.sql.thriftServer.incrementalCollect=true

beeline -u 'jdbc:hive2://thrift-host:10000/sales' -n analyst

The architectural fact that governs how it behaves in production: the Thrift server is one long-lived Spark application. Every JDBC connection becomes a session inside that single driver, and all of them share one SparkContext and one pool of executors. Sessions are isolated for temporary views and for most SQL configuration, but they are not isolated for resources. One analyst running an unfiltered scan of the fact table will slow every other connection, and one badly-shaped query can take the driver down and disconnect everybody at once.

Two mitigations matter. Set spark.scheduler.mode=FAIR and assign connections to scheduler pools so a long query cannot monopolise the executors - without it, the default FIFO ordering means whoever submitted first wins. And enable spark.sql.thriftServer.incrementalCollect so large result sets are streamed to the client partition by partition rather than collected into driver memory first; a BI tool that issues an unlimited SELECT * is otherwise a reliable way to exhaust the driver heap. Beyond that, treat it as multi-tenant infrastructure: size the driver generously, put a query timeout in front of it, and run separate Thrift servers for workloads that must not affect each other. For newer deployments the decoupled client-server path in Spark Connect addresses the same need with per-client isolation.

Functions and where UDF registration fits

Spark ships a large built-in function library, and it is worth searching before writing anything custom: SHOW FUNCTIONS lists what is available and DESCRIBE FUNCTION EXTENDED regexp_extract prints the signature with examples. Built-ins are known to the optimiser, participate in constant folding and pushdown, and are compiled into generated code. A user-defined function is none of those things - it is an opaque call the planner cannot see inside.

# session-scoped: visible in SQL for this SparkSession only
spark.udf.register("normalize_sku", lambda s: s.strip().upper() if s else None, "string")
spark.sql("SELECT normalize_sku(sku) FROM raw_events")

Registration through spark.udf.register is session-scoped and unpersisted: it is gone when the session ends, and it is invisible to a JDBC client connecting to a Thrift server that never ran your registration code. That asymmetry is the practical limit of UDFs in a SQL-first deployment - the analysts using SQL cannot see functions defined by the engineers using the API unless something registers them at startup for every session. Permanent functions backed by a JAR can be declared in the catalog with CREATE FUNCTION, which stores the class name and JAR location in the metastore so any session can resolve them, at the cost of getting that JAR onto the classpath everywhere.

The cost profile also differs by language. A JVM UDF is a method call inside the executor. A Python UDF has to move rows out of the JVM into a Python worker and the results back, and that boundary is usually the dominant cost of any query containing one. Prefer a built-in, then a JVM UDF, then a vectorised Python UDF, and treat a plain row-at-a-time Python UDF inside a large scan as something to justify.

CACHE TABLE, SET, and the SQL control surface

A useful amount of Spark's runtime control is reachable from SQL, which matters when SQL is all a client can send.

CACHE TABLE orders materialises a table or view into the cluster's storage memory. Note the semantic difference from the API: the SQL form is eager by default and blocks until the data is cached, where df.cache() is lazy and does nothing until an action runs. CACHE LAZY TABLE orders restores the deferred behaviour, and CACHE TABLE hot AS SELECT ... caches a query result under a name in one statement. UNCACHE TABLE releases one entry and CLEAR CACHE releases all of them. Whether caching is worth the memory it takes from execution is a separate question, worked through in the Spark overview.

SET spark.sql.shuffle.partitions = 400;   -- runtime, this session
SET spark.sql.shuffle.partitions;         -- read the current value
RESET spark.sql.shuffle.partitions;       -- back to the default

CACHE TABLE dim_customer;                 -- eager
ANALYZE TABLE sales.orders COMPUTE STATISTICS FOR ALL COLUMNS;

Not every configuration key is mutable this way. Spark separates static SQL configuration - spark.sql.warehouse.dir, spark.sql.catalogImplementation and others fixed when the SparkSession is built - from runtime configuration, and a SET against a static key raises an error telling you it cannot be modified. If a documented setting appears to have no effect after a SET, check whether it is static before assuming Spark ignored you.

SQL comment hints such as /*+ BROADCAST(dim) */ are also part of this surface; what the planner does with them, and why a hint that was correct last quarter is a liability this quarter, belongs to the optimizer article. ANALYZE TABLE is how table and column statistics get into the catalog in the first place, which is the precondition for any cost-based decision.

What Spark SQL is not

Spark SQL speaks a large dialect and supports real DDL and DML, which makes it easy to assume it is a database. It is a query engine over storage it does not control, and the gap shows up in specific, predictable places.

There are no indexes. Nothing in Spark corresponds to CREATE INDEX. Selective access comes from physical layout instead - partitioning so that irrelevant directories are never listed, sorting or clustering so that file-level min/max statistics let whole files be skipped, and bucketing so that a join can avoid a shuffle. A query that would use a B-tree in a relational database will scan in Spark unless the data was laid out for it in advance.

There are no enforced constraints. Primary keys, foreign keys and check constraints are not validated by the engine at write time. Whatever integrity your tables have comes from the pipelines that write them or from an expectations framework running alongside, and a schema that declares a key guarantees nothing about the data behind it.

There are no multi-statement transactions. There is no BEGIN and no ROLLBACK. Atomicity exists only at the granularity of a single write, and even that depends on the commit protocol underneath - a job that fails partway through a plain directory write on object storage can leave partial output behind.

UPDATE, DELETE and MERGE INTO are conditional. They parse against any table but execute only against a table whose connector implements row-level operations. Point them at a plain Parquet directory and the analyser rejects them. This is the single most common surprise for someone arriving from a warehouse background, and it leads directly to the last section.

None of this makes Spark SQL a weaker tool; it makes it a differently shaped one. It scans enormous volumes cheaply, joins across heterogeneous sources in one statement, and scales by adding executors. It is a poor choice for high-concurrency point lookups, for anything needing sub-second latency on a single row, and for a workload whose correctness depends on the engine refusing a bad write.

Where a table format takes over

Most of the gaps above are closed by putting a table format underneath the SQL rather than by changing the engine. Delta Lake, Apache Iceberg and Apache Hudi all do the same structural thing: they add a metadata layer over the files that records which files constitute the table at a given version, and they implement enough of the DataSource V2 catalog contract that Spark SQL can drive them with ordinary statements.

CREATE TABLE sales.orders (id BIGINT, amount DECIMAL(12,2), dt DATE)
USING delta
PARTITIONED BY (dt);

MERGE INTO sales.orders t
USING staging.orders_new s
  ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.amount = s.amount
WHEN NOT MATCHED THEN INSERT *;

What arrives with that USING clause is concrete: snapshot isolation, so a reader never sees a half-finished write; atomic commits, so a failed job leaves the table at its previous version instead of half-updated; schema enforcement on write; time travel to an earlier version; and row-level UPDATE, DELETE and MERGE. The SQL you write barely changes - the table name is the same - which is precisely why the seam is easy to miss until a MERGE fails on a table someone created without the format.

The costs are equally concrete. The metadata layer needs maintenance - compaction of small files, expiry of old snapshots - and untended metadata degrades planning time. Concurrent writers resolve conflicts optimistically, so two jobs touching the same partitions can leave one of them to retry or fail. And the catalog plumbing has to be right: the format's session catalog registered under spark_catalog, its extensions configured, and its JARs matched to your Spark version. The internals behind these guarantees are developed in Delta Lake and Iceberg.