Under the Hood: Inside Spark Structured Streaming’s Delta Source

Spark Structured Streaming is one of the major streaming technologies, enabling real-time insights from your data. With the rise of the Lakehouse architecture and the growing popularity of the Delta Lake format, Delta tables have become a common source for streaming data in Spark.

Our Cloud Data Platform team at msg believes in understanding technologies we use at a deep level. That’s why we decided to explore the open-source implementation of the Delta Source in Spark Structured Streaming. Diving into the code is not only intellectually rewarding and allows you to deepen your understanding of a technology — it also pays off in practice. When documentation is sparse or ambiguous, knowing how to navigate the codebase becomes invaluable. After all, the source code is the most reliable form of documentation.

This post shares what we learned by providing an overview of the internals of the Delta Source for Spark Structured Streaming. Whether you’re looking to deepen your understanding or troubleshoot real-world issues, this post could help you connect the dots beneath the surface.

Before delving into the intricacies of the Delta source, we’ll first look at its overall structure and how it interacts with the Spark MicroBatchExecution engine. Thus, this article is organized into three main parts:

• The Anatomy of the Delta Source for Spark Structured Streaming

• The Interaction of the Spark MicroBatchExecution with the Delta Source

• The Inner Workings of the Delta Source

Disclaimer: This article assumes a general understanding of Spark Structured Streaming and Delta Lake. For a refresher on the internals of the Delta transaction log, see Diving Into Delta Lake: Unpacking The Transaction Log.

The Anatomy of the Delta Source for Spark Structured Streaming

Figure 1 — Class Diagram of Delta Source (some details omitted for brevity)

To understand how Delta Lake integrates with Spark Structured Streaming, it’s helpful to start with the general interface that streaming sources adhere to. Figure 1 shows a class diagram which represents the relationship between the Delta source and other important classes and interfaces.

The Delta source (org.apache.spark.sql.delta.sources.DeltaSource) is implemented as a subclass of org.apache.spark.sql.delta.sources.DeltaSourceBase, which itself adheres to the org.apache.spark.sql.execution.streaming.Source trait. The trait defines the contract that any streaming source must fulfill to be compatible with Spark’s micro-batch execution engine.

Two particularly important methods in this contract are:

· **getBatch** — Returns data from the source for the interval (start, end]. If start is None, the batch starts from the beginning of the source.

· **getOffset** — Returns the maximum offset currently available for processing.

Spark Structured Streaming also provides controls for limiting how much data is read in each batch. These mechanisms are exposed through the maxBytesPerTrigger and maxFilesPerTrigger options, and are supported by the additional interface SupportsAdmissionControl. Notably, the SupportsAdmissionControl interface leverages latestOffset (rather than getOffset) to compute the latest offset available for processing, while taking the specified ReadLimit into account. This ReadLimit — returned by the getDefaultReadLimit method — enforces constraints based on maxBytesPerTrigger and maxFilesPerTrigger on how much data can be contained in a micro-batch. The Delta Source also implements SupportsTriggerAvailableNow, which enables the AvailableNow trigger to process all data present at the query start.

Together, these interfaces form the contract of how the Delta source integrates with Spark Structured Streaming.

The Interaction of the Spark MicroBatchExecution with the Delta Source

Figure 2— Interaction between MicroBatchExecution engine and Delta Source (some details omitted for brevity)

At the heart of Spark Structured Streaming lies the MicroBatchExecution engine, which coordinates the execution of streaming queries in a micro-batch fashion. The engine interacts with the streaming source via the previously mentioned Source interface. This means that, from the perspective of the execution engine, Delta behaves like any other source implementing this interface. The high level interaction is displayed in figure 2.

Figure 3 — Initilization of DeltaSource by MicroBatchExecution engine

