Sneak Peak Of DataFusion Presentation! Iceberg writes in DataFusion Comet

[00:00:00]Hello everyone and welcome back to the channel. I'm sorry for uh the weird lighting here. I'm recording this one a little bit late at night cuz I'm skipping town for the weekend. So, I got to fit this one in. Uh it has been a very smoky day in Chicago today due to the Canada wildfires. Everyone's making a really big deal about it. Personally, I've been training for a situation like this with a bunch of smoke in my lungs since college, so it hasn't really been too big of an issue for me. But basically to talk about this week's video, at some point next week I have to give a presentation on data fusion comet which I signed up for a few weeks ago thinking that I probably wasn't going to do it. Anyways, of course I am. So I decided to kill three birds with one stone in this video. One is that I had to actually make the presentation. Two is that I have to make this YouTube video and post one every Saturday. So here we are. And three is that most importantly I need to understand my AI slop that I've been writing for the last few weeks. So, let's go ahead and talk about it here. I'm going to do this presentation. It'll be a cool dry run and then we'll see if there's anything I need to improve. All righty. So, as you can see, I've got the plain old PowerPoint presentation coming up here.

[00:01:05]We're going to go through it, see how it looks. Basically, today we're going to be talking about Comet. I've spoken about it on the channel before, but basically, in order to talk about Comet, you've got to speak about Spark. Spark is pretty much the de facto batch processing framework uh that is open source, right? like it's it's pretty much everyone's go-to in their head uh when they're thinking about hey I need to run this big distributed job I have to crunch a bunch of data and uh you know ultimately I need to be able to orchestrate that job among some cluster of executives right so in spark generally speaking uh you know you've got some driver or a master process and then you've got all these worker nodes that are doing the actual execution of a bunch of rows of data for you now spark is pretty old at this point and I mean I I say old kind of satirically because it's only like I don't know 10 to 15 years old, but that's old in the computer science world. And so, uh, it is based on the JVM. Uh, it does a lot of its processing row-wise, and we're going to talk about that in a second.

[00:02:02]So, basically the gist is here, and I started to allude to this is that in 2026, it's kind of odd that we're still using Spark so much when it's doing row-wise uh, execution. So, what I mean by that is, let's imagine I issue some SQL query to Spark that is, you know, doing a couple of projections. It's doing some filtering. Internally, what Spark is running under the hood is it's maybe running some for loop that's going over a row of data, right? That might come out of some sort of iceberg table or parquet file or something like that.

[00:02:30]And for each row, it runs one operation, you know, maybe that code code was codegenerated and then it decides what to do with that row. Now, this code looks very standard and what it's what we're used to writing, you know, in our typical everyday uh back-end server side code, but it's not ultimately the best type of code to run in 2026 when you're trying to build a really high throughput data system. So, what should we be doing instead? Well, actually, the best thing to do these days is to try and execute uh you know, your computations over a bunch of column batches. We're going to go over why that is. So if you look at the code on the left over here, what you're basically going to see is that instead of doing one singular loop that's going to handle the entire row of data at a time, instead we're going to run a few different loops where each of them is going to do one of those operations. Now, at its core, you know, doing the same loop over and over again and, you know, doing one very tight loop where it's a single instruction right off the bat, that's really great for CPU branch prediction, right? When a CPU is doing the same thing over and over again, that's really good uh because it's not going to basically mispredict its instruction sets that it's going to be running. Even besides that though, there is some better performance that we can get right here. For one, I never necessarily said that we were operating on column batches just yet, but imagine that we were using something like Apache Arrow and organizing the data so that every single column of our data set was aligned contiguously in memory. In that case, when we iterate through the data like this and operate on it, what we're going to get is the ability to have really good CPU cache pre cache prefetching. We're going to pull in multiple uh elements of a data stream at once into a CPU's cache. And that way, we're going to basically have pre-cacheed the data that we need for every subsequent operation. To make matters even better, we're even going to be able to execute SIMD instructions because we're doing the same operation on the same data type multiple times in a row. SIMD is basically a CPU instruction that allows us to do uh you know single instruction multiple data meaning that we get free parallelism even on a single core. So we can get 4x or 8x uh performance improvements pretty trivially which is beautiful.

[00:04:37]That being said while I can go ahead and sit here and act like spark is antiquated and it's not that fast the truth is it's still incredibly useful.

