The Future of DSv2 in Apache Spark™

Summary

• DSv2 architecture benefits: Shifts connector complexity to Apache Spark, improving reliability and enabling features like catalog federation, MERGE operations, and aggregate pushdown.
• DSv1 limitations: The original data source API lacks abstractions, pushes complexity to connectors, and has inconsistent behavior across implementations.
• Advanced functionality unlocked: DSv2 enables catalog federation, storage-partitioned joins, stored procedures, and modern DML operations not available in DSv1.

The Future of DSv2 in Apache Spark™

Talk by Anton Okolnychyi, Senior Staff Software Engineer, Databricks
DSv2, Spark's next-generation Catalog API, is gaining traction among data source developers. It shifts complexity to Apache Spark™, improves connector reliability and unlocks new functionality such as catalog federation, MERGE operations, storage-partitioned joins, aggregate pushdown, stored procedures and more. this video covers the design of DSv2, current strengths and gaps and its evolving roadmap. It's intended for Spark users and developers working with data sources, whether custom-built or off-the-shelf.
Here's more to explore:
Databricks named a leader in the 2024 Gartner® Magic Quadrant™for Cloud DBMS: https://www.databricks.com/resources/analyst-paper/databricks-named-leader-by-gartner
An open, unified approach to your data, BI and AI workloads: https://www.databricks.com/product/databricks-sql
See all the product announcements from Data + AI Summit: https://www.databricks.com/events/dataaisummit-2025-announcements

FAQs

What is the main difference between DSv1 and DSv2?

DSv1 is the original data source API from Spark 1.3 that uses relation providers, while DSv2, added in Spark 3, revolves around catalog plugins and shifts complexity from connectors to Spark itself.

What problems does DSv2 solve?

DSv2 addresses DSv1's lack of abstractions, eliminates the burden of complex low-level APIs on connector developers, and provides consistent behavior across implementations rather than connector-specific interpretations.

What new capabilities does DSv2 enable?

This video covers DSv2's support for catalog federation, MERGE operations, storage-partitioned joins, aggregate pushdown, and stored procedures—advanced features not available in DSv1.

Who should watch this video?

This video is intended for Spark users and developers working with data sources, whether custom-built or off-the-shelf.

Full transcript