The entry point into running a stream is the runStream method of the StreamExecution class inherited to the MicroBatchExecution class. During initialization, the engine receives a logical plan that includes one or more streaming sources. These sources are initialized and added to the plan during the runStream execution (see figure 3). If the source is a Delta table, the instantiation is handled by org.apache.spark.sql.delta.sources.DeltaDataSource, which creates a DeltaSource object.

Once initialized, the engine repeatedly calls the runActivatedStream method for each micro-batch cycle. Here’s a breakdown of what happens during each cycle, specifically in the interaction between the engine and the Delta source:

1. Constructing the Next Batch

The engine prepares each batch using constructNextBatch, which involves:

Figure 4 — MicroBatchExecution engine retrieves the latest offset of the source (under consideration of the read limit) which serves as the upper bound of the next batch to be proceesed

• Retrieve the last processed offset — The engine first retrieves the previously processed end offset from an internal map called availableOffsets via getStartOffset method (see figure 4). (The availableOffsets map tracks the end offset available for the next execution, i.e., up to which point data should be processed by the upcoming cycle. At this point in the constructNextBatch method, we’re still at the start of the next cycle, so the current value in availableOffsets returned by getStartOffset reflects the end offset of the last processed batch.)
• Get the latest offset — The source’s latestOffset method is called with the previous offset passed as the start offset, returning the most recent available offset considering the configured ReadLimit (see figure 4) determined by the Source’s getDefaultReadLimit method.

Figure 5 — Write-ahead logging of next latest offset

• Write-ahead logging — The new latest offset returned is added back into the availableOffsets map and written to the OffsetSeqLog, which serves as a write-ahead log (see figure 5). The rational behind the write-ahead log is well explained in the source code: “In order to ensure that a given batch will always consist of the same data, we write to this log before any processing is done. Thus, the Nth record in this log indicated data that is currently being processed and the N-1th entry indicates which offsets have been durably committed to the sink.”

Figure 6 — Commit the previous already proceseed batch to the commit log

• Commit the previous batch — The batch id is committed to Spark’s commit log. In addition, the source’s commit method is called with the previous offset, finalizing processing of the last batch (see figure 6).

2. Running the Batch

With the batch prepared, Spark moves on to processing the data:

Figure 7 — MicroBatchExecution engine retrieves the batch to be processed

• Retrieving the batch from the Delta source — The engine calls the source’s getBatch method, passing in the previous committed offset (or None if the stream is new) as the starting point, and the latest offset as the endpoint (see figure 7). The getBatch method returns a DataFrame containing all data between these two offsets.
• Processing the batch — Spark processes the DataFrame returned by getBatch as defined in the logical plan (e.g., applying transformations, writing to sinks, etc.).
• Committing the batch — Once processing is complete, the engine adds the availableOffset to the committedOffsets map to reflect that this batch has been successfully completed. Additionally, the batch id is written to the commit log.

So far, we’ve looked at the Delta Source from two angles:

1. Structure — It’s built on Spark’s Source interface, extended by DeltaSourceBase, and enhanced with SupportsAdmissionControl and SupportsTriggerAvailableNow. These define how it integrates with Spark Structured Streaming and control batch sizes through options like maxBytesPerTrigger and maxFilesPerTrigger.

2. Interaction with the MicroBatchExecution engine — We focused on how the MicroBatchExecution engine interacts with the Delta Source, from discovering offsets to retrieving and processing data returned by getBatch.

With this context in place, we can now dive into the implementation of the Delta source itself.

The Inner Workings of the Delta Source

Figure 8—The inner workings of the Delta Source getBatch method

This section explores the implementation of the DeltaSource — focusing on one of its most important components: the getBatch method. It’s where most of the heavy lifting happens in terms of reading and interpreting the Delta transaction log.

How getBatch Works

At a high level, getBatch delegates to getFileChangesAndCreateDataFrame, which in turn performs two steps:

1. Calls getFileChanges — retrieves the list of relevant Delta log entries based on the requested version range.
2. Calls createDataFrame — constructs a DataFrame based on the list of retrieved delta log entries.