[00:04:46]It does a ton of really nice stuff for us that is a pain in the butt to reimplement in another language whether that might be C++ or Rust or Zigg or something else. And the reason for that is that it does a lot of these things and they're battle uh you know they're battle tested they're proven and you know reimplementing them in some other language doesn't really buy us big performance benefits. I mean it absolutely does buy us big performance benefits to you know build a better query optimizer or do better adaptive query execution or something like that [snorts] but the language itself and the execution stack that we're using there is going to be less impactful than just you know building better logic. So the biggest thing the biggest reason that we don't want to ditch Spark is that a ton of people still use Spark. uh everyone including myself is still basically using Spark as the reference engine for their open open table format for example whether it's iceberg or hoodie or pyman or delta lake or anything like that people are building their tools with spark as a first class citizen and so a lot of people are building their pipelines because spark is already well integrated with any new dating data tooling that tends to come out so as a result of that uh one trend that we've seen pop up basically in the 2020s and onwards or I guess in the last few years is this concept of a Spark accelerator. If I'm already running a Spark job, people want to basically be able to provide me with some software that's going to allow me to trivially accelerate that job without having to change all my code around, worry whether I'm going to get the same output data, uh, you know, completely change the way that I deploy and monitor my system. you know, if I'm a big data engineer, I'm going to have tons of tooling already living on Spark and around that ecosystem, and I'm not going to want to change it all up. So, the idea of the Spark accelerator is to basically just go ahead and say, "All right, we're [snorts] going to keep, you know, the vast majority of Spark, whether that's going to be the scheduling, the SQL parsing logic, the way that we handle maybe floatingoint arithmetic or reg x expressions or something like that. But under the hood, we're going to make it faster." And there are many different ways that you can make Spark faster. For example, I could be using GPUs to process more data in parallel at a time.

[00:06:58]That would be a good example as far as Rapids goes. Nvidia maintains that project, but it's ultimately open source. The one closed source project on this list is Photon, which was created by Data Bricks in 2022. They have a great white paper about it though, so you can start to learn about that project.

[00:07:13]Two more that have popped up more recently are going to be project gluten which is an Apache project uh which was started by Intel is doing spark acceleration but mainly using metavlloxx which is like this embedded C++ query engine as their backing engine to try and accelerate the data movement.

[00:07:29]Finally the one that we're going to be talking about the most in this video was started by Apple called data fusion comet and that is as you can read on the slide spark on data fusion. So basically the idea here is uh you know every single Spark job gets turned into a Spark plan especially the Spark SQL jobs and the Spark dataf frame API jobs are going to be turning into common operators like projection filtering and things like that and [snorts] these accelerators are generally trying to reimplement those so that you can replace a existing spark operator with a new one and then speed up the execution time there.

[00:08:04]Okay. So at a high level what is comet's architecture? Like I said, you're basically keeping the majority of Spark, but then on the executives themselves, you have the ability to run both Java code, but then you also embed a data fusion instance in there. Keep in mind that data fusion is effectively just an embedded library that can do a lot of data processing really efficiently using rust, sim uh columner execution over arrow batches, things like that. And so because you can embed a data fusion library on these exeutors, what you can actually do is on a given exeutor, you [clears throat] can basically call data fusion functions from the Java code over JNI or you know over data fusion's bindings or something like that.

[00:08:45]Cool. So to give an example here of what this actually looks like in practice, let's imagine I am a SQL query on Spark and I look like the query plan on the left. So maybe you're going to scan parquet and because spark's internal processing format is row-wise, you're going to have to take those column batches that uh parquet is stored in and after you deserialize them from your Zstandard or your snappy compression or something like that, you are going to go ahead and convert those to Spark internal rows. You're going to maybe do a projection in Spark. You're going to do a filter. Maybe you'll do a hash aggregate. And it's going to look something like that code that we saw earlier, which is one single for loop.

[00:09:20]You basically run a code generation in order to try and inline all of these operations to operate on a row at a time and then you know do what you have to do. However, comet is a little bit different. It's going to start out with the same Spark plan that Spark generates, but then it's going to run its own logic in order to modify it to a comet plan. So after scanning parquet like Spark normally would, what comet is then going to do is say, "All right, well, I'm already dealing with column batches, right? Parquet can be pretty trivially deserialized into arrow batches in memory without doing a lot of work because arrow batches themselves are columnar. And so then data fusion can take in those arrow batches, do projections, do filters, do hash aggregations. And again that is embedded data fusion within a single node. If you need to do shuffles or you need to pass data from one executive to another, all the data fusion has to do is basically pass those arrow batches back to the JVM and let Spark do its normal shuffle operations over an opaque bite array.

