Inside OcientAIQ™: The Spark connector that pushes work down instead of pulling data up

By Jason Arnold, Co-Founder, Director & Distinguished Engineer

Welcome back. In a recent post in this series, we compared the OcientAIQ Unified Data Platform’s graph library against Spark and showed what happens when you move the computation to the data instead of the data to the computation.

This time I’m staying close to Spark but flipping the relationship. This post doesn’t say “stop using Spark and run it in OcientAIQ.” It says “you’re already using Spark, so here’s how our connector keeps the boundary between the two as thin and as parallel as possible.”

This is the one post in our series where the star is not SQL syntax or an analytics API. It is a piece of integration plumbing: the OcientAIQ Spark connector. Plumbing is not glamorous, but a unified data platform lives or dies by how well it talks to the rest of the stack, and most Spark connectors are thinner than people realize. I want to walk through what ours actually does, line up the parts that are different from a generic JDBC bridge, and be precise about why each one matters at scale.

If you stop reading here, you have a working connector.

It is a catalog, not just a reader/writer

The first thing that separates this from a generic JDBC source is that it implements Spark’s DataSource V2 CatalogPlugin, not just the read and write interfaces.

Register it as a catalog.

Now OcientAIQ is a first-class catalog inside Spark SQL, so you can do the below.

Under the hood, listTables, loadTable, createTable, alterTable, and dropTable are all backed by information_schema queries and real DDL against the cluster. The connector is careful about the distinction between a base table, a view, and a system table, because Spark’s behavior differs for each, and it resolves a one-part name like events against your default schema by asking the engine for current_user. A generic JDBC source gives you spark.read and spark.write and nothing else; you cannot create a table, you cannot list a schema, and DROP TABLE is not a concept. Making OcientAIQ a catalog means your existing Spark SQL, including DDL, just works against it.

Pro Tip: Because it is a real catalog, schema discovery is cached per (url, schema, table) and explicitly invalidated whenever the connector creates or drops a table. So a CREATE TABLE followed immediately by a read will not hand you a stale schema, but steady-state reads do not pay a metadata round trip every time.

The connector compiles it into something close to the below.

The filtering, grouping, and ranking happen inside OcientAIQ, on Ocient’s storage indexes and columnar layout, and only the surviving rows come back to Spark. The amount of data crossing the connector is the size of the answer, not the size of the table.

One detail worth pulling out, because it is the part that keeps the answers correct. The connector tells Spark that every pushdown is partial. When a scan is split into many partitions (more on that next), each partition runs its own LIMIT 10 or its own partial aggregation, and Spark must re-apply the global LIMIT or the final aggregation after collecting the partition results. Telling Spark “I fully handled this” when the work is spread across partitions would give wrong answers. So we push the work down to shrink each partition’s output, and let Spark finish the global step. Correct first, fast second.

This is my favorite part, and it is the part that is genuinely hard to find elsewhere.

A typical partitioned JDBC read in Spark works like this: you give it a numeric partitionColumn, a lowerBound, an upperBound, and a numPartitions, and Spark slices the range into N WHERE col >= x AND col < y queries that run in parallel. Our connector does that too, but it adds three things on top.

It discovers the bounds for you, and it does so after pushdown. If you give it a partitionColumn and numPartitions but no bounds, it runs a MIN/MAX query to find them. Critically, it runs that bounds query with your pushed-down WHERE filters already applied. If you are reading one day out of a ten-year table, the partition bounds are computed over that one day, not the whole decade, so the partitions are tight instead of mostly empty. And if you set ocient.minRowsPerPartition, it will also count the rows and dial numPartitions down when you asked for more partitions than the data can usefully fill, so you do not end up with a swarm of empty tasks.

It spreads partitions across the SQL nodes of the cluster. An OcientAIQ cluster has multiple SQL nodes. The connector asks the JDBC driver for the full list of them (via the driver’s secondary interface discovery), and assigns partition i to node i mod numNodes, rewriting the JDBC URL so that partition actually connects to that node and pins there with the driver’s force flag. So you are not hammering one coordinator with N parallel queries; you are spreading N parallel queries across the whole front end of the cluster.

And it gives you a knob for how parallel each partition’s query is inside its node. The connector appends USING PARALLELISM N to every read query. It is worth being precise about what that clause does, because it is the opposite of what people assume: USING PARALLELISM does not turn parallelism on, it caps it. Left to its own devices, an OcientAIQ query runs at the system’s full configured parallelism. The clause is how you ask for less than that, which the engine itself does for small internal lookups (a SQL node or loader node lookup runs with USING PARALLELISM 1 because spinning up the full machine for a tiny catalog query is wasteful). The connector exposes that same dial through the ocient.parallelism option. The default is 1, which is deliberately conservative: each Spark partition’s query runs single-threaded inside its node, and you scale by adding Spark partitions. Raise ocient.parallelism and each partition’s query opens up to use more of its node, so you can tune the balance between many small partitions and fewer wide ones for your cluster and your table. (Catalog and bounds-discovery queries are left alone and run at the engine default.)