Uh let's get started. Uh thanks for coming today. Um. I'm a software engineer at Data Bricks. If we ever met in any of the open source community, make sure to say hi after the session. It's amazing to connect with people in person, especially after you worked for a while in in u those communities. Today we'll focus on DSV2 in Spark. I will cover why it exists, what's available in 4. 0, and what the community is working on right now.
So what do we actually mean by the data source API? Essentially, this is an SPI that allows developers to plug in custom code for reading from and writing to external storage systems from Spark. The primary goal is to abstract how the data is physically stored from its in-memory representation in Spark. So if you ever created a file or read a file using Spark, you interacted with either a DSV1 or DSV2 uh connector. When I say DSV1, I refer to the relation provider and all of the interfaces and classes around it.
Um such as B relation. This is the original data source API in Spark that's available from 1. 3. DSV2 on the other hand was added much later in Spark 3 and it revolves around catalog plugin. So if you implemented a catalog plug-in or you work with one, you use DSV2. So what's wrong with DSV1? Why did the community decide to rethink how connectors should integrate with Spark? In my view, the biggest problem with DSP1 is the lack of abstractions.
In order to create a DSP1 connector, you have to work with very low-level APIs and depend on Spark internals. Essentially, it pushes a lot of complexity into the connector itself. To make things worse, it also pushes a lot of decisions into the connector so that different connectors can interpret the same concepts differently. An example of that is the famously inconsistent overwrite mode. Depending on which connector you used, you would have a completely different behavior.
Finally, it was really hard to evolve. That's the reason why you don't have the proper notion of the catalog. you don't have any advanced DML or things like optimized rights that are available in in modern spark. So all of those reasons motivated the community to create DSV2. And before talking about its future, I want to quickly recap what's available today. One of the features that's unique to DSV2 is the ability to use partition transforms.
If you worked with DSV1, then the only way you could partition is by providing data columns and that were part of your table schema. That's no longer the requirement in DSV2. You can actually partition by arbitrary expressions as long as your connector supports them. Iceberg, for instance, uses this to implement hidden partitioning. DSV2 also supports atomic create table as select and replace table as select. The atomic bit is really important because it's about reliability.
It's about isolation. So you either see the new table definition and all of the data that was supposed to be there or nothing at all. So no intermediate steps, no partial results. You only see everything once it's completed. So this is really important for isolation and making sure you're not leaking anything into your downstream jobs. There are lots of things to make batteries efficient. So you have columner and row oriented reads.
You have predicate um limit aggregate push down. You have schema pruning. You have different types of runtime filtering. Connectors can also expose stats and then Spark can use those stats to do costbased optimization. So a lot of community of effort went into making sure that the batch reads are efficient. DSV2 also supports time travel. So you can either specify the version of the table that you would like to query or a timestamp that would represent the point in time as of which you would like to query the table.
You have both the SQL and the data set APIs. DSV2 supports batch appends, dynamic partition overrides and overrides with a condition. You also have structured streaming support so you can stream in and out of your table. Spark provides you all the tools to implement exactly once end to end guarantees as well. A lot of the community effort went into allowing connectors to output properly sized, properly clustered and properly ordered data.
So specifically a connector can request spark to distribute the data in a particular fashion before it passes that data to the connector. So clustering is about how you split your data across different tasks and then on top of that you can also request how you want to order that data within each task. Connectors can control both the distribution and the ordering of the data before it's being passed um into the right connectors can also control the size of that task.
Right? They can indicate what uh what's the target size of the tasks they prefer. Previously, it was on users to pick the right number of shuffle partitions to make sure that your rights are having enough parallelism but not generating uh two too small files and was really hard because like you didn't necessarily know all the table config and whatnot. Today it's the connector responsibility that has a little bit more information and like users are no longer required to do that themselves.
Uh, Spark would also uh once it knows the target task size, it would try to split whatever is skewed and then coales the tiny tasks to try to meet that target task size that the connector provided. Another big one is storage partition joins. It's a more it's a lot more powerful and more generic implementation of the DSV1 bucketed join. It's no longer limited to bucketing. You can use it with any clustering. And the idea is still the same though.
Like if you're doing a join and both sides are clustered in a compatible way and the word compatible here is important. So it doesn't have to be exactly the same. It has to be compatible. Uh then you can skip the shuffle. So this is really important if you're running at scale. And Spark will try to be smart about it. So it will try try to make the clusterings compatible at runtime. So sometimes it may um roll up some of the some of the portions of the data.
Sometimes it may actually add empty task on the other side. So it will try to make the cluster incompatible at runtime. It will also address any skew. So it will be able to split one side and then replicate the other so that you're no longer like limited by the time it takes to process um like your hotspot. DSV2 also comes with built-in support for delete, update, and merge. You have two execution modes. You can either ask for rowle changes.
That's what you want to have if you're working with deletion vectors in Delta or iceberg. or you can ask for group level changes. That's what you need if you're rewriting entire files. This is what's called copy and writing iceberg. All of the these advanced DML commands have uh subquery support. So basically delete update have subquery support. Merge kind of has that by definition. You have ro lineage. You have the same optimized rights that I just covered.
And you also have SPJ storage partition joints. If you can benefit from SPJ and hash join, your mergers will be shuffless and sortless. That's the key how you can scale that to tens of pabytes in in a single job. Spark 4. 0 also comes with built-in support for DSV2 store procedures. So connectors can now expose predefined routines to spark and those can be called uh through SQL. A lot of connectors use store procedures for table maintenance and uh I believe the originally the idea was borrowed from trino that did it for hive tables but today it's been done for hoodie for iceberg.
The problem though that both Iceberg and Hoodie used to have their own Spark SQL extensions to make it work. So you'd have to define your own parser rules, logical physical plans, your own strategy. Everything was very fragile and then like was really really messy. Um that's no longer needed because Spark has built-in support for store procedures and it's a lot more powerful than either of those connectors had before.
There are a lot more other features that we won't be able to cover for the sake of time, but the point that I was trying to make is that DSV2 today is already a very comprehensive API and it's production ready even though it is evolving. Now, let's talk about what's coming next. One of the things that the community is working on is default values. Instead of always providing an explicit value for a column, you would be able to assign an expression.
Spark would evaluate that expression and pass back the value every time new data gets added into the table. Default values are extremely handy if you are evolving the schema. So instead of simply returning null for all of the records that existed before, you can now have a reasonable initial default. The way we do this in Spark matters as well. There are two requirements. First, it must be it must follow the SQL standard.
Second, it must be portable. So if you use Spark to create a delterise brick table, the table happens to have default values. they have to be portable. So other engines like Trino or Flink should be able to still benefit from that. That's why you have the column default value API in DSV2. It essentially provides three different things. The first one is the original SQL string that was provided by the user. Obviously that's in Spark SQL dialect and it's helpful mostly for debugging purposes.
Connectors cannot depend on them directly because it's a black box. You don't know what's inside. you can't evaluate it. You don't know if it's strictly ANC behavior or is it like really something specific to Spark. Therefore, column default value also provides you a DSV2 expression that represents that default value. So, it's a structural representation that connectors can traverse, explore, inspect, validate, and maybe convert into something that they understand and then can evaluate if needed.
Um the SV2 expressions are also called public expressions in Spark. Uh they have stable definitions and they strictly follow the SQL standard. Um so by definition they are portable. Whenever Spark converts from catalyst to DSV2, it will also normalize these expressions. So in this example, the user provided default value or defined default value is now which is a synonym for current time stamp. now is not part of the SQL standard but current current time stamp is.
So once the conversion happens, Spark would also normalize these. And finally, you have what's called the initial default. That's always a literal that represents the value of the expression um at the point when the column was added to the table. You can change the right default, but you can never change the initial default after the column is added. This is what's used to backfill all of your previous data. The back fill can be lazy, so you can still insert it every time you read it, but that's the value that you always have to insert for all of the records that existed before.
Another feature that the community is working on is check constraints. So check in constraints prevent invalid states or invalid values from being inserted into the table so that you don't have any negative prices, any invalid statuses, anything like that. The important part here is that Spark would be able to fully enforce and validate these, right? So it's not just syntax, it's actually functionality in Spark to make sure that you never end up with scruped data.
Every insert, every update, every merge, these check constraints will be validated. So users would be able to define a check constraint next to the column definition if it covers just one column or um at the end. So technically the only difference between a table and the column constraint is the number of columns that you reference. From the API perspective, it's all the same. There would be also a possibility to add check constraints to existing tables and Spark would again follow the SQL standard and would do a validation of all of the data that's currently in the table to ensure that all of the data that's today in the table meets the check condition and also the way it's being wired between the connector and spark would ensure that it's atomic as well.
So if there are any concurrent jobs that write into the table while you're doing the validation connectors have all the information to make like uh the proper choice. If this operation would succeed then you know that despite all of the things that are happening concurrently the check constraint is valid. Uh we also added a bunch of informationational constraints in particular the primary key unique and foreign key. These areformational only unlike check spark will not enforce them but it will use them for query optimization purposes.
Um the way we designed this is also kind of important because the DSV2 APIs have a way to represent two states of constraints. In the first state we will know that a system did the validation and knows that the constraint is true. That's guaranteed to be true. And the other state is the system did not do the validation but the user thinks that the constraint is true. Those are two different levels of guarantees. Both of them are exposed to spark.
So spark can make the choice that um web to optimize a query or not right whether like on like how you configure the session. In the future, Spark may be extended to also do the validation, but it's really hard to imagine like ensuring the primary key um at the big scale every time you write it. So maybe it can be done something that you would do periodically if you will. Another one that I'm really excited about is built-in support for change data capture.
So CDC is all about tracking and exposing role level changes. Um it underpins use cases like incremental consumption, efficient materialized view refresh, auditing and others. So far table formulas like delta and iceberg had their own ways of doing CDC and honestly both of them were a bit clunky. The reason for that they were trying to work around the absence of the built-in functionality for CDC and Spark. Um the Spark community is committed to fixing that.
Uh so we're designing the proper DSV2 APIs that would allow connectors to both capture and expose um role level changes. Spark would also have the logic for squashing the changes across the sequence of transactions um and presenting like a net view of what changed between point A and point B in the table history. I'll walk through an example just in a second. Spark would also be able to reconstruct pre and post update images from the raw deletes and inserts that delta and iceberg capture.
Um there is no concept of update in delta and iceberg. Update is simply a delete followed by insert. So you need some logic that would reconstruct those pre and past images um after the fact. users would have SQL and data set APIs for doing CDC and we're really hoping that we would support both structured streaming and batch. This is something that's still being discussed in the community at a very early stage but it will look like something like this.
So imagine you have a table called orders. Um there are three columns email price and status and let's assume there are just two records in that table. This is that we're in initial state. There is record A that's pending and there's record B that's um with status paid. In order for CDC to work, you need a few mated columns. The first one is called row ID. Row ID needs to uniquely identify every record in the table. This can be either like a primary key that uses data columns or it can be a synthetic synthetically generated row ID as well.
Doesn't really matter as long as it uniquely identifies every record in the table. Row ID must be assigned when the data is first inserted into the table and it must never change. If you update columns or do anything to the row, row ID is preserved. It stays as it was when it was inserted. The second column that's important is called row version. It's essentially a relative version of the row. And unlike the row ID, it actually changes every time you update the row.
So if you update couple of records, you must update the row version as well. Um it does not change if you take one row and do compaction and then you move it to another file. In that case, there is no change. So the row version has to be preserved. Let's imagine there is a sequence of transactions. So let's say we have an update that modifies record A and sets the status as paid. Then we have an insert that basically adds one fresh record C to the table so that we have three different records.
And then we have one more update that again updates record A. Right. So basically we have two updates. Both of them modify record A and then we have in between them one insert that adds record C. If we ask Spark for changes across these three transactions, right? Then we would get something like that. For the first transaction, you would have a delete and insert because again there is no concept of update in Delta and Hisburg.
So you would have two records in your change log. One delete, one insert. You see that the status column is different, right? And the row version is updated as well. Row ID remains the same. The row version changed, right? Because the record was updated. For the second transaction, it's just an insert. So there's one um entry in the in the change log. For the third transaction, we again have a delete and insert because that was an update.
So overall if you ask for row changes then you would have something like this where you would have four records for record A and then one for record C. There is nothing for record B because it was not modified right? So that's why it's not in the change log. This is the view of the CDC log that you would want if you want to explore every change that happened in your table. But the problem was that so like if you're doing auditing this is good but if you're really interested in what changed then this is not helpful because it has those intermediate steps as well.
So that's where the squashing comes in. So Spark would be able to squash these changes across these three transactions and remove all intermediate states. So that the output would look something like this where you got rid of all of the intermediate states and you just present okay this row was like this. It now looks like that. So you you go directly from pending to refunded for insert there's just one was one. So there's nothing to squash.
So this is the view of the data that you would want if you're doing like an incremental refresh. You're only interested in the state that was before and in this final state that you want to like replicate. you want to distinguish between a true delete, a delete as that was part of an update, right? So you want to reconstruct your updates and that's possible as well. So, Spark would be able to do that for any connectors that choose to support CDC this way.
Um, and basically our delete and insert is will be turned into update pre-image and update post image. If it was a true delete, we would still keep it as as this is true delete. There's also a lot of work planned for delete, update, and merge. big focus is isolation and reliability. So today if you execute a merge statement, Spark would remember the version of the table that was scanned and the predicates that were used and would pass that back to the writer.
So writer can do the validation to make sure that nothing that happened concurrently kind of violate the isolation level. The problem today that Spark does not inspect the sub queries. So if there are let's say if you query three tables in your source of the merge spark does not like it's blind spot for spark it does not keep track for whatever changes there right so like in technically that if that impacts the isolation of merge as a whole if you want to have a truly serializable isolation for a merge as a transaction you have to address that so basically the community is working on that so that merge actually behaves like a multi-table single write transaction and you can configure it to be either snapshot isolated or serializable.
We're also doing a bunch of work for metrics and assertions to make sure that we're not corrupting any table. That's something that uh we used to have at Apple a little bit. I know that a lot of folks in the Delta community have a lot of robust checks. So we're partnering together to make sure that the DML and Spark does not corrupt any data and we're kind of sure that if anything goes wrong, we have ways to debug where exactly went wrong.
Um there's a bunch of other like performance optimizations that we plan also like automatic schema evolution and and everything. So that's something that multiple communities are working on and collaborating in Spark. We also committed to finishing the view catalog. The view catalog is the API has been merged like maybe a couple of years now but there is no implementation but we committed to finishing that. So essentially this is really powerful. would allow you to plug in views that are like defined by other matter stores, not necessarily HMS and uh that maybe under the hood use a different representation for views like for instance ice views that can be cross cross engine views as well.
So I'll finish with a quick summary before we open for questions. The first and foremost point that I was trying to make today is that DSV2 is not only the future but al also the present. So it it is really powerful today as it stands even though it's evolving. And if you're plan on building a connector or you have one, I would strongly encourage you to consider migrating. The beauty of the SV2 is that it has higher level APIs and it shifts the complexity into Spark.
So if there is something that you want to have in your connector, but there is nothing like in Spark to support that, please don't just build it in the connector, maybe work with Spark. Let's make it a little bit more generic to make sure it works for other connectors and then work as a community because so far we've seen it many times, right? It's more or less the same logic, slightly different implementation, slightly different behavior.
Users have to learn all those differences. So hopefully with DSV2 we can move all of that complexity to Spark so that users have unified behavior across connectors. And the final point that I'm kind of really excited about is that the Delta community decided to migrate to DSV2 as well. So they are in the process of doing this migration for Delta Spark. to Delta Spark would be on DSV2 as well and essentially three communities would be working together iceberg, Delta Spark and we would encourage everyone who's interested in this space to collaborate with us.
That's about it. Thank you.

Learn more about the Databricks Data and AI platform.