[00:10:21]Why do I personally care about Comet?

[00:10:24]That's a really good question. I mean, it's really cool. Uh, but other than that, it actually impacts my day-to-day at work in so far as it accelerates iceberg workloads. Now, basically, I have uh a couple of iceberg jobs that are using multiple terabytes of RAM, and my manager is starting to get really mad at me about it, and I would prefer to speed those up, use less RAM for less amounts of time, and ideally just use less memory and less CPU in general because we're on prem and I don't want to, you know, use up my entire Kubernetes cluster for my iceberg ETL jobs.

[00:10:59]Okay, so what can Comet already do with iceberg? Well, right now it's capable of just reading iceberg data sets and running normal SQL queries on them. So, basically, Spark would plan the query like it normally would. You're going to see that there's some sort of iceberg scan operator at which point the comet code is going to take over and say rather than just doing this iceberg scan, uh, you know, Java is going to go ahead and plan that iceberg scan out and Comet is going to reflectively access all of the information about the file scan tasks that Java just planned out.

[00:11:31]So, Java is going to continue to do all of that complex planning logic. And then what it's going to do is pass down a bunch of file addresses, maybe some bite ranges to read those for down to the Rust layer. And those can be handed down to iceberg rust, which is just a library. It's a re-implementation of iceberg, but in the Rust library, so that rust code can read iceberg tables.

[00:11:53]From there, that rust code is going to read all of those files into arrow buffers and then pass them into data fusion for subsequent computation.

[00:12:01]This is actually a surprisingly simplistic approach to get this done and it works really well. Generally speaking, the planning is not the bottleneck. So, it's really not too important to make sure that iceberg rust is doing the planning. The useful part of you know this whole point of getting comet is that we have these um arrow batches that we can ultimately push through data fusion. And so if you're able to hand all of this metadata over to the rust side uh which was computed by Java and then let rust side do the actual heavy computation row over row that is going to be much more efficient and in reality it's column by column not rowby row.

[00:12:39]Let's just talk about iceberg in general uh and what it is. I mean, iceberg is an open table format. And I realize that I should now put this slide before the other one. Uh, hence why you may have been confused. Iceberg is an open table format, which in reality just means that it has some sort of spec that describes a metadata layer which describes what t uh what files make up a particular table that can be queried in SQL or a variety of other tools. Basically, iceberg is a format supported by Spark, Trino, Presto, uh, Dreo, Starox, you name it.

[00:13:12]and that is part of its popularity. But in addition, its metadata format also lends really nicely to doing things like time travel. Uh you know, pruning out a large portion of the data set by only reading metadata. It's better than Hive in so far as it in a lot of the file names that are in the table. So you don't have to go to S3 and emit LS operations which can be slow or Hadoop and do something like that. Uh you can emit row tombstones and soon enough column tombstones. uh it is eventually going to have a very flexible data file format API. You can do things like lazy partition evolution or partition spec evolution. And of course, and most importantly, you can modify data across multiple files or add new files and delete files at the same time and do so atomically by just updating your pointer to a new metadata set.

[00:14:00]So again, why do I care about Comet?