Handling Initial Snapshots vs. Log Tailing

Figure 9—Two paths for constructing iterator of indexed Delta log files exist

As you can see in the code snippet (see figure 9), the logic inside getFileChanges has two paths depending on whether the stream is (re)started (if-branch in code snippet) or continuing from the last offset of the previous execution cycle (else-branch in code snippet).

When a stream is started, Spark first processes all existing data in the table. This initial version of the table is referred to as the initial snapshot. In this case, the Delta log is read in two phases:

1. Initial snapshot: the full state of the table is reconstructed using getSnapshotAt(startVersion) method.
2. Subsequent changes: new log entries from startVersion + 1 onward are tailed from the Delta log by the filterAndIndexDeltaLogs method.

If the stream is already running and the initial snapshot has been processed during the start of the stream, the second path is taken (see else-branch in code snippet). That is, Spark simply tails the Delta log using filterAndIndexDeltaLogs without prior reconstruction of the table state via the getSnapshotAt method.

Let’s look more closely at how this tailing process works.

Tailing the Delta Log

Figure 10—Listing Delta log files

Under the hood, filterAndIndexDeltaLogs relies on a LogStore implementation, such as HadoopFileSystemLogStore, to physically access the Delta table’s _delta_log directory (see figure 10). It lists all log files from the Delta log directory starting from the specified delta log version.

Figure 10 — Reading and parsing delta log files as list of delta log actions

Next, each Delta log file is parsed into a list of Action objects (see figure 10). A single log file can contain multiple actions (see figure 11), such as:

• AddFile — refers to a parquet file containing added (or updated) data
• RemoveFile — marks a parquet file as deleted
• Metadata — contains schema changes or table properties

Figure 11 — A Delta log file for an update operation contains Add and Remove action entries alongside commit meta data

The LogStore is used again to read the physical files from underlying storage (e.g., HDFS, S3, etc.) that were listed in the previous step. The result is a rewindable iterator over all Delta log action entries from the relevant log files, grouped by commit/version.

Next, Spark filters these log entries in two passes:

First pass — Validate Commit and Decide Skipping The validateCommitAndDecideSkipping method inspects each delta log commit and decides whether it can be ignored. It checks:

• Whether the commit contains updates/delete actions.
• If the commit includes updates or deletes rather than just appends, the method throws an error unless explicitly allowed. This behavior is configurable through the options ignoreChanges, ignoreDeletes, or skipChangeCommits.
• If the commit contains updates and not just appends and skipChangesCommits is set to True, the commit is marked to be skipped.
• If a Metadata action is found, it validates schema compatibility using checkReadIncompatibleSchemaChanges.

Second pass — Filtering and Indexing

Figure 12—List of Delta log actions is filtered for data-changing Add actions and actions to be skipped

Next, Spark iterates again over the list of delta log files via the filterAndGetIndexedFiles method, this time applying the skipping decisions from the previous validateCommitAndDecideSkipping. It collects only the relevant AddFile actions where dataChange = True (see figure 12). Thus, commits that only contain non-data-changing actions (e.g., delta log entry written by OPTIMIZE command) are marked to be skipped since these have dataChange = False. The resulting AddFile actions are wrapped as IndexedFile objects, which consists of:

• The delta log version number
• The position (index) of the action within that delta log file
• The actual AddFile action itself

For example, if Delta log version 1 contains three AddFile actions, this results in three IndexedFile instances with values (version=1,index=0, AddFile1), (version=1,index=1, AddFile2) and (version=1,index=2, AddFile3).

But Wait — What About OPTIMIZE and VACUUM?

You might be wondering: OPTIMIZE only reorganizes data. Hence, it sets dataChange to false, so filterAndGetIndexedFiles skips those actions. But if a stream ignores OPTIMIZE commits and the original (pre-OPTIMIZE) files are later removed by VACUUM, how does streaming still work?