There is a resilience payoff to the node assignment too. Each partition reader is handed the full node list and a starting index. If the assigned node is unreachable, the reader rotates to the next node, and it blacklists the failed node for an hour so subsequent partitions in the same job do not keep retrying a dead endpoint. A single node hiccup degrades throughput slightly instead of failing the job.

Pro Tip: If you do not pass a partitionColumn, you get a single-partition read, which is fine for small results but leaves all that parallelism on the table for large ones. For big extracts, pass a partitionColumn (a high-cardinality numeric, date, or timestamp column works best), set numPartitions, and let the connector auto-discover the bounds. Then decide whether you want throughput from many Spark partitions, from a higher ocient.parallelism per partition, or a mix of both.

Writing is where the connector does the most work that you will never see, and it is the capability I would point to first if someone asked what makes it different.

A vanilla Spark JDBC write opens one connection per executor task, batches some INSERTs, and commits each task independently. If the job fails halfway, you are left with a partially written table and no clean way to roll back, because each task already committed its own slice. You write a dedup or a staging-table dance to work around it.

Our writer does three things differently.

It distributes writers across SQL nodes, the same way reads are distributed. The batch write fetches the sorted list of SQL nodes once, and each writer task connects to node partitionId mod numNodes with force=true. So a write with 16 Spark partitions against a 4-node front end puts 4 parallel writers on each SQL node, not 16 writers piled onto one.

It makes the entire multi-task write one atomic Ocient transaction. This is the piece that needs engine support, and OcientAIQ has it. When the batch write starts, it mints a job-level UUID and hands it to the driver as a transaction group prefix. Each writer task then derives a task-scoped UUID under that same job prefix. The engine understands this structure: a 128-bit UUID partitioned into job, task, and statement-sequence segments, with a configurable prefix length, so the cluster can recognize that sixteen writer connections spread across four nodes are all parts of one logical transaction. When every task has finished its batches, a single job-control connection commits the whole thing, and if anything fails, the whole thing rolls back, across every task and every node. There is no half-written table. The Spark write either lands completely or not at all.

It has two write paths, and the fast one partitions work across loader nodes too. Each writer task builds rows through pre-compiled per-column converter functions (created once per task, then a tight per-row loop), and how those rows reach the table depends on the path the JDBC driver picks.

The default path is a multi-row INSERT. The driver groups same-shape parameterized INSERTs together and rewrites them into combined INSERT ... VALUES (...), (...), ... statements, chunked so each statement carries a bounded number of rows, and sends them over the writer’s SQL-node connection. This is already a big win over one-row-per-round-trip JDBC, and it is what every write gets for free.

The second path is bulk load, and it is where the work fans out across a whole additional tier of the cluster. Set enableBulkLoad=true, and once a writer’s batch crosses a row threshold (25,000 rows by default) the driver switches strategy. Instead of pushing rows through the SQL node, it serializes the batch into JSON chunks of 60,000 rows each, and stages those chunks for the cluster to ingest through a temporary pipeline (CREATE PIPELINE jdbc_bulk_<uuid>) that loads the files, monitors itself to completion, and cleans up afterward.

There are two staging modes. In the default SFTP mode, the driver discovers the cluster’s active streamloader nodes from the system catalog and picks one by round-robin, a deterministic rotation that spreads consecutive batches evenly across the loaders rather than piling onto one (with the same one-hour blacklist on a node that fails to accept a connection), and ships the chunks to that node over SFTP. In S3 mode, the chunks go to an object store that the cluster reads from directly, so no specific loader node is pinned. That object store does not have to be AWS: the S3 client takes a configurable endpoint and path-style access, so any S3-compatible store works.

If a bulk load fails for a retriable reason, the driver by default falls back to the multi-row path so the write still lands (you can opt out of that fallback if you would rather the batch fail loudly). So in SFTP mode a single Spark write is parallel across two tiers of the cluster at once: writer tasks spread across the SQL nodes by the connector, and each task’s batches spread across the loader nodes by the driver.

Putting a write together.

Sixteen partitions become sixteen writers, spread round-robin across the cluster’s SQL nodes and all enrolled in one transaction. With enableBulkLoad=true, each writer’s batches are staged as chunked files (across the streamloader nodes in SFTP mode, or to an S3-compatible object store the cluster reads directly) and ingested through temporary pipelines, so the write fans out across more than just the SQL tier, and the whole thing commits together at the end.

The series has spent whole posts on arrays, tuples, and the rich type system, so it would be a shame if all of that flattened to strings the moment it touched Spark. It does not.

The connector’s dialect maps types in both directions, and it goes well past the primitives:

• Ocient arrays become Spark ArrayType and back, recursively, so an array of arrays or an array of tuples round-trips.

• Ocient tuples become Spark structs and back. To make this work cleanly, the JDBC driver in “spark mode” reports a TUPLE column to Spark’s reader as a STRUCT, both in the type name and the JDBC type code, so Spark’s own catalyst type inference does the right thing instead of choking on a type it has never heard of.