[00:14:03]Well, I want to be able to speed up iceberg data pipelines. And it's great because Comet has already done a lot of the work for me. It [snorts] is already able to read iceberg and then once it's able to read iceberg tables into arrow batches, it can pretty much do any arbitrary form of SQL computation. It's not 100% of the way there, but it's getting a lot of the way there. And you know, things are continuing to speed up, which is great. Most importantly, it only supports reads and not writes. So there are two different places where I am going to be writing a lot of iceberg data. for one, just in normal ETL jobs and data pipelines, right? If you're going to read in iceberg data or you're going to read in data from some other source and you want to write it out to an iceberg table, you want that to be efficient. Number two is that if I have a bunch of small files in my iceberg table, typically what I want to do is compact them together so that I can get better row group sizing, better file sizing, less overhead when I actually query the data. Compaction is really just reading in iceberg data and writing it back out using iceberg's typical writers. However, right now we use a ton of resources devoted to compaction. We have big tables. We have particularly wide tables which Java's paret writers are not very good at and it is using up a lot of our compute. You may notice that someone like S3 tables who's probably doing a ton of this might be pretty interested in it as well. So the point is if I'm going to eventually write my data out to parquet, I don't want to have to do all this processing in comet and then convert my data back to rows so so that I can just convert it back to columns again to write it to parquet. So what do iceberg writes currently look like in comet? Because they haven't yet been implemented, the master branch of comet looks something like this. As you can see on the left side of the screen, we'll be able to scan iceberg tables using comet. take arrow batches, project them through data fusion, filter them through data fusion, and then right before we write that query result back to an iceberg table, all of a sudden we have to take those arrow batches and we have to convert them to Spark's internal row format. The reason for this is that the actual Java iceberg write operator, you know, the one that ships with iceberg, takes in Spark's internal form uh row format as an input. And [snorts] so implicitly, it's going to write those back out to column batches when you write to parquet. Uh but Unfortunately, we're not smart enough to do anything about it at the moment. And this is wholly inefficient.

[00:16:27]So, what am I trying to ultimately push here for uh push for here with my AI slop review that isn't really AI slop at this point, but it definitely started out that way. Well, I want to be able to scan Comet or rather scan iceberg with Comet, continue to do all of my data processing within Data Fusion. And then ultimately, when it comes to writing out that data, I don't want to have to do any additional transpose between those column batches back to rows. I want to take those arrow column batches and write them directly back out to Parquet.

[00:16:57][snorts] You'd be surprised for particularly wide tables, this has a major impact on performance, and we'll look at some benchmarks at the end.

[00:17:05]So what are the goals of my highle architecture? Well, basically I want to be able to use data fusion to write parquet files because it's great for that especially when arrow buffers are the input. Number two is I don't want to transpose data because that's stupid and wasteful. Number three is I want to keep the JVM continuing to handle the writing of iceberg metadata files and any edge cases associated with committing commit retries talking to the catalog. I don't really need that in Rust. It's not necessarily a bad thing if it happened, but Java is sufficiently fine for this because writing metadata files are generally small and talking to the catalog is IObound. So, it's not really where we're going to be blocked on getting good query performance. And number four, which is also really important, is that if you're going to build a Spark accelerator, you have to make sure that you're producing equivalent results to Spark itself. So, we need to ensure that we're getting par with Spark's actual existing writer implementation.

[00:18:02]Okay, so let's take a look a little bit deeper into Spark writes in Iceberg. Uh, you don't have to be a Spark expert here. I'm certainly not, but I'm going to do my best to explain how these work.

[00:18:12]The way that iceberg writes in Spark currently work is that they are known as a V2 command exec. The v2 writer pattern is basically uh some sort of interface over the concept of a bunch of executives committing a bunch of tasks individually and then once all of those tasks are committed for example files getting written to object store they can communicate back with the driver that would actually commit those files to the iceberg table. In the case of iceberg the driver is going to maybe write some metadata files communicate with the catalog and say hey we're ready to commit these cool. The way that that works is that if you have um some sort of spark query plan which is going to be aligned uh you know represented as some sort of tree the v2 write operator whether it's iceberg or delta or hoodie or anything like that is going to be right at the end of it and it's going to be outside of the part of spark uh plans that are basically handled in adaptive query execution. So I know this is going to start getting a little bit confusing but basically uh it is part of the existing spark paradigm that when you have these v2 writer operations those are not going to be touched in adaptive query execution everything upstream of that whether it's the projections the filters the joins anything like that that can be modified at runtime due to AQE uh but otherwise these won't and so what is kind of the implication there is that comet really only looks at what's inside of the AQE block in order to determine whether it's going to accelerate it. And you may note that within the iceberg v2 writer block, which I show on the bottom right of our screen, I care about accelerating part of it. I care about accelerating the data file writing part, but not so much about the committing. And yet it's still packaged into one single operator. So what do we actually do instead? I introduce a single toggle in data fusion comet which is actually splitting the iceberg write and the iceberg commit step into two logical different operators. The iceberg commit uh maintains its v2 command exec status and is outside of adaptive query execution.

[00:20:20]However, we add a physical iceberg writer operator within AQE at the bottom of the plan to write to. And so because that's within AQE, what we can now do is actually accelerate it within Comet.