This is where the initial snapshot mentioned at the beginning of the section becomes important again.

Figure 13 — Two paths for constructing iterator of indexed Delta log files exist

While running streams ignore OPTIMIZE commits, a freshly (re)started stream considers the full table snapshot. It retrieves the table snapshot via the getSnapshotAt method (see figure 13). This snapshot reflects the latest valid state of the table, including any reorganized files introduced by OPTIMIZE (despite dataChange = False), and ensures completeness even if earlier files have been garbage-collected by running VACUUM.

In short:

• Running stream → skips OPTIMIZE commits using filters
• Restarted stream → reconstructs full table state including OPTIMIZE commits

Reconstructing the Table State

Figure 14 —Finding the last Delta log checkpoint for a specific Delta log version

To build this snapshot, Delta leverages the SnapshotManagement class — specifically the method findLastCompleteCheckpointBefore (see figure 14). This locates the most recent checkpoint in the Delta log, which serves as a summarized view of all the table’s log files at a given version.

Figure 15 — List Delta log files and checkpoints starting from last checkpoint version

Delta then retrieves the delta version from the checkpoint and lists all Delta log files from that point forward (see figure 15). Internally, this listing is once again performed via LogStore.listFrom, as previously mentioned in the Tailing the Delta Log section. If enabled through configuration, Delta can also include minor compaction files in this listing.

Finally, Delta replays these log entries — together with the last checkpoint, if available — to reconstruct the state of the table.

Replaying the Delta Log: InMemoryLogReplay

Figure 16— Reconstructing table state from list of Delta log actions

The reconstruction of the table state is performed by the InMemoryLogReplay class. This class processes each Delta log file version-by-version and interprets the different action types — such as AddFile, RemoveFile, Metadata, and others — to build an active set of files part of the table state at a given version (see figure 16).

For example, an OPTIMIZE operation typically introduces a new AddFile entry for a compacted file while introducing several RemoveFile actions for the smaller files replaced by the compacted file. During log replay, any RemoveFile action removes the corresponding file from the active set, so the resulting snapshot contains only the latest, compacted file.

Figure 17— Filtering list of Delta log actions for Add actions only

Figure 18 — Sorting list of Delta log Add actions by modification time

Once the snapshot has been constructed, it contains a list of all active files. These are further refined before being passed to the streaming engine:

• Only AddFile actions are retained; other action types are discarded (see figure 17).
• The list is sorted by each file’s modification timestamp to ensure deterministic ordering (see figure 18).

This log-replay logic not only clarifies the handling of OPTIMIZE operations but also has noteworthy implications for how UPDATE commits are managed. Let’s explore this as a quick quiz.

Imagine you have a Delta table with some records and a Spark Structured Streaming job reading from it. At some point, a single row in this table is updated. In the Delta transaction log, this update appears as new Delta log file containing:

• An AddFile action refering to the Parquet file with the updated row.
• A RemoveFile action marking the Parquet file containing the old version of the row as deleted.

Now, depending on when the update happens and how the stream is configured, the behavior of the stream will vary. Let’s look at three scenarios — try to guess the outcome of each before reading the answers at the end.

Scenario 1 — Stream starts after the update The stream is started after the update has already been committed.

Scenario 2 — Update occurs while the stream is running (default settings) The stream is already running when the update happens, with no special configuration applied.

Scenario 3 — Update occurs while the stream is running (custom settings) The stream is running when the update happens and is configured skipChangeCommits=true (ignore transactions that delete or modify existing records).

Answers:

1. Scenario 1 → Processes only the latest AddFile and emits only the updated record.
2. Scenario 2 → Fails with an error by default, as updates are not allowed unless explicitly handled.
3. Scenario 3 → Emits only the original (non-updated) record and ignores the updated record.

Why This Happens The different outcomes in the three scenarios come down to whether the stream is starting or already running when the update occurs.