• Spark maps are stored as an array of two-field tuples and reconstructed on read.

• Spark ML Vector and Matrix types (the VectorUDT and MatrixUDT from org.apache.spark.ml.linalg) map onto Ocient’s native MATRIX type. The connector detects the exact underlying struct shape of those UDTs and converts a Spark dense or sparse vector into an Ocient matrix on write, and back into a Spark dense matrix on read, handling the row-major to column-major flip that Spark ML expects. If you are doing feature engineering in Ocient and modeling in Spark MLlib, the vectors move across intact.

• The year-month and day-time interval types and TIMESTAMP_NTZ (all on the Spark 3.5 baseline) and Spark 4’s Variant are all handled, with Variant reached through the reflection mechanism so it works on Spark 4 without breaking the Spark 3 build.

For the complex Spark types that do not have a one-to-one Ocient column type, the connector writes a TYPE HINT onto the column at create time that records the full Spark catalog type string, and reads it back to reconstruct the exact Spark type later. So a deeply nested STRUCT<a: STRING, b: ARRAY<INT>> you write from Spark comes back as the same Spark type, not as an approximation.

Pro Tip: This is the same philosophy as the rest of the engine that we have covered in this series. The type system is first-class, so the connector’s job is to preserve it across the boundary rather than degrade it. If you have been using arrays and tuples in your Ocient tables, they are usable from Spark as arrays and structs without a translation layer you have to write yourself.

I want to be careful and fair here, the same way the graph post was. Plenty of databases ship a Spark connector, and the better ones do real pushdown. I am not going to pretend everyone else returns SELECT *. But there are a few patterns I see repeatedly that are worth naming, in the order they tend to matter:

• Most connectors are a read/write source, not a catalog. You get spark.read and spark.write, and table management lives somewhere else. Implementing the V2 CatalogPlugin so that CREATE, DROP, ALTER, and SHOW TABLES flow through Spark SQL into the warehouse is less common, and it is the difference between “Spark can query this database” and “this database is a catalog inside Spark.”

• Treating the database as one endpoint is the norm. A connector that fans out across Spark partitions but funnels every partition through a single coordinator is leaving the cluster’s front end mostly idle. Spreading partitions across all of the cluster’s SQL nodes, exposing the engine’s own per-query parallelism as a tunable dial, and partitioning bulk writes across the loader tier as well are things most bridges do not do.

• Per-task-commit writes are the default almost everywhere, because cross-executor atomicity needs the database to understand that many connections form one transaction. Without an engine-level transaction-grouping mechanism, the connector cannot promise all-or-nothing, so it does not. The job/task UUID prefixing that makes our write atomic is engine support that has to exist on both sides; it is not something a connector can fake.

• Type fidelity erodes at the edges. Flat primitive columns are easy. Arrays, structs, maps, ML vectors, and variant data are where connectors quietly downgrade to strings or JSON blobs. Preserving them losslessly, in both directions, is work, and it only pays off if the engine has the rich types to map onto in the first place.

None of this is a knock on Spark, which is doing exactly what it should: pushing as much as the data source is willing to take and handling the rest itself. The point is that the source has to be willing to take a lot, and has to give Spark accurate signals about what it took. That contract, full pushdown, work spread across every tier of the cluster, atomic distributed writes, lossless types, is what the connector is for.

The OcientAIQ Spark connector is a DataSource V2 implementation, published for both Spark 3.5+ and Spark 4.x from one codebase, that tries to make the seam between Spark and OcientAIQ as thin as possible:

• It is a full catalog, so Spark SQL DDL and table listing work against Ocient directly, not just reads and writes.

• Reads push columns, filters, aggregates, limits, and TopN down into the engine, so what crosses the wire is the answer, not the table.

• Reads are spread across the cluster’s SQL nodes, with auto-discovered, filter-aware partition bounds, automatic node failover, and a per-partition ocient.parallelism dial that caps or opens up how much of each node a partition’s query uses.

• Writes are distributed across SQL nodes, node-pinned, and atomic: an entire multi-executor DataFrame write is one Ocient transaction via the engine’s job/task UUID transaction grouping, with a bulk-load path that stages each writer’s batches (across the loader nodes over SFTP, or to an S3-compatible object store) and ingests them through temporary pipelines.

• The rich type system survives the round trip: arrays, tuples as structs, maps, Spark ML vectors and matrices onto Ocient’s native MATRIX, plus variant, intervals, and TIMESTAMP_NTZ, with a TYPE HINT mechanism that reconstructs exact nested Spark types.

If you already have a Spark stack and an OcientAIQ cluster, the connector is the part that decides whether moving data between them is a tax or a non-event. Pull com.ocient:ocient-spark3_2.12 or com.ocient:ocient-spark4_2.13 from our Maven repository, register it as a catalog, and point a partitioned read at one of your large tables. Watch the compiled query in the logs and you will see your filters and your column list and your USING PARALLELISM right there in the SQL. The data that comes back is the data you asked for, and nothing more.