[00:20:32]Under the hood, I'm basically just using the actual Java classes uh that were used in the V2 writer exec, but I'm basically breaking them up. So the iceberg write is running a data writer.

[00:20:44]So for example a partition data writer which is exactly what would be going on in the previous uh operator and once it's done it's serializing the task commit class which is Java passing that to the iceberg commit operator which might be running a batch append for example which is a type of iceberg commit operation defined in the iceberg repo and then ultimately that goes to the catalog and commits the right.

[00:21:08]Cool. So if we want to talk about how comet actually accelerates these spark rights, basically what we have is you know we have a plan rule running in comet that looks for iceberg rights and if we're allowed to accelerate them and I'll talk about what circumstances we can do so on the next slide. Then we basically take a protocol buffer we populate a bunch of settings from the right metadata like hey what is the data directory path where these writes all have to be written to? What are some settings about the parquet files? What is the target file size? What is the row group target file size? Are there any column stats that we want to add? Things like that. And we pass that down to our Rust layer, which is just going to be calling methods in iceberg rust. Iceberg Rust, fortunately, is able to then, you know, take those settings, receive a bunch of arrow batches at query runtime, and then start persisting those as paret files on S3. And then when it's done doing so, it can actually create its own iceberg manifest, which is a standardized JSON format to basically describe what was just written. Keep that in memory and then pass that back over through JNI back to Java.

[00:22:20]Okay, like I mentioned, one really key piece of all of this is being able to fall back when we are not able to handle a particular write or we don't want to.

[00:22:29]There are a lot of cases where someone's iceberg setup is made in a way or they configure certain properties such that we can't perfectly mimic it in iceberg rust. So for example, when we are passing down settings to iceberg rust, we have to include a lot of things. We need to include basically the location where we need to write files. We need to include the format of the file that we want to write. We need to include the partition spec. We need to include the table schema. We need to tell it what compression to use, what compression codec, what compression level like Zstandard 1 verse three is actually not aligned by a default between Java and Rust. Uh there are all these different properties that we need to be able to translate between those two different libraries. So first off, I have extensive testing in order to do that.

[00:23:13]Number two is that we don't want to write any files or we won't we won't want to accelerate any file writing where we can't perfectly replicate those settings. So because iceberg rust is a relatively new library compared to iceberg Java, iceberg Java has created a lot of writer settings over the years that simply cannot yet be replicated in iceberg rust. One examples of uh one example of those might be column stats.

[00:23:37]In iceberg Java, you can basically specify exact names for columns which will receive column stats. In iceberg rust, they all just receive column stats. Another example might be bloom filter settings. For example, in Java, you can specify a false positive rate.

[00:23:52]Uh on the or a false positive probability on the iceberg rust side, you simply cannot. Uh merge on read support is much better on Java. You can write positional deletes and deletion vectors. To my knowledge, iceberg rust only has support for equality deletes.

[00:24:06]Maybe that's changed by now.

[00:24:10]All right, before we finish up this video, I'm going to quickly talk about benchmarks. Uh you can see my shitty methodology here which is running it on my local computer in order to try and see what was better. And basically I've kept up three different margins. This is a very simple query where all I'm really doing is reading in a bunch of data from parquet writing it back out to parquet.

[00:24:30]In the first query which is a I'm doing that in just normal spark and you can see that whether it's 100 columns or one column it takes a decent amount of time.

[00:24:39]Number two is I only accelerate the read operation using comet which is basically using comet as it currently stands. You can see that I can still get a pretty nice performance boost when there are a lot of columns.

[00:24:52]Uh that being said at the end of the day we ultimately still have to do another transpose from those arrow batches into Spark's internal row to write to parquet. So the benefit is not as good as we would hope. Number three is that we uh you know both read and write arrow batches uh and use comet for the entire pipeline, right? So we're reading parquet and we're writing out arrow batches to parquet as output. [snorts] And you can see there that's when the dramatic performance improvement really kicks in. In the case where we have 100 columns, which is where I would expect there to be more of a performance benefit because transposes are more costly from rows to columns, uh we see a 3.6x performance improvement. And when there's only one column, even despite there being not really any transpose penalty, we still get a performance improvement, which I imagine I can largely attribute to things like SIMD, cash locality, things like that. So, moving on. I appreciate you guys coming.

[00:25:42]Uh, I hope it was an interesting presentation. Uh, this one's just for my YouTube channel. So, thank you all and have a great day and I will see you in the next