• Starting stream → Follows the snapshot reconstruction path. Spark rebuilds the full table state via getSnapshotAt, which includes only the most recent valid files. In this case, an update’s old file is removed during replay, so the stream sees only the updated record.
• Running stream → Follows the log tailing path without snapshot reconstruction. The engine reads only new Delta log entries since the last processed version and applies commit filtering. Here, how updates are handled depends on the stream configuration:
• Default → Update commits trigger an error because updates are treated as changes which are disallowed by default.
• skipChangeCommits=true → The update commit is ignored, so only the original record is emitted.

Final Steps: Filtering Add Actions

Figure 19— Filtering list of Delta log Add actions so that startVersion < x <= endVersion

Earlier, we saw that the Delta Source can follow two paths when retrieving AddFile entries (see figure 19):

1. Snapshot reconstruction for a (re)started stream.
2. Log tailing for a running stream.

Regardless of which path is taken, both ultimately produce a list of AddFile actions representing the table’s current active files. The final step is to narrow this list down to the files relevant for the current micro-batch — specifically, those where startVersion < x <= endVersion (see figure 19).

With the relevant AddFile entries selected for the current micro-batch, the focus shifts back from Delta’s internal log processing to Spark’s execution layer. The next step is transforming this filtered set of Delta files into a Spark DataFrame that the streaming query can actually process.

Bridging Delta and Spark: From Delta AddFiles to Spark DataFrame

Figure 20—Creating a Spark Dataframe from list of Delta log add actions

After obtaining a list of relevant AddFiles from the Delta log, the createDataFrame method crosses the boundary from Delta-specific internals to the Spark execution engine — from a list of Delta AddFile actions to a Spark DataFrame.

The engine first wraps the files in a TahoeBatchFileIndex, then builds a HadoopFsRelation using the current snapshot and file index. This relation is wrapped in a streaming-aware logical plan and returned as a Spark DataFrame ready for processing in the current micro-batch.

Wrapping Up

In this deep dive, we traced the Delta Source for Spark Structured Streaming from its high-level architecture down to its internal mechanics.

We began by looking at the DeltaSource class from an API perspective, built on Spark’s Source interface with additional capabilities like SupportsAdmissionControl and SupportsTriggerAvailableNow. These define how the Delta Source integrates with the Spark engine, control batch size through read limits, and enable features like the AvailableNow trigger.

Next, we explored the interaction between the Delta Source and MicroBatchExecution, Spark’s engine for executing micro-batch streaming queries. We saw how the engine repeatedly calls the Delta Source to determine available offsets, retrieve the next batch of data for a specific offset range, and commit processed batches.

From there, we dug into the two execution paths inside the Delta Source getBatch method and how they affect operations like OPTIMIZE and UPDATE:

• Snapshot reconstruction for (re)started streams rebuilding the complete table state.
• Log tailing for running streams, which processes only appends since the last committed batch.

Finally, we explored how a list of Delta’s AddFile entries is transformed into a Spark DataFrame ready for processing by the MicroBatchExecution engine.

By understanding these layers — from interface contracts to engine interaction, to Delta-specific log processing — you can improve your ability to debug complex streaming scenarios and reason about the behavior of Delta and Spark — especially in situations where documentation, blog posts, and other online resources are sparse. And truth be told, diving into the source code isn’t just educational — it’s a curious little adventure!

Want to learn more?

Feel free to visit our website.

About us

msg’s Data & Analytics unit empowers organizations to become data-driven leaders through comprehensive, technology-agnostic support across the entire data lifecycle. From crafting tailored data strategies and governance frameworks to implementing cutting-edge use cases with advanced analytics and AI, we focus on delivering tangible business value. Our expertise includes designing and building scalable cloud data platforms on AWS, Azure, GCP, and Kubernetes, utilizing leading technologies like Databricks, MS Fabric, Confluent, etc. ensuring our clients gain a decisive competitive edge through optimized processes, data-informed decisions, and innovative, data-powered business models.