This is the full developer documentation for Delta Lake
# Welcome to the Delta Lake documentation
> Learn how to use Delta Lake
[Delta Lake](https://delta.io) is an [open source project](https://github.com/delta-io/delta) that enables building a [Lakehouse architecture](https://www.databricks.com/blog/2020/01/30/what-is-a-data-lakehouse.html) on top of [data lakes](https://www.databricks.com/discover/data-lakes). Delta Lake provides [ACID transactions](/concurrency-control), scalable metadata handling, and unifies [streaming](/delta-streaming) and [batch](/delta-batch) data processing on top of existing data lakes, such as S3, ADLS, GCS, and HDFS.
Specifically, Delta Lake offers:
* [ACID transactions](/concurrency-control) on Spark: Serializable isolation levels ensure that readers never see inconsistent data.
* Scalable metadata handling: Leverages Spark distributed processing power to handle all the metadata for petabyte-scale tables with billions of files at ease.
* [Streaming](/delta-streaming) and [batch](/delta-batch) unification: A table in Delta Lake is a batch table as well as a streaming source and sink. Streaming data ingest, batch historic backfill, interactive queries all just work out of the box.
* Schema enforcement: Automatically handles schema variations to prevent insertion of bad records during ingestion.
* [Time travel](/delta-batch#query-an-older-snapshot-of-a-table-time-travel): Data versioning enables rollbacks, full historical audit trails, and reproducible machine learning experiments.
* [Upserts](/delta-update#upsert-into-a-table-using-merge) and [deletes](/delta-update#delete-from-a-table): Supports merge, update and delete operations to enable complex use cases like change-data-capture, slowly-changing-dimension (SCD) operations, streaming upserts, and so on.
* Vibrant connector ecosystem: Delta Lake has connectors read and write Delta tables from various data processing engines like Apache Spark, Apache Flink, Apache Hive, Apache Trino, AWS Athena, and more.
To get started follow the [quickstart guide](/quick-start) to learn how to use Delta Lake with Apache Spark.
# Best practices
> Learn best practices when using Delta Lake.
## Choose the right partition column
[Section titled “Choose the right partition column”](#choose-the-right-partition-column)
You can partition a Delta table by a column. The most commonly used partition column is `date`. Follow these two rules of thumb for deciding on what column to partition by:
* If the cardinality of a column will be very high, do not use that column for partitioning. For example, if you partition by a column `userId` and if there can be 1M distinct user IDs, then that is a bad partitioning strategy.
* Amount of data in each partition: You can partition by a column if you expect data in that partition to be at least 1 GB.
## Compact files
[Section titled “Compact files”](#compact-files)
If you continuously write data to a Delta table, it will over time accumulate a large number of files, especially if you add data in small batches. This can have an adverse effect on the efficiency of table reads, and it can also affect the performance of your file system. Ideally, a large number of small files should be rewritten into a smaller number of larger files on a regular basis. This is known as compaction.
You can compact a table by repartitioning it to smaller number of files. In addition, you can specify the option `dataChange` to be `false` indicates that the operation does not change the data, only rearranges the data layout. This would ensure that other concurrent operations are minimally affected due to this compaction operation.
For example, you can compact a table into 16 files:
* Scala
```scala
val path = "..."
val numFiles = 16
spark.read
.format("delta")
.load(path)
.repartition(numFiles)
.write
.option("dataChange", "false")
.format("delta")
.mode("overwrite")
.save(path)
```
* Python
```python
path = "..."
numFiles = 16
(spark.read
.format("delta")
.load(path)
.repartition(numFiles)
.write
.option("dataChange", "false")
.format("delta")
.mode("overwrite")
.save(path))
```
If your table is partitioned and you want to repartition just one partition based on a predicate, you can read only the partition using `where` and write back to that using `replaceWhere`:
* Scala
```scala
val path = "..."
val partition = "year = '2019'"
val numFilesPerPartition = 16
spark.read
.format("delta")
.load(path)
.where(partition)
.repartition(numFilesPerPartition)
.write
.option("dataChange", "false")
.format("delta")
.mode("overwrite")
.option("replaceWhere", partition)
.save(path)
```
* Python
```python
path = "..."
partition = "year = '2019'"
numFilesPerPartition = 16
(spark.read
.format("delta")
.load(path)
.where(partition)
.repartition(numFilesPerPartition)
.write
.option("dataChange", "false")
.format("delta")
.mode("overwrite")
.option("replaceWhere", partition)
.save(path))
```
Caution
Using `dataChange = false` on an operation that changes data can corrupt the data in the table.
Note
This operation does not remove the old files. To remove them, run the [VACUUM](/delta-utility/#remove-files-no-longer-referenced-by-a-delta-table) command.
## Replace the content or schema of a table
[Section titled “Replace the content or schema of a table”](#replace-the-content-or-schema-of-a-table)
Sometimes you may want to replace a Delta table. For example:
* You discover the data in the table is incorrect and want to replace the content.
* You want to rewrite the whole table to do incompatible schema changes (such as changing column types).
While you can delete the entire directory of a Delta table and create a new table on the same path, it’s *not recommended* because:
* Deleting a directory is not efficient. A directory containing very large files can take hours or even days to delete.
* You lose all of content in the deleted files; it’s hard to recover if you delete the wrong table.
* The directory deletion is not atomic. While you are deleting the table a concurrent query reading the table can fail or see a partial table.
If you don’t need to change the table schema, you can [delete](/delta-update/#delete-from-a-table) data from a Delta table and insert your new data, or [update](/delta-update/#update-a-table) the table to fix the incorrect values.
If you want to change the table schema, you can replace the whole table atomically. For example:
* Python
```python
dataframe.write \
.format("delta") \
.mode("overwrite") \
.option("overwriteSchema", "true") \
.partitionBy() \
.saveAsTable("") # Managed table
dataframe.write \
.format("delta") \
.mode("overwrite") \
.option("overwriteSchema", "true") \
.option("path", "") \
.partitionBy() \
.saveAsTable("") # External table
```
* SQL
```sql
REPLACE TABLE USING DELTA PARTITIONED BY () AS SELECT ... -- Managed table
REPLACE TABLE USING DELTA PARTITIONED BY () LOCATION "" AS SELECT ... -- External table
```
* Scala
```scala
dataframe.write
.format("delta")
.mode("overwrite")
.option("overwriteSchema", "true")
.partitionBy()
.saveAsTable("") // Managed table
dataframe.write
.format("delta")
.mode("overwrite")
.option("overwriteSchema", "true")
.option("path", "")
.partitionBy()
.saveAsTable("") // External table
```
There are multiple benefits with this approach:
* Overwriting a table is much faster because it doesn’t need to list the directory recursively or delete any files.
* The old version of the table still exists. If you delete the wrong table you can easily retrieve the old data using [Time Travel](/delta-batch/#query-an-older-snapshot-of-a-table-time-travel).
* It’s an atomic operation. Concurrent queries can still read the table while you are deleting the table.
* Because of Delta Lake ACID transaction guarantees, if overwriting the table fails, the table will be in its previous state.
In addition, if you want to delete old files to save storage cost after overwriting the table, you can use [VACUUM](/delta-utility/#remove-files-no-longer-referenced-by-a-delta-table) to delete them. It’s optimized for file deletion and usually faster than deleting the entire directory.
## Spark caching
[Section titled “Spark caching”](#spark-caching)
You should not use Spark caching for the following reasons:
* You lose any data skipping that can come from additional filters added on top of the cached `DataFrame`.
* The data that gets cached may not be updated if the table is accessed using a different identifier (for example, you do `spark.table(x).cache()` but then write to the table using `spark.write.save(/some/path)`.
# Google BigQuery connector
> Learn how to read Delta Lake tables from Google BigQuery.
Google BigQuery supports reading Delta Lake (reader version 3 with [Deletion Vectors](/delta-deletion-vectors) and [Column Mapping](/delta-column-mapping/)). Please refer to [Delta Lake BigLake tables documentation](https://cloud.google.com/bigquery/docs/create-delta-lake-table) for more details.
# Concurrency control
> Learn about the ACID transaction guarantees between reads and writes provided by Delta Lake.
Delta Lake provides ACID transaction guarantees between reads and writes. This means that:
* For supported [storage systems](/delta-storage), multiple writers across multiple clusters can simultaneously modify a table partition and see a consistent snapshot view of the table and there will be a serial order for these writes.
* Readers continue to see a consistent snapshot view of the table that the Apache Spark job started with, even when a table is modified during a job.
## Optimistic concurrency control
[Section titled “Optimistic concurrency control”](#optimistic-concurrency-control)
Delta Lake uses [optimistic concurrency control](https://en.wikipedia.org/wiki/Optimistic_concurrency_control) to provide transactional guarantees between writes. Under this mechanism, writes operate in three stages:
1. **Read**: Reads (if needed) the latest available version of the table to identify which files need to be modified (that is, rewritten).
2. **Write**: Stages all the changes by writing new data files.
3. **Validate and commit**: Before committing the changes, checks whether the proposed changes conflict with any other changes that may have been concurrently committed since the snapshot that was read. If there are no conflicts, all the staged changes are committed as a new versioned snapshot, and the write operation succeeds. However, if there are conflicts, the write operation fails with a concurrent modification exception rather than corrupting the table as would happen with the write operation on a Parquet table.
## Write conflicts
[Section titled “Write conflicts”](#write-conflicts)
The following table describes which pairs of write operations can conflict. Compaction refers to [file compaction operation](/best-practices/#compact-files) written with the option `dataChange` set to `false`.
| | INSERT | UPDATE, DELETE, MERGE INTO | COMPACTION |
| ------------------------------ | --------------- | -------------------------- | ------------ |
| **INSERT** | Cannot conflict | | |
| **UPDATE, DELETE, MERGE INTO** | Can conflict | Can conflict | |
| **COMPACTION** | Cannot conflict | Can conflict | Can conflict |
## Avoid conflicts using partitioning and disjoint command conditions
[Section titled “Avoid conflicts using partitioning and disjoint command conditions”](#avoid-conflicts-using-partitioning-and-disjoint-command-conditions)
In all cases marked “can conflict”, whether the two operations will conflict depends on whether they operate on the same set of files. You can make the two sets of files disjoint by partitioning the table by the same columns as those used in the conditions of the operations. For example, the two commands `UPDATE table WHERE date > '2010-01-01' ...` and `DELETE table WHERE date < '2010-01-01'` will conflict if the table is not partitioned by date, as both can attempt to modify the same set of files. Partitioning the table by `date` will avoid the conflict. Hence, partitioning a table according to the conditions commonly used on the command can reduce conflicts significantly. However, partitioning a table by a column that has high cardinality can lead to other performance issues due to large number of subdirectories.
## Conflict exceptions
[Section titled “Conflict exceptions”](#conflict-exceptions)
When a transaction conflict occurs, you will observe one of the following exceptions:
### ConcurrentAppendException
[Section titled “ConcurrentAppendException”](#concurrentappendexception)
This exception occurs when a concurrent operation adds files in the same partition (or anywhere in an unpartitioned table) that your operation reads. The file additions can be caused by `INSERT`, `DELETE`, `UPDATE`, or `MERGE` operations.
This exception is often thrown during concurrent `DELETE`, `UPDATE`, or `MERGE` operations. While the concurrent operations may be physically updating different partition directories, one of them may read the same partition that the other one concurrently updates, thus causing a conflict. You can avoid this by making the separation explicit in the operation condition. Consider the following example.
* Scala
```scala
// Target 'deltaTable' is partitioned by date and country
deltaTable.as("t")
.merge(
source.as("s"),
"s.user_id = t.user_id AND s.date = t.date AND s.country = t.country"
)
.whenMatched()
.updateAll()
.whenNotMatched()
.insertAll()
.execute()
```
Suppose you run the above code concurrently for different dates or countries. Since each job is working on an independent partition on the target Delta table, you don’t expect any conflicts. However, the condition is not explicit enough and can scan the entire table and can conflict with concurrent operations updating any other partitions. Instead, you can rewrite your statement to add specific date and country to the merge condition, as shown in the following example.
* Scala
```scala
// Target 'deltaTable' is partitioned by date and country
deltaTable.as("t")
.merge(
source.as("s"),
"s.user_id = t.user_id AND s.date = t.date AND s.country = t.country AND t.date = '" + + "' AND t.country = '" + + "'"
)
.whenMatched()
.updateAll()
.whenNotMatched()
.insertAll()
.execute()
```
This operation is now safe to run concurrently on different dates and countries.
### ConcurrentDeleteReadException
[Section titled “ConcurrentDeleteReadException”](#concurrentdeletereadexception)
This exception occurs when a concurrent operation deleted a file that your operation read. Common causes are a `DELETE`, `UPDATE`, or `MERGE` operation that rewrites files.
### ConcurrentDeleteDeleteException
[Section titled “ConcurrentDeleteDeleteException”](#concurrentdeletedeleteexception)
This exception occurs when a concurrent operation deleted a file that your operation also deletes. This could be caused by two concurrent compaction operations rewriting the same files.
### MetadataChangedException
[Section titled “MetadataChangedException”](#metadatachangedexception)
This exception occurs when a concurrent transaction updates the metadata of a Delta table. Common causes are `ALTER TABLE` operations or writes to your Delta table that update the schema of the table.
### ConcurrentTransactionException
[Section titled “ConcurrentTransactionException”](#concurrenttransactionexception)
If a streaming query using the same checkpoint location is started multiple times concurrently and tries to write to the Delta table at the same time. You should never have two streaming queries use the same checkpoint location and run at the same time.
### ProtocolChangedException
[Section titled “ProtocolChangedException”](#protocolchangedexception)
This exception can occur in the following cases:
* When your Delta table is upgraded to a new version. For future operations to succeed you may need to upgrade your Delta Lake version.
* When multiple writers are creating or replacing a table at the same time.
* When multiple writers are writing to an empty path at the same time.
# Delta Lake APIs
> Learn about the APIs provided by Delta Lake.
Note
Some Delta Lake APIs are still evolving and are indicated with the **Evolving** qualifier or annotation in the API docs.
## Delta Spark
[Section titled “Delta Spark”](#delta-spark)
Delta Spark is a library for reading and writing Delta tables using Apache Spark™. For most read and write operations on Delta tables, you can use Apache Spark reader and writer APIs. For examples, see [Table batch reads and writes](/delta-batch/) and [Table streaming reads and writes](/delta-streaming/).
However, there are some operations that are specific to Delta Lake and you must use Delta Lake APIs. For examples, see [Table utility commands](/delta-utility/).
* [Scala API docs](/api/latest/scala/spark/io/delta/tables/index.html)
* [Java API docs](/api/latest/java/spark/index.html)
* [Python API docs](/api/latest/python/spark/index.html)
## Delta Kernel
[Section titled “Delta Kernel”](#delta-kernel)
Delta Kernel is a library for operating on Delta tables. Specifically, it provides simple and narrow APIs for reading and writing to Delta tables without the need to understand the [Delta protocol](https://github.com/delta-io/delta/blob/master/PROTOCOL.md) details. You can use this library to do the following:
* Read Delta tables from your applications.
* Build a connector for a distributed engine like Apache Spark™, Apache Flink, or Trino for reading massive Delta tables.
More details refer [here](https://github.com/delta-io/delta/blob/branch-3.0/kernel/USER_GUIDE.md).
* [Java API docs](/api/latest/java/kernel/index.html)
## Delta Rust
[Section titled “Delta Rust”](#delta-rust)
This [library](https://docs.rs/deltalake/latest/deltalake/) allows Rust (with Python bindings) low level access to Delta tables and is intended to be used with data processing frameworks like `datafusion`, `ballista`, `rust-dataframe`, `vega`, etc.
## Delta Standalone
[Section titled “Delta Standalone”](#delta-standalone)
Caution
The Delta Standalone is deprecated in favor of [Delta Kernel](/delta-kernel/) which has support for reading from or writing into Delta tables with advanced features.
Delta Standalone, formerly known as the Delta Standalone Reader (DSR), is a JVM library to read and write Delta tables. Unlike Delta-Spark, this library doesn’t use Spark to read or write tables and it has only a few transitive dependencies. It can be used by any application that cannot use a Spark cluster. More details refer [here](https://github.com/delta-io/delta/blob/master/connectors/README.md).
* [Java API docs](/api/3.3.2/java/standalone/index.html)
## Delta Flink
[Section titled “Delta Flink”](#delta-flink)
Flink/Delta Connector is a JVM library to read and write data from Apache Flink applications to Delta tables utilizing the Delta Standalone JVM library. More details refer [here](https://github.com/delta-io/delta/blob/master/connectors/flink/README.md).
* [Java API docs](/api/3.3.2/java/flink/index.html)
# AWS Athena Delta Connector
> Learn how to set up an integration to enable you to read Delta tables from AWS Athena.
# AWS Athena Delta Connector
[Section titled “AWS Athena Delta Connector”](#aws-athena-delta-connector)
Since Athena [version 3](https://docs.aws.amazon.com/athena/latest/ug/engine-versions-reference-0003.html), Athena natively supports reading Delta Lake tables. For details on using the native Delta Lake connector, see [Querying Delta Lake tables](https://docs.aws.amazon.com/athena/latest/ug/delta-lake-tables.html). For Athena versions lower than [version 3](https://docs.aws.amazon.com/athena/latest/ug/engine-versions-reference-0003.html), you can use the manifest-based approach detailed in [Presto, Trino, and Athena to Delta Lake integration using manifests](/presto-integration).
# Table batch reads and writes
> Learn how to perform batch reads and writes on Delta tables.
Delta Lake supports most of the options provided by Apache Spark DataFrame read and write APIs for performing batch reads and writes on tables.
For many Delta Lake operations on tables, you enable integration with Apache Spark DataSourceV2 and Catalog APIs (since 3.0) by setting configurations when you create a new `SparkSession`. See [Configure SparkSession](#configure-sparksession).
## Create a table
[Section titled “Create a table”](#create-a-table)
Delta Lake supports creating two types of tables—tables defined in the metastore and tables defined by path.
To work with metastore-defined tables, you must enable integration with Apache Spark DataSourceV2 and Catalog APIs by setting configurations when you create a new `SparkSession`. See [Configure SparkSession](#configure-sparksession).
You can create tables in the following ways:
* **SQL DDL commands**: You can use standard SQL DDL commands supported in Apache Spark (for example, `CREATE TABLE` and `REPLACE TABLE`) to create Delta tables.
* SQL
```sql
CREATE TABLE IF NOT EXISTS default.people10m (
id INT,
firstName STRING,
middleName STRING,
lastName STRING,
gender STRING,
birthDate TIMESTAMP,
ssn STRING,
salary INT
) USING DELTA
CREATE OR REPLACE TABLE default.people10m (
id INT,
firstName STRING,
middleName STRING,
lastName STRING,
gender STRING,
birthDate TIMESTAMP,
ssn STRING,
salary INT
) USING DELTA
```
SQL also supports creating a table at a path, without creating an entry in the Hive metastore.
* SQL
```sql
-- Create or replace table with path
CREATE OR REPLACE TABLE delta.`/tmp/delta/people10m` (
id INT,
firstName STRING,
middleName STRING,
lastName STRING,
gender STRING,
birthDate TIMESTAMP,
ssn STRING,
salary INT
) USING DELTA
```
* **`DataFrameWriter` API**: If you want to simultaneously create a table and insert data into it from Spark DataFrames or Datasets, you can use the Spark `DataFrameWriter` ([Scala or Java](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/DataFrameWriter.html) and [Python](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/io.html)).
* Python
```python
# Create table in the metastore using DataFrame's schema and write data to it
df.write.format("delta").saveAsTable("default.people10m")
# Create or replace partitioned table with path using DataFrame's schema and write/overwrite data to it
df.write.format("delta").mode("overwrite").save("/tmp/delta/people10m")
```
* Scala
```scala
// Create table in the metastore using DataFrame's schema and write data to it
df.write.format("delta").saveAsTable("default.people10m")
// Create table with path using DataFrame's schema and write data to it
df.write.format("delta").mode("overwrite").save("/tmp/delta/people10m")
```
You can also create Delta tables using the Spark `DataFrameWriterV2` API.
* **`DeltaTableBuilder` API**: You can also use the `DeltaTableBuilder` API in Delta Lake to create tables. Compared to the DataFrameWriter APIs, this API makes it easier to specify additional information like column comments, table properties, and [generated columns](#use-generated-columns).
Note
This feature is new and is in Preview.
* Python
```python
# Create table in the metastore
DeltaTable.createIfNotExists(spark) \
.tableName("default.people10m") \
.addColumn("id", "INT") \
.addColumn("firstName", "STRING") \
.addColumn("middleName", "STRING") \
.addColumn("lastName", "STRING", comment = "surname") \
.addColumn("gender", "STRING") \
.addColumn("birthDate", "TIMESTAMP") \
.addColumn("ssn", "STRING") \
.addColumn("salary", "INT") \
.execute()
# Create or replace table with path and add properties
DeltaTable.createOrReplace(spark) \
.addColumn("id", "INT") \
.addColumn("firstName", "STRING") \
.addColumn("middleName", "STRING") \
.addColumn("lastName", "STRING", comment = "surname") \
.addColumn("gender", "STRING") \
.addColumn("birthDate", "TIMESTAMP") \
.addColumn("ssn", "STRING") \
.addColumn("salary", "INT") \
.property("description", "table with people data") \
.location("/tmp/delta/people10m") \
.execute()
```
* Scala
```scala
// Create table in the metastore
DeltaTable.createOrReplace(spark)
.tableName("default.people10m")
.addColumn("id", "INT")
.addColumn("firstName", "STRING")
.addColumn("middleName", "STRING")
.addColumn(
DeltaTable.columnBuilder("lastName")
.dataType("STRING")
.comment("surname")
.build())
.addColumn("lastName", "STRING", comment = "surname")
.addColumn("gender", "STRING")
.addColumn("birthDate", "TIMESTAMP")
.addColumn("ssn", "STRING")
.addColumn("salary", "INT")
.execute()
// Create or replace table with path and add properties
DeltaTable.createOrReplace(spark)
.addColumn("id", "INT")
.addColumn("firstName", "STRING")
.addColumn("middleName", "STRING")
.addColumn(
DeltaTable.columnBuilder("lastName")
.dataType("STRING")
.comment("surname")
.build())
.addColumn("lastName", "STRING", comment = "surname")
.addColumn("gender", "STRING")
.addColumn("birthDate", "TIMESTAMP")
.addColumn("ssn", "STRING")
.addColumn("salary", "INT")
.property("description", "table with people data")
.location("/tmp/delta/people10m")
.execute()
```
See the [API documentation](/delta-apidoc/) for details.
### Partition data
[Section titled “Partition data”](#partition-data)
You can partition data to speed up queries or DML that have predicates involving the partition columns. To partition data when you create a Delta table, specify a partition by columns. The following example partitions by gender.
* SQL
```sql
-- Create table in the metastore
CREATE TABLE default.people10m (
id INT,
firstName STRING,
middleName STRING,
lastName STRING,
gender STRING,
birthDate TIMESTAMP,
ssn STRING,
salary INT
)
USING DELTA
PARTITIONED BY (gender)
```
* Python
```python
df.write.format("delta").partitionBy("gender").saveAsTable("default.people10m")
DeltaTable.create(spark) \
.tableName("default.people10m") \
.addColumn("id", "INT") \
.addColumn("firstName", "STRING") \
.addColumn("middleName", "STRING") \
.addColumn("lastName", "STRING", comment = "surname") \
.addColumn("gender", "STRING") \
.addColumn("birthDate", "TIMESTAMP") \
.addColumn("ssn", "STRING") \
.addColumn("salary", "INT") \
.partitionedBy("gender") \
.execute()
```
* Scala
```scala
df.write.format("delta").partitionBy("gender").saveAsTable("default.people10m")
DeltaTable.createOrReplace(spark)
.tableName("default.people10m")
.addColumn("id", "INT")
.addColumn("firstName", "STRING")
.addColumn("middleName", "STRING")
.addColumn(
DeltaTable.columnBuilder("lastName")
.dataType("STRING")
.comment("surname")
.build())
.addColumn("lastName", "STRING", comment = "surname")
.addColumn("gender", "STRING")
.addColumn("birthDate", "TIMESTAMP")
.addColumn("ssn", "STRING")
.addColumn("salary", "INT")
.partitionedBy("gender")
.execute()
```
To determine whether a table contains a specific partition, use the statement `SELECT COUNT(*) > 0 FROM WHERE = `. If the partition exists, `true` is returned. For example:
* SQL
```sql
SELECT COUNT(*) > 0 AS `Partition exists` FROM default.people10m WHERE gender = "M"
```
* Python
```python
display(spark.sql("SELECT COUNT(*) > 0 AS `Partition exists` FROM default.people10m WHERE gender = 'M'"))
```
* Scala
```scala
display(spark.sql("SELECT COUNT(*) > 0 AS `Partition exists` FROM default.people10m WHERE gender = 'M'"))
```
### Control data location
[Section titled “Control data location”](#control-data-location)
For tables defined in the metastore, you can optionally specify the `LOCATION` as a path. Tables created with a specified `LOCATION` are considered unmanaged by the metastore. Unlike a managed table, where no path is specified, an unmanaged table’s files are not deleted when you `DROP` the table.
When you run `CREATE TABLE` with a `LOCATION` that *already* contains data stored using Delta Lake, Delta Lake does the following:
* If you specify *only the table name and location*, for example:
* SQL
```sql
CREATE TABLE default.people10m
USING DELTA
LOCATION '/tmp/delta/people10m'
```
the table in the metastore automatically inherits the schema, partitioning, and table properties of the existing data. This functionality can be used to “import” data into the metastore.
* If you specify *any configuration* (schema, partitioning, or table properties), Delta Lake verifies that the specification exactly matches the configuration of the existing data.
Important
If the specified configuration does not *exactly* match the configuration of the data, Delta Lake throws an exception that describes the discrepancy.
Note
The metastore is not the source of truth about the latest information of a Delta table. In fact, the table definition in the metastore may not contain all the metadata like schema and properties. It contains the location of the table, and the table’s transaction log at the location is the source of truth. If you query the metastore from a system that is not aware of this Delta-specific customization, you may see incomplete or stale table information.
### Use generated columns
[Section titled “Use generated columns”](#use-generated-columns)
Note
This feature is new and is in Preview.
Delta Lake supports generated columns which are a special type of columns whose values are automatically generated based on a user-specified function over other columns in the Delta table. When you write to a table with generated columns and you do not explicitly provide values for them, Delta Lake automatically computes the values. For example, you can automatically generate a date column (for partitioning the table by date) from the timestamp column; any writes into the table need only specify the data for the timestamp column. However, if you explicitly provide values for them, the values must satisfy the [constraint](/delta-constraints/) `( <=> ) IS TRUE` or the write will fail with an error.
Important
Tables created with generated columns have a higher table writer protocol version than the default. See [How does Delta Lake manage feature compatibility?](/versioning/) to understand table protocol versioning and what it means to have a higher version of a table protocol version.
The following example shows how to create a table with generated columns:
* Python
```python
DeltaTable.create(spark) \
.tableName("default.people10m") \
.addColumn("id", "INT") \
.addColumn("firstName", "STRING") \
.addColumn("middleName", "STRING") \
.addColumn("lastName", "STRING", comment = "surname") \
.addColumn("gender", "STRING") \
.addColumn("birthDate", "TIMESTAMP") \
.addColumn("dateOfBirth", DateType(), generatedAlwaysAs="CAST(birthDate AS DATE)") \
.addColumn("ssn", "STRING") \
.addColumn("salary", "INT") \
.partitionedBy("gender") \
.execute()
```
* Scala
```scala
DeltaTable.create(spark)
.tableName("default.people10m")
.addColumn("id", "INT")
.addColumn("firstName", "STRING")
.addColumn("middleName", "STRING")
.addColumn(
DeltaTable.columnBuilder("lastName")
.dataType("STRING")
.comment("surname")
.build())
.addColumn("lastName", "STRING", comment = "surname")
.addColumn("gender", "STRING")
.addColumn("birthDate", "TIMESTAMP")
.addColumn(
DeltaTable.columnBuilder("dateOfBirth")
.dataType(DateType)
.generatedAlwaysAs("CAST(dateOfBirth AS DATE)")
.build())
.addColumn("ssn", "STRING")
.addColumn("salary", "INT")
.partitionedBy("gender")
.execute()
```
Generated columns are stored as if they were normal columns. That is, they occupy storage.
The following restrictions apply to generated columns:
* A generation expression can use any SQL functions in Spark that always return the same result when given the same argument values, except the following types of functions:
* User-defined functions.
* Aggregate functions.
* Window functions.
* Functions returning multiple rows.
* For Delta Lake 1.1.0 and above, `MERGE` operations support generated columns when you set `spark.databricks.delta.schema.autoMerge.enabled` to true.
Delta Lake may be able to generate partition filters for a query whenever a partition column is defined by one of the following expressions:
* `CAST(col AS DATE)` and the type of `col` is `TIMESTAMP`.
* `YEAR(col)` and the type of `col` is `TIMESTAMP`.
* Two partition columns defined by `YEAR(col), MONTH(col)` and the type of `col` is `TIMESTAMP`.
* Three partition columns defined by `YEAR(col), MONTH(col), DAY(col)` and the type of `col` is `TIMESTAMP`.
* Four partition columns defined by `YEAR(col), MONTH(col), DAY(col), HOUR(col)` and the type of `col` is `TIMESTAMP`.
* `SUBSTRING(col, pos, len)` and the type of `col` is `STRING`
* `DATE_FORMAT(col, format)` and the type of `col` is `TIMESTAMP`.
* `DATE_TRUNC(format, col)` and the type of the `col` is `TIMESTAMP` or `DATE`.
* `TRUNC(col, format)` and type of the `col` is either `TIMESTAMP` or `DATE`.
If a partition column is defined by one of the preceding expressions, and a query filters data using the underlying base column of a generation expression, Delta Lake looks at the relationship between the base column and the generated column, and populates partition filters based on the generated partition column if possible. For example, given the following table:
* Python
```python
DeltaTable.create(spark) \
.tableName("default.events") \
.addColumn("eventId", "BIGINT") \
.addColumn("data", "STRING") \
.addColumn("eventType", "STRING") \
.addColumn("eventTime", "TIMESTAMP") \
.addColumn("eventDate", "DATE", generatedAlwaysAs="CAST(eventTime AS DATE)") \
.partitionedBy("eventType", "eventDate") \
.execute()
```
If you then run the following query:
* Python
```python
spark.sql('SELECT * FROM default.events WHERE eventTime >= "2020-10-01 00:00:00" <= "2020-10-01 12:00:00"')
```
Delta Lake automatically generates a partition filter so that the preceding query only reads the data in partition `date=2020-10-01` even if a partition filter is not specified.
As another example, given the following table:
* Python
```python
DeltaTable.create(spark) \
.tableName("default.events") \
.addColumn("eventId", "BIGINT") \
.addColumn("data", "STRING") \
.addColumn("eventType", "STRING") \
.addColumn("eventTime", "TIMESTAMP") \
.addColumn("year", "INT", generatedAlwaysAs="YEAR(eventTime)") \
.addColumn("month", "INT", generatedAlwaysAs="MONTH(eventTime)") \
.addColumn("day", "INT", generatedAlwaysAs="DAY(eventTime)") \
.partitionedBy("eventType", "year", "month", "day") \
.execute()
```
If you then run the following query:
* Python
```python
spark.sql('SELECT * FROM default.events WHERE eventTime >= "2020-10-01 00:00:00" <= "2020-10-01 12:00:00"')
```
Delta Lake automatically generates a partition filter so that the preceding query only reads the data in partition `year=2020/month=10/day=01` even if a partition filter is not specified.
You can use an [EXPLAIN](https://spark.apache.org/docs/latest/sql-ref-syntax-qry-explain.html) clause and check the provided plan to see whether Delta Lake automatically generates any partition filters.
### Use identity columns
[Section titled “Use identity columns”](#use-identity-columns)
Important
Declaring an identity column on a Delta table disables concurrent transactions. Only use identity columns in use cases where concurrent writes to the target table are not required.
Delta Lake identity columns are supported in Delta Lake 3.3 and above. They are a type of generated column that assigns unique values for each record inserted into a table. The following example shows how to declare an identity column during a create table command:
* Python
```python
from delta.tables import DeltaTable, IdentityGenerator
from pyspark.sql.types import LongType
DeltaTable.create()
.tableName("table_name")
.addColumn("id_col1", dataType=LongType(), generatedAlwaysAs=IdentityGenerator())
.addColumn("id_col2", dataType=LongType(), generatedAlwaysAs=IdentityGenerator(start=-1, step=1))
.addColumn("id_col3", dataType=LongType(), generatedByDefaultAs=IdentityGenerator())
.addColumn("id_col4", dataType=LongType(), generatedByDefaultAs=IdentityGenerator(start=-1, step=1))
.execute()
```
* Scala
```scala
import io.delta.tables.DeltaTable
import org.apache.spark.sql.types.LongType
DeltaTable.create(spark)
.tableName("table_name")
.addColumn(
DeltaTable.columnBuilder(spark, "id_col1")
.dataType(LongType)
.generatedAlwaysAsIdentity().build())
.addColumn(
DeltaTable.columnBuilder(spark, "id_col2")
.dataType(LongType)
.generatedAlwaysAsIdentity(start = -1L, step = 1L).build())
.addColumn(
DeltaTable.columnBuilder(spark, "id_col3")
.dataType(LongType)
.generatedByDefaultAsIdentity().build())
.addColumn(
DeltaTable.columnBuilder(spark, "id_col4")
.dataType(LongType)
.generatedByDefaultAsIdentity(start = -1L, step = 1L).build())
.execute()
```
Note
SQL APIs for identity columns are not supported yet.
You can optionally specify the following:
* A starting value.
* A step size, which can be positive or negative.
Both the starting value and step size default to `1`. You cannot specify a step size of `0`.
Values assigned by identity columns are unique and increment in the direction of the specified step, and in multiples of the specified step size, but are not guaranteed to be contiguous. For example, with a starting value of `0` and a step size of `2`, all values are positive even numbers but some even numbers might be skipped.
When the identity column is specified to be `generated by default as identity`, insert operations can specify values for the identity column. Specify it to be `generated always as identity` to override the ability to manually set values.
Identity columns only support `LongType`, and operations fail if the assigned value exceeds the range supported by `LongType`.
You can use `ALTER TABLE table_name ALTER COLUMN column_name SYNC IDENTITY` to synchronize the metadata of an identity column with the actual data. When you write your own values to an identity column, it might not comply with the metadata. This option evaluates the state and updates the metadata to be consistent with the actual data. After this command, the next automatically assigned identity value will start from `start + (n + 1) * step`, where `n` is the smallest value that satisfies `start + n * step >= max()` (for a positive step).
#### CTAS and identity columns
[Section titled “CTAS and identity columns”](#ctas-and-identity-columns)
You cannot define schema, identity column constraints, or any other table specifications when using a `CREATE TABLE table_name AS SELECT` (CTAS) statement.
To create a new table with an identity column and populate it with existing data, do the following:
1. Create a table with the correct schema, including the identity column definition and other table properties.
2. Run an insertion operation.
The following example define the identity column to be `generated by default as identity`. If data inserted into the table includes valid values for the identity column, these values are used.
* Python
```python
from delta.tables import DeltaTable, IdentityGenerator
from pyspark.sql.types import LongType, DateType
DeltaTable.create(spark)
.tableName("new_table")
.addColumn("id", dataType=LongType(), generatedByDefaultAs=IdentityGenerator(start=5, step=1))
.addColumn("event_date", dataType=DateType())
.addColumn("some_value", dataType=LongType())
.execute()
# Insert records including existing IDs
old_table_df = spark.table("old_table").select("id", "event_date", "some_value")
old_table_df.write
.format("delta")
.mode("append")
.saveAsTable("new_table")
# Insert records and generate new IDs
new_records_df = spark.table("new_records").select("event_date", "some_value")
new_records_df.write
.format("delta")
.mode("append")
.saveAsTable("new_table")
```
* Scala
```scala
import org.apache.spark.sql.types._
import io.delta.tables.DeltaTable
DeltaTable.createOrReplace(spark)
.tableName("new_table")
.addColumn(
DeltaTable.columnBuilder(spark, "id")
.dataType(LongType)
.generatedByDefaultAsIdentity(start = 5L, step = 1L)
.build())
.addColumn(
DeltaTable.columnBuilder(spark, "event_date")
.dataType(DateType)
.nullable(true)
.build())
.addColumn(
DeltaTable.columnBuilder(spark, "some_value")
.dataType(LongType)
.nullable(true)
.build())
.execute()
// Insert records including existing IDs
val oldTableDF = spark.table("old_table").select("id", "event_date", "some_value")
oldTableDF.write
.format("delta")
.mode("append")
.saveAsTable("new_table")
// Insert records and generate new IDs
val newRecordsDF = spark.table("new_records").select("event_date", "some_value")
newRecordsDF.write
.format("delta")
.mode("append")
.saveAsTable("new_table")
```
#### Identity column limitations
[Section titled “Identity column limitations”](#identity-column-limitations)
The following limitations exist when working with identity columns:
* Concurrent transactions are not supported on tables with identity columns enabled.
* You cannot partition a table by an identity column.
* You cannot `ADD`, `REPLACE`, or `CHANGE` an identity column.
* You cannot update the value of an identity column for an existing record.
Note
To change the `IDENTITY` value for an existing record, you must delete the record and `INSERT` it as a new record.
### Specify default values for columns
[Section titled “Specify default values for columns”](#specify-default-values-for-columns)
Delta enables the specification of [default expressions](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#default-columns) for columns in Delta tables. When users write to these tables without explicitly providing values for certain columns, or when they explicitly use the `DEFAULT` SQL keyword for a column, Delta automatically generates default values for those columns. For more information, please refer to the dedicated documentation page.
### Use special characters in column names
[Section titled “Use special characters in column names”](#use-special-characters-in-column-names)
By default, special characters such as spaces and any of the characters `,;{}()\n\t=` are not supported in table column names. To include these special characters in a table’s column name, enable column mapping.
### Default table properties
[Section titled “Default table properties”](#default-table-properties)
Delta Lake configurations set in the SparkSession override the default [table properties](/table-properties/) for new Delta Lake tables created in the session. The prefix used in the SparkSession is different from the configurations used in the table properties.
| Delta Lake conf | SparkSession conf |
| --------------- | --------------------------------------------------- |
| `delta.` | `spark.databricks.delta.properties.defaults.` |
For example, to set the `delta.appendOnly = true` property for all new Delta Lake tables created in a session, set the following:
```sql
SET spark.databricks.delta.properties.defaults.appendOnly = true
```
## Read a table
[Section titled “Read a table”](#read-a-table)
You can load a Delta table as a DataFrame by specifying a table name or a path:
* SQL
```sql
SELECT * FROM default.people10m -- query table in the metastore
SELECT * FROM delta.`/tmp/delta/people10m` -- query table by path
```
* Python
```python
spark.table("default.people10m") # query table in the metastore
spark.read.format("delta").load("/tmp/delta/people10m") # query table by path
```
* Scala
```scala
spark.table("default.people10m") // query table in the metastore
spark.read.format("delta").load("/tmp/delta/people10m") // create table by path
import io.delta.implicits._
spark.read.delta("/tmp/delta/people10m")
```
The DataFrame returned automatically reads the most recent snapshot of the table for any query; you never need to run `REFRESH TABLE`. Delta Lake automatically uses partitioning and statistics to read the minimum amount of data when there are applicable predicates in the query.
## Query an older snapshot of a table (time travel)
[Section titled “Query an older snapshot of a table (time travel)”](#query-an-older-snapshot-of-a-table-time-travel)
Delta Lake time travel allows you to query an older snapshot of a Delta table. Time travel has many use cases, including:
* Re-creating analyses, reports, or outputs (for example, the output of a machine learning model). This could be useful for debugging or auditing, especially in regulated industries.
* Writing complex temporal queries.
* Fixing mistakes in your data.
* Providing snapshot isolation for a set of queries for fast changing tables.
This section describes the supported methods for querying older versions of tables, data retention concerns, and provides examples.
Note
The timestamp of each version N depends on the timestamp of the log file corresponding to the version N in Delta table log. Hence, time travel by timestamp can break if you copy the entire Delta table directory to a new location. Time travel by version will be unaffected.
### Syntax
[Section titled “Syntax”](#syntax)
This section shows how to query an older version of a Delta table.
#### SQL `AS OF` syntax
[Section titled “SQL AS OF syntax”](#sql-as-of-syntax)
* SQL
```sql
SELECT * FROM table_name TIMESTAMP AS OF timestamp_expression
SELECT * FROM table_name VERSION AS OF version
```
- `timestamp_expression` can be any one of:
* `'2018-10-18T22:15:12.013Z'`, that is, a string that can be cast to a timestamp
* `cast('2018-10-18 13:36:32 CEST' as timestamp)`
* `'2018-10-18'`, that is, a date string
* `current_timestamp() - interval 12 hours`
* `date_sub(current_date(), 1)`
* Any other expression that is or can be cast to a timestamp
- `version` is a long value that can be obtained from the output of `DESCRIBE HISTORY table_spec`.
Neither `timestamp_expression` nor `version` can be subqueries.
##### Example
[Section titled “Example”](#example)
* SQL
```sql
SELECT * FROM default.people10m TIMESTAMP AS OF '2018-10-18T22:15:12.013Z'
SELECT * FROM delta.`/tmp/delta/people10m` VERSION AS OF 123
```
#### DataFrameReader options
[Section titled “DataFrameReader options”](#dataframereader-options)
DataFrameReader options allow you to create a DataFrame from a Delta table that is fixed to a specific version of the table.
* Python
```python
df1 = spark.read.format("delta").option("timestampAsOf", timestamp_string).load("/tmp/delta/people10m")
df2 = spark.read.format("delta").option("versionAsOf", version).load("/tmp/delta/people10m")
```
For `timestamp_string`, only date or timestamp strings are accepted. For example, `"2019-01-01"` and `"2019-01-01T00:00:00.000Z"`.
A common pattern is to use the latest state of the Delta table throughout the execution of a job to update downstream applications.
Because Delta tables auto update, a DataFrame loaded from a Delta table may return different results across invocations if the underlying data is updated. By using time travel, you can fix the data returned by the DataFrame across invocations:
* Python
```python
history = spark.sql("DESCRIBE HISTORY delta.`/tmp/delta/people10m`")
latest_version = history.selectExpr("max(version)").collect()
df = spark.read.format("delta").option("versionAsOf", latest_version[0][0]).load("/tmp/delta/people10m")
```
##### Examples
[Section titled “Examples”](#examples)
* Fix accidental deletes to a table for the user 111:
* Python
```python
yesterday = spark.sql("SELECT CAST(date_sub(current_date(), 1) AS STRING)").collect()[0][0]
df = spark.read.format("delta").option("timestampAsOf", yesterday).load("/tmp/delta/events")
df.where("userId = 111").write.format("delta").mode("append").save("/tmp/delta/events")
```
* Fix accidental incorrect updates to a table:
* Python
```python
yesterday = spark.sql("SELECT CAST(date_sub(current_date(), 1) AS STRING)").collect()[0][0]
df = spark.read.format("delta").option("timestampAsOf", yesterday).load("/tmp/delta/events")
df.createOrReplaceTempView("my_table_yesterday")
spark.sql('''
MERGE INTO delta.`/tmp/delta/events` target
USING my_table_yesterday source
ON source.userId = target.userId
WHEN MATCHED THEN UPDATE SET *
''')
```
* Query the number of new customers added over the last week:
* Python
```python
last_week = spark.sql("SELECT CAST(date_sub(current_date(), 7) AS STRING)").collect()[0][0]
df = spark.read.format("delta").option("timestampAsOf", last_week).load("/tmp/delta/events")
last_week_count = df.select("userId").distinct().count()
count = spark.read.format("delta").load("/tmp/delta/events").select("userId").distinct().count()
new_customers_count = count - last_week_count
```
### Data retention
[Section titled “Data retention”](#data-retention)
To time travel to a previous version, you must retain *both* the log and the data files for that version.
The data files backing a Delta table are *never* deleted automatically; data files are deleted only when you run [VACUUM](/delta-utility#remove-files-no-longer-referenced-by-a-delta-table). `VACUUM` *does not* delete Delta log files; log files are automatically cleaned up after checkpoints are written.
By default you can time travel to a Delta table up to 30 days old unless you have:
* Run `VACUUM` on your Delta table.
* Changed the data or log file retention periods using the following [table properties](/table-properties/):
* `delta.logRetentionDuration = "interval "`: controls how long the history for a table is kept. The default is `interval 30 days`.
Each time a checkpoint is written, Delta automatically cleans up log entries older than the retention interval. If you set this config to a large enough value, many log entries are retained. This should not impact performance as operations against the log are constant time. Operations on history are parallel but will become more expensive as the log size increases.
* `delta.deletedFileRetentionDuration = "interval "`: controls how long ago a file must have been deleted *before being a candidate for* `VACUUM`. The default is `interval 7 days`.
To access 30 days of historical data even if you run `VACUUM` on the Delta table, set `delta.deletedFileRetentionDuration = "interval 30 days"`. This setting may cause your storage costs to go up.
Note
Due to log entry cleanup, instances can arise where you cannot time travel to a version that is less than the retention interval. Delta Lake requires all consecutive log entries since the previous checkpoint to time travel to a particular version. For example, with a table initially consisting of log entries for versions \[0, 19] and a checkpoint at verison 10, if the log entry for version 0 is cleaned up, then you cannot time travel to versions \[1, 9]. Increasing the table property `delta.logRetentionDuration` can help avoid these situations.
## Write to table
[Section titled “Write to table”](#write-to-table)
### Append
[Section titled “Append”](#append)
To atomically add new data to an existing Delta table, use `append` mode:
* SQL
```sql
INSERT INTO default.people10m SELECT * FROM morePeople
```
* Python
```python
df.write.format("delta").mode("append").save("/tmp/delta/people10m")
df.write.format("delta").mode("append").saveAsTable("default.people10m")
```
* Scala
```scala
df.write.format("delta").mode("append").save("/tmp/delta/people10m")
df.write.format("delta").mode("append").saveAsTable("default.people10m")
import io.delta.implicits._
df.write.mode("append").delta("/tmp/delta/people10m")
```
### Overwrite
[Section titled “Overwrite”](#overwrite)
To atomically replace all the data in a table, use `overwrite` mode:
* SQL
```sql
INSERT OVERWRITE TABLE default.people10m SELECT * FROM morePeople
```
* Python
```python
df.write.format("delta").mode("overwrite").save("/tmp/delta/people10m")
df.write.format("delta").mode("overwrite").saveAsTable("default.people10m")
```
* Scala
```scala
df.write.format("delta").mode("overwrite").save("/tmp/delta/people10m")
df.write.format("delta").mode("overwrite").saveAsTable("default.people10m")
import io.delta.implicits._
df.write.mode("overwrite").delta("/tmp/delta/people10m")
```
Delta Lake provides several options to selectively overwrite only part of a table. The following table summarizes them. For most use cases, we recommend using `replaceUsing` or `replaceWhere`. Use `replaceOn` only when your use case requires complex or null-safe matching conditions.
| Option | Use case | Available since |
| ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------ |
| `replaceWhere` | Atomically overwrite rows that match a predicate. Use for replacements with a fixed matching condition, such as `colA = 5` or `int_col IN (1, 2, 3)`. | Delta Lake 1.1.0 (DataFrames); 2.4.0 (SQL) |
| `replaceUsing` | Dynamic data overwrite by column equality. Replaces all rows that match the specified columns, based on equality comparison of the column values, in the provided data set. | Delta Lake 4.3.0 |
| `replaceOn` | Dynamic data overwrite by Boolean expression. Use for replacements with a complex or null-safe matching condition, such as `s.colA <=> t.colA AND s.colB <=> t.colB`. | Delta Lake 4.3.0 |
| `partitionOverwriteMode` | Dynamic partition overwrite, which overwrites all the existing data in each partition for which the write will commit new data. Legacy since Delta Lake 4.3.0; not recommended for new workloads. | Delta Lake 2.0 |
#### Overwrite rows matching a predicate with `replaceWhere`
[Section titled “Overwrite rows matching a predicate with replaceWhere”](#overwrite-rows-matching-a-predicate-with-replacewhere)
`replaceWhere` atomically overwrites the rows that match a predicate, leaving all other rows unchanged. Use it for replacements with a fixed matching condition, such as `colA = 5` or `int_col IN (1, 2, 3)`. This feature is available with DataFrames in Delta Lake 1.1.0 and above and supported in SQL in Delta Lake 2.4.0 and above.
The following command atomically replaces events in January in the target table, which is partitioned by `start_date`, with the data in `replace_data`:
* SQL
```sql
INSERT INTO TABLE events REPLACE WHERE start_data >= '2017-01-01' AND end_date <= '2017-01-31' SELECT * FROM replace_data
```
* Python
```python
replace_data.write \
.format("delta") \
.mode("overwrite") \
.option("replaceWhere", "start_date >= '2017-01-01' AND end_date <= '2017-01-31'") \
.save("/tmp/delta/events")
```
* Scala
```scala
replace_data.write
.format("delta")
.mode("overwrite")
.option("replaceWhere", "start_date >= '2017-01-01' AND end_date <= '2017-01-31'")
.save("/tmp/delta/events")
```
This sample code writes out the data in `replace_data`, validates that it all matches the predicate, and performs an atomic replacement. If you want to write out data that doesn’t all match the predicate, to replace the matching rows in the target table, you can disable the constraint check by setting `spark.databricks.delta.replaceWhere.constraintCheck.enabled` to false:
* SQL
```sql
SET spark.databricks.delta.replaceWhere.constraintCheck.enabled=false
```
* Python
```python
spark.conf.set("spark.databricks.delta.replaceWhere.constraintCheck.enabled", False)
```
* Scala
```scala
spark.conf.set("spark.databricks.delta.replaceWhere.constraintCheck.enabled", false)
```
In Delta Lake 1.0.0 and below, `replaceWhere` overwrites data matching a predicate over partition columns only. The following command atomically replaces the month in January in the target table, which is partitioned by `date`, with the data in `df`:
* Python
```python
df.write \
.format("delta") \
.mode("overwrite") \
.option("replaceWhere", "birthDate >= '2017-01-01' AND birthDate <= '2017-01-31'") \
.save("/tmp/delta/people10m")
```
* Scala
```scala
df.write
.format("delta")
.mode("overwrite")
.option("replaceWhere", "birthDate >= '2017-01-01' AND birthDate <= '2017-01-31'")
.save("/tmp/delta/people10m")
```
In Delta Lake 1.1.0 and above, if you want to fall back to the old behavior, you can disable the `spark.databricks.delta.replaceWhere.dataColumns.enabled` flag:
* SQL
```sql
SET spark.databricks.delta.replaceWhere.dataColumns.enabled=false
```
* Python
```python
spark.conf.set("spark.databricks.delta.replaceWhere.dataColumns.enabled", False)
```
* Scala
```scala
spark.conf.set("spark.databricks.delta.replaceWhere.dataColumns.enabled", false)
```
#### Dynamically overwrite data with `replaceUsing` and `replaceOn`
[Section titled “Dynamically overwrite data with replaceUsing and replaceOn”](#dynamically-overwrite-data-with-replaceusing-and-replaceon)
Dynamic data overwrites selectively replace the data that matches the specified key columns or Boolean expression, leaving all other data unchanged. The `replaceUsing` option matches on key columns, while `replaceOn` matches on a Boolean expression. Partitioned tables, unpartitioned tables, and tables with liquid clustering are all supported. Both options are available with DataFrames in Delta Lake 4.3.0 and above.
Dynamic partition overwrites (the legacy [`partitionOverwriteMode`](#dynamic-partition-overwrites-with-partitionoverwritemode-legacy-since-delta-lake-430) option) are a subset of dynamic data overwrite behavior. Dynamic partition overwrites replace all the existing data in each partition for which the write will commit new data, and leave all other partitions unchanged. Only partitioned tables are supported in dynamic partition overwrite mode.
##### replaceUsing
[Section titled “replaceUsing”](#replaceusing)
`replaceUsing` atomically deletes the rows in the target table that match a row in the incoming DataFrame on the specified columns, then inserts the incoming DataFrame rows. Matching is based on equality comparison of the specified column values. Provide a comma-separated list of columns. Each column must exist in both the table and the incoming DataFrame, at the same ordinal position. The following command replaces rows whose `event_id` and `start_date` match between the incoming DataFrame and the target table:
* Python
```python
source_data.write \
.format("delta") \
.mode("overwrite") \
.option("replaceUsing", "event_id, start_date") \
.saveAsTable("events")
```
* Scala
```scala
source_data.write
.format("delta")
.mode("overwrite")
.option("replaceUsing", "event_id, start_date")
.saveAsTable("events")
```
If the source query is empty, no data is deleted.
Like a `JOIN USING`, `replaceUsing` matches rows with regular equality, where `NULL` is not equal to anything. Rows with `NULL` values in the specified columns don’t match and aren’t removed from the target table. For null-safe matching, or other logic that equality cannot express, use `replaceOn` instead.
For example, suppose the `students` table is clustered by `country`:
| name | country |
| ----- | ------- |
| Dylan | US |
| Doug | UK |
| Julia | IT |
| Adam | `NULL` |
and the `new_students` DataFrame holds the replacement rows:
| name | country |
| ------ | ------- |
| Peter | FR |
| Jennie | UK |
| Eva | `NULL` |
* Python
```python
# Replace rows in students whose country matches a row in new_students.
new_students.write \
.format("delta") \
.mode("overwrite") \
.option("replaceUsing", "country") \
.saveAsTable("students")
```
* Scala
```scala
// Replace rows in students whose country matches a row in new_students.
new_students.write
.format("delta")
.mode("overwrite")
.option("replaceUsing", "country")
.saveAsTable("students")
```
After the write, `students` contains:
| name | country |
| ------ | ------- |
| Dylan | US |
| Julia | IT |
| Jennie | UK |
| Peter | FR |
| Adam | `NULL` |
| Eva | `NULL` |
`Doug` is removed because `UK` matches a row in the incoming DataFrame. `Adam` keeps his row because his `NULL` country never matches, and `Eva`’s `NULL` row is appended for the same reason.
##### replaceOn
[Section titled “replaceOn”](#replaceon)
`replaceOn` replaces rows when they match a user-defined condition, unlike `replaceUsing`, which replaces rows when the specified columns compare equal under equality. Use `replaceOn` when you need matching logic that `replaceUsing` does not support, such as treating `NULL` values as equal with the `<=>` operator.
Optionally, use the `targetAlias` option to specify an alias for the target table and the `.as()` or `.alias()` APIs to specify an alias for the source data.
The following command replaces rows whose `event_id` and `start_date` match between the incoming DataFrame and the target table:
* Python
```python
source_data.alias("s").write \
.format("delta") \
.mode("overwrite") \
.option("targetAlias", "t") \
.option("replaceOn", "s.event_id <=> t.event_id AND s.start_date <=> t.start_date") \
.saveAsTable("events")
```
* Scala
```scala
source_data.as("s").write
.format("delta")
.mode("overwrite")
.option("targetAlias", "t")
.option("replaceOn", "s.event_id <=> t.event_id AND s.start_date <=> t.start_date")
.saveAsTable("events")
```
If the source query is empty, no data is deleted.
For example, suppose the `students` table contains:
| name | row\_origin |
| ------ | ----------- |
| Alice | table |
| `NULL` | table |
| Bob | table |
and the `people` DataFrame holds the replacement rows:
| name | row\_origin |
| ------ | ----------- |
| Alice | query |
| `NULL` | query |
| Delta | query |
* Python
```python
# Replace rows in students whose name matches a row in people, using a null-safe comparison.
people.alias("s").write \
.format("delta") \
.mode("overwrite") \
.option("targetAlias", "t") \
.option("replaceOn", "t.name <=> s.name") \
.saveAsTable("students")
```
* Scala
```scala
// Replace rows in students whose name matches a row in people, using a null-safe comparison.
people.as("s").write
.format("delta")
.mode("overwrite")
.option("targetAlias", "t")
.option("replaceOn", "t.name <=> s.name")
.saveAsTable("students")
```
After the write, `students` contains:
| name | row\_origin |
| ------ | ----------- |
| Alice | query |
| `NULL` | query |
| Bob | table |
| Delta | query |
Because `<=>` is null-safe, the `NULL` row matches and is replaced. `Bob` has no match in the incoming DataFrame, so his row is kept, and `Delta` is appended.
Note
The `replaceOn` and `replaceUsing` options can’t be combined with `replaceWhere` or dynamic partition overwrite (`partitionOverwriteMode`) in the same write.
#### Dynamic Partition Overwrites with `partitionOverwriteMode` (legacy since Delta Lake 4.3.0)
[Section titled “Dynamic Partition Overwrites with partitionOverwriteMode (legacy since Delta Lake 4.3.0)”](#dynamic-partition-overwrites-with-partitionoverwritemode-legacy-since-delta-lake-430)
Delta Lake 2.0 and above supports *dynamic* partition overwrite mode for partitioned tables.
When in dynamic partition overwrite mode, we overwrite all existing data in each logical partition for which the write will commit new data. Any existing logical partitions for which the write does not contain data will remain unchanged. This mode is only applicable when data is being written in overwrite mode: either `INSERT OVERWRITE` in SQL, or a DataFrame write with `df.write.mode("overwrite")`.
Configure dynamic partition overwrite mode by setting the Spark session configuration `spark.sql.sources.partitionOverwriteMode` to `dynamic`. You can also enable this by setting the `DataFrameWriter` option `partitionOverwriteMode` to `dynamic`. If present, the query-specific option overrides the mode defined in the session configuration. The default for `partitionOverwriteMode` is `static`.
* SQL
```sql
SET spark.sql.sources.partitionOverwriteMode=dynamic;
INSERT OVERWRITE TABLE default.people10m SELECT * FROM morePeople;
```
* Python
```python
df.write \
.format("delta") \
.mode("overwrite") \
.option("partitionOverwriteMode", "dynamic") \
.saveAsTable("default.people10m")
```
* Scala
```scala
df.write
.format("delta")
.mode("overwrite")
.option("partitionOverwriteMode", "dynamic")
.saveAsTable("default.people10m")
```
Note
Dynamic partition overwrite conflicts with the option `replaceWhere` for partitioned tables.
* If dynamic partition overwrite is enabled in the Spark session configuration, and `replaceWhere` is provided as a `DataFrameWriter` option, then Delta Lake overwrites the data according to the `replaceWhere` expression (query-specific options override session configurations).
* You’ll receive an error if the `DataFrameWriter` options have both dynamic partition overwrite and `replaceWhere` enabled.
Important
Validate that the data written with dynamic partition overwrite touches only the expected partitions. A single row in the incorrect partition can lead to unintentionally overwriting an entire partition.
If a partition has been accidentally overwritten, you can use [Restore a Delta table to an earlier state](/delta-utility/#restore-a-delta-table-to-an-earlier-state) to undo the change.
For Delta Lake support for updating tables, see [Table deletes, updates, and merges](/delta-update/).
### Limit rows written in a file
[Section titled “Limit rows written in a file”](#limit-rows-written-in-a-file)
You can use the SQL session configuration `spark.sql.files.maxRecordsPerFile` to specify the maximum number of records to write to a single file for a Delta Lake table. Specifying a value of zero or a negative value represents no limit.
You can also use the DataFrameWriter option `maxRecordsPerFile` when using the DataFrame APIs to write to a Delta Lake table. When `maxRecordsPerFile` is specified, the value of the SQL session configuration `spark.sql.files.maxRecordsPerFile` is ignored.
* Python
```python
df.write.format("delta") \
.mode("append") \
.option("maxRecordsPerFile", "10000") \
.save("/tmp/delta/people10m")
```
* Scala
```scala
df.write.format("delta")
.mode("append")
.option("maxRecordsPerFile", "10000")
.save("/tmp/delta/people10m")
```
### Idempotent writes
[Section titled “Idempotent writes”](#idempotent-writes)
Sometimes a job that writes data to a Delta table is restarted due to various reasons (for example, job encounters a failure). The failed job may or may not have written the data to Delta table before terminating. In the case where the data is written to the Delta table, the restarted job writes the same data to the Delta table which results in duplicate data.
To address this, Delta tables support the following `DataFrameWriter` options to make the writes idempotent:
* `txnAppId`: A unique string that you can pass on each `DataFrame` write. For example, this can be the name of the job.
* `txnVersion`: A monotonically increasing number that acts as transaction version. This number needs to be unique for data that is being written to the Delta table(s). For example, this can be the epoch seconds of the instant when the query is attempted for the first time. Any subsequent restarts of the same job needs to have the same value for `txnVersion`.
The above combination of options needs to be unique for each new data that is being ingested into the Delta table and the `txnVersion` needs to be higher than the last data that was ingested into the Delta table. For example:
* Last successfully written data contains option values as `dailyETL:23423` (`txnAppId:txnVersion`).
* Next write of data should have `txnAppId = dailyETL` and `txnVersion` as at least `23424` (one more than the last written data `txnVersion`).
* Any attempt to write data with `txnAppId = dailyETL` and `txnVersion` as `23422` or less is ignored because the `txnVersion` is less than the last recorded `txnVersion` in the table.
* Attempt to write data with `txnAppId:txnVersion` as `anotherETL:23424` is successful writing data to the table as it contains a different `txnAppId` compared to the same option value in last ingested data.
You can also configure idempotent writes by setting the Spark session configuration `spark.databricks.delta.write.txnAppId` and `spark.databricks.delta.write.txnVersion`. In addition, you can set `spark.databricks.delta.write.txnVersion.autoReset.enabled` to true to automatically reset `spark.databricks.delta.write.txnVersion` after every write. When both the writer options and session configuration are set, we will use the writer option values.
Warning
This solution assumes that the data being written to Delta table(s) in multiple retries of the job is same. If a write attempt in a Delta table succeeds but due to some downstream failure there is a second write attempt with same txn options but different data, then that second write attempt will be ignored. This can cause unexpected results.
#### Example
[Section titled “Example”](#example-1)
* SQL
```sql
SET spark.databricks.delta.write.txnAppId = ...;
SET spark.databricks.delta.write.txnVersion = ...;
SET spark.databricks.delta.write.txnVersion.autoReset.enabled = true; -- if set to true, this will reset txnVersion after every write
```
* Python
```python
app_id = ... # A unique string that is used as an application ID.
version = ... # A monotonically increasing number that acts as transaction version.
dataFrame.write.format(...).option("txnVersion", version).option("txnAppId", app_id).save(...)
```
* Scala
```scala
val appId = ... // A unique string that is used as an application ID.
version = ... // A monotonically increasing number that acts as transaction version.
dataFrame.write.format(...).option("txnVersion", version).option("txnAppId", appId).save(...)
```
### Set user-defined commit metadata
[Section titled “Set user-defined commit metadata”](#set-user-defined-commit-metadata)
You can specify user-defined strings as metadata in commits made by these operations, either using the DataFrameWriter option `userMetadata` or the SparkSession configuration `spark.databricks.delta.commitInfo.userMetadata`. If both of them have been specified, then the option takes preference. This user-defined metadata is readable in the [history](/delta-utility/#retrieve-delta-table-history) operation.
* SQL
```sql
SET spark.databricks.delta.commitInfo.userMetadata=overwritten-for-fixing-incorrect-data
INSERT OVERWRITE default.people10m SELECT * FROM morePeople
```
* Python
```python
df.write.format("delta") \
.mode("overwrite") \
.option("userMetadata", "overwritten-for-fixing-incorrect-data") \
.save("/tmp/delta/people10m")
```
* Scala
```scala
df.write.format("delta")
.mode("overwrite")
.option("userMetadata", "overwritten-for-fixing-incorrect-data")
.save("/tmp/delta/people10m")
```
## Schema validation
[Section titled “Schema validation”](#schema-validation)
Delta Lake automatically validates that the schema of the DataFrame being written is compatible with the schema of the table. Delta Lake uses the following rules to determine whether a write from a DataFrame to a table is compatible:
* All DataFrame columns must exist in the target table. If there are columns in the DataFrame not present in the table, an exception is raised. Columns present in the table but not in the DataFrame are set to null.
* DataFrame column data types must match the column data types in the target table. If they don’t match, an exception is raised.
* DataFrame column names cannot differ only by case. This means that you cannot have columns such as “Foo” and “foo” defined in the same table. While you can use Spark in case sensitive or insensitive (default) mode, Parquet is case sensitive when storing and returning column information. Delta Lake is case-preserving but insensitive when storing the schema and has this restriction to avoid potential mistakes, data corruption, or loss issues.
Delta Lake support DDL to add new columns explicitly and the ability to update schema automatically.
If you specify other options, such as `partitionBy`, in combination with append mode, Delta Lake validates that they match and throws an error for any mismatch. When `partitionBy` is not present, appends automatically follow the partitioning of the existing data.
## Update table schema
[Section titled “Update table schema”](#update-table-schema)
Delta Lake lets you update the schema of a table. The following types of changes are supported:
* Adding new columns (at arbitrary positions)
* Reordering existing columns
You can make these changes explicitly using DDL or implicitly using DML.
Important
When you update a Delta table schema, streams that read from that table terminate. If you want the stream to continue you must restart it.
### Explicitly update schema
[Section titled “Explicitly update schema”](#explicitly-update-schema)
You can use the following DDL to explicitly change the schema of a table.
#### Add columns
[Section titled “Add columns”](#add-columns)
* SQL
```sql
ALTER TABLE table_name ADD COLUMNS (col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name], ...)
```
By default, nullability is `true`.
To add a column to a nested field, use:
* SQL
```sql
ALTER TABLE table_name ADD COLUMNS (col_name.nested_col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name], ...)
```
##### Example
[Section titled “Example”](#example-2)
If the schema before running `ALTER TABLE boxes ADD COLUMNS (colB.nested STRING AFTER field1)` is:
```plaintext
- root
| - colA
| - colB
| +-field1
| +-field2
```
the schema after is:
```plaintext
- root
| - colA
| - colB
| +-field1
| +-nested
| +-field2
```
Note
Adding nested columns is supported only for structs. Arrays and maps are not supported.
#### Change column comment or ordering
[Section titled “Change column comment or ordering”](#change-column-comment-or-ordering)
* SQL
```sql
ALTER TABLE table_name ALTER [COLUMN] col_name col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name]
```
To change a column in a nested field, use:
* SQL
```sql
ALTER TABLE table_name ALTER [COLUMN] col_name.nested_col_name nested_col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name]
```
##### Example
[Section titled “Example”](#example-3)
If the schema before running `ALTER TABLE boxes CHANGE COLUMN colB.field2 field2 STRING FIRST` is:
```plaintext
- root
| - colA
| - colB
| +-field1
| +-field2
```
the schema after is:
```plaintext
- root
| - colA
| - colB
| +-field2
| +-field1
```
#### Replace columns
[Section titled “Replace columns”](#replace-columns)
* SQL
```sql
ALTER TABLE table_name REPLACE COLUMNS (col_name1 col_type1 [COMMENT col_comment1], ...)
```
##### Example
[Section titled “Example”](#example-4)
When running the following DDL:
* SQL
```sql
ALTER TABLE boxes REPLACE COLUMNS (colC STRING, colB STRUCT, colA STRING)
```
if the schema before is:
```plaintext
- root
| - colA
| - colB
| +-field1
| +-field2
```
the schema after is:
```plaintext
- root
| - colC
| - colB
| +-field2
| +-nested
| +-field1
| - colA
```
#### Rename columns
[Section titled “Rename columns”](#rename-columns)
Note
This feature is available in Delta Lake 1.2.0 and above. This feature is currently experimental.
To rename columns without rewriting any of the columns’ existing data, you must enable column mapping for the table. See [enable column mapping](/delta-column-mapping/).
To rename a column:
* SQL
```sql
ALTER TABLE table_name RENAME COLUMN old_col_name TO new_col_name
```
To rename a nested field:
* SQL
```sql
ALTER TABLE table_name RENAME COLUMN col_name.old_nested_field TO new_nested_field
```
##### Example
[Section titled “Example”](#example-5)
When you run the following command:
```sql
ALTER TABLE boxes RENAME COLUMN colB.field1 TO field001
```
If the schema before is:
```plaintext
- root
| - colA
| - colB
| +-field1
| +-field2
```
Then the schema after is:
```plaintext
- root
| - colA
| - colB
| +-field001
| +-field2
```
#### Drop columns
[Section titled “Drop columns”](#drop-columns)
Note
This feature is available in Delta Lake 2.0 and above. This feature is currently experimental.
To drop columns as a metadata-only operation without rewriting any data files, you must enable column mapping for the table. See [enable column mapping](/delta-column-mapping/).
Important
Dropping a column from metadata does not delete the underlying data for the column in files.
To drop a column:
* SQL
```sql
ALTER TABLE table_name DROP COLUMN col_name
```
To drop multiple columns:
* SQL
```sql
ALTER TABLE table_name DROP COLUMNS (col_name_1, col_name_2)
```
#### Change column type or name
[Section titled “Change column type or name”](#change-column-type-or-name)
You can change a column’s type or name or drop a column by rewriting the table. To do this, use the `overwriteSchema` option:
##### Change a column type
[Section titled “Change a column type”](#change-a-column-type)
* Python
```python
spark.read.table(...) \
.withColumn("birthDate", col("birthDate").cast("date")) \
.write \
.format("delta") \
.mode("overwrite")
.option("overwriteSchema", "true") \
.saveAsTable(...)
```
##### Change a column name
[Section titled “Change a column name”](#change-a-column-name)
* Python
```python
spark.read.table(...) \
.withColumnRenamed("dateOfBirth", "birthDate") \
.write \
.format("delta") \
.mode("overwrite") \
.option("overwriteSchema", "true") \
.saveAsTable(...)
```
### Automatic schema update
[Section titled “Automatic schema update”](#automatic-schema-update)
Delta Lake can automatically update the schema of a table as part of a DML transaction (either appending or overwriting), and make the schema compatible with the data being written.
#### Add columns
[Section titled “Add columns”](#add-columns-1)
Columns present in the source data but missing from the table are automatically added during a write transaction when schema evolution is enabled. The added columns are appended to the end of the struct they are present in. The case is preserved when a new column is appended.
##### Enable schema evolution (Delta Lake 4.3 and above)
[Section titled “Enable schema evolution (Delta Lake 4.3 and above)”](#enable-schema-evolution-delta-lake-43-and-above)
On Delta Lake 4.3 and above, use `WITH SCHEMA EVOLUTION` in a SQL `INSERT` statement or `.withSchemaEvolution()` in the DataFrame API to enable automatic schema evolution for a single operation. When enabled, the schema of the target Delta Lake table is automatically updated to accommodate additional columns or widened types of the source data. For a list of the supported type changes, see [Supported type changes](https://docs.delta.io/delta-type-widening/). For example:
* SQL
```sql
INSERT WITH SCHEMA EVOLUTION INTO target_table
SELECT * FROM source_table
```
* Python
```python
(spark.read
.table("source_table")
.write
.mode("append")
.withSchemaEvolution()
.saveAsTable("target_table")
)
```
* Scala
```scala
spark.read
.table("source_table")
.write
.mode("append")
.withSchemaEvolution()
.saveAsTable("target_table")
```
If the source data contains columns that don’t exist in the target table, those columns are automatically added to the `target_table` schema. Existing rows receive `NULL` values for the new columns.
##### Enable schema evolution in earlier versions
[Section titled “Enable schema evolution in earlier versions”](#enable-schema-evolution-in-earlier-versions)
In Delta Lake versions earlier than 4.3, enable schema evolution using one of the following alternatives. Both are also available in Delta Lake 4.3 and above.
Set the `mergeSchema` option on an individual `write` or `writeStream` operation:
* Python
```python
(spark.read
.table("source_table")
.write
.option("mergeSchema", "true")
.mode("append")
.saveAsTable("target_table")
)
```
* Scala
```scala
spark.read
.table("source_table")
.write
.option("mergeSchema", "true")
.mode("append")
.saveAsTable("target_table")
```
Alternatively, set the Spark configuration `spark.databricks.delta.schema.autoMerge.enabled` to `true` to enable schema evolution for all write operations in the current `SparkSession`:
* Python
```python
spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", True)
```
* Scala
```scala
spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", true)
```
* SQL
```sql
SET spark.databricks.delta.schema.autoMerge.enabled=true
```
Important
Enabling schema evolution session-wide is not recommended because it can lead to unintended schema changes across multiple operations and makes it harder to reason about which operations evolve the schema. Instead, enable schema evolution for each individual operation using `WITH SCHEMA EVOLUTION`, `.withSchemaEvolution()`, or the `mergeSchema` option.
When both an operation-level setting (the `mergeSchema` option or `WITH SCHEMA EVOLUTION` / `.withSchemaEvolution()`) and the session-wide Spark configuration are specified, the operation-level setting takes precedence.
#### `NullType` columns
[Section titled “NullType columns”](#nulltype-columns)
Because Parquet doesn’t support `NullType`, `NullType` columns are dropped from the DataFrame when writing into Delta tables, but are still stored in the schema. When a different data type is received for that column, Delta Lake merges the schema to the new data type. If Delta Lake receives a `NullType` for an existing column, the old schema is retained and the new column is dropped during the write.
`NullType` in streaming is not supported. Since you must set schemas when using streaming this should be very rare. `NullType` is also not accepted for complex types such as `ArrayType` and `MapType`.
## Replace table schema
[Section titled “Replace table schema”](#replace-table-schema)
By default, overwriting the data in a table does not overwrite the schema. When overwriting a table using mode `overwrite` without `replaceWhere`, you may still want to overwrite the schema of the data being written. You replace the schema and partitioning of the table by setting the `overwriteSchema` option to `true`:
* Python
```python
df.write.option("overwriteSchema", "true")
```
## Views on tables
[Section titled “Views on tables”](#views-on-tables)
Delta Lake supports the creation of views on top of Delta tables just like you might with a data source table.
The core challenge when you operate with views is resolving the schemas. If you alter a Delta table schema, you must recreate derivative views to account for any additions to the schema. For instance, if you add a new column to a Delta table, you must make sure that this column is available in the appropriate views built on top of that base table.
## Table properties
[Section titled “Table properties”](#table-properties)
You can store your own metadata as a table property using `TBLPROPERTIES` in `CREATE` and `ALTER`. You can then `SHOW` that metadata. For example:
* SQL
```sql
ALTER TABLE default.people10m SET TBLPROPERTIES ('department' = 'accounting', 'delta.appendOnly' = 'true');
-- Show the table's properties.
SHOW TBLPROPERTIES default.people10m;
-- Show just the 'department' table property.
SHOW TBLPROPERTIES default.people10m ('department');
```
`TBLPROPERTIES` are stored as part of Delta table metadata. You cannot define new `TBLPROPERTIES` in a `CREATE` statement if a Delta table already exists in a given location.
In addition, to tailor behavior and performance, Delta Lake supports certain Delta table properties:
* Block deletes and updates in a Delta table: `delta.appendOnly=true`.
* Configure the [time travel](#query-an-older-snapshot-of-a-table-time-travel) retention properties: `delta.logRetentionDuration=` and `delta.deletedFileRetentionDuration=`. For details, see [Data retention](#data-retention).
* Configure the number of columns for which statistics are collected: `delta.dataSkippingNumIndexedCols=n`. This property indicates to the writer that statistics are to be collected only for the first `n` columns in the table. Also the data skipping code ignores statistics for any column beyond this column index. This property takes affect only for new data that is written out.
Note
Modifying a Delta table property is a write operation that will conflict with other [concurrent write operations](/concurrency-control), causing them to fail. We recommend that you modify a table property only when there are no concurrent write operations on the table.
You can also set `delta.`-prefixed properties during the first commit to a Delta table using Spark configurations. For example, to initialize a Delta table with the property `delta.appendOnly=true`, set the Spark configuration `spark.databricks.delta.properties.defaults.appendOnly` to `true`. For example:
* SQL
```sql
spark.sql("SET spark.databricks.delta.properties.defaults.appendOnly = true")
```
* Python
```python
spark.conf.set("spark.databricks.delta.properties.defaults.appendOnly", "true")
```
* Scala
```scala
spark.conf.set("spark.databricks.delta.properties.defaults.appendOnly", "true")
```
See also the [Delta table properties reference](/table-properties/).
## Syncing table schema and properties to the Hive metastore
[Section titled “Syncing table schema and properties to the Hive metastore”](#syncing-table-schema-and-properties-to-the-hive-metastore)
You can enable asynchronous syncing of table schema and properties to the metastore by setting `spark.databricks.delta.catalog.update.enabled` to `true`. Whenever the Delta client detects that either of these two were changed due to an update, it will sync the changes to the metastore.
The schema is stored in the table properties in HMS. If the schema is small, it will be stored directly under the key `spark.sql.sources.schema`:
* JSON
```json
{
"spark.sql.sources.schema": "{'name':'col1','type':'string','nullable':true, 'metadata':{}},{'name':'col2','type':'string','nullable':true,'metadata':{}}"
}
```
If Schema is large, the schema will be broken down into multiple parts. Appending them together should give the correct schema. For example:
* JSON
```json
{
"spark.sql.sources.schema.numParts": "4",
"spark.sql.sources.schema.part.1": "{'name':'col1','type':'string','nullable':tr",
"spark.sql.sources.schema.part.2": "ue, 'metadata':{}},{'name':'co",
"spark.sql.sources.schema.part.3": "l2','type':'string','nullable':true,'meta",
"spark.sql.sources.schema.part.4": "data':{}}"
}
```
## Table metadata
[Section titled “Table metadata”](#table-metadata)
Delta Lake has rich features for exploring table metadata.
It supports `SHOW COLUMNS` and `DESCRIBE TABLE`.
It also provides the following unique commands:
### `DESCRIBE DETAIL`
[Section titled “DESCRIBE DETAIL”](#describe-detail)
Provides information about schema, partitioning, table size, and so on. For details, see [Retrieve Delta table details](/delta-utility/#retrieve-delta-table-history).
### `DESCRIBE HISTORY`
[Section titled “DESCRIBE HISTORY”](#describe-history)
Provides provenance information, including the operation, user, and so on, and operation metrics for each write to a table. Table history is retained for 30 days. For details, see [Retrieve Delta table history](/delta-utility/#retrieve-delta-table-history).
## Configure SparkSession
[Section titled “Configure SparkSession”](#configure-sparksession)
For many Delta Lake operations, you enable integration with Apache Spark DataSourceV2 and Catalog APIs (since 3.0) by setting the following configurations when you create a new `SparkSession`.
* Python
```python
from pyspark.sql import SparkSession
spark = SparkSession \
.builder \
.appName("...") \
.master("...") \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
.getOrCreate()
```
* Scala
```scala
import org.apache.spark.sql.SparkSession
val spark = SparkSession
.builder()
.appName("...")
.master("...")
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
.getOrCreate()
```
* Java
```java
import org.apache.spark.sql.SparkSession;
SparkSession spark = SparkSession
.builder()
.appName("...")
.master("...")
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
.getOrCreate();
```
Alternatively, you can add configurations when submitting your Spark application using `spark-submit` or when starting `spark-shell` or `pyspark` by specifying them as command-line parameters.
```bash
spark-submit --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog" ...
```
```bash
pyspark --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"
```
## Configure storage credentials
[Section titled “Configure storage credentials”](#configure-storage-credentials)
Delta Lake uses Hadoop FileSystem APIs to access storage systems. The credentials for storage systems usually can be set through Hadoop configurations. Delta Lake provides multiple ways to set Hadoop configurations similar to Apache Spark.
### Spark configurations
[Section titled “Spark configurations”](#spark-configurations)
When you start a Spark application on a cluster, you can set the Spark configurations in the form of `spark.hadoop.*` to pass your custom Hadoop configurations. For example, setting a value for `spark.hadoop.a.b.c` will pass the value as a Hadoop configuration `a.b.c`, and Delta Lake will use it to access Hadoop FileSystem APIs.
See [Spark documentation](http://spark.apache.org/docs/latest/configuration.html#custom-hadoophive-configuration) for more details.
### SQL session configurations
[Section titled “SQL session configurations”](#sql-session-configurations)
Spark SQL will pass all of the current [SQL session configurations](http://spark.apache.org/docs/latest/configuration.html#runtime-sql-configuration) to Delta Lake, and Delta Lake will use them to access Hadoop FileSystem APIs. For example, `SET a.b.c=x.y.z` will tell Delta Lake to pass the value `x.y.z` as a Hadoop configuration `a.b.c`, and Delta Lake will use it to access Hadoop FileSystem APIs.
### DataFrame options
[Section titled “DataFrame options”](#dataframe-options)
Besides setting Hadoop file system configurations through the Spark (cluster) configurations or SQL session configurations, Delta supports reading Hadoop file system configurations from `DataFrameReader` and `DataFrameWriter` options (that is, option keys that start with the `fs.` prefix) when the table is read or written, by using `DataFrameReader.load(path)` or `DataFrameWriter.save(path)`.
For example, you can pass your storage credentials through DataFrame options:
* Python
```python
df1 = spark.read.format("delta") \
.option("fs.azure.account.key..dfs.core.windows.net", "") \
.read("...")
df2 = spark.read.format("delta") \
.option("fs.azure.account.key..dfs.core.windows.net", "") \
.read("...")
df1.union(df2).write.format("delta") \
.mode("overwrite") \
.option("fs.azure.account.key..dfs.core.windows.net", "") \
.save("...")
```
* Scala
```scala
val df1 = spark.read.format("delta")
.option("fs.azure.account.key..dfs.core.windows.net", "")
.read("...")
val df2 = spark.read.format("delta")
.option("fs.azure.account.key..dfs.core.windows.net", "")
.read("...")
df1.union(df2).write.format("delta")
.mode("overwrite")
.option("fs.azure.account.key..dfs.core.windows.net", "")
.save("...")
```
You can find the details of the Hadoop file system configurations for your storage in [Storage configuration](/delta-storage/).
# Use catalog-managed tables
> Learn how to enable and use catalog-managed commits in Delta Lake.
Note
This feature is available in Delta Lake 4.0.1 and above.
[Catalog-Managed Tables](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#catalog-managed-tables) introduce a `catalogManaged` reader-writer table feature that changes how Delta Lake discovers and accesses tables. With this feature enabled, the catalog coordinates commit atomicity, allowing for features like multi-table transactions that are difficult to achieve with filesystem-only primitives.
## Overview
[Section titled “Overview”](#overview)
By default, Delta Lake relies entirely on the filesystem for read-time discovery and write-time commit atomicity. Each table manages its own transaction logs and conflict detection independently. Catalog-managed tables shift this responsibility to the managing catalog, which allows the catalog to orchestrate commits across multiple tables within a single transaction boundary while maintaining Delta Lake’s ACID guarantees.
Important
Filesystem-based access to catalog-managed tables is not supported. Delta clients must discover and access these tables through the managing catalog, not by direct path-based access. This ensures consistency across distributed environments. Users can use any catalog implementation that supports the Delta Catalog-Managed Table protocol.
## Requirements
[Section titled “Requirements”](#requirements)
* Catalog-managed tables requires the following Delta protocols:
* Reader version 3 or above.
* Writer version 7 or above.
* The [In-Commit Timestamps](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#in-commit-timestamps) table feature must be enabled, as commit publishing can occur asynchronously and file modification timestamps may not reflect actual commit times.
* The [VACUUM Protocol Check](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#vacuum-protocol-check) table feature must be enabled to provide proper vacuum operations on catalog-managed tables.
## Enable catalog-managed commits
[Section titled “Enable catalog-managed commits”](#enable-catalog-managed-commits)
You can enable catalog-managed commits for new tables when using a catalog that supports this feature, such as [Unity Catalog](https://www.unitycatalog.io/).
### Enable catalog-managed commits for new tables
[Section titled “Enable catalog-managed commits for new tables”](#enable-catalog-managed-commits-for-new-tables)
Enable the `catalogManaged` table feature by setting the following table property when creating a table:
* SQL
```sql
CREATE TABLE sales_data (
sale_id BIGINT,
amount DECIMAL(10,2),
sale_date DATE
)
TBLPROPERTIES ('delta.feature.catalogManaged' = 'supported');
```
Warning
When you enable catalog-managed commits, the table protocol version is upgraded. After upgrading, the table will not be readable by Delta Lake clients that do not support catalog-managed tables.
## Check if catalog-managed commits are enabled
[Section titled “Check if catalog-managed commits are enabled”](#check-if-catalog-managed-commits-are-enabled)
To verify whether a table has catalog-managed commits enabled:
* SQL
```sql
DESCRIBE DETAIL sales_data;
```
If enabled, `catalogManaged` appears in the `tableFeatures` column.
## Limitations
[Section titled “Limitations”](#limitations)
* Catalog-managed tables cannot be enabled on existing tables. Once enabled, the feature cannot be disabled.
* `CREATE OR REPLACE TABLE` is not supported for tables with catalog-managed commits enabled.
# Change data feed
> Learn how to get row-level change information from Delta tables using the Delta change data feed.
Change Data Feed (CDF) feature allows Delta tables to track row-level changes between versions of a Delta table. When enabled on a Delta table, the runtime records “change events” for all the data written into the table. This includes the row data along with metadata indicating whether the specified row was inserted, deleted, or updated.
You can read the change events in batch queries using DataFrame APIs (that is, `df.read`) and in streaming queries using DataFrame APIs (that is, `df.readStream`).
## Use cases
[Section titled “Use cases”](#use-cases)
Change Data Feed is not enabled by default. The following use cases should drive when you enable the change data feed.
* **Silver and Gold tables**: Improve Delta performance by processing only row-level changes following initial `MERGE`, `UPDATE`, or `DELETE` operations to accelerate and simplify ETL and ELT operations.
* **Transmit changes**: Send a change data feed to downstream systems such as Kafka or RDBMS that can use it to incrementally process in later stages of data pipelines.
* **Audit trail table**: Capture the change data feed as a Delta table provides perpetual storage and efficient query capability to see all changes over time, including when deletes occur and what updates were made.
## Enable change data feed
[Section titled “Enable change data feed”](#enable-change-data-feed)
You must explicitly enable the change data feed option using one of the following methods:
* **New table**: Set the table property `delta.enableChangeDataFeed = true` in the `CREATE TABLE` command.
* SQL
```sql
CREATE TABLE student (id INT, name STRING, age INT) TBLPROPERTIES (delta.enableChangeDataFeed = true)
```
* **Existing table**: Set the table property `delta.enableChangeDataFeed = true` in the `ALTER TABLE` command.
* SQL
```sql
ALTER TABLE myDeltaTable SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
```
* **All new tables**:
* SQL
```sql
set spark.databricks.delta.properties.defaults.enableChangeDataFeed = true;
```
Important
Once you enable the change data feed option for a table, you can no longer write to the table using Delta Lake 1.2.1 or below. You can always read the table.
Only changes made *after* you enable the change data feed are recorded; past changes to a table are not captured.
### Change data storage
[Section titled “Change data storage”](#change-data-storage)
Delta Lake records change data for `UPDATE`, `DELETE`, and `MERGE` operations in the `_change_data` folder under the Delta table directory. These records may be skipped when Delta Lake detects it can efficiently compute the change data feed directly from the transaction log. In particular, insert-only operations and full partition deletes will not generate data in the `_change_data` directory.
The files in the `_change_data` folder follow the retention policy of the table. Therefore, if you run the [VACUUM](/delta-utility/#remove-files-no-longer-referenced-by-a-delta-table) command, change data feed data is also deleted.
## Read changes in batch queries
[Section titled “Read changes in batch queries”](#read-changes-in-batch-queries)
You can provide either version or timestamp for the start and end. The start and end versions and timestamps are inclusive in the queries. To read the changes from a particular start version to the *latest* version of the table, specify only the starting version or timestamp.
You specify a version as an integer and a timestamps as a string in the format `yyyy-MM-dd[ HH:mm:ss[.SSS]]`.
If you provide a version lower or timestamp older than one that has recorded change events, that is, when the change data feed was enabled, an error is thrown indicating that the change data feed was not enabled.
* SQL
```sql
-- version as ints or longs e.g. changes from version 0 to 10
SELECT * FROM table_changes('tableName', 0, 10)
-- timestamp as string formatted timestamps
SELECT * FROM table_changes('tableName', '2021-04-21 05:45:46', '2021-05-21 12:00:00')
-- providing only the startingVersion/timestamp
SELECT * FROM table_changes('tableName', 0)
-- database/schema names inside the string for table name, with backticks for escaping dots and special characters
SELECT * FROM table_changes('dbName.`dotted.tableName`', '2021-04-21 06:45:46' , '2021-05-21 12:00:00')
-- path based tables
SELECT * FROM table_changes_by_path('\path', '2021-04-21 05:45:46')
```
* Python
```python
# version as ints or longs
spark.read.format("delta") \
.option("readChangeFeed", "true") \
.option("startingVersion", 0) \
.option("endingVersion", 10) \
.table("myDeltaTable")
# timestamps as formatted timestamp
spark.read.format("delta") \
.option("readChangeFeed", "true") \
.option("startingTimestamp", '2021-04-21 05:45:46') \
.option("endingTimestamp", '2021-05-21 12:00:00') \
.table("myDeltaTable")
# providing only the startingVersion/timestamp
spark.read.format("delta") \
.option("readChangeFeed", "true") \
.option("startingVersion", 0) \
.table("myDeltaTable")
# path based tables
spark.read.format("delta") \
.option("readChangeFeed", "true") \
.option("startingTimestamp", '2021-04-21 05:45:46') \
.load("pathToMyDeltaTable")
```
* Scala
```scala
// version as ints or longs
spark.read.format("delta")
.option("readChangeFeed", "true")
.option("startingVersion", 0)
.option("endingVersion", 10)
.table("myDeltaTable")
// timestamps as formatted timestamp
spark.read.format("delta")
.option("readChangeFeed", "true")
.option("startingTimestamp", "2021-04-21 05:45:46")
.option("endingTimestamp", "2021-05-21 12:00:00")
.table("myDeltaTable")
// providing only the startingVersion/timestamp
spark.read.format("delta")
.option("readChangeFeed", "true")
.option("startingVersion", 0)
.table("myDeltaTable")
// path based tables
spark.read.format("delta")
.option("readChangeFeed", "true")
.option("startingTimestamp", "2021-04-21 05:45:46")
.load("pathToMyDeltaTable")
```
## Read changes in streaming queries
[Section titled “Read changes in streaming queries”](#read-changes-in-streaming-queries)
* Python
```python
# providing a starting version
spark.readStream.format("delta") \
.option("readChangeFeed", "true") \
.option("startingVersion", 0) \
.table("myDeltaTable")
# providing a starting timestamp
spark.readStream.format("delta") \
.option("readChangeFeed", "true") \
.option("startingTimestamp", "2021-04-21 05:35:43") \
.load("/pathToMyDeltaTable")
# not providing a starting version/timestamp will result in the latest snapshot being fetched first
spark.readStream.format("delta") \
.option("readChangeFeed", "true") \
.table("myDeltaTable")
```
* Scala
```scala
// providing a starting version
spark.readStream.format("delta")
.option("readChangeFeed", "true")
.option("startingVersion", 0)
.table("myDeltaTable")
// providing a starting timestamp
spark.readStream.format("delta")
.option("readChangeFeed", "true")
.option("startingVersion", "2021-04-21 05:35:43")
.load("/pathToMyDeltaTable")
// not providing a starting version/timestamp will result in the latest snapshot being fetched first
spark.readStream.format("delta")
.option("readChangeFeed", "true")
.table("myDeltaTable")
```
To get the change data while reading the table, set the option `readChangeFeed` to `true`. The `startingVersion` or `startingTimestamp` are optional and if not provided the stream returns the latest snapshot of the table at the time of streaming as an `INSERT` and future changes as change data. Options like rate limits (`maxFilesPerTrigger`, `maxBytesPerTrigger`) and `excludeRegex` are also supported when reading change data.
Note
Rate limiting can be atomic for versions other than the starting snapshot version. That is, the entire commit version will be rate limited or the entire commit will be returned.
By default if a user passes in a version or timestamp exceeding the last commit on a table, the error `timestampGreaterThanLatestCommit` will be thrown. CDF can handle the out of range version case, if the user sets the following configuration to `true`.
* Python
```sql
set spark.databricks.delta.changeDataFeed.timestampOutOfRange.enabled = true;
```
If you provide a start version greater than the last commit on a table or a start timestamp newer than the last commit on a table, then when the preceding configuration is enabled, an empty read result is returned.
If you provide an end version greater than the last commit on a table or an end timestamp newer than the last commit on a table, then when the preceding configuration is enabled in batch read mode, all changes between the start version and the last commit are be returned.
## What is the schema for the change data feed?
[Section titled “What is the schema for the change data feed?”](#what-is-the-schema-for-the-change-data-feed)
When you read from the change data feed for a table, the schema for the latest table version is used.
Note
Most schema change and evolution operations are fully supported. Tables with column mapping enabled do not support all use cases and demonstrate different behavior. See [Change data feed limitations for tables with column mapping enabled](#change-data-feed-limitations-for-tables-with-column-mapping-enabled).
In addition to the data columns from the schema of the Delta table, change data feed contains metadata columns that identify the type of change event:
| Column name | Type | Values |
| :------------------ | :-------- | :-------------------------------------------------------------------- |
| `_change_type` | String | `insert`, `update_preimage` , `update_postimage`, `delete` [(1)](#-1) |
| `_commit_version` | Long | The Delta log or table version containing the change. |
| `_commit_timestamp` | Timestamp | The timestamp associated when the commit was created. |
[]()**(1)** `preimage` is the value before the update, `postimage` is the value after the update.
## Change data feed limitations for tables with column mapping enabled
[Section titled “Change data feed limitations for tables with column mapping enabled”](#change-data-feed-limitations-for-tables-with-column-mapping-enabled)
With column mapping enabled on a Delta table, you can drop or rename columns in the table without rewriting data files for existing data. With column mapping enabled, change data feed has limitations after performing non-additive schema changes such as renaming or dropping a column, changing data type, or nullability changes.
Important
In Delta Lake 2.0 and before, tables with column mapping enabled do not support streaming reads or batch reads on change data feed.
In Delta Lake 2.1, tables with column mapping enabled support batch reads on change data feed as long as there are no non-additive schema changes. Streaming reads of change data feed of tables with column mapping enabled is not supported.
In Delta Lake 2.2, tables with column mapping enabled support both batch and streaming reads on change data feed as long as there are no non-additive schema changes.
In Delta Lake 2.3 and above, you can perform batch reads on change data feed for tables with column mapping enabled that have experienced non-additive schema changes. Instead of using the schema of the latest version of the table, read operations use the schema of the end version of the table specified in the query. Queries still fail if the version range specified spans a non-additive schema change.
In Delta Lake 3.0 and above, you can perform streaming read on change data feed for tables with column mapping enabled that have experienced non-additive schema changes by enabling [schema tracking](/delta-streaming/#tracking-non-additive-schema-changes).
## Frequently asked questions (FAQ)
[Section titled “Frequently asked questions (FAQ)”](#frequently-asked-questions-faq)
### What is the overhead of enabling the change data feed?
[Section titled “What is the overhead of enabling the change data feed?”](#what-is-the-overhead-of-enabling-the-change-data-feed)
There is no significant impact. The change data records are generated in line during the query execution process, and are generally much smaller than the total size of rewritten files.
### What is the retention policy for change records?
[Section titled “What is the retention policy for change records?”](#what-is-the-retention-policy-for-change-records)
Change records follow the same retention policy as out-of-date table versions, and will be cleaned up through VACUUM if they are outside the specified retention period.
### When do new records become available in the change data feed?
[Section titled “When do new records become available in the change data feed?”](#when-do-new-records-become-available-in-the-change-data-feed)
Change data is committed along with the Delta Lake transaction, and will become available at the same time as the new data is available in the table.
# Use liquid clustering for Delta tables
> Learn about liquid clustering in Delta Lake.
Liquid clustering improves the existing partitioning and `ZORDER` techniques by simplifying data layout decisions in order to optimize query performance. Liquid clustering provides flexibility to redefine clustering columns without rewriting existing data, allowing data layout to evolve alongside analytic needs over time.
Note
This feature is available in Delta Lake 3.1.0 and above. See [Limitations](#limitations).
## What is liquid clustering used for?
[Section titled “What is liquid clustering used for?”](#what-is-liquid-clustering-used-for)
The following are examples of scenarios that benefit from clustering:
* Tables often filtered by high cardinality columns.
* Tables with significant skew in data distribution.
* Tables that grow quickly and require maintenance and tuning effort.
* Tables with access patterns that change over time.
* Tables where a typical partition column could leave the table with too many or too few partitions.
## Enable liquid clustering
[Section titled “Enable liquid clustering”](#enable-liquid-clustering)
You can enable liquid clustering on an existing table or during table creation. Clustering is not compatible with partitioning or `ZORDER`. Once enabled, run `OPTIMIZE` jobs as usual to incrementally cluster data. See [How to trigger clustering](#how-to-trigger-clustering).
To enable liquid clustering, add the `CLUSTER BY` phrase to a table creation statement, as in the examples below:
Note
In Delta Lake 3.2 and above, you can use DeltaTable API in Python or Scala to enable liquid clustering.
* SQL
```sql
-- Create an empty table
CREATE TABLE table1(col0 int, col1 string) USING DELTA CLUSTER BY (col0);
-- Using a CTAS statement (Delta 3.3+)
CREATE EXTERNAL TABLE table2 CLUSTER BY (col0) -- specify clustering after table name, not in subquery
LOCATION 'table_location'
AS SELECT * FROM table1;
```
* Python
```python
# Create an empty table
DeltaTable.create()
.tableName("table1")
.addColumn("col0", dataType = "INT")
.addColumn("col1", dataType = "STRING")
.clusterBy("col0")
.execute()
```
* Scala
```scala
// Create an empty table
DeltaTable.create()
.tableName("table1")
.addColumn("col0", dataType = "INT")
.addColumn("col1", dataType = "STRING")
.clusterBy("col0")
.execute()
```
Warning
Tables created with liquid clustering have `Clustering` and `DomainMetadata` table features enabled (both writer features) and use Delta writer version 7 and reader version 1. Table protocol versions cannot be downgraded. See [How does Delta Lake manage feature compatibility?](/versioning/).
In Delta Lake 3.3 and above you can enable liquid clustering on an existing unpartitioned Delta table using the following syntax:
* SQL
```sql
ALTER TABLE
CLUSTER BY ()
```
Important
Default behavior does not apply clustering to previously written data. To force reclustering for all records, you must use `OPTIMIZE FULL`. See [Recluster entire table](#recluster-entire-table).
## Choose clustering columns
[Section titled “Choose clustering columns”](#choose-clustering-columns)
Clustering columns can be defined in any order. If two columns are correlated, you only need to add one of them as a clustering column.
If you’re converting an existing table, consider the following recommendations:
| Current data optimization technique | Recommendation for clustering columns |
| --------------------------------------------------------------------------- | ------------------------------------------------------------------------------------ |
| Hive-style partitioning | Use partition columns as clustering columns. |
| Z-order indexing | Use the `ZORDER BY` columns as clustering columns. |
| Hive-style partitioning and Z-order | Use both partition columns and `ZORDER BY` columns as clustering columns. |
| Generated columns to reduce cardinality (for example, date for a timestamp) | Use the original column as a clustering column, and don’t create a generated column. |
## Write data to a clustered table
[Section titled “Write data to a clustered table”](#write-data-to-a-clustered-table)
You must use a Delta writer client that supports `Clustering` and `DomainMetadata` table features.
## How to trigger clustering
[Section titled “How to trigger clustering”](#how-to-trigger-clustering)
Use the `OPTIMIZE` command on your table, as in the following example:
* SQL
`sql OPTIMIZE table_name;`
Liquid clustering is incremental, meaning that data is only rewritten as necessary to accommodate data that needs to be clustered. Already clustered data files with different clustering columns are not rewritten.
### Recluster entire table
[Section titled “Recluster entire table”](#recluster-entire-table)
In Delta Lake 3.3 and above, you can force reclustering of all records in a table with the following syntax:
* SQL
`sql OPTIMIZE table_name FULL;`
Important
Running `OPTIMIZE FULL` reclusters all existing data as necessary. For large tables that have not previously been clustered on the specified columns, this operation might take hours.
Run `OPTIMIZE FULL` when you change clustering columns. If you have previously run `OPTIMIZE FULL` and there has been no change to clustering columns, `OPTIMIZE FULL` runs the same as `OPTIMIZE`. Always use `OPTIMIZE FULL` to ensure that data layout reflects the current clustering columns.
## Read data from a clustered table
[Section titled “Read data from a clustered table”](#read-data-from-a-clustered-table)
You can read data in a clustered table using any Delta Lake client. For best query results, include clustering columns in your query filters, as in the following example:
* SQL
```sql
```
## Change clustering columns
[Section titled “Change clustering columns”](#change-clustering-columns)
You can change clustering columns for a table at any time by running an `ALTER TABLE` command, as in the following example:
* SQL
`sql ALTER TABLE table_name CLUSTER BY (new_column1, new_column2);`
When you change clustering columns, subsequent `OPTIMIZE` and write operations use the new clustering approach, but existing data is not rewritten.
You can also turn off clustering by setting the columns to `NONE`, as in the following example:
* SQL
`sql ALTER TABLE table_name CLUSTER BY NONE;`
Setting cluster columns to `NONE` does not rewrite data that has already been clustered, but prevents future `OPTIMIZE` operations from using clustering columns.
## See how table is clustered
[Section titled “See how table is clustered”](#see-how-table-is-clustered)
You can use `DESCRIBE DETAIL` commands to see the clustering columns for a table, as in the following examples:
* SQL
`sql DESCRIBE DETAIL table_name;`
## Limitations
[Section titled “Limitations”](#limitations)
The following limitations exist:
* You can only specify columns with statistics collected for clustering columns. By default, the first 32 columns in a Delta table have statistics collected.
* You can specify up to 4 clustering columns.
Important
In Delta Lake 3.1, users needs to enable the feature flag `spark.databricks.delta.clusteredTable.enableClusteringTablePreview` to use liquid clustering. The following features are not supported in this preview:
* ZCube based incremental clustering
* `ALTER TABLE ... CLUSTER BY` to change clustering columns
* `DESCRIBE DETAIL` to inspect the current clustering columns
In Delta Lake 3.2, the preview flag is removed and the above features are supported.
# Delta column mapping
> Learn about column mapping in Delta.
Note
This feature is available in Delta Lake 1.2.0 and above. This feature is currently experimental with [known limitations](#known-limitations).
[Column mapping feature](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#column-mapping) allows Delta table columns and the underlying Parquet file columns to use different names. This enables Delta schema evolution operations such as `RENAME COLUMN` and `DROP COLUMNS` on a Delta table without the need to rewrite the underlying Parquet files. It also allows users to name Delta table columns by using [characters that are not allowed](#supported-characters-in-column-names) by Parquet, such as spaces, so that users can directly ingest CSV or JSON data into Delta without the need to rename columns due to previous character constraints.
## How to enable Delta Lake column mapping
[Section titled “How to enable Delta Lake column mapping”](#how-to-enable-delta-lake-column-mapping)
Important
Enabling column mapping for a table upgrades the Delta [table version](/versioning/#what-is-a-protocol-version). This protocol upgrade is irreversible. Tables with column mapping enabled can only be read in Delta Lake 1.2 and above.
Column mapping requires the following Delta protocols:
* Reader version 2 or above.
* Writer version 5 or above.
For a Delta table with the required protocol versions, you can enable column mapping by setting `delta.columnMapping.mode` to `name`.
You can use the following command to upgrade the table version and enable column mapping:
* SQL
```sql
ALTER TABLE SET TBLPROPERTIES (
'delta.minReaderVersion' = '2',
'delta.minWriterVersion' = '5',
'delta.columnMapping.mode' = 'name'
)
```
Note
You cannot turn off column mapping after you enable it. If you try to set `'delta.columnMapping.mode' = 'none'`, you’ll get an error.
## Rename a column
[Section titled “Rename a column”](#rename-a-column)
When column mapping is enabled for a Delta table, you can rename a column:
* SQL
```sql
ALTER TABLE RENAME COLUMN old_col_name TO new_col_name
```
For more examples, see [Rename columns](/delta-batch/#rename-columns).
## Drop columns
[Section titled “Drop columns”](#drop-columns)
When column mapping is enabled for a Delta table, you can drop one or more columns:
* SQL
```sql
ALTER TABLE table_name DROP COLUMN col_name;
ALTER TABLE table_name DROP COLUMNS (col_name_1, col_name_2, ...);
```
For more details, see [Drop columns](/delta-batch/#drop-columns).
## Supported characters in column names
[Section titled “Supported characters in column names”](#supported-characters-in-column-names)
When column mapping is enabled for a Delta table, you can include spaces as well as any of these characters in the table’s column names: `,;{}()\n\t=`.
## Known limitations
[Section titled “Known limitations”](#known-limitations)
* Enabling column mapping on tables might break downstream operations that rely on Delta change data feed. See [Change data feed limitations for tables with column mapping enabled](/delta-change-data-feed/#change-data-feed-limitations-for-tables-with-column-mapping-enabled).
* In Delta Lake 2.1 and below, [Spark Structured Streaming](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html) reads are explicitly blocked on a column mapping enabled table.
* In Delta Lake 2.2 and above, [Spark Structured Streaming](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html) reads are explicitly blocked on a column mapping enabled table that underwent column renaming or column dropping.
* In Delta Lake 3.0 and above, [Spark Structured Streaming](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html) reads require schema tracking to be enabled on a column mapping enabled table that underwent column renaming or column dropping. See [Tracking non-additive schema changes](/delta-streaming/#schema-tracking)
* The Delta table protocol specifies two modes of column mapping, by `name` and by `id`. Delta Lake 2.1 and below do not support `id` mode.
```plaintext
```
# Constraints
> Learn how Delta tables apply constraints.
Delta tables support standard SQL constraint management clauses that ensure that the quality and integrity of data added to a table is automatically verified. When a constraint is violated, Delta Lake throws an `InvariantViolationException` to signal that the new data can’t be added.
Important
Adding a constraint automatically upgrades the table writer protocol version. See [How does Delta Lake manage feature compatibility?](/versioning/) to understand table protocol versioning and what it means to upgrade the protocol version.
Two types of constraints are supported:
* `NOT NULL`: indicates that values in specific columns cannot be null.
* `CHECK`: indicates that a specified Boolean expression must be true for each input row.
### `NOT NULL` constraint
[Section titled “NOT NULL constraint”](#not-null-constraint)
You specify `NOT NULL` constraints in the schema when you create a table and drop `NOT NULL` constraints using the `ALTER TABLE CHANGE COLUMN` command.
* SQL
```sql
CREATE TABLE default.people10m (
id INT NOT NULL,
firstName STRING,
middleName STRING NOT NULL,
lastName STRING,
gender STRING,
birthDate TIMESTAMP,
ssn STRING,
salary INT
) USING DELTA;
ALTER TABLE default.people10m CHANGE COLUMN middleName DROP NOT NULL;
```
If you specify a `NOT NULL` constraint on a column nested within a struct, the parent struct is also constrained to not be null. However, columns nested within array or map types do not accept `NOT NULL` constraints.
### `CHECK` constraint
[Section titled “CHECK constraint”](#check-constraint)
You manage `CHECK` constraints using the `ALTER TABLE ADD CONSTRAINT` and `ALTER TABLE DROP CONSTRAINT` commands. `ALTER TABLE ADD CONSTRAINT` verifies that all existing rows satisfy the constraint before adding it to the table.
* SQL
```sql
CREATE TABLE default.people10m (
id INT,
firstName STRING,
middleName STRING,
lastName STRING,
gender STRING,
birthDate TIMESTAMP,
ssn STRING,
salary INT
) USING DELTA;
ALTER TABLE default.people10m ADD CONSTRAINT dateWithinRange CHECK (birthDate > '1900-01-01');
ALTER TABLE default.people10m DROP CONSTRAINT dateWithinRange;
```
`CHECK` constraints are table properties in the output of the `DESCRIBE DETAIL` and `SHOW TBLPROPERTIES` commands.
* SQL
```sql
ALTER TABLE default.people10m ADD CONSTRAINT validIds CHECK (id > 1 and id < 99999999);
DESCRIBE DETAIL default.people10m;
SHOW TBLPROPERTIES default.people10m;
```
# Delta default column values
> Learn about default column values in Delta.
Note
This feature is available in Delta Lake 3.1.0 and above and is enabled using the `allowColumnDefaults` writer [table feature](/versioning/#what-are-table-features).
Delta enables the specification of [default expressions](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#default-columns) for columns in Delta tables. When users write to these tables without explicitly providing values for certain columns, or when they explicitly use the DEFAULT SQL keyword for a column, Delta automatically generates default values for those columns.
This information is stored in the [StructField](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#struct-field) corresponding to the column of interest.
## How to enable Delta Lake default column values
[Section titled “How to enable Delta Lake default column values”](#how-to-enable-delta-lake-default-column-values)
Important
Enabling default column values for a Delta table will upgrade its [protocol version](/versioning/#what-is-a-protocol-version) to support [table features](/versioning/#what-are-table-features). This protocol upgrade is irreversible. Tables with default column values enabled can only be written to in Delta Lake 3.1 and above.
You can enable default column values for a table by setting `delta.feature.allowColumnDefaults` to `enabled`:
* SQL
```sql
ALTER TABLE SET TBLPROPERTIES (
'delta.feature.allowColumnDefaults' = 'enabled'
)
```
## How to use default columns in SQL commands
[Section titled “How to use default columns in SQL commands”](#how-to-use-default-columns-in-sql-commands)
* For SQL commands that perform table writes, such as `INSERT`, `UPDATE`, and `MERGE` commands, the `DEFAULT` keyword resolves to the most recently assigned default value for the corresponding column (or NULL if no default value exists). For instance, the following SQL command will use the default value for the second column in the table: `INSERT INTO t VALUES (16, DEFAULT);`
* It is also possible for INSERT commands to specify lists of fewer columns than the target table, in which case the engine will assign default values for the remaining columns (or NULL for any columns where no defaults yet exist).
Important
The metadata discussed here apply solely to write operations, not read operations.
* The `ALTER TABLE ... ADD COLUMN` command that introduces a new column to an existing table may not specify a default value for the new column. For instance, the following SQL command is not supported in Delta Lake: `ALTER TABLE t ADD COLUMN c INT DEFAULT 16;`
* It is permissible, however, to assign or update default values for columns that were created in previous commands. For example, the following SQL command is valid: `ALTER TABLE t ALTER COLUMN c SET DEFAULT 16;`
# What are deletion vectors?
> Learn about deletion vectors in Delta Lake.
Note
This feature is available in Delta Lake 2.3.0 and above. This feature is in experimental support mode.
Deletion vectors are a storage optimization feature that can be enabled on Delta Lake tables. By default, when a single row in a data file is deleted, the entire Parquet file containing the record must be rewritten. With deletion vectors enabled for the table, some Delta operations use deletion vectors to mark existing rows as removed without rewriting the Parquet file. Subsequent reads on the table resolve current table state by applying the deletions noted by deletion vectors to the most recent table version.
Support for deletion vectors was incrementally added with each Delta Lake version. The table below depicts the supported operations for each Delta Lake version.
| Operation | First available Delta Lake version | Enabled by default since Delta Lake version |
| --------- | ---------------------------------- | ------------------------------------------- |
| `SCAN` | 2.3.0 | 2.3.0 |
| `DELETE` | 2.4.0 | 2.4.0 |
| `UPDATE` | 3.0.0 | 3.1.0 |
| `MERGE` | 3.1.0 | 3.1.0 |
## Enable deletion vectors
[Section titled “Enable deletion vectors”](#enable-deletion-vectors)
You enable support for deletion vectors on a Delta Lake table by setting a Delta Lake table property:
* SQL
```sql
ALTER TABLE SET TBLPROPERTIES('delta.enableDeletionVectors' = true);
```
Warning
When you enable deletion vectors, the table protocol version is upgraded. After upgrading, the table will not be readable by Delta Lake clients that do not support deletion vectors. See [How does Delta Lake manage feature compatibility?](/versioning/).
In Delta Lake 3.0 and above, you can drop the deletion vectors table feature to enable compatibility with other Delta clients. See [Drop Delta table features](/delta-drop-feature/).
## Apply changes to Parquet data files
[Section titled “Apply changes to Parquet data files”](#apply-changes-to-parquet-data-files)
Deletion vectors indicate changes to rows as soft-deletes that logically modify existing Parquet data files in the Delta Lake tables. These changes are applied physically when data files are rewritten, as triggered by one of the following events:
* A DML command with deletion vectors disabled (by a command flag or a table property) is run on the table.
* An `OPTIMIZE` command is run on the table.
* `REORG TABLE ... APPLY (PURGE)` is run against the table.
`UPDATE`, `MERGE`, and `OPTIMIZE` do not have strict guarantees for resolving changes recorded in deletion vectors, and some changes recorded in deletion vectors might not be applied if target data files contain no updated records, or would not otherwise be candidates for file compaction. `REORG TABLE ... APPLY (PURGE)` rewrites all data files containing records with modifications recorded using deletion vectors. See [Apply changes with REORG TABLE](#apply-changes-with-reorg-table)
Note
Modified data might still exist in the old files. You can run `VACUUM` to physically delete the old files. `REORG TABLE ... APPLY (PURGE)` creates a new version of the table at the time it completes, which is the timestamp you must consider for the retention threshold for your `VACUUM` operation to fully remove deleted files.
### Apply changes with REORG TABLE
[Section titled “Apply changes with REORG TABLE”](#apply-changes-with-reorg-table)
Reorganize a Delta Lake table by rewriting files to purge soft-deleted data, such as rows marked as deleted by deletion vectors with `REORG TABLE`:
* SQL
```sql
REORG TABLE events APPLY (PURGE);
-- If you have a large amount of data and only want to purge a subset of it, you can specify an optional partition predicate using `WHERE`:
REORG TABLE events WHERE date >= '2022-01-01' APPLY (PURGE);
REORG TABLE events
WHERE date >= current_timestamp() - INTERVAL '1' DAY
APPLY (PURGE);
```
Note
* `REORG TABLE` only rewrites files that contain soft-deleted data. - When resulting files of the purge are small, `REORG TABLE` will coalesce them into larger ones. See [OPTIMIZE](/optimizations-oss/) for more info. - `REORG TABLE` is *idempotent*, meaning that if it is run twice on the same dataset, the second run has no effect. - After running `REORG TABLE`, the soft-deleted data may still exist in the old files. You can run [VACUUM](/delta-utility/#vacuum) to physically delete the old files.
# Drop Delta table features
> Learn how to drop table features in Delta Lake to downgrade reader and writer protocol requirements and resolve compatibility issues.
This article describes how to drop Delta Lake table features and downgrade protocol versions.
Note
This feature is available in Delta Lake 4.0.0 and above. A legacy implementation of the feature is available since Delta Lake 3.0.0. Not all Delta table features can be dropped. See [What Delta table features can be dropped?](#what-delta-table-features-can-be-dropped)
You should only use this functionality to support compatibility with earlier Delta Lake versions, Delta Sharing, or other Delta Lake reader or writer clients.
## How can I drop a Delta table feature?
[Section titled “How can I drop a Delta table feature?”](#how-can-i-drop-a-delta-table-feature)
To remove a Delta table feature, you run an `ALTER TABLE DROP FEATURE ` command.
## What Delta table features can be dropped?
[Section titled “What Delta table features can be dropped?”](#what-delta-table-features-can-be-dropped)
You can drop the following Delta table features:
* `deletionVectors`. See [What are deletion vectors?](/delta-deletion-vectors/). Drop support for deletion vectors is available in Delta Lake 4.0.0 and above.
* `typeWidening-preview`. See [Delta type widening](/delta-type-widening/). Type widening is available in preview in Delta Lake 3.2.0 and above.
* `typeWidening`. See [Delta type widening](/delta-type-widening/). Type widening is available in preview in Delta Lake 4.0.0 and above.
* `v2Checkpoint`. See [V2 Checkpoint Spec](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#v2-spec). Drop support for V2 Checkpoints is available in Delta Lake 3.1.0 and above.
* `columnMapping`. See [Delta column mapping](/delta-column-mapping/). Drop support for column mapping is available in Delta Lake 3.3.0 and above.
* `vacuumProtocolCheck`. See [Vacuum Protocol Check Spec](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#vacuum-protocol-check). Drop support for vacuum protocol check is available in Delta Lake 3.3.0 and above.
* `checkConstraints`. See [Constraints](/delta-constraints/). Drop support for check constraints is available in Delta Lake 3.3.0 and above.
* `inCommitTimestamp`. See [In-Commit Timestamps](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#in-commit-timestamps). Drop support for In-Commit Timestamp is available in Delta Lake 3.3.0 and above.
* `checkpointProtection`. See [Checkpoint Protection Spec](https://github.com/delta-io/delta/blob/master/protocol_rfcs/checkpoint-protection.md). Drop support for checkpoint protection is available in Delta Lake 4.0.0 and above.
You cannot drop other [Delta table features](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#valid-feature-names-in-table-features).
## What happens when a table feature is dropped?
[Section titled “What happens when a table feature is dropped?”](#what-happens-when-a-table-feature-is-dropped)
When you drop a table feature, Delta Lake performs a series of atomic operations:
* Disable table properties that use the table feature.
* Rewrite data files as necessary to remove all traces of the table feature from the data files backing the table in the current version.
* Create a set of protected checkpoints that allow reader clients to interpret table history correctly.
* Add the writer table feature checkpointProtection to the table protocol.
* Downgrade the table protocol to the lowest reader and writer versions that support all remaining table features.
## What is the checkpointProtection table feature?
[Section titled “What is the checkpointProtection table feature?”](#what-is-the-checkpointprotection-table-feature)
When you drop a feature, Delta Lake rewrites data and metadata in the table’s history as protected checkpoints to respect the protocol downgrade. After the downgrade, the table should always be readable by more clients. This is because the protocol for the table now reflects that support for the dropped feature is no longer required to read the table. The protected checkpoints and the checkpointProtection feature accomplish the following:
* Reader clients that understand the dropped table feature can access all available table history.
* Reader clients that do not support the dropped table feature only need to read the table history starting from the protocol downgrade version.
* Writer clients do not rewrite checkpoints prior to the protocol downgrade.
* Table maintenance operations respect requirements set by `checkpointProtection`, which mark protocol downgrade checkpoints as protected.
* While you can only drop one table feature with each DROP FEATURE command, a table can have multiple protected checkpoints and dropped features in its table history.
The table feature `checkpointProtection` should not block read-only access from Delta Lake clients. To fully downgrade the table and remove the `checkpointProtection` table feature, you must use TRUNCATE HISTORY. The recommendation is to only use this pattern if you need to write to tables with external Delta clients that do not support checkpointProtection.
## Fully downgrade table protocols for legacy clients
[Section titled “Fully downgrade table protocols for legacy clients”](#fully-downgrade-table-protocols-for-legacy-clients)
If integrations with external Delta Lake clients require writes that don’t support the checkpointProtection table feature, you must use TRUNCATE HISTORY to fully remove all traces of the disabled table features and fully downgrade the table protocol.
It is recommended to always test the default behavior for DROP FEATURE before proceeding with TRUNCATE HISTORY. Running TRUNCATE HISTORY removes all table history greater than 24 hours.
Full table downgrade occurs in two steps that must occur at least 24 hours apart.
### Step 1: Prepare to drop a table feature
[Section titled “Step 1: Prepare to drop a table feature”](#step-1-prepare-to-drop-a-table-feature)
During the first stage, the user prepares to drop the table feature. The following describes what happens during this stage:
1. You run the `ALTER TABLE DROP FEATURE TRUNCATE HISTORY` command.
2. Table properties that specifically enable a table feature have values set to disable the feature.
3. Table properties that control behaviors associated with the dropped feature have options set to default values before the feature was introduced.
4. As necessary, data and metadata files are rewritten respecting the updated table properties.
5. The command finishes running and returns an error message informing the user they must wait 24 hours to proceed with feature removal.
After first disabling a feature, you can continue writing to the target table before completing the protocol downgrade, but you cannot use the table feature you are removing.
Note
If you leave the table in this state, operations against the table do not use the table feature, but the protocol still supports the table feature. Until you complete the final downgrade step, the table is not readable by Delta clients that do not understand the table feature.
### Step 2: Downgrade the protocol and drop a table feature
[Section titled “Step 2: Downgrade the protocol and drop a table feature”](#step-2-downgrade-the-protocol-and-drop-a-table-feature)
To fully remove all transaction history associated with the feature and downgrade the protocol:
1. After at least 24 hours have passed, you run the `ALTER TABLE DROP FEATURE TRUNCATE HISTORY` command.
2. The client confirms that no transactions in the specified retention threshold use the table feature, then truncates the table history to that threshold.
3. The protocol is downgraded, dropping the table feature.
4. If the table features that are present in the table can be represented by a legacy protocol version, the `minReaderVersion` and `minWriterVersion` for the table are downgraded to the lowest version that supports exactly all remaining features in use by the Delta table.
Important
Running `ALTER TABLE DROP FEATURE TRUNCATE HISTORY` removes all transaction log data older than 24 hours. After dropping a Delta table feature, you do not have access to table history or time travel.
See [How does Delta Lake manage feature compatibility?](/versioning/).
# Frequently asked questions (FAQ)
> Find answers to commonly asked questions about Delta Lake.
## What is Delta Lake?
[Section titled “What is Delta Lake?”](#what-is-delta-lake)
[Delta Lake](https://delta.io/) is an [open source storage layer](https://github.com/delta-io/delta) that brings reliability to [data lakes](https://databricks.com/discover/data-lakes/introduction). Delta Lake provides ACID transactions, scalable metadata handling, and unifies streaming and batch data processing. Delta Lake runs on top of your existing data lake and is fully compatible with Apache Spark APIs.
## How is Delta Lake related to Apache Spark?
[Section titled “How is Delta Lake related to Apache Spark?”](#how-is-delta-lake-related-to-apache-spark)
Delta Lake sits on top of Apache Spark. The format and the compute layer helps to simplify building big data pipelines and increase the overall efficiency of your pipelines.
## What format does Delta Lake use to store data?
[Section titled “What format does Delta Lake use to store data?”](#what-format-does-delta-lake-use-to-store-data)
Delta Lake uses versioned Parquet files to store your data in your cloud storage. Apart from the versions, Delta Lake also stores a transaction log to keep track of all the commits made to the table or blob store directory to provide ACID transactions.
## How can I read and write data with Delta Lake?
[Section titled “How can I read and write data with Delta Lake?”](#how-can-i-read-and-write-data-with-delta-lake)
You can use your favorite Apache Spark APIs to read and write data with Delta Lake. See [Read a table](/delta-batch/#read-a-table) and [Write to a table](/delta-batch/#write-to-table).
## Where does Delta Lake store the data?
[Section titled “Where does Delta Lake store the data?”](#where-does-delta-lake-store-the-data)
When writing data, you can specify the location in your cloud storage. Delta Lake stores the data in that location in Parquet format.
## Can I copy my Delta Lake table to another location?
[Section titled “Can I copy my Delta Lake table to another location?”](#can-i-copy-my-delta-lake-table-to-another-location)
Yes you can copy your Delta Lake table to another location. Remember to copy files without changing the timestamps to ensure that the time travel with timestamps will be consistent.
## Can I stream data directly into and from Delta tables?
[Section titled “Can I stream data directly into and from Delta tables?”](#can-i-stream-data-directly-into-and-from-delta-tables)
Yes, you can use Structured Streaming to directly write data into Delta tables and read from Delta tables. See [Stream data into Delta tables](/delta-streaming/#delta-table-as-a-sink) and [Stream data from Delta tables](/delta-streaming/#delta-table-as-a-source).
## Does Delta Lake support writes or reads using the Spark Streaming DStream API?
[Section titled “Does Delta Lake support writes or reads using the Spark Streaming DStream API?”](#does-delta-lake-support-writes-or-reads-using-the-spark-streaming-dstream-api)
Delta does not support the DStream API. We recommend [Table streaming reads and writes](/delta-streaming/).
## When I use Delta Lake, will I be able to port my code to other Spark platforms easily?
[Section titled “When I use Delta Lake, will I be able to port my code to other Spark platforms easily?”](#when-i-use-delta-lake-will-i-be-able-to-port-my-code-to-other-spark-platforms-easily)
Yes. When you use Delta Lake, you are using open Apache Spark APIs so you can easily port your code to other Spark platforms. To port your code, replace `delta` format with `parquet` format.
## Does Delta Lake support multi-table transactions?
[Section titled “Does Delta Lake support multi-table transactions?”](#does-delta-lake-support-multi-table-transactions)
Delta Lake does not support multi-table transactions and foreign keys. Delta Lake supports transactions at the *table* level.
## How can I change the type of a column?
[Section titled “How can I change the type of a column?”](#how-can-i-change-the-type-of-a-column)
Changing a column’s type or dropping a column requires rewriting the table. For an example, see [Change column type](/delta-batch/#change-column-type-or-name).
# Delta Kernel
> Learn how to build connectors to read and write Delta tables.
The Delta Kernel project is a set of libraries ([Java](#kernel-java) and [Rust](#kernel-rust)) for building Delta connectors that can read from and write into Delta tables without the need to understand the [Delta protocol details](https://github.com/delta-io/delta/blob/master/PROTOCOL.md).
You can use this library to do the following:
* Read data from small Delta tables in a single thread in a single process.
* Read data from large Delta tables using multiple threads in a single process.
* Build a complex connector for a distributed processing engine and read very large Delta tables.
* Insert data into a Delta table either from a single process or a complex distributed engine.
Here is an example of a simple table scan with a filter:
* Java
```java
Engine myEngine = DefaultEngine.create() ; // define a engine (more details below)
Table myTable = Table.forPath("/delta/table/path"); // define what table to scan
Snapshot mySnapshot = myTable.getLatestSnapshot(myEngine); // define which version of table to scan
Scan myScan = mySnapshot.getScanBuilder(myEngine) // specify the scan details
.withFilters(myEngine, scanFilter)
.build();
CloseableIterator physicalData = // read the Parquet data files
.. read from Parquet data files ...
Scan.transformPhysicalData(...) // returns the table data
```
A complete version of the above example program and more examples of reading from and writing into a Delta table are available [here](https://github.com/delta-io/delta/tree/master/kernel/examples).
Notice that there are two sets of public APIs to build connectors.
* **Table APIs** - Interfaces like [`Table`](https://delta-io.github.io/delta/snapshot/kernel-api/java/index.html?io/delta/kernel/Table.html) and [`Snapshot`](https://delta-io.github.io/delta/snapshot/kernel-api/java/index.html?io/delta/kernel/Snapshot.html) that allow you to read (and soon write to) Delta tables
* **Engine APIs** - The [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java//index.html?io/delta/kernel/engine/Engine.html) interface allows you to plug in connector-specific optimizations to compute-intensive components in the Kernel. For example, Delta Kernel provides a *default* Parquet file reader via the `DefaultEngine`, but you may choose to replace that default with a custom `Engine` implementation that has a faster Parquet reader for your connector/processing engine.
## Kernel Java
[Section titled “Kernel Java”](#kernel-java)
## What is Delta Kernel?
[Section titled “What is Delta Kernel?”](#what-is-delta-kernel)
Delta Kernel is a library for operating on Delta tables. Specifically, it provides simple and narrow APIs for reading and writing to Delta tables without the need to understand the [Delta protocol](https://github.com/delta-io/delta/blob/master/PROTOCOL.md) details. You can use this library to do the following:
* Read and write Delta tables from your applications.
* Build a connector for a distributed engine like [Apache Spark™](https://github.com/apache/spark), [Apache Flink](https://github.com/apache/flink), or [Trino](https://github.com/trinodb/trino) for reading or writing massive Delta tables.
## Set up Delta Kernel for your project
[Section titled “Set up Delta Kernel for your project”](#set-up-delta-kernel-for-your-project)
You need to `io.delta:delta-kernel-api` and `io.delta:delta-kernel-defaults` dependencies. Following is an example Maven `pom` file dependency list.
The `delta-kernel-api` module contains the core of the Kernel that abstracts out the Delta protocol to enable reading and writing into Delta tables. It makes use of the `Engine` interface that is being passed to the Kernel API by the connector for heavy-lift operations such as reading/writing Parquet or JSON files, evaluating expressions or file system operations such as listing contents of the Delta Log directory, etc. Kernel supplies a default implementation of `Engine` in module `delta-kernel-defaults`. The connectors can implement their own version of `Engine` to make use of their native implementation of functionalities the `Engine` provides. For example: the connector can make use of their Parquet reader instead of using the reader from the `DefaultEngine`. More details on this [later](#step-2-build-your-own-engine).
```xml
io.delta
delta-kernel-api
${delta-kernel.version}
io.delta
delta-kernel-defaults
${delta-kernel.version}
```
If your connector is not using the [`DefaultEngine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) provided by the Kernel, the dependency `delta-kernel-defaults` from the above list can be skipped.
## Read a Delta table in a single process
[Section titled “Read a Delta table in a single process”](#read-a-delta-table-in-a-single-process)
In this section, we will walk through how to build a very simple single-process Delta connector that can read a Delta table using the default [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) implementation provided by Delta Kernel.
You can either write this code yourself in your project, or you can use the [examples](https://github.com/delta-io/delta/tree/master/kernel/examples) present in the Delta code repository.
### Step 1: Full scan on a Delta table
[Section titled “Step 1: Full scan on a Delta table”](#step-1-full-scan-on-a-delta-table)
The main entry point is [`io.delta.kernel.Table`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Table.html) which is a programmatic representation of a Delta table. Say you have a Delta table at the directory `myTablePath`. You can create a `Table` object as follows:
```java
import io.delta.kernel.*;
import io.delta.kernel.defaults.*;
import org.apache.hadoop.conf.Configuration;
String myTablePath = ; // fully qualified table path. Ex: file:/user/tables/myTable
Configuration hadoopConf = new Configuration();
Engine myEngine = DefaultEngine.create(hadoopConf);
Table myTable = Table.forPath(myEngine, myTablePath);
```
Note the default [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) we are creating to bootstrap the `myTable` object. This object allows you to plug in your own libraries for computationally intensive operations like Parquet file reading, JSON parsing, etc. You can ignore it for now. We will discuss more about this later when we discuss how to build more complex connectors for distributed processing engines.
From this `myTable` object you can create a [`Snapshot`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Snapshot.html) object which represents the consistent state (a.k.a. a snapshot consistency) in a specific version of the table.
```java
Snapshot mySnapshot = myTable.getLatestSnapshot(myEngine);
```
Now that we have a consistent snapshot view of the table, we can query more details about the table. For example, you can get the version and schema of this snapshot.
```java
long version = mySnapshot.getVersion(myEngine);
StructType tableSchema = mySnapshot.getSchema(myEngine);
```
Next, to read the table data, we have to *build* a [`Scan`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Scan.html) object. In order to build a `Scan` object, create a [`ScanBuilder`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/ScanBuilder.html) object which optionally allows selecting a subset of columns to read or setting a query filter. For now, ignore these optional settings.
```java
Scan myScan = mySnapshot.getScanBuilder(myEngine).build()
// Common information about scanning for all data files to read.
Row scanState = myScan.getScanState(myEngine)
// Information about the list of scan files to read
CloseableIterator scanFiles = myScan.getScanFiles(myEngine)
```
This [`Scan`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Scan.html) object has all the necessary metadata to start reading the table. There are two crucial pieces of information needed for reading data from a file in the table.
* `myScan.getScanFiles(Engine)`: Returns scan files as columnar batches (represented as an iterator of `FilteredColumnarBatch`es, more on that later) where each selected row in the batch has information about a single file containing the table data.
* `myScan.getScanState(Engine)`: Returns the snapshot-level information needed for reading any file. Note that this is a single row and common to all scan files.
For each scan file the physical data must be read from the file. The columns to read are specified in the scan file state. Once the physical data is read, you have to call [`ScanFile.transformPhysicalData(…)`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Scan.html#transformPhysicalData-io.delta.kernel.engine.Engine-io.delta.kernel.data.Row-io.delta.kernel.data.Row-io.delta.kernel.utils.CloseableIterator-) with the scan state and the physical data read from scan file. This API takes care of transforming (e.g. adding partition columns) the physical data into logical data of the table. Here is an example of reading all the table data in a single thread.
```java
CloserableIterator fileIter = scanObject.getScanFiles(myEngine);
Row scanStateRow = scanObject.getScanState(myEngine);
while(fileIter.hasNext()) {
FilteredColumnarBatch scanFileColumnarBatch = fileIter.next();
// Get the physical read schema of columns to read from the Parquet data files
StructType physicalReadSchema =
ScanStateRow.getPhysicalDataReadSchema(engine, scanStateRow);
try (CloseableIterator scanFileRows = scanFileColumnarBatch.getRows()) {
while (scanFileRows.hasNext()) {
Row scanFileRow = scanFileRows.next();
// From the scan file row, extract the file path, size and modification time metadata
// needed to read the file.
FileStatus fileStatus = InternalScanFileUtils.getAddFileStatus(scanFileRow);
// Open the scan file which is a Parquet file using connector's own
// Parquet reader or default Parquet reader provided by the Kernel (which
// is used in this example).
CloseableIterator physicalDataIter =
engine.getParquetHandler().readParquetFiles(
singletonCloseableIterator(fileStatus),
physicalReadSchema,
Optional.empty() /* optional predicate the connector can apply to filter data from the reader */
);
// Now the physical data read from the Parquet data file is converted to a table
// logical data. Logical data may include the addition of partition columns and/or
// subset of rows deleted
try (
CloseableIterator transformedData =
Scan.transformPhysicalData(
engine,
scanStateRow,
scanFileRow,
physicalDataIter)) {
while (transformedData.hasNext()) {
FilteredColumnarBatch logicalData = transformedData.next();
ColumnarBatch dataBatch = logicalData.getData();
// Not all rows in `dataBatch` are in the selected output.
// An optional selection vector determines whether a row with a
// specific row index is in the final output or not.
Optional selectionVector = dataReadResult.getSelectionVector();
// access the data for the column at ordinal 0
ColumnVector column0 = dataBatch.getColumnVector(0);
for (int rowIndex = 0; rowIndex < column0.getSize(); rowIndex++) {
// check if the row is selected or not
if (!selectionVector.isPresent() || // there is no selection vector, all records are selected
(!selectionVector.get().isNullAt(rowId) && selectionVector.get().getBoolean(rowId))) {
// Assuming the column type is String.
// If it is a different type, call the relevant function on the `ColumnVector`
System.out.println(column0.getString(rowIndex));
}
}
// access the data for column at ordinal 1
ColumnVector column1 = dataBatch.getColumnVector(1);
for (int rowIndex = 0; rowIndex < column1.getSize(); rowIndex++) {
// check if the row is selected or not
if (!selectionVector.isPresent() || // there is no selection vector, all records are selected
(!selectionVector.get().isNullAt(rowId) && selectionVector.get().getBoolean(rowId))) {
// Assuming the column type is Long.
// If it is a different type, call the relevant function on the `ColumnVector`
System.out.println(column1.getLong(rowIndex));
}
}
// .. more ..
}
}
}
}
}
```
A few working examples to read Delta tables within a single process are available [here](https://github.com/delta-io/delta/tree/master/kernel/examples).
Important
All the Delta protocol-level details are encoded in the rows returned by `Scan.getScanFiles` API, but you do not have to understand them in order to read the table data correctly. All you need is to get the Parquet file status from each scan file row and read the data from the Parquet file into the `ColumnarBatch` format. The physical data is converted into the logical data of the table using [`Scan.transformPhysicalData`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Scan.html#transformPhysicalData-io.delta.kernel.engine.Engine-io.delta.kernel.data.Row-io.delta.kernel.data.Row-io.delta.kernel.utils.CloseableIterator-). Transformation to logical data is dictated by the protocol and the metadata of the table and the scan file. As the Delta protocol evolves this transformation step will evolve with it and your code will not have to change to accommodate protocol changes. This is the major advantage of the abstractions provided by Delta Kernel.
Note
Observe that the same `Engine` instance `myEngine` is passed multiple times whenever a call to Delta Kernel API is made. The reason for passing this instance for every call is because it is the connector context, it should maintained outside of the Delta Kernel APIs to give the connector control over the `Engine`.
### Step 2: Improve scan performance with file skipping
[Section titled “Step 2: Improve scan performance with file skipping”](#step-2-improve-scan-performance-with-file-skipping)
We have explored how to do a full table scan. However, the real advantage of using the Delta format is that you can skip files using your query filters. To make this possible, Delta Kernel provides an [expression framework](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/expressions/package-summary.html) to encode your filters and provide them to Delta Kernel to skip files during the scan file generation. For example, say your table is partitioned by `columnX`, you want to query only the partition `columnX=1`. You can generate the expression and use it to build the scan as follows:
* Java
```java
import io.delta.kernel.expressions.*;
import io.delta.kernel.defaults.engine.*;
Engine myEngine = DefaultEngine.create(new Configuration());
Predicate filter = new Predicate(
"=",
Arrays.asList(new Column("columnX"), Literal.ofInt(1)));
Scan myFilteredScan = mySnapshot.buildScan(engine)
.withFilter(myEngine, filter)
.build()
// Subset of the given filter that is not guaranteed to be satisfied by
// Delta Kernel when it returns data. This filter is used by Delta Kernel
// to do data skipping as much as possible. The connector should use this filter
// on top of the data returned by Delta Kernel in order for further filtering.
Optional remainingFilter = myFilteredScan.getRemainingFilter();
```
The scan files returned by `myFilteredScan.getScanFiles(myEngine)` will have rows representing files only of the required partition. Similarly, you can provide filters for non-partition columns, and if the data in the table is well clustered by those columns, then Delta Kernel will be able to skip files as much as possible.
## Create a Delta table
[Section titled “Create a Delta table”](#create-a-delta-table)
In this section, we will walk through how to build a Delta connector that can create a Delta table using the default [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) implementation provided by Delta Kernel.
You can either write this code yourself in your project, or you can use the [examples](https://github.com/delta-io/delta/tree/master/kernel/examples) present in the Delta code repository.
The main entry point is [`io.delta.kernel.Table`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Table.html) which is a programmatic representation of a Delta table. Say you want to create Delta table at the directory `myTablePath`. You can create a `Table` object as follows:
* Java
```java
package io.delta.kernel.examples;
import io.delta.kernel.*;
import io.delta.kernel.types.*;
import io.delta.kernel.utils.CloseableIterable;
String myTablePath = ;
Configuration hadoopConf = new Configuration();
Engine myEngine = DefaultEngine.create(hadoopConf);
Table myTable = Table.forPath(myEngine, myTablePath);
```
Note the default [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) we are creating to bootstrap the `myTable` object. This object allows you to plug in your own libraries for computationally intensive operations like Parquet file reading, JSON parsing, etc. You can ignore it for now. We will discuss more about this later when we discuss how to build more complex connectors for distributed processing engines.
From this `myTable` object you can create a [`TransactionBuilder`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/TransactionBuilder.html) object which allows you to construct a [`Transaction`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Transaction.html) object
```java
TransactionBuilder txnBuilder =
myTable.createTransactionBuilder(
myEngine,
"Examples", /* engineInfo - connector can add its own identifier which is noted in the Delta Log */
Operation.CREATE_TABLE /* What is the operation we are trying to perform. This is noted in the Delta Log */
);
```
Now that you have the `TransactionBuilder` object, you can set the table schema and partition columns of the table.
```java
StructType mySchema = new StructType()
.add("id", IntegerType.INTEGER)
.add("name", StringType.STRING)
.add("city", StringType.STRING)
.add("salary", DoubleType.DOUBLE);
// Partition columns are optional. Use it only if you are creating a partitioned table.
List myPartitionColumns = Collections.singletonList("city");
// Set the schema of the new table on the transaction builder
txnBuilder = txnBuilder
.withSchema(engine, mySchema);
// Set the partition columns of the new table only if you are creating
// a partitioned table; otherwise, this step can be skipped.
txnBuilder = txnBuilder
.withPartitionColumns(engine, examplePartitionColumns);
```
`TransactionBuilder` allows setting additional properties of the table such as enabling a certain Delta feature or setting identifiers for idempotent writes. We will be visiting these in the next sections. The next step is to build `Transaction` out of the `TransactionBuilder` object.
```java
// Build the transaction
Transaction txn = txnBuilder.build(engine);
```
`Transaction` object allows the connector to optionally add any data and finally commit the transaction. A successful commit ensures that the table is created with the given schema. In this example, we are just creating a table and not adding any data as part of the table.
```java
// Commit the transaction.
// As we are just creating the table and not adding any data, the `dataActions` is empty.
TransactionCommitResult commitResult =
txn.commit(
engine,
CloseableIterable.emptyIterable() /* dataActions */
);
```
The [`TransactionCommitResult`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/TransactionCommitResult.html) contains the what version the transaction is committed as and whether the table is ready for a checkpoint. As we are creating a table the version will be `0`. We will be discussing later on what a checkpoint is and what it means for the table to be ready for the checkpoint.
A few working examples to create partitioned and un-partitioned Delta tables are available [here](https://github.com/delta-io/delta/tree/master/kernel/examples).
## Create a table and insert data into it
[Section titled “Create a table and insert data into it”](#create-a-table-and-insert-data-into-it)
In this section, we will walk through how to build a Delta connector that can create a Delta table and insert data into the table (similar to `CREATE TABLE AS ` construct in SQL) using the default [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) implementation provided by Delta Kernel.
You can either write this code yourself in your project, or you can use the [examples](https://github.com/delta-io/delta/tree/master/kernel/examples) present in the Delta code repository.
The first step is to construct a `Transaction`. Below is the code for that. For more details on what each step of the code means, please read the [create table](#create-a-delta-table) section.
```plaintext
package io.delta.kernel.examples;
import io.delta.kernel.*;
import io.delta.kernel.types.*;
import io.delta.kernel.utils.CloseableIterable;
String myTablePath = ;
Configuration hadoopConf = new Configuration();
Engine myEngine = DefaultEngine.create(hadoopConf);
Table myTable = Table.forPath(myEngine, myTablePath);
StructType mySchema = new StructType()
.add("id", IntegerType.INTEGER)
.add("name", StringType.STRING)
.add("city", StringType.STRING)
.add("salary", DoubleType.DOUBLE);
// Partition columns are optional. Use it only if you are creating a partitioned table.
List myPartitionColumns = Collections.singletonList("city");
TransactionBuilder txnBuilder =
myTable.createTransactionBuilder(
myEngine,
"Examples", /* engineInfo - connector can add its own identifier which is noted in the Delta Log */
Operation.WRITE /* What is the operation we are trying to perform? This is noted in the Delta Log */
);
// Set the schema of the new table on the transaction builder
txnBuilder = txnBuilder
.withSchema(engine, mySchema);
// Set the partition columns of the new table only if you are creating
// a partitioned table; otherwise, this step can be skipped.
txnBuilder = txnBuilder
.withPartitionColumns(engine, examplePartitionColumns);
// Build the transaction
Transaction txn = txnBuilder.build(engine);
```
Now that we have the [`Transaction`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Transaction.html) object, the next step is generating the data that confirms the table schema and partitioned according to the table partitions.
```java
StructType dataSchema = txn.getSchema(engine)
// Optional for un-partitioned tables
List partitionColumnNames = txn.getPartitionColumns(engine)
```
Using the data schema and partition column names the connector can plan the query and generate data. At tasks that actually have the data to write to the table, the connector can ask the Kernel to transform the data given in the table schema into physical data that can actually be written to the Parquet data files. For partitioned tables, the data needs to be first partitioned by the partition columns, and then the connector should ask the Kernel to transform the data for each partition separately. The partitioning step is needed because any given data file in the Delta table contains data belonging to exactly one partition.
Get the state of the transaction. The transaction state contains the information about how to convert the data in the table schema into physical data that needs to be written. The transformations depend on the protocol and features the table has.
```java
Row txnState = txn.getTransactionState(engine);
```
Prepare the data.
```java
// The data generated by the connector to write into a table
CloseableIterator data = ...
// Create partition value map
Map partitionValues =
Collections.singletonMap(
"city", // partition column name
// partition value. Depending upon the partition column type, the
// partition value should be created. In this example, the partition
// column is of type StringType, so we are creating a string literal.
Literal.ofString(city)
);
```
The connector data is passed as an iterator of [`FilteredColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/FilteredColumnarBatch.html). Each of the `FilteredColumnarBatch` contains a [`ColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnarBatch.html) which actually contains the data in columnar access format and an optional section vector that allows the connector to specify which rows from the `ColumnarBatch` to write to the table.
Partition values are passed as a map of the partition column name to the partition value. For an un-partitioned table, the map should be empty as it has no partition columns.
```plaintext
// Transform the logical data to physical data that needs to be written to the Parquet
// files
CloseableIterator physicalData =
Transaction.transformLogicalData(engine, txnState, data, partitionValues);
```
The above code converts the given data for partitions into an iterator of `FilteredColumnarBatch` that needs to be written to the Parquet data files. In order to write the data files, the connector needs to get the [`WriteContext`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/DataWriteContext.html) from Kernel, which tells the connector where to write the data files and what columns to collect statistics from each data file.
```java
// Get the write context
DataWriteContext writeContext = Transaction.getWriteContext(engine, txnState, partitionValues);
```
Now, the connector has the physical data that needs to be written to Parquet data files, and where those files should be written, it can start writing the data files.
```java
CloseableIterator dataFiles = engine.getParquetHandler()
.writeParquetFiles(
writeContext.getTargetDirectory(),
physicalData,
writeContext.getStatisticsColumns()
);
```
In the above code, the connector is making use of the `Engine` provided `ParquetHandler` to write the data, but the connector can choose its own Parquet file writer to write the data. Also note that the return of the above call is an iterator of [`DataFileStatus`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/utils/DataFileStatus.html) for each data file written. It basically contains the file path, file metadata, and optional file-level statistics for columns specified by the [`WriteContext.getStatisticsColumns()`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/DataWriteContext.html#getStatisticsColumns--))
Convert each `DataFileStatus` into a Delta log action that can be written to the Delta table log.
```java
CloseableIterator dataActions =
Transaction.generateAppendActions(engine, txnState, dataFiles, writeContext);
```
The next step is constructing [`CloseableIterable`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/utils/CloseableIterable.html) out of the all the Delta log actions generated above. The reason for constructing an `Iterable` is that the transaction committing involves accessing the list of Delta log actions more than one time (in order to resolve conflicts when there are multiple writes to the table). Kernel provides a [utility method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/utils/CloseableIterable.html#inMemoryIterable-io.delta.kernel.utils.CloseableIterator-) to create an in-memory version of `CloseableIterable`. This interface also gives the connector an option to implement a custom implementation that spills the data actions to disk when the contents are too big to fit in memory.
```java
// Create a iterable out of the data actions. If the contents are too big to fit in memory,
// the connector may choose to write the data actions to a temporary file and return an
// iterator that reads from the file.
CloseableIterable dataActionsIterable = CloseableIterable.inMemoryIterable(dataActions);
```
The final step is committing the transaction!
```java
TransactionCommitStatus commitStatus = txn.commit(engine, dataActionsIterable)
```
The [`TransactionCommitResult`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/TransactionCommitResult.html) contains the what version the transaction is committed as and whether the table is ready for a checkpoint. As we are creating a table the version will be `0`. We will be discussing later on what a checkpoint is and what it means for the table to be ready for the checkpoint.
A few working examples to create and insert data into partitioned and un-partitioned Delta tables are available [here](https://github.com/delta-io/delta/tree/master/kernel/examples).
## Blind append into an existing Delta table
[Section titled “Blind append into an existing Delta table”](#blind-append-into-an-existing-delta-table)
In this section, we will walk through how to build a Delta connector that inserts data into an existing Delta table (similar to `INSERT INTO ` construct in SQL) using the default [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) implementation provided by Delta Kernel.
You can either write this code yourself in your project, or you can use the [examples](https://github.com/delta-io/delta/tree/master/kernel/examples) present in the Delta code repository. The steps are exactly similar to [Create table and insert data into it](#create-a-table-and-insert-data-into-it) except that we won’t be providing any schema or partition columns when building the `TransactionBuilder`
```java
// Create a `Table` object with the given destination table path
Table table = Table.forPath(engine, tablePath);
// Create a transaction builder to build the transaction
TransactionBuilder txnBuilder =
table.createTransactionBuilder(
engine,
"Examples", /* engineInfo */
Operation.WRITE
);
/ Build the transaction - no need to provide the schema as the table already exists.
Transaction txn = txnBuilder.build(engine);
// Get the transaction state
Row txnState = txn.getTransactionState(engine);
List dataActions = new ArrayList<>();
// Generate the sample data for three partitions. Process each partition separately.
// This is just an example. In a real-world scenario, the data may come from different
// partitions. Connectors already have the capability to partition by partition values
// before writing to the table
// In the test data `city` is a partition column
for (String city : Arrays.asList("San Francisco", "Campbell", "San Jose")) {
FilteredColumnarBatch batch1 = generatedPartitionedDataBatch(
5 /* offset */, city /* partition value */);
FilteredColumnarBatch batch2 = generatedPartitionedDataBatch(
5 /* offset */, city /* partition value */);
FilteredColumnarBatch batch3 = generatedPartitionedDataBatch(
10 /* offset */, city /* partition value */);
CloseableIterator data =
toCloseableIterator(Arrays.asList(batch1, batch2, batch3).iterator());
// Create partition value map
Map partitionValues =
Collections.singletonMap(
"city", // partition column name
// partition value. Depending upon the parition column type, the
// partition value should be created. In this example, the partition
// column is of type StringType, so we are creating a string literal.
Literal.ofString(city));
// First transform the logical data to physical data that needs to be written
// to the Parquet
// files
CloseableIterator physicalData =
Transaction.transformLogicalData(engine, txnState, data, partitionValues);
// Get the write context
DataWriteContext writeContext =
Transaction.getWriteContext(engine, txnState, partitionValues);
// Now write the physical data to Parquet files
CloseableIterator dataFiles = engine.getParquetHandler()
.writeParquetFiles(
writeContext.getTargetDirectory(),
physicalData,
writeContext.getStatisticsColumns());
// Now convert the data file status to data actions that needs to be written to the Delta
// table log
CloseableIterator partitionDataActions = Transaction.generateAppendActions(
engine,
txnState,
dataFiles,
writeContext);
// Now add all the partition data actions to the main data actions list. In a
// distributed query engine, the partition data is written to files at tasks on executor
// nodes. The data actions are collected at the driver node and then written to the
// Delta table log using the `Transaction.commit`
while (partitionDataActions.hasNext()) {
dataActions.add(partitionDataActions.next());
}
}
// Create a iterable out of the data actions. If the contents are too big to fit in memory,
// the connector may choose to write the data actions to a temporary file and return an
// iterator that reads from the file.
CloseableIterable dataActionsIterable = CloseableIterable.inMemoryIterable(
toCloseableIterator(dataActions.iterator()));
// Commit the transaction.
TransactionCommitResult commitResult = txn.commit(engine, dataActionsIterable);
```
## Idempotent Blind Appends to a Delta Table
[Section titled “Idempotent Blind Appends to a Delta Table”](#idempotent-blind-appends-to-a-delta-table)
Idempotent writes allow the connector to make sure the data belonging to a particular transaction version and application id is inserted into the table at most once. In incremental processing systems (e.g. streaming systems), track progress using their own application-specific versions need to record what progress has been made in order to avoid duplicating data in the face of failures and retries during writes. By setting the transaction identifier, the Delta table can ensure that the data with the same identifier is not written multiple times. For more information refer to the Delta protocol section [Transaction Identifiers](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#transaction-identifiers)
To make the data append idempotent, set the transaction identifier on the [`TransactionBuilder`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/TransactionBuilder.html#withTransactionId-io.delta.kernel.engine.Engine-java.lang.String-long-)
```java
// Set the transaction identifiers for idempotent writes
// Delta/Kernel makes sure that there exists only one transaction in the Delta log
// with the given application id and txn version
txnBuilder =
txnBuilder.withTransactionId(
engine,
"my app id", /* application id */
100 /* monotonically increasing txn version with each new data insert */
);
```
That’s all the connector need to do for idempotent blind appends.
## Checkpointing a Delta table
[Section titled “Checkpointing a Delta table”](#checkpointing-a-delta-table)
[Checkpoints](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#checkpoints) are an optimization in Delta Log in order to construct the state of the Delta table faster. It basically contains the state of the table at the version the checkpoint is created. Delta Kernel allows the connector to optionally make the checkpoints. It is created for every few commits (configurable table property) on the table.
The result of `Transaction.commit` returns a `TransactionCommitResult` that contains the version the transaction is committed as and whether the table is [read for checkpoint](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/TransactionCommitResult.html#isReadyForCheckpoint--). Creating a checkpoint takes time as it needs to construct the entire state of the table. If the connector doesn’t want to checkpoint by itself but uses other connectors that are faster in creating a checkpoint, it can skip the checkpointing step.
If it wants to checkpoint, the `Table` object has an [API](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Table.html#checkpoint-io.delta.kernel.engine.Engine-long-) to checkpoint the table.
```java
TransactionCommitResult commitResult = txn.commit(engine, dataActionsIterable);
if (commitResult.isReadyForCheckpoint()) {
// Checkpoint the table
Table.forPath(engine, tablePath).checkpoint(engine, commitResult.getVersion());
}
```
## Build a Delta connector for a distributed processing engine
[Section titled “Build a Delta connector for a distributed processing engine”](#build-a-delta-connector-for-a-distributed-processing-engine)
Unlike simple applications that just read the table in a single process, building a connector for complex processing engines like Apache Spark™ and Trino can require quite a bit of additional effort. For example, to build a connector for an SQL engine you have to do the following
* Understand the APIs provided by the engine to build connectors and how Delta Kernel can be used to provide the information necessary for the connector + engine to operate on a Delta table.
* Decide what libraries to use to do computationally expensive operations like reading Parquet files, parsing JSON, computing expressions, etc. Delta Kernel provides all the extension points to allow you to plug in any library without having to understand all the low-level details of the Delta protocol.
* Deal with details specific to distributed engines. For example,
* Serialization of Delta table metadata provided by Delta Kernel.
* Efficiently transforming data read from Parquet into the engine in-memory processing format.
In this section, we are going to outline the steps needed to build a connector.
### Step 0: Validate the prerequisites
[Section titled “Step 0: Validate the prerequisites”](#step-0-validate-the-prerequisites)
In the previous section showing how to read a simple table, we were briefly introduced to the [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html). This is the main extension point where you can plug in your implementations of computationally-expensive operations like reading Parquet files, parsing JSON, etc. For the simple case, we were using a default implementation of the helper that works in most cases. However, for building a high-performance connector for a complex processing engine, you will very likely need to provide your own implementation using the libraries that work with your engine. So before you start building your connector, it is important to understand these requirements and plan for building your own engine.
Here are the libraries/capabilities you need to build a connector that can read the Delta table
* Perform file listing and file reads from your storage/file system.
* Read Parquet files in columnar data, preferably in an in-memory columnar format.
* Parse JSON data
* Read JSON files
* Evaluate expressions on in-memory columnar batches
For each of these capabilities, you can choose to build your own implementation or reuse the default implementation.
### Step 1: Set up Delta Kernel in your connector project
[Section titled “Step 1: Set up Delta Kernel in your connector project”](#step-1-set-up-delta-kernel-in-your-connector-project)
In the Delta Kernel project, there are multiple dependencies you can choose to depend on.
1. Delta Kernel core APIs - This is a must-have dependency, which contains all the main APIs like Table, Snapshot, and Scan that you will use to access the metadata and data of the Delta table. This has very few dependencies reducing the chance of conflicts with any dependencies in your connector and engine. This also provides the [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) interface which allows you to plug in your implementations of computationally expensive operations, but it does not provide any implementation of this interface.
2. Delta Kernel default- This has a default implementation called [`DefaultEngine`](https://delta-io.github.io/delta/snapshot/kernel-defaults/java/io/delta/kernel/defaults/engine/DefaultEngine.html) and additional dependencies such as `Hadoop`. If you wish to reuse all or parts of this implementation, then you can optionally depend on this.
#### Set up Java projects
[Section titled “Set up Java projects”](#set-up-java-projects)
As discussed above, you can import one or both of the artifacts as follows:
```xml
io.delta
delta-kernel-api
${delta-kernel.version}
io.delta
delta-kernel-defaults
${delta-kernel.version}
```
### Step 2: Build your own Engine
[Section titled “Step 2: Build your own Engine”](#step-2-build-your-own-engine)
In this section, we are going to explore the [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) interface and walk through how to implement your own implementation so that you can plug in your connector/engine-specific implementations of computationally-intensive operations, threading model, resource management, etc.
> \[!IMPORTANT] During the validation process, if you believe that all the dependencies of the default `Engine` implementation can work with your connector and engine, then you can skip this step and jump to Step 3 of implementing your connector using the default engine. If later you have the need to customize the helper for your connector, you can revisit this step.
#### Step 2.1: Implement the `Engine` interface
[Section titled “Step 2.1: Implement the Engine interface”](#step-21-implement-the-engine-interface)
The [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) interface combines a bunch of sub-interfaces each of which is designed for a specific purpose. Here is a brief overview of the subinterfaces. See the API docs (Java) for a more detailed view.
```java
interface Engine {
/**
* Get the connector provided {@link ExpressionHandler}.
* @return An implementation of {@link ExpressionHandler}.
*/
ExpressionHandler getExpressionHandler();
/**
* Get the connector provided {@link JsonHandler}.
* @return An implementation of {@link JsonHandler}.
*/
JsonHandler getJsonHandler();
/**
* Get the connector provided {@link FileSystemClient}.
* @return An implementation of {@link FileSystemClient}.
*/
FileSystemClient getFileSystemClient();
/**
* Get the connector provided {@link ParquetHandler}.
* @return An implementation of {@link ParquetHandler}.
*/
ParquetHandler getParquetHandler();
}
```
To build your own `Engine` implementation, you can choose to either use the default implementations of each sub-interface or completely build every one from scratch.
```java
class MyEngine extends DefaultEngine {
FileSystemClient getFileSystemClient() {
// Build a new implementation from scratch
return new MyFileSystemClient();
}
// For all other sub-clients, use the default implementations provided by the `DefaultEngine`.
}
```
Next, we will walk through how to implement each interface.
#### Step 2.2: Implement `FileSystemClient` interface
[Section titled “Step 2.2: Implement FileSystemClient interface”](#step-22-implement-filesystemclient-interface)
The [`FileSystemClient`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/FileSystemClient.html) interface contains basic file system operations like listing directories, resolving paths into a fully qualified path and reading bytes from files. Implementation of this interface must take care of the following when interacting with storage systems such as S3, Hadoop, or ADLS:
* Credentials and permissions: The connector must populate its `FileSystemClient` with the necessary configurations and credentials for the client to retrieve the necessary data from the storage system. For example, an implementation based on Hadoop’s FileSystem abstractions can be passed S3 credentials via the Hadoop configurations.
* Decryption: If file system objects are encrypted, then the implementation must decrypt the data before returning the data.
#### Step 2.3: Implement `ParquetHandler`
[Section titled “Step 2.3: Implement ParquetHandler”](#step-23-implement-parquethandler)
As the name suggests, this interface contains everything related to reading and writing Parquet files. It has been designed such that a connector can plug in a wide variety of implementations, from a simple single-threaded reader to a very advanced multi-threaded reader with pre-fetching and advanced connector-specific expression pushdown. Let’s explore the methods to implement, and the guarantees associated with them.
##### Method `readParquetFiles(CloseableIterator fileIter, StructType physicalSchema, java.util.Optional predicate)`
[Section titled “Method readParquetFiles(CloseableIterator\ fileIter, StructType physicalSchema, java.util.Optional\ predicate)”](#method-readparquetfilescloseableiteratorfilestatus-fileiter-structtype-physicalschema-javautiloptionalpredicate-predicate)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ParquetHandler.html#readParquetFiles-io.delta.kernel.utils.CloseableIterator-io.delta.kernel.types.StructType-)) takes as input `FileStatus`s which contains metadata such as file path, size etc. of the Parquet file to read. The columns to be read from the Parquet file are defined by the physical schema. To implement this method, you may have to first implement your own [`ColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnarBatch.html) and [`ColumnVector`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnVector.html) which is used to represent the in-memory data generated from the Parquet files.
When identifying the columns to read, note that there are multiple types of columns in the physical schema (represented as a [`StructType`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/types/StructType.html)).
* Data columns: Columns that are expected to be read from the Parquet file. Based on the `StructField` object defining the column, read the column in the Parquet file that matches the same name or field id. If the column has a field id (stored as `parquet.field.id` in the `StructField` metadata) then the field id should be used to match the column in the Parquet file. Otherwise, the column name should be used for matching.
* Metadata columns: These are special columns that must be populated using metadata about the Parquet file ([`StructField#isMetadataColumn`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/types/StructField.html#isMetadataColumn--) tells whether a column in `StructType` is a metadata column). To understand how to populate such a column, first match the column name against the set of standard metadata column name constants. For example,
* `StructFileld#isMetadataColumn()` returns true and the column name is `StructField.METADATA_ROW_INDEX_COLUMN_NAME`, then you have to a generate column vector populated with the actual index of each row in the Parquet file (that is, not indexed by the possible subset of rows returned after Parquet data skipping).
##### Requirements and guarantees
[Section titled “Requirements and guarantees”](#requirements-and-guarantees)
Any implementation must adhere to the following guarantees.
* The schema of the returned `ColumnarBatch`es must match the physical schema.
* If a data column is not found and the `StructField.isNullable = true`, then return a `ColumnVector` of nulls. Throw an error if it is not nullable.
* The output iterator must maintain ordering as the input iterator. That is, if `file1` is before `file2` in the input iterator, then columnar batches of `file1` must be before those of `file2` in the output iterator.
##### Method `writeParquetFiles(String directoryPath, CloseableIterator dataIter, java.util.List statsColumns)`
[Section titled “Method writeParquetFiles(String directoryPath, CloseableIterator\ dataIter, java.util.List\ statsColumns)”](#method-writeparquetfilesstring-directorypath-closeableiteratorfilteredcolumnarbatch-dataiter-javautillistcolumn-statscolumns)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ParquetHandler.html#writeParquetFiles-java.lang.String-io.delta.kernel.utils.CloseableIterator-java.util.List-) takes given data writes it into one or more Parquet files into the given directory. The data is given as an iterator of [FilteredColumnarBatches](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/FilteredColumnarBatch.html) which contains a [ColumnarBatch](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnarBatch.html) and an optional selection vector containing one entry for each row in `ColumnarBatch` indicating whether a row is selected or not selected. The `ColumnarBatch` also contains the schema of the data. This schema should be converted to Parquet schema, including any field IDs present [`FieldMetadata`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/types/FieldMetadata.html) for each column `StructField`.
There is also the parameter `statsColumns`, which is a hint to the Parquet writer on what set of columns to collect stats for each file. The statistics include `min`, `max` and `null_count` for each column in the `statsColumns` list. Statistics collection is optional, but when present it is used by Kernel to persist the stats as part of the Delta table commit. This will help read queries prune un-needed data files based on the query predicate.
For each written data file, the caller is expecting a [`DataFileStatus`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/utils/DataFileStatus.html) object. It contains the data file path, size, modification time, and optional column statistics.
#### Method `writeParquetFileAtomically(String filePath, CloseableIterator data)`
[Section titled “Method writeParquetFileAtomically(String filePath, CloseableIterator\ data)”](#method-writeparquetfileatomicallystring-filepath-closeableiteratorfilteredcolumnarbatch-data)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ParquetHandler.html#writeParquetFiles-java.lang.String-io.delta.kernel.utils.CloseableIterator-java.util.List-) writes the given `data` into Parquet file at location `filePath`. The write is an atomic write i.e., either a Parquet file is created with all given content or no Parquet file is created at all. This should not create a file with partial content in it.
The default implementation makes use of [`LogStore`](https://github.com/delta-io/delta/blob/master/storage/src/main/java/io/delta/storage/LogStore.java) implementations from the [`delta-storage`](https://github.com/delta-io/delta/tree/master/storage) module to accomplish the atomicity. A connector that wants to implement their own version of `ParquetHandler` can take a look at the default implementation for details.
##### Performance suggestions
[Section titled “Performance suggestions”](#performance-suggestions)
* The representation of data as `ColumnVector`s and `ColumnarBatch`es can have a significant impact on the query performance and it’s best to read the Parquet file data directly into vectors and batches of the engine-native format to avoid potentially costly in-memory data format conversion. Create a Kernel `ColumnVector` and `ColumnarBatch` wrappers around the engine-native format equivalent classes.
#### Step 2.4: Implement `ExpressionHandler` interface
[Section titled “Step 2.4: Implement ExpressionHandler interface”](#step-24-implement-expressionhandler-interface)
The [`ExpressionHandler`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ExpressionHandler.html) interface has all the methods needed for handling expressions that may be applied on columnar data.
##### Method `getEvaluator(StructType batchSchema, Expression expresion, DataType outputType)`
[Section titled “Method getEvaluator(StructType batchSchema, Expression expresion, DataType outputType)”](#method-getevaluatorstructtype-batchschema-expression-expresion-datatype-outputtype)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ExpressionHandler.html#getEvaluator-io.delta.kernel.types.StructType-io.delta.kernel.expressions.Expression-io.delta.kernel.types.DataType-) generates an object of type [`ExpressionEvaluator`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/expressions/ExpressionEvaluator.html) that can evaluate the `expression` on a batch of row data to produce a result of a single column vector. To generate this function, the `getEvaluator()` method takes as input the expression and the schema of the `ColumnarBatch`es of data on which the expressions will be applied. The same object can be used to evaluate multiple columnar batches of input with the same schema and expression the evaluator is created for.
##### Method `getPredicateEvaluator(StructType inputSchema, Predicate predicate)`
[Section titled “Method getPredicateEvaluator(StructType inputSchema, Predicate predicate)”](#method-getpredicateevaluatorstructtype-inputschema-predicate-predicate)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ExpressionHandler.html#createSelectionVector-boolean:A-int-int-) is for creating an expression evaluator for `Predicate` type expressions. The `Predicate` type expressions return a boolean value as output.
The returned object is of type [`PredicateEvaluator`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/expressions/PredicateEvaluator.html). This is a special interface for evaluating Predicate on input batch returns a selection vector containing one value for each row in input batch indicating whether the row has passed the predicate or not. Optionally it takes an existing selection vector along with the input batch for evaluation. The result selection vector is combined with the given existing selection vector and a new selection vector is returned. This mechanism allows running an input batch through several predicate evaluations without rewriting the input batch to remove rows that do not pass the predicate after each predicate evaluation. The new selection should be the same or more selective as the existing selection vector. For example, if a row is marked as unselected in the existing selection vector, then it should remain unselected in the returned selection vector even when the given predicate returns true for the row.
##### Method `createSelectionVector(boolean[] values, int from, int to)`
[Section titled “Method createSelectionVector(boolean\[\] values, int from, int to)”](#method-createselectionvectorboolean-values-int-from-int-to)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ExpressionHandler.html#createSelectionVector-boolean:A-int-int-) allows creating `ColumnVector` for boolean type values given as input. This allows the connector to maintain all `ColumnVector`s created in the desired memory format.
##### Requirements and guarantees
[Section titled “Requirements and guarantees”](#requirements-and-guarantees-1)
Any implementation must adhere to the following guarantees.
* Implementation must handle all possible variations of expressions. If the implementation encounters an expression type that it does not know how to handle, then it must throw a specific language-dependent exception.
* Java: [NotSupportedException](https://docs.oracle.com/javaee/7/api/javax/resource/NotSupportedException.html)
* The `ColumnarBatch`es on which the generated `ExpressionEvaluator` is going to be used are guaranteed to have the schema provided during generation. Hence, it is safe to bind the expression evaluation logic to column ordinals instead of column names, thus making the actual evaluation faster.
#### Step 2.5: Implement `JsonHandler`
[Section titled “Step 2.5: Implement JsonHandler”](#step-25-implement-jsonhandler)
[This](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/JsonHandler.html) engine interface allows the connector to use plug-in their own JSON handling code and expose it to the Delta Kernel.
##### Method `readJsonFiles(CloseableIterator fileIter, StructType physicalSchema, java.util.Optional predicate)`
[Section titled “Method readJsonFiles(CloseableIterator\ fileIter, StructType physicalSchema, java.util.Optional\ predicate)”](#method-readjsonfilescloseableiteratorfilestatus-fileiter-structtype-physicalschema-javautiloptionalpredicate-predicate)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/JsonHandler.html#readJsonFiles-io.delta.kernel.utils.CloseableIterator-io.delta.kernel.types.StructType-) takes as input `FileStatus`s of the JSON files and returns the data in a series of columnar batches. The columns to be read from the JSON file are defined by the physical schema, and the return batches must match that schema. To implement this method, you may have to first implement your own [`ColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnarBatch.html)and [`ColumnVector`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnVector.html) which is used to represent the in-memory data generated from the JSON files.
When identifying the columns to read, note that there are multiple types of columns in the physical schema (represented as a [`StructType`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/types/StructType.html)).
##### Method `parseJson(ColumnVector jsonStringVector, StructType outputSchema, java.util.Optional selectionVector)`
[Section titled “Method parseJson(ColumnVector jsonStringVector, StructType outputSchema, java.util.Optional\ selectionVector)”](#method-parsejsoncolumnvector-jsonstringvector-structtype-outputschema-javautiloptionalcolumnvector-selectionvector)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/JsonHandler.html#parseJson-io.delta.kernel.data.ColumnVector-io.delta.kernel.types.StructType-) allows parsing a `ColumnVector` of string values which are in JSON format into the output format specified by the `outputSchema`. If a given column in `outputSchema` is not found, then a null value is returned. It optionally takes a selection vector which indicates what entries in the input `ColumnVector` of strings to parse. If an entry is not selected then a `null` value is returned as parsed output for that particular entry in the output.
##### Method `deserializeStructType(String structTypeJson)`
[Section titled “Method deserializeStructType(String structTypeJson)”](#method-deserializestructtypestring-structtypejson)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/JsonHandler.html#deserializeStructType-java.lang.String-) allows parsing JSON encoded (according to [Delta schema serialization rules](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#schema-serialization-format)) `StructType` schema into a `StructType`. Most implementations of `JsonHandler` do not need to implement this method and instead use the one in the [default `JsonHandler`](https://delta-io.github.io/delta/snapshot/kernel-defaults/java/io/delta/kernel/defaults/engine/DefaultJsonHandler.html) implementation.
#### Method `writeJsonFileAtomically(String filePath, CloseableIterator data, boolean overwrite)`
[Section titled “Method writeJsonFileAtomically(String filePath, CloseableIterator\ data, boolean overwrite)”](#method-writejsonfileatomicallystring-filepath-closeableiteratorrow-data-boolean-overwrite)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/JsonHandler.html#writeJsonFileAtomically-java.lang.String-io.delta.kernel.utils.CloseableIterator-boolean-) writes the given `data` into a JSON file at location `filePath`. The write is an atomic write i.e., either a JSON file is created with all given content or no Parquet file is created at all. This should not create a file with partial content in it.
The default implementation makes use of [`LogStore`](https://github.com/delta-io/delta/blob/master/storage/src/main/java/io/delta/storage/LogStore.java) implementations from the [`delta-storage`](https://github.com/delta-io/delta/tree/master/storage) module to accomplish the atomicity. A connector that wants to implement their own version of `JsonHandler` can take a look at the default implementation for details.
The implementation is expected to handle the serialization rules (converting the `Row` object to JSON string) as described in the [API Javadoc](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/JsonHandler.html#writeJsonFileAtomically-java.lang.String-io.delta.kernel.utils.CloseableIterator-boolean-).
#### Step 2.6: Implement `ColumnarBatch` and `ColumnVector`
[Section titled “Step 2.6: Implement ColumnarBatch and ColumnVector”](#step-26-implement-columnarbatch-and-columnvector)
[`ColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnarBatch.html) and [`ColumnVector`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnVector.html) are two interfaces to represent the data read into memory from files. This representation can have a significant impact on query performance. Each engine likely has a native representation of in-memory data with which it applies data transformation operations. For example, in Apache Spark™, the row data is internally represented as `UnsafeRow` for efficient processing. So it’s best to read the Parquet file data directly into vectors and batches of the native format to avoid potentially costly in-memory data format conversions. So the recommended approach is to build wrapper classes that extend the two interfaces but internally use engine-native classes to store the data. When the connector has to forward the columnar batches received from the kernel to the engine, it has to be smart enough to skip converting vectors and batches that are already in the engine-native format.
### Step 3: Build read support in your connector
[Section titled “Step 3: Build read support in your connector”](#step-3-build-read-support-in-your-connector)
In this section, we are going to walk through the likely sequence of Kernel API calls your connector will have to make to read a table. The exact timing of making these calls in your connector in the context of connector-engine interactions depends entirely on the engine-connector APIs and is therefore beyond the scope of this guide. However, we will try to provide broad guidelines that are likely (but not guaranteed) to apply to your connector-engine setup. For this purpose, we are going to assume that the engine goes through the following phases when processing a read/scan query - logical plan analysis, physical plan generation, and physical plan execution. Based on these broad characterizations, a typical control and data flow for reading a Delta table is going to be as follows:
| Step | Typical query phase when this step occurs |
| ------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Resolve the table snapshot to query | Logical plan analysis phase when the plan’s schema and other details need to be resolved and validated |
| Resolve files to scan based on query parameters | Physical plan generation, when the final parameters of the scan are available. For example: Schema of data to read after pruning away unused columns. Query filters to apply after filter rearrangement |
| Distribute the file information to workers | Physical plan execution, only if it is a distributed engine. |
| Read the columnar data using the file information | Physical plan execution, when the data is being processed by the engine |
Let’s understand the details of each step.
#### Step 3.1: Resolve the table snapshot to query
[Section titled “Step 3.1: Resolve the table snapshot to query”](#step-31-resolve-the-table-snapshot-to-query)
The first step is to resolve the consistent snapshot and the schema associated with it. This is often required by the connector/ engine to resolve and validate the logical plan of the scan query (if the concept of logical plan exists in your engine). To achieve this, the connector has to do the following.
* Resolve the table path from the query: If the path is directly available, then this is easy. Otherwise, if it is a query based on a catalog table (for example, a Delta table defined in Hive Metastore), then the connector has to resolve the table path from the catalog.
* Initialize the `Engine` object: Create a new instance of the `Engine` that you have chosen in Step 2.
* Initialize the Kernel objects and get the schema: Assuming the query is on the latest available version/snapshot of the table, you can get the table schema as follows:
```java
import io.delta.kernel.*;
import io.delta.kernel.defaults.engine.*;
Engine myEngine = new MyEngine();
Table myTable = Table.forPath(myTablePath);
Snapshot mySnapshot = myTable.getLatestSnapshot(myEngine);
StructType mySchema = mySnapshot.getSchema(myEngine);
```
If you want to query a specific version of the table (that is, not the schema), then you can get the required snapshot as `myTable.getSnapshot(version)`.
#### Step 3.2: Resolve files to scan
[Section titled “Step 3.2: Resolve files to scan”](#step-32-resolve-files-to-scan)
Next, we need to build a Scan object using more information from the query. Here we are going to assume that the connector/engine has been able to extract the following details from the query (say, after optimizing the logical plan):
* Read schema: The columns in the table that the query needs to read. This may be the full set of columns or a subset of columns.
* Query filters: The filters on partitions or data columns that can be used skip reading table data.
To provide this information to Kernel, you have to do the following:
* Convert the engine-specific schema and filter expressions to Kernel schema and expressions: For schema, you have to create a `StructType` object. For the filters, you have to create an `Expression` object using all the available subclasses of `Expression`.
* Build the scan with the converted information: Build the scan as follows:
```java
import io.delta.kernel.expressions.*;
import io.delta.kernel.types.*;
StructType readSchema = ... ; // convert engine schema
Predicate filterExpr = ... ; // convert engine filter expression
Scan myScan = mySnapshot.buildScan(engine)
.withFilter(myEngine, filterExpr)
.withReadSchema(myEngine, readSchema)
.build()
```
* Resolve the information required to file reads: The generated Scan object has two sets of information.
* Scan files: `myScan.getScanFiles()` returns an iterator of `ColumnarBatch`es. Each batch in the iterator contains rows and each row has information about a single file that has been selected based on the query filter.
* Scan state: `myScan.getScanState()` returns a `Row` that contains all the information that is common across all the files that need to be read.
````java
Row myScanStateRow = myScan.getScanState();
CloseableIterator myScanFilesAsBatches = myScan.getScanFiles();
```java
Row myScanStateRow = myScan.getScanState();
CloseableIterator myScanFilesAsBatches = myScan.getScanFiles();
while (myScanFilesAsBatches.hasNext()) {
FilteredColumnarBatch scanFileBatch = myScanFilesAsBatches.next();
CloseableIterator myScanFilesAsRows = scanFileBatch.getRows();
}
````
As we will soon see, reading the columnar data from a selected file will need to use both, the scan state row, and a scan file row with the file information.
##### Requirements and guarantees
[Section titled “Requirements and guarantees”](#requirements-and-guarantees-2)
Here are the details you need to ensure when defining this scan.
* The provided `readSchema` must be the exact schema of the data that the engine will expect when executing the query. Any mismatch in the schema defined during this query planning and the query execution will result in runtime failures. Hence you must build the scan with the readSchema only after the engine has finalized the logical plan after any optimizations like column pruning.
* When applicable (for example, with Java Kernel APIs), you have to make sure to call the close() method as you consume the `ColumnarBatch`es of scan files (that is, either serialize the rows or use them to read the table data).
#### Step 3.3: Distribute the file information to the workers
[Section titled “Step 3.3: Distribute the file information to the workers”](#step-33-distribute-the-file-information-to-the-workers)
If you are building a connector for a distributed engine like Spark/Presto/Trino/Flink, then your connector has to send all the scan metadata from the query planning machine (henceforth called the driver) to task execution machines (henceforth called the workers). You will have to serialize and deserialize the scan state and scan file rows. It is the connector job to implement serialization and deserialization utilities for a [`Row`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/Row.html). If the connector wants to split reading one scan file into multiple tasks, it can add additional connector specific split context to the task. At the task, the connector can use its own Parquet reader to read the specific part of the file indicated by the split info.
##### Custom `Row` Serializer/Deserializer
[Section titled “Custom Row Serializer/Deserializer”](#custom-row-serializerdeserializer)
Here are steps on how to build your own serializer/deserializer such that it will work with any [`Row`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/Row.html) of any schema.
* Serializing
* First serialize the row schema, that is, `StructType` object.
* Then, use the schema to identify types of each column/ordinal in the `Row` and use that to serialize all the values one by one.
* Deserializing
* Define your own class that extends the Row interface. It must be able to handle complex types like arrays, nested structs and maps.
* First deserialize the schema.
* Then, use the schema to deserialize the values and put them in an instance of your custom Row class.
```java
import io.delta.kernel.utils.*;
// In the driver where query planning is being done
Byte[] scanStateRowBytes = RowUtils.serialize(scanStateRow);
Byte[] scanFileRowBytes = RowUtils.serialize(scanFileRow);
// Optionally the connector adds a split info to the task (scan file, scan state) to
// split reading of a Parquet file into multiple tasks. The task gets split info
// along with the scan file row and scan state row.
Split split = ...; // connector specific class, not related to Kernel
// Send these over to the worker
// In the worker when data will be read, after rowBytes have been sent over
Row scanStateRow = RowUtils.deserialize(scanStateRowBytes);
Row scanFileRow = RowUtils.deserialize(scanFileRowBytes);
Split split = ... deserialize split info ...;
```
#### Step 3.4: Read the columnar data
[Section titled “Step 3.4: Read the columnar data”](#step-34-read-the-columnar-data)
Finally, we are ready to read the columnar data. You will have to do the following:
* Read the physical data from Parquet file as indicated by the scan file row, scan state, and optionally the split info
* Convert the physical data into logical data of the table using the Kernel’s APIs.
```java
Row scanStateRow = ... ;
Row scanFileRow = ... ;
Split split = ...;
// Additional option predicate such as dynamic filters the connector wants to
// pass to the reader when reading files.
Predicate optPredicate = ...;
// Get the physical read schema of columns to read from the Parquet data files
StructType physicalReadSchema =
ScanStateRow.getPhysicalDataReadSchema(engine, scanStateRow);
// From the scan file row, extract the file path, size and modification metadata
// needed to read the file.
FileStatus fileStatus = InternalScanFileUtils.getAddFileStatus(scanFileRow);
// Open the scan file which is a Parquet file using connector's own
// Parquet reader which supports reading specific parts (split) of the file.
// If the connector doesn't have its own Parquet reader, it can use the
// default Parquet reader provider which at the moment doesn't support reading
// a specific part of the file, but reads the entire file from the beginning.
CloseableIterator physicalDataIter =
connectParquetReader.readParquetFile(
fileStatus
physicalReadSchema,
split, // what part of the Parquet file to read data from
optPredicate /* additional predicate the connector can apply to filter data from the reader */
);
// Now the physical data read from the Parquet data file is converted to logical data
// the table represents.
// Logical data may include the addition of partition columns and/or
// subset of rows deleted
CloseableIterator transformedData =
Scan.transformPhysicalData(
engine,
scanState,
scanFileRow,
physicalDataIter));
```
* Resolve the data in the batches: Each [`FilteredColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/FilteredColumnarBatch.html) has two components:
* Columnar batch (returned by `FilteredColumnarBatch.getData()`): This is the data read from the files having the schema matching the readSchema provided when the Scan object was built in the earlier step.
* Optional selection vector (returned by `FilteredColumnarBatch.getSelectionVector()`): Optionally, a boolean vector that will define which rows in the batch are valid and should be consumed by the engine.
If the selection vector is present, then you will have to apply it to the batch to resolve the final consumable data.
* Convert to engine-specific data format: Each connector/engine has its own native row / columnar batch formats and interfaces. To return the read data batches to the engine, you have to convert them to fit those engine-specific formats and/or interfaces. Here are a few tips that you can follow to make this efficient.
* Matching the engine-specific format: Some engines may expect the data in an in-memory format that may be different from the data produced by `getData()`. So you will have to do the data conversion for each column vector in the batch as needed.
* Matching the engine-specific interfaces: You may have to implement wrapper classes that extend the engine-specific interfaces and appropriately encapsulate the row data.
For best performance, you can implement your own Parquet reader and other `Engine` implementations to make sure that every `ColumnVector` generated is already in the engine-native format thus eliminating any need to convert.
Now you should be able to read the Delta table correctly.
### Step 4: Build append support in your connector
[Section titled “Step 4: Build append support in your connector”](#step-4-build-append-support-in-your-connector)
In this section, we are going to walk through the likely sequence of Kernel API calls your connector will have to make to append data to a table. The exact timing of making these calls in your connector in the context of connector-engine interactions depends entirely on the engine-connector APIs and is, therefore, beyond the scope of this guide. However, we will try to provide broad guidelines that are likely (but not guaranteed) to apply to your connector-engine setup. For this purpose, we are going to assume that the engine goes through the following phases when processing a write query - logical plan analysis, physical plan generation, and physical plan execution. Based on these broad characterizations, a typical control and data flow for reading a Delta table is going to be as follows:
| Step | Typical query phase when this step occurs |
| -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Determine the schema of the data that needs to be written to the table. Schema is derived from the existing table or from the parent operation of the `write` operator in the query plan when the table doesn’t exist yet. | Logical plan analysis phase when the plan’s schema (`write` operator schema matches the table schema, etc.) and other details need to be resolved and validated. |
| Determine the physical partitioning of the data based on the table schema and partition columns either from the existing table or from the query plan (for new tables) | Physical plan generation, where the number of writer tasks, data schema and partitioning is determined |
| Distribute the writer tasks definitions (which include the transaction state) to workers. | Physical plan execution, only if it is a distributed engine. |
| Tasks write the data to data files and send the data file info to the driver. | Physical plan execution, when the data is actually written to the table location |
| Finalize the query. Here, all the info of the data files written by the tasks is aggregated and committed to the transaction created at the beginning of the physical execution. | Finalize the query. This happens on the driver where the query has started. |
Let’s understand the details of each step.
#### Step 4.1: Determine the schema of the data that needs to be written to the table
[Section titled “Step 4.1: Determine the schema of the data that needs to be written to the table”](#step-41-determine-the-schema-of-the-data-that-needs-to-be-written-to-the-table)
The first step is to resolve the output data schema. This is often required by the connector/ engine to resolve and validate the logical plan of the query (if the concept of logical plan exists in your engine). To achieve this, the connector has to do the following. At a high level query plan is a tree of operators where the leaf-level operators generate or read data from storage/tables and feed it upwards towards the parent operator nodes. This data transfer happens until it reaches the root operator node where the query is finalized (either the results are sent to the client or data is written to another table).
* Create the `Table` object
* From the `Table` object try to get the schema.
* If the table is not found
* the query includes creating the table (e.g., `CREATE TABLE AS` SQL query);
* the schema is derived from the operator above the `write` that feeds the data to the `write` operator.
* the query doesn’t include creating new table, an exception is thrown saying the table is not found
* If the table already exists
* get the schema from the table and check if it matches the schema of the `write` operator. If not throw an exception.
* Create a `TransactionBuilder` - this basically begins the steps of transaction construction.
```java
import io.delta.kernel.*;
import io.delta.kernel.defaults.engine.*;
Engine myEngine = new MyEngine();
Table myTable = Table.forPath(myTablePath);
StructType writeOperatorSchema = // ... derived from the query operator tree ...
StructType dataSchema;
boolean isNewTable = false;
try {
Snapshot mySnapshot = myTable.getLatestSnapshot(myEngine);
dataSchema = mySnapshot.getSchema(myEngine);
// .. check dataSchema and writeOperatorSchema match ...
} catch(TableNotFoundException e) {
isNewTable = true;
dataSchema = writeOperatorSchema;
}
TransactionBuilder txnBuilder =
myTable.createTransactionBuilder(
myEngine,
"Examples", /* engineInfo - connector can add its own identifier which is noted in the Delta Log */
Operation /* What is the operation we are trying to perform? This is noted in the Delta Log */
);
if (isNewTable) {
// For a new table set the table schema in the transaction builder
txnBuilder = txnBuilder.withSchema(engine, dataSchema)
}
```
#### Step 4.2: Determine the physical partitioning of the data based on the table schema and partition columns
[Section titled “Step 4.2: Determine the physical partitioning of the data based on the table schema and partition columns”](#step-42-determine-the-physical-partitioning-of-the-data-based-on-the-table-schema-and-partition-columns)
Partition columns are found either from the query (for new tables, the query defines the partition columns) or from the existing table.
```java
TransactionBuilder txnBuilder = ... from the last step ...
Transaction txn;
List partitionColumns = ...
if (newTable) {
partitionColumns = ... derive from the query parameters (ex. PARTITION BY clause in SQL) ...
txnBuilder = txnBuilder.withPartitionColumns(engine, partitionColumns);
txn = txnBuilder.build(engine);
} else {
txn = txnBuilder.build(engine);
partitionColumns = txn.getPartitionColumns(engine);
}
```
At the end of this step, we have the `Transaction` and schema of the data to generate and its partitioning.
#### Step 4.3: Distribute the writer tasks definitions (which include the transaction state) to workers
[Section titled “Step 4.3: Distribute the writer tasks definitions (which include the transaction state) to workers”](#step-43-distribute-the-writer-tasks-definitions-which-include-the-transaction-state-to-workers)
If you are building a connector for a distributed engine like Spark/Presto/Trino/Flink, then your connector has to send all the writer metadata from the query planning machine (henceforth called the driver) to task execution machines (henceforth called the workers). You will have to serialize and deserialize the transaction state. It is the connector job to implement serialization and deserialization utilities for a [`Row`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/Row.html). More details on a custom `Row` SerDe are found [here](#custom-row-serializerdeserializer).
```java
Row txnState = txn.getState(engine);
String jsonTxnState = serializeToJson(txnState);
```
#### Step 4.4: Tasks write the data to data files and send the data file info to the driver
[Section titled “Step 4.4: Tasks write the data to data files and send the data file info to the driver”](#step-44-tasks-write-the-data-to-data-files-and-send-the-data-file-info-to-the-driver)
In this step (which is executed on the worker nodes inside each task):
* Deserialize the transaction state
* Writer operator within the task gets the data from its parent operator.
* The data is converted into a `FilteredColumnarBatch`. Each [`FilteredColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/FilteredColumnarBatch.html) has two components:
* Columnar batch (returned by `FilteredColumnarBatch.getData()`): This is the data read from the files having the schema matching the readSchema provided when the Scan object was built in the earlier step.
* Optional selection vector (returned by `FilteredColumnarBatch.getSelectionVector()`): Optionally, a boolean vector that will define which rows in the batch are valid and should be consumed by the engine.
* The connector can create `FilteredColumnBatch` wrapper around data in its own in-memory format.
* Check if the data is partitioned or not. If not partitioned, partition the data by partition values.
* For each partition generate the map of the partition column to the partition value
* Use Kernel to convert the partitioned data into physical data that should go into the data files
* Write the physical data into one or more data files.
* Convert data file statues into a Delta log actions
* Serialize the Delta log action `Row` objects and send them to the driver node
```plaintext
Row txnState = ... deserialize from JSON string sent by the driver ...
CloseableIterator data = ... generate data ...
// If the table is un-partitioned then this is an empty map
Map partitionValues = ... prepare the partition values ...
// First transform the logical data to physical data that needs to be written
// to the Parquet files
CloseableIterator physicalData =
Transaction.transformLogicalData(engine, txnState, data, partitionValues);
// Get the write context
DataWriteContext writeContext = Transaction.getWriteContext(engine, txnState, partitionValues);
// Now write the physical data to Parquet files
CloseableIterator dataFiles =
engine.getParquetHandler()
.writeParquetFiles(
writeContext.getTargetDirectory(),
physicalData,
writeContext.getStatisticsColumns());
// Now convert the data file status to data actions that needs to be written to the Delta table log
CloseableIterator partitionDataActions =
Transaction.generateAppendActions(
engine,
txnState,
dataFiles,
writeContext);
.... serialize `partitionDataActions` and send them to driver node
```
#### Step 4.5: Finalize the query
[Section titled “Step 4.5: Finalize the query”](#step-45-finalize-the-query)
At the driver node, the delta log actions from all the tasks are received and committed to the transaction. The tasks send the Delta log actions as a serialized JSON and deserialize them back to `Row` objects.
```plaintext
// Create a iterable out of the data actions. If the contents are too big to fit in memory,
// the connector may choose to write the data actions to a temporary file and return an
// iterator that reads from the file.
CloseableIterable dataActionsIterable = CloseableIterable.inMemoryIterable(
toCloseableIterator(dataActions.iterator()));
// Commit the transaction.
TransactionCommitResult commitResult = txn.commit(engine, dataActionsIterable);
// Optional step
if (commitResult.isReadyForCheckpoint()) {
// Checkpoint the table
Table.forPath(engine, tablePath).checkpoint(engine, commitResult.getVersion());
}
```
Thats it. Now you should be able to append data to Delta tables using the Kernel APIs.
## Migration guide
[Section titled “Migration guide”](#migration-guide)
Kernel APIs are still evolving and new features are being added. Kernel authors try to make the API changes backward compatible as much as they can with each new release, but sometimes it is hard to maintain the backward compatibility for a project that is evolving rapidly.
This section provides guidance on how to migrate your connector to the latest version of Delta Kernel. With each new release the [examples](https://github.com/delta-io/delta/tree/master/kernel/examples) are kept up-to-date with the latest API changes. You can refer to the examples to understand how to use the new APIs.
### Migration from Delta Lake version 3.1.0 to 3.2.0
[Section titled “Migration from Delta Lake version 3.1.0 to 3.2.0”](#migration-from-delta-lake-version-310-to-320)
Following are API changes in Delta Kernel 3.2.0 that may require changes in your connector.
#### Rename `TableClient` to [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html)
[Section titled “Rename TableClient to Engine”](#rename-tableclient-to-engine)
The `TableClient` interface has been renamed to `Engine`. This is the most significant API change in this release. The `TableClient` interface name is not exactly representing the functionality it provides. At a high level it provides capabilities such as reading Parquet files, JSON files, evaluating expressions on data and file system functionality. These are basically the heavy lift operations that Kernel depends on as a separate interface to allow the connectors to substitute their own custom implementation of the same functionality (e.g. custom Parquet reader). Essentially, these functionalities are the core of the `engine` functionalities. By renaming to `Engine`, we are representing the interface functionality with a proper name that is easy to understand.
The `DefaultTableClient` has been renamed to [`DefaultEngine`](https://delta-io.github.io/delta/snapshot/kernel-defaults/java/io/delta/kernel/defaults/engine/DefaultEngine.html).
#### [`Table.forPath(Engine engine, String tablePath)`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Table.html#forPath-io.delta.kernel.engine.Engine-java.lang.String-) behavior change
[Section titled “Table.forPath(Engine engine, String tablePath) behavior change”](#tableforpathengine-engine-string-tablepath-behavior-change)
Earlier when a non-existent table path is passed, the API used to throw `TableNotFoundException`. Now it doesn’t throw the exception. Instead, it returns a `Table` object. When trying to get a `Snapshot` from the table object it throws the `TableNotFoundException`.
#### [`FileSystemClient.resolvePath`](https://delta-io.github.io/delta/snapshot/kernel-defaults/java/io/delta/kernel/defaults/engine/DefaultFileSystemClient.html#resolvePath-java.lang.String-) behavior change
[Section titled “FileSystemClient.resolvePath behavior change”](#filesystemclientresolvepath-behavior-change)
Earlier when a non-existent path is passed, the API used to throw `FileNotFoundException`. Now it doesn’t throw the exception. It still resolves the given path into a fully qualified path.
## Kernel Rust
[Section titled “Kernel Rust”](#kernel-rust)
The Rust Kernel is a set of libraries for building Delta connectors in native languages. Work in progress.
## More Information
[Section titled “More Information”](#more-information)
* [Talk](https://www.youtube.com/watch?v=KVUMFv7470I) explaining the rationale behind Kernel and the API design (slides are available [here](https://docs.google.com/presentation/d/1PGSSuJ8ndghucSF9GpYgCi9oeRpWolFyehjQbPh92-U/edit) which are kept up-to-date with the changes).
* [User guide](https://github.com/delta-io/delta/blob/master/kernel/USER_GUIDE.md) on the step-by-step process of using Kernel in a standalone Java program or in a distributed processing connector for reading and writing to Delta tables.
* Example [Java programs](https://github.com/delta-io/delta/tree/master/kernel/examples) that illustrate how to read and write Delta tables using the Kernel APIs.
* Table and default Engine API Java [documentation](/api/latest/java/kernel/index.html)
* [Migration guide](https://github.com/delta-io/delta/blob/master/kernel/USER_GUIDE.md#migration-guide)
# Delta Kernel Java User Guide
> Learn how to build connectors to read and write Delta tables using Delta Kernel Java.
## What is Delta Kernel?
[Section titled “What is Delta Kernel?”](#what-is-delta-kernel)
Delta Kernel is a library for operating on Delta tables. Specifically, it provides simple and narrow APIs for reading and writing to Delta tables without the need to understand the [Delta protocol](https://github.com/delta-io/delta/blob/master/PROTOCOL.md) details. You can use this library to do the following:
* Read and write Delta tables from your applications.
* Build a connector for a distributed engine like [Apache Spark™](https://github.com/apache/spark), [Apache Flink](https://github.com/apache/flink), or [Trino](https://github.com/trinodb/trino) for reading or writing massive Delta tables.
## Set up Delta Kernel for your project
[Section titled “Set up Delta Kernel for your project”](#set-up-delta-kernel-for-your-project)
You need to `io.delta:delta-kernel-api` and `io.delta:delta-kernel-defaults` dependencies. Following is an example Maven `pom` file dependency list.
The `delta-kernel-api` module contains the core of the Kernel that abstracts out the Delta protocol to enable reading and writing into Delta tables. It makes use of the `Engine` interface that is being passed to the Kernel API by the connector for heavy-lift operations such as reading/writing Parquet or JSON files, evaluating expressions or file system operations such as listing contents of the Delta Log directory, etc. Kernel supplies a default implementation of `Engine` in module `delta-kernel-defaults`. The connectors can implement their own version of `Engine` to make use of their native implementation of functionalities the `Engine` provides. For example: the connector can make use of their Parquet reader instead of using the reader from the `DefaultEngine`. More details on this [later](#step-2-build-your-own-engine).
```xml
io.delta
delta-kernel-api
${delta-kernel.version}
io.delta
delta-kernel-defaults
${delta-kernel.version}
```
If your connector is not using the [`DefaultEngine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) provided by the Kernel, the dependency `delta-kernel-defaults` from the above list can be skipped.
## Read a Delta table in a single process
[Section titled “Read a Delta table in a single process”](#read-a-delta-table-in-a-single-process)
In this section, we will walk through how to build a very simple single-process Delta connector that can read a Delta table using the default [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) implementation provided by Delta Kernel.
You can either write this code yourself in your project, or you can use the [examples](https://github.com/delta-io/delta/tree/master/kernel/examples) present in the Delta code repository.
### Step 1: Full scan on a Delta table
[Section titled “Step 1: Full scan on a Delta table”](#step-1-full-scan-on-a-delta-table)
The main entry point is [`io.delta.kernel.Table`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Table.html) which is a programmatic representation of a Delta table. Say you have a Delta table at the directory `myTablePath`. You can create a `Table` object as follows:
```java
import io.delta.kernel.*;
import io.delta.kernel.defaults.*;
import org.apache.hadoop.conf.Configuration;
String myTablePath = ; // fully qualified table path. Ex: file:/user/tables/myTable
Configuration hadoopConf = new Configuration();
Engine myEngine = DefaultEngine.create(hadoopConf);
Table myTable = Table.forPath(myEngine, myTablePath);
```
Note the default [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) we are creating to bootstrap the `myTable` object. This object allows you to plug in your own libraries for computationally intensive operations like Parquet file reading, JSON parsing, etc. You can ignore it for now. We will discuss more about this later when we discuss how to build more complex connectors for distributed processing engines.
From this `myTable` object you can create a [`Snapshot`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Snapshot.html) object which represents the consistent state (a.k.a. a snapshot consistency) in a specific version of the table.
```java
Snapshot mySnapshot = myTable.getLatestSnapshot(myEngine);
```
Now that we have a consistent snapshot view of the table, we can query more details about the table. For example, you can get the version and schema of this snapshot.
```java
long version = mySnapshot.getVersion();
StructType tableSchema = mySnapshot.getSchema();
```
Next, to read the table data, we have to *build* a [`Scan`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Scan.html) object. In order to build a `Scan` object, create a [`ScanBuilder`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/ScanBuilder.html) object which optionally allows selecting a subset of columns to read or setting a query filter. For now, ignore these optional settings.
```java
Scan myScan = mySnapshot.getScanBuilder().build()
// Common information about scanning for all data files to read.
Row scanState = myScan.getScanState(myEngine)
// Information about the list of scan files to read
CloseableIterator scanFiles = myScan.getScanFiles(myEngine)
```
This [`Scan`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Scan.html) object has all the necessary metadata to start reading the table. There are two crucial pieces of information needed for reading data from a file in the table.
* `myScan.getScanFiles(Engine)`: Returns scan files as columnar batches (represented as an iterator of `FilteredColumnarBatch`es, more on that later) where each selected row in the batch has information about a single file containing the table data.
* `myScan.getScanState(Engine)`: Returns the snapshot-level information needed for reading any file. Note that this is a single row and common to all scan files.
For each scan file the physical data must be read from the file. The columns to read are specified in the scan file state. Once the physical data is read, you have to call [`ScanFile.transformPhysicalData(...)`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Scan.html#transformPhysicalData-io.delta.kernel.engine.Engine-io.delta.kernel.data.Row-io.delta.kernel.data.Row-io.delta.kernel.utils.CloseableIterator-) with the scan state and the physical data read from scan file. This API takes care of transforming (e.g. adding partition columns) the physical data into logical data of the table. Here is an example of reading all the table data in a single thread.
```java
CloserableIterator fileIter = scanObject.getScanFiles(myEngine);
Row scanStateRow = scanObject.getScanState(myEngine);
while(fileIter.hasNext()) {
FilteredColumnarBatch scanFileColumnarBatch = fileIter.next();
// Get the physical read schema of columns to read from the Parquet data files
StructType physicalReadSchema =
ScanStateRow.getPhysicalDataReadSchema(engine, scanStateRow);
try (CloseableIterator scanFileRows = scanFileColumnarBatch.getRows()) {
while (scanFileRows.hasNext()) {
Row scanFileRow = scanFileRows.next();
// From the scan file row, extract the file path, size and modification time metadata
// needed to read the file.
FileStatus fileStatus = InternalScanFileUtils.getAddFileStatus(scanFileRow);
// Open the scan file which is a Parquet file using connector's own
// Parquet reader or default Parquet reader provided by the Kernel (which
// is used in this example).
CloseableIterator physicalDataIter =
engine.getParquetHandler().readParquetFiles(
singletonCloseableIterator(fileStatus),
physicalReadSchema,
Optional.empty() /* optional predicate the connector can apply to filter data from the reader */
);
// Now the physical data read from the Parquet data file is converted to a table
// logical data. Logical data may include the addition of partition columns and/or
// subset of rows deleted
try (
CloseableIterator transformedData =
Scan.transformPhysicalData(
engine,
scanStateRow,
scanFileRow,
physicalDataIter)) {
while (transformedData.hasNext()) {
FilteredColumnarBatch logicalData = transformedData.next();
ColumnarBatch dataBatch = logicalData.getData();
// Not all rows in `dataBatch` are in the selected output.
// An optional selection vector determines whether a row with a
// specific row index is in the final output or not.
Optional selectionVector = dataReadResult.getSelectionVector();
// access the data for the column at ordinal 0
ColumnVector column0 = dataBatch.getColumnVector(0);
for (int rowIndex = 0; rowIndex < column0.getSize(); rowIndex++) {
// check if the row is selected or not
if (!selectionVector.isPresent() || // there is no selection vector, all records are selected
(!selectionVector.get().isNullAt(rowId) && selectionVector.get().getBoolean(rowId))) {
// Assuming the column type is String.
// If it is a different type, call the relevant function on the `ColumnVector`
System.out.println(column0.getString(rowIndex));
}
}
// access the data for column at ordinal 1
ColumnVector column1 = dataBatch.getColumnVector(1);
for (int rowIndex = 0; rowIndex < column1.getSize(); rowIndex++) {
// check if the row is selected or not
if (!selectionVector.isPresent() || // there is no selection vector, all records are selected
(!selectionVector.get().isNullAt(rowId) && selectionVector.get().getBoolean(rowId))) {
// Assuming the column type is Long.
// If it is a different type, call the relevant function on the `ColumnVector`
System.out.println(column1.getLong(rowIndex));
}
}
// .. more ..
}
}
}
}
}
```
A few working examples to read Delta tables within a single process are available [here](https://github.com/delta-io/delta/tree/master/kernel/examples).
important
All the Delta protocol-level details are encoded in the rows returned by `Scan.getScanFiles` API, but you do not have to understand them in order to read the table data correctly. All you need is to get the Parquet file status from each scan file row and read the data from the Parquet file into the `ColumnarBatch` format. The physical data is converted into the logical data of the table using [`Scan.transformPhysicalData`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Scan.html#transformPhysicalData-io.delta.kernel.engine.Engine-io.delta.kernel.data.Row-io.delta.kernel.data.Row-io.delta.kernel.utils.CloseableIterator-). Transformation to logical data is dictated by the protocol and the metadata of the table and the scan file. As the Delta protocol evolves this transformation step will evolve with it and your code will not have to change to accommodate protocol changes. This is the major advantage of the abstractions provided by Delta Kernel.
Note
Observe that the same `Engine` instance `myEngine` is passed multiple times whenever a call to Delta Kernel API is made. The reason for passing this instance for every call is because it is the connector context; it should be maintained outside of the Delta Kernel APIs to give the connector control over the `Engine`.
### Step 2: Improve scan performance with file skipping
[Section titled “Step 2: Improve scan performance with file skipping”](#step-2-improve-scan-performance-with-file-skipping)
We have explored how to do a full table scan. However, the real advantage of using the Delta format is that you can skip files using your query filters. To make this possible, Delta Kernel provides an [expression framework](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/expressions/package-summary.html) to encode your filters and provide them to Delta Kernel to skip files during the scan file generation. For example, say your table is partitioned by `columnX`, you want to query only the partition `columnX=1`. You can generate the expression and use it to build the scan as follows:
```java
import io.delta.kernel.expressions.*;
import io.delta.kernel.defaults.engine.*;
Engine myEngine = DefaultEngine.create(new Configuration());
Predicate filter = new Predicate(
"=",
Arrays.asList(new Column("columnX"), Literal.ofInt(1)));
Scan myFilteredScan = mySnapshot.getScanBuilder().withFilter(filter).build()
// Subset of the given filter that is not guaranteed to be satisfied by
// Delta Kernel when it returns data. This filter is used by Delta Kernel
// to do data skipping as much as possible. The connector should use this filter
// on top of the data returned by Delta Kernel in order for further filtering.
Optional remainingFilter = myFilteredScan.getRemainingFilter();
```
The scan files returned by `myFilteredScan.getScanFiles(myEngine)` will have rows representing files only of the required partition. Similarly, you can provide filters for non-partition columns, and if the data in the table is well clustered by those columns, then Delta Kernel will be able to skip files as much as possible.
## Create a Delta table
[Section titled “Create a Delta table”](#create-a-delta-table)
In this section, we will walk through how to build a Delta connector that can create a Delta table using the default [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) implementation provided by Delta Kernel.
You can either write this code yourself in your project, or you can use the [examples](https://github.com/delta-io/delta/tree/master/kernel/examples) present in the Delta code repository.
The main entry point is [`io.delta.kernel.Table`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Table.html) which is a programmatic representation of a Delta table. Say you want to create Delta table at the directory `myTablePath`. You can create a `Table` object as follows:
```java
package io.delta.kernel.examples;
import io.delta.kernel.*;
import io.delta.kernel.types.*;
import io.delta.kernel.utils.CloseableIterable;
String myTablePath = ;
Configuration hadoopConf = new Configuration();
Engine myEngine = DefaultEngine.create(hadoopConf);
Table myTable = Table.forPath(myEngine, myTablePath);
```
Note the default [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) we are creating to bootstrap the `myTable` object. This object allows you to plug in your own libraries for computationally intensive operations like Parquet file reading, JSON parsing, etc. You can ignore it for now. We will discuss more about this later when we discuss how to build more complex connectors for distributed processing engines.
From this `myTable` object you can create a [`TransactionBuilder`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/TransactionBuilder.html) object which allows you to construct a [`Transaction`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Transaction.html) object
```java
TransactionBuilder txnBuilder =
myTable.createTransactionBuilder(
myEngine,
"Examples", /* engineInfo - connector can add its own identifier which is noted in the Delta Log */
Operation.CREATE_TABLE /* What is the operation we are trying to perform. This is noted in the Delta Log */
);
```
Now that you have the `TransactionBuilder` object, you can set the table schema and partition columns of the table.
```java
StructType mySchema = new StructType()
.add("id", IntegerType.INTEGER)
.add("name", StringType.STRING)
.add("city", StringType.STRING)
.add("salary", DoubleType.DOUBLE);
// Partition columns are optional. Use it only if you are creating a partitioned table.
List myPartitionColumns = Collections.singletonList("city");
// Set the schema of the new table on the transaction builder
txnBuilder = txnBuilder
.withSchema(engine, mySchema);
// Set the partition columns of the new table only if you are creating
// a partitioned table; otherwise, this step can be skipped.
txnBuilder = txnBuilder
.withPartitionColumns(engine, examplePartitionColumns);
```
`TransactionBuilder` allows setting additional properties of the table such as enabling a certain Delta feature or setting identifiers for idempotent writes. We will be visiting these in the next sections. The next step is to build `Transaction` out of the `TransactionBuilder` object.
```java
// Build the transaction
Transaction txn = txnBuilder.build(engine);
```
`Transaction` object allows the connector to optionally add any data and finally commit the transaction. A successful commit ensures that the table is created with the given schema. In this example, we are just creating a table and not adding any data as part of the table.
```java
// Commit the transaction.
// As we are just creating the table and not adding any data, the `dataActions` is empty.
TransactionCommitResult commitResult =
txn.commit(
engine,
CloseableIterable.emptyIterable() /* dataActions */
);
```
The [`TransactionCommitResult`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/TransactionCommitResult.html) contains the what version the transaction is committed as and whether the table is ready for a checkpoint. As we are creating a table the version will be `0`. We will be discussing later on what a checkpoint is and what it means for the table to be ready for the checkpoint.
A few working examples to create partitioned and un-partitioned Delta tables are available [here](https://github.com/delta-io/delta/tree/master/kernel/examples).
## Create a table and insert data into it
[Section titled “Create a table and insert data into it”](#create-a-table-and-insert-data-into-it)
In this section, we will walk through how to build a Delta connector that can create a Delta table and insert data into the table (similar to `CREATE TABLE AS ` construct in SQL) using the default [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) implementation provided by Delta Kernel.
You can either write this code yourself in your project, or you can use the [examples](https://github.com/delta-io/delta/tree/master/kernel/examples) present in the Delta code repository.
The first step is to construct a `Transaction`. Below is the code for that. For more details on what each step of the code means, please read the [create table](#create-a-delta-table) section.
```plaintext
package io.delta.kernel.examples;
import io.delta.kernel.*;
import io.delta.kernel.types.*;
import io.delta.kernel.utils.CloseableIterable;
String myTablePath = ;
Configuration hadoopConf = new Configuration();
Engine myEngine = DefaultEngine.create(hadoopConf);
Table myTable = Table.forPath(myEngine, myTablePath);
StructType mySchema = new StructType()
.add("id", IntegerType.INTEGER)
.add("name", StringType.STRING)
.add("city", StringType.STRING)
.add("salary", DoubleType.DOUBLE);
// Partition columns are optional. Use it only if you are creating a partitioned table.
List myPartitionColumns = Collections.singletonList("city");
TransactionBuilder txnBuilder =
myTable.createTransactionBuilder(
myEngine,
"Examples", /* engineInfo - connector can add its own identifier which is noted in the Delta Log */
Operation.WRITE /* What is the operation we are trying to perform? This is noted in the Delta Log */
);
// Set the schema of the new table on the transaction builder
txnBuilder = txnBuilder
.withSchema(engine, mySchema);
// Set the partition columns of the new table only if you are creating
// a partitioned table; otherwise, this step can be skipped.
txnBuilder = txnBuilder
.withPartitionColumns(engine, examplePartitionColumns);
// Build the transaction
Transaction txn = txnBuilder.build(engine);
```
Now that we have the [`Transaction`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Transaction.html) object, the next step is generating the data that confirms the table schema and partitioned according to the table partitions.
```java
StructType dataSchema = txn.getSchema(engine)
// Optional for un-partitioned tables
List partitionColumnNames = txn.getPartitionColumns(engine)
```
Using the data schema and partition column names the connector can plan the query and generate data. At tasks that actually have the data to write to the table, the connector can ask the Kernel to transform the data given in the table schema into physical data that can actually be written to the Parquet data files. For partitioned tables, the data needs to be first partitioned by the partition columns, and then the connector should ask the Kernel to transform the data for each partition separately. The partitioning step is needed because any given data file in the Delta table contains data belonging to exactly one partition.
Get the state of the transaction. The transaction state contains the information about how to convert the data in the table schema into physical data that needs to be written. The transformations depend on the protocol and features the table has.
```java
Row txnState = txn.getTransactionState(engine);
```
Prepare the data.
```java
// The data generated by the connector to write into a table
CloseableIterator data = ...
// Create partition value map
Map partitionValues =
Collections.singletonMap(
"city", // partition column name
// partition value. Depending upon the partition column type, the
// partition value should be created. In this example, the partition
// column is of type StringType, so we are creating a string literal.
Literal.ofString(city)
);
```
The connector data is passed as an iterator of [`FilteredColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/FilteredColumnarBatch.html). Each of the `FilteredColumnarBatch` contains a [`ColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnarBatch.html) which actually contains the data in columnar access format and an optional section vector that allows the connector to specify which rows from the `ColumnarBatch` to write to the table.
Partition values are passed as a map of the partition column name to the partition value. For an un-partitioned table, the map should be empty as it has no partition columns.
```plaintext
// Transform the logical data to physical data that needs to be written to the Parquet
// files
CloseableIterator physicalData =
Transaction.transformLogicalData(engine, txnState, data, partitionValues);
```
The above code converts the given data for partitions into an iterator of `FilteredColumnarBatch` that needs to be written to the Parquet data files. In order to write the data files, the connector needs to get the [`WriteContext`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/DataWriteContext.html) from Kernel, which tells the connector where to write the data files and what columns to collect statistics from each data file.
```java
// Get the write context
DataWriteContext writeContext = Transaction.getWriteContext(engine, txnState, partitionValues);
```
Now, the connector has the physical data that needs to be written to Parquet data files, and where those files should be written, it can start writing the data files.
```java
CloseableIterator dataFiles = engine.getParquetHandler()
.writeParquetFiles(
writeContext.getTargetDirectory(),
physicalData,
writeContext.getStatisticsColumns()
);
```
In the above code, the connector is making use of the `Engine` provided `ParquetHandler` to write the data, but the connector can choose its own Parquet file writer to write the data. Also note that the return of the above call is an iterator of [`DataFileStatus`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/utils/DataFileStatus.html) for each data file written. It basically contains the file path, file metadata, and optional file-level statistics for columns specified by the [`WriteContext.getStatisticsColumns()`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/DataWriteContext.html#getStatisticsColumns--)
Convert each `DataFileStatus` into a Delta log action that can be written to the Delta table log.
```java
CloseableIterator dataActions =
Transaction.generateAppendActions(engine, txnState, dataFiles, writeContext);
```
The next step is constructing [`CloseableIterable`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/utils/CloseableIterable.html) out of the all the Delta log actions generated above. The reason for constructing an `Iterable` is that the transaction committing involves accessing the list of Delta log actions more than one time (in order to resolve conflicts when there are multiple writes to the table). Kernel provides a [utility method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/utils/CloseableIterable.html#inMemoryIterable-io.delta.kernel.utils.CloseableIterator-) to create an in-memory version of `CloseableIterable`. This interface also gives the connector an option to implement a custom implementation that spills the data actions to disk when the contents are too big to fit in memory.
```java
// Create a iterable out of the data actions. If the contents are too big to fit in memory,
// the connector may choose to write the data actions to a temporary file and return an
// iterator that reads from the file.
CloseableIterable dataActionsIterable = CloseableIterable.inMemoryIterable(dataActions);
```
The final step is committing the transaction!
```java
TransactionCommitStatus commitStatus = txn.commit(engine, dataActionsIterable)
```
The [`TransactionCommitResult`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/TransactionCommitResult.html) contains the what version the transaction is committed as and whether the table is ready for a checkpoint. As we are creating a table the version will be `0`. We will be discussing later on what a checkpoint is and what it means for the table to be ready for the checkpoint.
A few working examples to create and insert data into partitioned and un-partitioned Delta tables are available [here](https://github.com/delta-io/delta/tree/master/kernel/examples).
## Blind append into an existing Delta table
[Section titled “Blind append into an existing Delta table”](#blind-append-into-an-existing-delta-table)
In this section, we will walk through how to build a Delta connector that inserts data into an existing Delta table (similar to `INSERT INTO ` construct in SQL) using the default [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) implementation provided by Delta Kernel.
You can either write this code yourself in your project, or you can use the [examples](https://github.com/delta-io/delta/tree/master/kernel/examples) present in the Delta code repository. The steps are exactly similar to [Create table and insert data into it](#create-a-table-and-insert-data-into-it) except that we won’t be providing any schema or partition columns when building the `TransactionBuilder`
```java
// Create a `Table` object with the given destination table path
Table table = Table.forPath(engine, tablePath);
// Create a transaction builder to build the transaction
TransactionBuilder txnBuilder =
table.createTransactionBuilder(
engine,
"Examples", /* engineInfo */
Operation.WRITE
);
/ Build the transaction - no need to provide the schema as the table already exists.
Transaction txn = txnBuilder.build(engine);
// Get the transaction state
Row txnState = txn.getTransactionState(engine);
List dataActions = new ArrayList<>();
// Generate the sample data for three partitions. Process each partition separately.
// This is just an example. In a real-world scenario, the data may come from different
// partitions. Connectors already have the capability to partition by partition values
// before writing to the table
// In the test data `city` is a partition column
for (String city : Arrays.asList("San Francisco", "Campbell", "San Jose")) {
FilteredColumnarBatch batch1 = generatedPartitionedDataBatch(
5 /* offset */, city /* partition value */);
FilteredColumnarBatch batch2 = generatedPartitionedDataBatch(
5 /* offset */, city /* partition value */);
FilteredColumnarBatch batch3 = generatedPartitionedDataBatch(
10 /* offset */, city /* partition value */);
CloseableIterator data =
toCloseableIterator(Arrays.asList(batch1, batch2, batch3).iterator());
// Create partition value map
Map partitionValues =
Collections.singletonMap(
"city", // partition column name
// partition value. Depending upon the parition column type, the
// partition value should be created. In this example, the partition
// column is of type StringType, so we are creating a string literal.
Literal.ofString(city));
// First transform the logical data to physical data that needs to be written
// to the Parquet
// files
CloseableIterator physicalData =
Transaction.transformLogicalData(engine, txnState, data, partitionValues);
// Get the write context
DataWriteContext writeContext =
Transaction.getWriteContext(engine, txnState, partitionValues);
// Now write the physical data to Parquet files
CloseableIterator dataFiles = engine.getParquetHandler()
.writeParquetFiles(
writeContext.getTargetDirectory(),
physicalData,
writeContext.getStatisticsColumns());
// Now convert the data file status to data actions that needs to be written to the Delta
// table log
CloseableIterator partitionDataActions = Transaction.generateAppendActions(
engine,
txnState,
dataFiles,
writeContext);
// Now add all the partition data actions to the main data actions list. In a
// distributed query engine, the partition data is written to files at tasks on executor
// nodes. The data actions are collected at the driver node and then written to the
// Delta table log using the `Transaction.commit`
while (partitionDataActions.hasNext()) {
dataActions.add(partitionDataActions.next());
}
}
// Create a iterable out of the data actions. If the contents are too big to fit in memory,
// the connector may choose to write the data actions to a temporary file and return an
// iterator that reads from the file.
CloseableIterable dataActionsIterable = CloseableIterable.inMemoryIterable(
toCloseableIterator(dataActions.iterator()));
// Commit the transaction.
TransactionCommitResult commitResult = txn.commit(engine, dataActionsIterable);
```
## Idempotent Blind Appends to a Delta Table
[Section titled “Idempotent Blind Appends to a Delta Table”](#idempotent-blind-appends-to-a-delta-table)
Idempotent writes allow the connector to make sure the data belonging to a particular transaction version and application id is inserted into the table at most once. In incremental processing systems (e.g. streaming systems), track progress using their own application-specific versions need to record what progress has been made in order to avoid duplicating data in the face of failures and retries during writes. By setting the transaction identifier, the Delta table can ensure that the data with the same identifier is not written multiple times. For more information refer to the Delta protocol section [Transaction Identifiers](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#transaction-identifiers)
To make the data append idempotent, set the transaction identifier on the [`TransactionBuilder`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/TransactionBuilder.html#withTransactionId-io.delta.kernel.engine.Engine-java.lang.String-long-)
```java
// Set the transaction identifiers for idempotent writes
// Delta/Kernel makes sure that there exists only one transaction in the Delta log
// with the given application id and txn version
txnBuilder =
txnBuilder.withTransactionId(
engine,
"my app id", /* application id */
100 /* monotonically increasing txn version with each new data insert */
);
```
That’s all the connector need to do for idempotent blind appends.
## Checkpointing a Delta table
[Section titled “Checkpointing a Delta table”](#checkpointing-a-delta-table)
[Checkpoints](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#checkpoints) are an optimization in Delta Log in order to construct the state of the Delta table faster. It basically contains the state of the table at the version the checkpoint is created. Delta Kernel allows the connector to optionally make the checkpoints. It is created for every few commits (configurable table property) on the table.
The result of `Transaction.commit` returns a `TransactionCommitResult` that contains the version the transaction is committed as and whether the table is [read for checkpoint](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/TransactionCommitResult.html#isReadyForCheckpoint--). Creating a checkpoint takes time as it needs to construct the entire state of the table. If the connector doesn’t want to checkpoint by itself but uses other connectors that are faster in creating a checkpoint, it can skip the checkpointing step.
If it wants to checkpoint, the `Table` object has an [API](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Table.html#checkpoint-io.delta.kernel.engine.Engine-long-) to checkpoint the table.
```java
TransactionCommitResult commitResult = txn.commit(engine, dataActionsIterable);
if (commitResult.isReadyForCheckpoint()) {
// Checkpoint the table
Table.forPath(engine, tablePath).checkpoint(engine, commitResult.getVersion());
}
```
## Build a Delta connector for a distributed processing engine
[Section titled “Build a Delta connector for a distributed processing engine”](#build-a-delta-connector-for-a-distributed-processing-engine)
Unlike simple applications that just read the table in a single process, building a connector for complex processing engines like Apache Spark™ and Trino can require quite a bit of additional effort. For example, to build a connector for an SQL engine you have to do the following
* Understand the APIs provided by the engine to build connectors and how Delta Kernel can be used to provide the information necessary for the connector + engine to operate on a Delta table.
* Decide what libraries to use to do computationally expensive operations like reading Parquet files, parsing JSON, computing expressions, etc. Delta Kernel provides all the extension points to allow you to plug in any library without having to understand all the low-level details of the Delta protocol.
* Deal with details specific to distributed engines. For example,
* Serialization of Delta table metadata provided by Delta Kernel.
* Efficiently transforming data read from Parquet into the engine in-memory processing format.
In this section, we are going to outline the steps needed to build a connector.
### Step 0: Validate the prerequisites
[Section titled “Step 0: Validate the prerequisites”](#step-0-validate-the-prerequisites)
In the previous section showing how to read a simple table, we were briefly introduced to the [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html). This is the main extension point where you can plug in your implementations of computationally-expensive operations like reading Parquet files, parsing JSON, etc. For the simple case, we were using a default implementation of the helper that works in most cases. However, for building a high-performance connector for a complex processing engine, you will very likely need to provide your own implementation using the libraries that work with your engine. So before you start building your connector, it is important to understand these requirements and plan for building your own engine.
Here are the libraries/capabilities you need to build a connector that can read the Delta table
* Perform file listing and file reads from your storage/file system.
* Read Parquet files in columnar data, preferably in an in-memory columnar format.
* Parse JSON data
* Read JSON files
* Evaluate expressions on in-memory columnar batches
For each of these capabilities, you can choose to build your own implementation or reuse the default implementation.
### Step 1: Set up Delta Kernel in your connector project
[Section titled “Step 1: Set up Delta Kernel in your connector project”](#step-1-set-up-delta-kernel-in-your-connector-project)
In the Delta Kernel project, there are multiple dependencies you can choose to depend on.
1. Delta Kernel core APIs - This is a must-have dependency, which contains all the main APIs like Table, Snapshot, and Scan that you will use to access the metadata and data of the Delta table. This has very few dependencies reducing the chance of conflicts with any dependencies in your connector and engine. This also provides the [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) interface which allows you to plug in your implementations of computationally expensive operations, but it does not provide any implementation of this interface.
2. Delta Kernel default- This has a default implementation called [`DefaultEngine`](https://delta-io.github.io/delta/snapshot/kernel-defaults/java/io/delta/kernel/defaults/engine/DefaultEngine.html) and additional dependencies such as `Hadoop`. If you wish to reuse all or parts of this implementation, then you can optionally depend on this.
#### Set up Java projects
[Section titled “Set up Java projects”](#set-up-java-projects)
As discussed above, you can import one or both of the artifacts as follows:
```xml
io.delta
delta-kernel-api
${delta-kernel.version}
io.delta
delta-kernel-defaults
${delta-kernel.version}
```
### Step 2: Build your own Engine
[Section titled “Step 2: Build your own Engine”](#step-2-build-your-own-engine)
In this section, we are going to explore the [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) interface and walk through how to implement your own implementation so that you can plug in your connector/engine-specific implementations of computationally-intensive operations, threading model, resource management, etc.
important
During the validation process, if you believe that all the dependencies of the default `Engine` implementation can work with your connector and engine, then you can skip this step and jump to Step 3 of implementing your connector using the default engine. If later you have the need to customize the helper for your connector, you can revisit this step.
#### Step 2.1: Implement the `Engine` interface
[Section titled “Step 2.1: Implement the Engine interface”](#step-21-implement-the-engine-interface)
The [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html) interface combines a bunch of sub-interfaces each of which is designed for a specific purpose. Here is a brief overview of the subinterfaces. See the API docs (Java) for a more detailed view.
```java
interface Engine {
/**
* Get the connector provided {@link ExpressionHandler}.
* @return An implementation of {@link ExpressionHandler}.
*/
ExpressionHandler getExpressionHandler();
/**
* Get the connector provided {@link JsonHandler}.
* @return An implementation of {@link JsonHandler}.
*/
JsonHandler getJsonHandler();
/**
* Get the connector provided {@link FileSystemClient}.
* @return An implementation of {@link FileSystemClient}.
*/
FileSystemClient getFileSystemClient();
/**
* Get the connector provided {@link ParquetHandler}.
* @return An implementation of {@link ParquetHandler}.
*/
ParquetHandler getParquetHandler();
}
```
To build your own `Engine` implementation, you can choose to either use the default implementations of each sub-interface or completely build every one from scratch.
```java
class MyEngine extends DefaultEngine {
FileSystemClient getFileSystemClient() {
// Build a new implementation from scratch
return new MyFileSystemClient();
}
// For all other sub-clients, use the default implementations provided by the `DefaultEngine`.
}
```
Next, we will walk through how to implement each interface.
#### Step 2.2: Implement `FileSystemClient` interface
[Section titled “Step 2.2: Implement FileSystemClient interface”](#step-22-implement-filesystemclient-interface)
The [`FileSystemClient`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/FileSystemClient.html) interface contains basic file system operations like listing directories, resolving paths into a fully qualified path and reading bytes from files. Implementation of this interface must take care of the following when interacting with storage systems such as S3, Hadoop, or ADLS:
* Credentials and permissions: The connector must populate its `FileSystemClient` with the necessary configurations and credentials for the client to retrieve the necessary data from the storage system. For example, an implementation based on Hadoop’s FileSystem abstractions can be passed S3 credentials via the Hadoop configurations.
* Decryption: If file system objects are encrypted, then the implementation must decrypt the data before returning the data.
#### Step 2.3: Implement `ParquetHandler`
[Section titled “Step 2.3: Implement ParquetHandler”](#step-23-implement-parquethandler)
As the name suggests, this interface contains everything related to reading and writing Parquet files. It has been designed such that a connector can plug in a wide variety of implementations, from a simple single-threaded reader to a very advanced multi-threaded reader with pre-fetching and advanced connector-specific expression pushdown. Let’s explore the methods to implement, and the guarantees associated with them.
##### Method `readParquetFiles(CloseableIterator fileIter, StructType physicalSchema, java.util.Optional predicate)`
[Section titled “Method readParquetFiles(CloseableIterator\ fileIter, StructType physicalSchema, java.util.Optional\ predicate)”](#method-readparquetfilescloseableiteratorfilestatus-fileiter-structtype-physicalschema-javautiloptionalpredicate-predicate)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ParquetHandler.html#readParquetFiles-io.delta.kernel.utils.CloseableIterator-io.delta.kernel.types.StructType-) takes as input `FileStatus`s which contains metadata such as file path, size etc. of the Parquet file to read. The columns to be read from the Parquet file are defined by the physical schema. To implement this method, you may have to first implement your own [`ColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnarBatch.html) and [`ColumnVector`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnVector.html) which is used to represent the in-memory data generated from the Parquet files.
When identifying the columns to read, note that there are multiple types of columns in the physical schema (represented as a [`StructType`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/types/StructType.html)).
* Data columns: Columns that are expected to be read from the Parquet file. Based on the `StructField` object defining the column, read the column in the Parquet file that matches the same name or field id. If the column has a field id (stored as `parquet.field.id` in the `StructField` metadata) then the field id should be used to match the column in the Parquet file. Otherwise, the column name should be used for matching.
* Metadata columns: These are special columns that must be populated using metadata about the Parquet file ([`StructField#isMetadataColumn`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/types/StructField.html#isMetadataColumn--) tells whether a column in `StructType` is a metadata column). To understand how to populate such a column, first match the column name against the set of standard metadata column name constants. For example,
* `StructFileld#isMetadataColumn()` returns true and the column name is `StructField.METADATA_ROW_INDEX_COLUMN_NAME`, then you have to a generate column vector populated with the actual index of each row in the Parquet file (that is, not indexed by the possible subset of rows returned after Parquet data skipping).
##### Requirements and guarantees
[Section titled “Requirements and guarantees”](#requirements-and-guarantees)
Any implementation must adhere to the following guarantees.
* The schema of the returned `ColumnarBatch`es must match the physical schema.
* If a data column is not found and the `StructField.isNullable = true`, then return a `ColumnVector` of nulls. Throw an error if it is not nullable.
* The output iterator must maintain ordering as the input iterator. That is, if `file1` is before `file2` in the input iterator, then columnar batches of `file1` must be before those of `file2` in the output iterator.
##### Method `writeParquetFiles(String directoryPath, CloseableIterator dataIter, java.util.List statsColumns)`
[Section titled “Method writeParquetFiles(String directoryPath, CloseableIterator\ dataIter, java.util.List\ statsColumns)”](#method-writeparquetfilesstring-directorypath-closeableiteratorfilteredcolumnarbatch-dataiter-javautillistcolumn-statscolumns)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ParquetHandler.html#writeParquetFiles-java.lang.String-io.delta.kernel.utils.CloseableIterator-java.util.List-) takes given data writes it into one or more Parquet files into the given directory. The data is given as an iterator of [FilteredColumnarBatches](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/FilteredColumnarBatch.html) which contains a [ColumnarBatch](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnarBatch.html) and an optional selection vector containing one entry for each row in `ColumnarBatch` indicating whether a row is selected or not selected. The `ColumnarBatch` also contains the schema of the data. This schema should be converted to Parquet schema, including any field IDs present [`FieldMetadata`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/types/FieldMetadata.html) for each column `StructField`.
There is also the parameter `statsColumns`, which is a hint to the Parquet writer on what set of columns to collect stats for each file. The statistics include `min`, `max` and `null_count` for each column in the `statsColumns` list. Statistics collection is optional, but when present it is used by Kernel to persist the stats as part of the Delta table commit. This will help read queries prune un-needed data files based on the query predicate.
For each written data file, the caller is expecting a [`DataFileStatus`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/utils/DataFileStatus.html) object. It contains the data file path, size, modification time, and optional column statistics.
#### Method `writeParquetFileAtomically(String filePath, CloseableIterator data)`
[Section titled “Method writeParquetFileAtomically(String filePath, CloseableIterator\ data)”](#method-writeparquetfileatomicallystring-filepath-closeableiteratorfilteredcolumnarbatch-data)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ParquetHandler.html#writeParquetFiles-java.lang.String-io.delta.kernel.utils.CloseableIterator-java.util.List-) writes the given `data` into Parquet file at location `filePath`. The write is an atomic write i.e., either a Parquet file is created with all given content or no Parquet file is created at all. This should not create a file with partial content in it.
The default implementation makes use of [`LogStore`](https://github.com/delta-io/delta/blob/master/storage/src/main/java/io/delta/storage/LogStore.java) implementations from the [`delta-storage`](https://github.com/delta-io/delta/tree/master/storage) module to accomplish the atomicity. A connector that wants to implement their own version of `ParquetHandler` can take a look at the default implementation for details.
##### Performance suggestions
[Section titled “Performance suggestions”](#performance-suggestions)
* The representation of data as `ColumnVector`s and `ColumnarBatch`es can have a significant impact on the query performance and it’s best to read the Parquet file data directly into vectors and batches of the engine-native format to avoid potentially costly in-memory data format conversion. Create a Kernel `ColumnVector` and `ColumnarBatch` wrappers around the engine-native format equivalent classes.
#### Step 2.4: Implement `ExpressionHandler` interface
[Section titled “Step 2.4: Implement ExpressionHandler interface”](#step-24-implement-expressionhandler-interface)
The [`ExpressionHandler`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ExpressionHandler.html) interface has all the methods needed for handling expressions that may be applied on columnar data.
##### Method `getEvaluator(StructType batchSchema, Expression expresion, DataType outputType)`
[Section titled “Method getEvaluator(StructType batchSchema, Expression expresion, DataType outputType)”](#method-getevaluatorstructtype-batchschema-expression-expresion-datatype-outputtype)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ExpressionHandler.html#getEvaluator-io.delta.kernel.types.StructType-io.delta.kernel.expressions.Expression-io.delta.kernel.types.DataType-) generates an object of type [`ExpressionEvaluator`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/expressions/ExpressionEvaluator.html) that can evaluate the `expression` on a batch of row data to produce a result of a single column vector. To generate this function, the `getEvaluator()` method takes as input the expression and the schema of the `ColumnarBatch`es of data on which the expressions will be applied. The same object can be used to evaluate multiple columnar batches of input with the same schema and expression the evaluator is created for.
##### Method `getPredicateEvaluator(StructType inputSchema, Predicate predicate)`
[Section titled “Method getPredicateEvaluator(StructType inputSchema, Predicate predicate)”](#method-getpredicateevaluatorstructtype-inputschema-predicate-predicate)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ExpressionHandler.html#createSelectionVector-boolean:A-int-int-) is for creating an expression evaluator for `Predicate` type expressions. The `Predicate` type expressions return a boolean value as output.
The returned object is of type [`PredicateEvaluator`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/expressions/PredicateEvaluator.html). This is a special interface for evaluating Predicate on input batch returns a selection vector containing one value for each row in input batch indicating whether the row has passed the predicate or not. Optionally it takes an existing selection vector along with the input batch for evaluation. The result selection vector is combined with the given existing selection vector and a new selection vector is returned. This mechanism allows running an input batch through several predicate evaluations without rewriting the input batch to remove rows that do not pass the predicate after each predicate evaluation. The new selection should be the same or more selective as the existing selection vector. For example, if a row is marked as unselected in the existing selection vector, then it should remain unselected in the returned selection vector even when the given predicate returns true for the row.
##### Method `createSelectionVector(boolean[] values, int from, int to)`
[Section titled “Method createSelectionVector(boolean\[\] values, int from, int to)”](#method-createselectionvectorboolean-values-int-from-int-to)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/ExpressionHandler.html#createSelectionVector-boolean:A-int-int-) allows creating `ColumnVector` for boolean type values given as input. This allows the connector to maintain all `ColumnVector`s created in the desired memory format.
##### Requirements and guarantees
[Section titled “Requirements and guarantees”](#requirements-and-guarantees-1)
Any implementation must adhere to the following guarantees.
* Implementation must handle all possible variations of expressions. If the implementation encounters an expression type that it does not know how to handle, then it must throw a specific language-dependent exception.
* Java: [NotSupportedException](https://docs.oracle.com/javaee/7/api/javax/resource/NotSupportedException.html)
* The `ColumnarBatch`es on which the generated `ExpressionEvaluator` is going to be used are guaranteed to have the schema provided during generation. Hence, it is safe to bind the expression evaluation logic to column ordinals instead of column names, thus making the actual evaluation faster.
#### Step 2.5: Implement `JsonHandler`
[Section titled “Step 2.5: Implement JsonHandler”](#step-25-implement-jsonhandler)
[This](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/JsonHandler.html) engine interface allows the connector to use plug-in their own JSON handling code and expose it to the Delta Kernel.
##### Method `readJsonFiles(CloseableIterator fileIter, StructType physicalSchema, java.util.Optional predicate)`
[Section titled “Method readJsonFiles(CloseableIterator\ fileIter, StructType physicalSchema, java.util.Optional\ predicate)”](#method-readjsonfilescloseableiteratorfilestatus-fileiter-structtype-physicalschema-javautiloptionalpredicate-predicate)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/JsonHandler.html#readJsonFiles-io.delta.kernel.utils.CloseableIterator-io.delta.kernel.types.StructType-) takes as input `FileStatus`s of the JSON files and returns the data in a series of columnar batches. The columns to be read from the JSON file are defined by the physical schema, and the return batches must match that schema. To implement this method, you may have to first implement your own [`ColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnarBatch.html)and [`ColumnVector`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnVector.html) which is used to represent the in-memory data generated from the JSON files.
When identifying the columns to read, note that there are multiple types of columns in the physical schema (represented as a [`StructType`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/types/StructType.html)).
##### Method `parseJson(ColumnVector jsonStringVector, StructType outputSchema, java.util.Optional selectionVector)`
[Section titled “Method parseJson(ColumnVector jsonStringVector, StructType outputSchema, java.util.Optional\ selectionVector)”](#method-parsejsoncolumnvector-jsonstringvector-structtype-outputschema-javautiloptionalcolumnvector-selectionvector)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/JsonHandler.html#parseJson-io.delta.kernel.data.ColumnVector-io.delta.kernel.types.StructType-) allows parsing a `ColumnVector` of string values which are in JSON format into the output format specified by the `outputSchema`. If a given column in `outputSchema` is not found, then a null value is returned. It optionally takes a selection vector which indicates what entries in the input `ColumnVector` of strings to parse. If an entry is not selected then a `null` value is returned as parsed output for that particular entry in the output.
##### Method `deserializeStructType(String structTypeJson)`
[Section titled “Method deserializeStructType(String structTypeJson)”](#method-deserializestructtypestring-structtypejson)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/JsonHandler.html#deserializeStructType-java.lang.String-) allows parsing JSON encoded (according to [Delta schema serialization rules](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#schema-serialization-format)) `StructType` schema into a `StructType`. Most implementations of `JsonHandler` do not need to implement this method and instead use the one in the [default `JsonHandler`](https://delta-io.github.io/delta/snapshot/kernel-defaults/java/io/delta/kernel/defaults/engine/DefaultJsonHandler.html) implementation.
#### Method `writeJsonFileAtomically(String filePath, CloseableIterator data, boolean overwrite)`
[Section titled “Method writeJsonFileAtomically(String filePath, CloseableIterator\ data, boolean overwrite)”](#method-writejsonfileatomicallystring-filepath-closeableiteratorrow-data-boolean-overwrite)
This [method](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/JsonHandler.html#writeJsonFileAtomically-java.lang.String-io.delta.kernel.utils.CloseableIterator-boolean-) writes the given `data` into a JSON file at location `filePath`. The write is an atomic write i.e., either a JSON file is created with all given content or no Parquet file is created at all. This should not create a file with partial content in it.
The default implementation makes use of [`LogStore`](https://github.com/delta-io/delta/blob/master/storage/src/main/java/io/delta/storage/LogStore.java) implementations from the [`delta-storage`](https://github.com/delta-io/delta/tree/master/storage) module to accomplish the atomicity. A connector that wants to implement their own version of `JsonHandler` can take a look at the default implementation for details.
The implementation is expected to handle the serialization rules (converting the `Row` object to JSON string) as described in the [API Javadoc](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/JsonHandler.html#writeJsonFileAtomically-java.lang.String-io.delta.kernel.utils.CloseableIterator-boolean-).
#### Step 2.6: Implement `ColumnarBatch` and `ColumnVector`
[Section titled “Step 2.6: Implement ColumnarBatch and ColumnVector”](#step-26-implement-columnarbatch-and-columnvector)
[`ColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnarBatch.html) and [`ColumnVector`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/ColumnVector.html) are two interfaces to represent the data read into memory from files. This representation can have a significant impact on query performance. Each engine likely has a native representation of in-memory data with which it applies data transformation operations. For example, in Apache Spark™, the row data is internally represented as `UnsafeRow` for efficient processing. So it’s best to read the Parquet file data directly into vectors and batches of the native format to avoid potentially costly in-memory data format conversions. So the recommended approach is to build wrapper classes that extend the two interfaces but internally use engine-native classes to store the data. When the connector has to forward the columnar batches received from the kernel to the engine, it has to be smart enough to skip converting vectors and batches that are already in the engine-native format.
### Step 3: Build read support in your connector
[Section titled “Step 3: Build read support in your connector”](#step-3-build-read-support-in-your-connector)
In this section, we are going to walk through the likely sequence of Kernel API calls your connector will have to make to read a table. The exact timing of making these calls in your connector in the context of connector-engine interactions depends entirely on the engine-connector APIs and is therefore beyond the scope of this guide. However, we will try to provide broad guidelines that are likely (but not guaranteed) to apply to your connector-engine setup. For this purpose, we are going to assume that the engine goes through the following phases when processing a read/scan query - logical plan analysis, physical plan generation, and physical plan execution. Based on these broad characterizations, a typical control and data flow for reading a Delta table is going to be as follows:
.. list-table:: :header-rows: 1 :widths: 30 70 Step Typical query phase when this step occurs Resolve the table snapshot to query Logical plan analysis phase when the plan’s schema and other details need to be resolved and validated Resolve files to scan based on query parameters Physical plan generation, when the final parameters of the scan are available. For example: Schema of data to read after pruning away unused columns Query filters to apply after filter rearrangement Distribute the file information to workers Physical plan execution, only if it is a distributed engine. Read the columnar data using the file information Physical plan execution, when the data is being processed by the engine
Let’s understand the details of each step.
#### Step 3.1: Resolve the table snapshot to query
[Section titled “Step 3.1: Resolve the table snapshot to query”](#step-31-resolve-the-table-snapshot-to-query)
The first step is to resolve the consistent snapshot and the schema associated with it. This is often required by the connector/ engine to resolve and validate the logical plan of the scan query (if the concept of logical plan exists in your engine). To achieve this, the connector has to do the following.
* Resolve the table path from the query: If the path is directly available, then this is easy. Otherwise, if it is a query based on a catalog table (for example, a Delta table defined in Hive Metastore), then the connector has to resolve the table path from the catalog.
* Initialize the `Engine` object: Create a new instance of the `Engine` that you have chosen in \[Step 2]\(#build-your-own Engine).
* Initialize the Kernel objects and get the schema: Assuming the query is on the latest available version/snapshot of the table, you can get the table schema as follows:
```java
import io.delta.kernel.*;
import io.delta.kernel.defaults.engine.*;
Engine myEngine = new MyEngine();
Table myTable = Table.forPath(myTablePath);
Snapshot mySnapshot = myTable.getLatestSnapshot(myEngine);
StructType mySchema = mySnapshot.getSchema(myEngine);
```
If you want to query a specific version of the table (that is, not the schema), then you can get the required snapshot as `myTable.getSnapshot(version)`.
#### Step 3.2: Resolve files to scan
[Section titled “Step 3.2: Resolve files to scan”](#step-32-resolve-files-to-scan)
Next, we need to build a Scan object using more information from the query. Here we are going to assume that the connector/engine has been able to extract the following details from the query (say, after optimizing the logical plan):
* Read schema: The columns in the table that the query needs to read. This may be the full set of columns or a subset of columns.
* Query filters: The filters on partitions or data columns that can be used skip reading table data.
To provide this information to Kernel, you have to do the following:
* Convert the engine-specific schema and filter expressions to Kernel schema and expressions: For schema, you have to create a `StructType` object. For the filters, you have to create an `Expression` object using all the available subclasses of `Expression`.
* Build the scan with the converted information: Build the scan as follows:
```java
import io.delta.kernel.expressions.*;
import io.delta.kernel.types.*;
StructType readSchema = ... ; // convert engine schema
Predicate filterExpr = ... ; // convert engine filter expression
Scan myScan = mySnapshot.getScanBuilder().withFilter(filterExpr).withReadSchema(readSchema).build();
```
* Resolve the information required to file reads: The generated Scan object has two sets of information.
* Scan files: `myScan.getScanFiles()` returns an iterator of `ColumnarBatch`es. Each batch in the iterator contains rows and each row has information about a single file that has been selected based on the query filter.
* Scan state: `myScan.getScanState()` returns a `Row` that contains all the information that is common across all the files that need to be read.
```java
Row myScanStateRow = myScan.getScanState();
CloseableIterator myScanFilesAsBatches = myScan.getScanFiles();
while (myScanFilesAsBatches.hasNext()) {
FilteredColumnarBatch scanFileBatch = myScanFilesAsBatches.next();
CloseableIterator myScanFilesAsRows = scanFileBatch.getRows();
}
```
As we will soon see, reading the columnar data from a selected file will need to use both, the scan state row, and a scan file row with the file information.
##### Requirements and guarantees
[Section titled “Requirements and guarantees”](#requirements-and-guarantees-2)
Here are the details you need to ensure when defining this scan.
* The provided `readSchema` must be the exact schema of the data that the engine will expect when executing the query. Any mismatch in the schema defined during this query planning and the query execution will result in runtime failures. Hence you must build the scan with the readSchema only after the engine has finalized the logical plan after any optimizations like column pruning.
* When applicable (for example, with Java Kernel APIs), you have to make sure to call the close() method as you consume the `ColumnarBatch`es of scan files (that is, either serialize the rows or use them to read the table data).
#### Step 3.3: Distribute the file information to the workers
[Section titled “Step 3.3: Distribute the file information to the workers”](#step-33-distribute-the-file-information-to-the-workers)
If you are building a connector for a distributed engine like Spark/Presto/Trino/Flink, then your connector has to send all the scan metadata from the query planning machine (henceforth called the driver) to task execution machines (henceforth called the workers). You will have to serialize and deserialize the scan state and scan file rows. It is the connector job to implement serialization and deserialization utilities for a [`Row`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/Row.html). If the connector wants to split reading one scan file into multiple tasks, it can add additional connector specific split context to the task. At the task, the connector can use its own Parquet reader to read the specific part of the file indicated by the split info.
##### Custom `Row` Serializer/Deserializer
[Section titled “Custom Row Serializer/Deserializer”](#custom-row-serializerdeserializer)
Here are steps on how to build your own serializer/deserializer such that it will work with any [`Row`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/Row.html) of any schema.
* Serializing
* First serialize the row schema, that is, `StructType` object.
* Then, use the schema to identify types of each column/ordinal in the `Row` and use that to serialize all the values one by one.
* Deserializing
* Define your own class that extends the Row interface. It must be able to handle complex types like arrays, nested structs and maps.
* First deserialize the schema.
* Then, use the schema to deserialize the values and put them in an instance of your custom Row class.
```java
import io.delta.kernel.utils.*;
// In the driver where query planning is being done
Byte[] scanStateRowBytes = RowUtils.serialize(scanStateRow);
Byte[] scanFileRowBytes = RowUtils.serialize(scanFileRow);
// Optionally the connector adds a split info to the task (scan file, scan state) to
// split reading of a Parquet file into multiple tasks. The task gets split info
// along with the scan file row and scan state row.
Split split = ...; // connector specific class, not related to Kernel
// Send these over to the worker
// In the worker when data will be read, after rowBytes have been sent over
Row scanStateRow = RowUtils.deserialize(scanStateRowBytes);
Row scanFileRow = RowUtils.deserialize(scanFileRowBytes);
Split split = ... deserialize split info ...;
```
#### Step 3.4: Read the columnar data
[Section titled “Step 3.4: Read the columnar data”](#step-34-read-the-columnar-data)
Finally, we are ready to read the columnar data. You will have to do the following:
* Read the physical data from Parquet file as indicated by the scan file row, scan state, and optionally the split info
* Convert the physical data into logical data of the table using the Kernel’s APIs.
```java
Row scanStateRow = ... ;
Row scanFileRow = ... ;
Split split = ...;
// Additional option predicate such as dynamic filters the connector wants to
// pass to the reader when reading files.
Predicate optPredicate = ...;
// Get the physical read schema of columns to read from the Parquet data files
StructType physicalReadSchema =
ScanStateRow.getPhysicalDataReadSchema(engine, scanStateRow);
// From the scan file row, extract the file path, size and modification metadata
// needed to read the file.
FileStatus fileStatus = InternalScanFileUtils.getAddFileStatus(scanFileRow);
// Open the scan file which is a Parquet file using connector's own
// Parquet reader which supports reading specific parts (split) of the file.
// If the connector doesn't have its own Parquet reader, it can use the
// default Parquet reader provider which at the moment doesn't support reading
// a specific part of the file, but reads the entire file from the beginning.
CloseableIterator physicalDataIter =
connectParquetReader.readParquetFile(
fileStatus
physicalReadSchema,
split, // what part of the Parquet file to read data from
optPredicate /* additional predicate the connector can apply to filter data from the reader */
);
// Now the physical data read from the Parquet data file is converted to logical data
// the table represents.
// Logical data may include the addition of partition columns and/or
// subset of rows deleted
CloseableIterator transformedData =
Scan.transformPhysicalData(
engine,
scanState,
scanFileRow,
physicalDataIter));
```
* Resolve the data in the batches: Each [`FilteredColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/FilteredColumnarBatch.html) has two components:
* Columnar batch (returned by `FilteredColumnarBatch.getData()`): This is the data read from the files having the schema matching the readSchema provided when the Scan object was built in the earlier step.
* Optional selection vector (returned by `FilteredColumnarBatch.getSelectionVector()`): Optionally, a boolean vector that will define which rows in the batch are valid and should be consumed by the engine.
If the selection vector is present, then you will have to apply it to the batch to resolve the final consumable data.
* Convert to engine-specific data format: Each connector/engine has its own native row / columnar batch formats and interfaces. To return the read data batches to the engine, you have to convert them to fit those engine-specific formats and/or interfaces. Here are a few tips that you can follow to make this efficient.
* Matching the engine-specific format: Some engines may expect the data in an in-memory format that may be different from the data produced by `getData()`. So you will have to do the data conversion for each column vector in the batch as needed.
* Matching the engine-specific interfaces: You may have to implement wrapper classes that extend the engine-specific interfaces and appropriately encapsulate the row data.
For best performance, you can implement your own Parquet reader and other `Engine` implementations to make sure that every `ColumnVector` generated is already in the engine-native format thus eliminating any need to convert.
Now you should be able to read the Delta table correctly.
### Step 4: Build append support in your connector
[Section titled “Step 4: Build append support in your connector”](#step-4-build-append-support-in-your-connector)
In this section, we are going to walk through the likely sequence of Kernel API calls your connector will have to make to append data to a table. The exact timing of making these calls in your connector in the context of connector-engine interactions depends entirely on the engine-connector APIs and is, therefore, beyond the scope of this guide. However, we will try to provide broad guidelines that are likely (but not guaranteed) to apply to your connector-engine setup. For this purpose, we are going to assume that the engine goes through the following phases when processing a write query - logical plan analysis, physical plan generation, and physical plan execution. Based on these broad characterizations, a typical control and data flow for reading a Delta table is going to be as follows:
* Typical steps and query phases
| Step | Typical query phase when this step occurs |
| -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Determine the schema of the data that needs to be written to the table. Schema is derived from the existing table or from the parent operation of the `write` operator in the query plan when the table doesn’t exist yet. | Logical plan analysis phase when the plan’s schema (`write` operator schema matches the table schema, etc.) and other details need to be resolved and validated. |
| Determine the physical partitioning of the data based on the table schema and partition columns either from the existing table or from the query plan (for new tables) | Physical plan generation, where the number of writer tasks, data schema and partitioning is determined |
| Distribute the writer tasks definitions (which include the transaction state) to workers. | Physical plan execution, only if it is a distributed engine. |
| Tasks write the data to data files and send the data file info to the driver. | Physical plan execution, when the data is actually written to the table location |
| Finalize the query. Here, all the info of the data files written by the tasks is aggregated and committed to the transaction created at the beginning of the physical execution. | Finalize the query. This happens on the driver where the query has started. |
Let’s understand the details of each step.
#### Step 4.1: Determine the schema of the data that needs to be written to the table
[Section titled “Step 4.1: Determine the schema of the data that needs to be written to the table”](#step-41-determine-the-schema-of-the-data-that-needs-to-be-written-to-the-table)
The first step is to resolve the output data schema. This is often required by the connector/ engine to resolve and validate the logical plan of the query (if the concept of logical plan exists in your engine). To achieve this, the connector has to do the following. At a high level query plan is a tree of operators where the leaf-level operators generate or read data from storage/tables and feed it upwards towards the parent operator nodes. This data transfer happens until it reaches the root operator node where the query is finalized (either the results are sent to the client or data is written to another table).
* Create the `Table` object
* From the `Table` object try to get the schema.
* If the table is not found
* the query includes creating the table (e.g., `CREATE TABLE AS` SQL query);
* the schema is derived from the operator above the `write` that feeds the data to the `write` operator.
* the query doesn’t include creating new table, an exception is thrown saying the table is not found
* If the table already exists
* get the schema from the table and check if it matches the schema of the `write` operator. If not throw an exception.
* Create a `TransactionBuilder` - this basically begins the steps of transaction construction.
```java
import io.delta.kernel.*;
import io.delta.kernel.defaults.engine.*;
Engine myEngine = new MyEngine();
Table myTable = Table.forPath(myTablePath);
StructType writeOperatorSchema = // ... derived from the query operator tree ...
StructType dataSchema;
boolean isNewTable = false;
try {
Snapshot mySnapshot = myTable.getLatestSnapshot(myEngine);
dataSchema = mySnapshot.getSchema(myEngine);
// .. check dataSchema and writeOperatorSchema match ...
} catch(TableNotFoundException e) {
isNewTable = true;
dataSchema = writeOperatorSchema;
}
TransactionBuilder txnBuilder =
myTable.createTransactionBuilder(
myEngine,
"Examples", /* engineInfo - connector can add its own identifier which is noted in the Delta Log */
Operation /* What is the operation we are trying to perform? This is noted in the Delta Log */
);
if (isNewTable) {
// For a new table set the table schema in the transaction builder
txnBuilder = txnBuilder.withSchema(engine, dataSchema)
}
```
#### Step 4.2: Determine the physical partitioning of the data based on the table schema and partition columns
[Section titled “Step 4.2: Determine the physical partitioning of the data based on the table schema and partition columns”](#step-42-determine-the-physical-partitioning-of-the-data-based-on-the-table-schema-and-partition-columns)
Partition columns are found either from the query (for new tables, the query defines the partition columns) or from the existing table.
```java
TransactionBuilder txnBuilder = ... from the last step ...
Transaction txn;
List partitionColumns = ...
if (newTable) {
partitionColumns = ... derive from the query parameters (ex. PARTITION BY clause in SQL) ...
txnBuilder = txnBuilder.withPartitionColumns(engine, partitionColumns);
txn = txnBuilder.build(engine);
} else {
txn = txnBuilder.build(engine);
partitionColumns = txn.getPartitionColumns(engine);
}
```
At the end of this step, we have the `Transaction` and schema of the data to generate and its partitioning.
#### Step 4.3: Distribute the writer tasks definitions (which include the transaction state) to workers
[Section titled “Step 4.3: Distribute the writer tasks definitions (which include the transaction state) to workers”](#step-43-distribute-the-writer-tasks-definitions-which-include-the-transaction-state-to-workers)
If you are building a connector for a distributed engine like Spark/Presto/Trino/Flink, then your connector has to send all the writer metadata from the query planning machine (henceforth called the driver) to task execution machines (henceforth called the workers). You will have to serialize and deserialize the transaction state. It is the connector job to implement serialization and deserialization utilities for a [`Row`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/Row.html). More details on a custom `Row` SerDe are found [here](#custom-row-serializerdeserializer).
```java
Row txnState = txn.getState(engine);
String jsonTxnState = serializeToJson(txnState);
```
#### Step 4.4: Tasks write the data to data files and send the data file info to the driver.
[Section titled “Step 4.4: Tasks write the data to data files and send the data file info to the driver.”](#step-44-tasks-write-the-data-to-data-files-and-send-the-data-file-info-to-the-driver)
In this step (which is executed on the worker nodes inside each task):
* Deserialize the transaction state
* Writer operator within the task gets the data from its parent operator.
* The data is converted into a `FilteredColumnarBatch`. Each [`FilteredColumnarBatch`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/data/FilteredColumnarBatch.html) has two components:
* Columnar batch (returned by `FilteredColumnarBatch.getData()`): This is the data read from the files having the schema matching the readSchema provided when the Scan object was built in the earlier step.
* Optional selection vector (returned by `FilteredColumnarBatch.getSelectionVector()`): Optionally, a boolean vector that will define which rows in the batch are valid and should be consumed by the engine.
* The connector can create `FilteredColumnBatch` wrapper around data in its own in-memory format.
* Check if the data is partitioned or not. If not partitioned, partition the data by partition values.
* For each partition generate the map of the partition column to the partition value
* Use Kernel to convert the partitioned data into physical data that should go into the data files
* Write the physical data into one or more data files.
* Convert data file statues into a Delta log actions
* Serialize the Delta log action `Row` objects and send them to the driver node
```plaintext
Row txnState = ... deserialize from JSON string sent by the driver ...
CloseableIterator data = ... generate data ...
// If the table is un-partitioned then this is an empty map
Map partitionValues = ... prepare the partition values ...
// First transform the logical data to physical data that needs to be written
// to the Parquet files
CloseableIterator physicalData =
Transaction.transformLogicalData(engine, txnState, data, partitionValues);
// Get the write context
DataWriteContext writeContext = Transaction.getWriteContext(engine, txnState, partitionValues);
// Now write the physical data to Parquet files
CloseableIterator dataFiles =
engine.getParquetHandler()
.writeParquetFiles(
writeContext.getTargetDirectory(),
physicalData,
writeContext.getStatisticsColumns());
// Now convert the data file status to data actions that needs to be written to the Delta table log
CloseableIterator partitionDataActions =
Transaction.generateAppendActions(
engine,
txnState,
dataFiles,
writeContext);
.... serialize `partitionDataActions` and send them to driver node
```
#### Step 4.5: Finalize the query.
[Section titled “Step 4.5: Finalize the query.”](#step-45-finalize-the-query)
At the driver node, the delta log actions from all the tasks are received and committed to the transaction. The tasks send the Delta log actions as a serialized JSON and deserialize them back to `Row` objects.
```plaintext
// Create a iterable out of the data actions. If the contents are too big to fit in memory,
// the connector may choose to write the data actions to a temporary file and return an
// iterator that reads from the file.
CloseableIterable dataActionsIterable = CloseableIterable.inMemoryIterable(
toCloseableIterator(dataActions.iterator()));
// Commit the transaction.
TransactionCommitResult commitResult = txn.commit(engine, dataActionsIterable);
// Optional step
if (commitResult.isReadyForCheckpoint()) {
// Checkpoint the table
Table.forPath(engine, tablePath).checkpoint(engine, commitResult.getVersion());
}
```
Thats it. Now you should be able to append data to Delta tables using the Kernel APIs.
## Migration guide
[Section titled “Migration guide”](#migration-guide)
Kernel APIs are still evolving and new features are being added. Kernel authors try to make the API changes backward compatible as much as they can with each new release, but sometimes it is hard to maintain the backward compatibility for a project that is evolving rapidly.
This section provides guidance on how to migrate your connector to the latest version of Delta Kernel. With each new release the [examples](https://github.com/delta-io/delta/tree/master/kernel/examples) are kept up-to-date with the latest API changes. You can refer to the examples to understand how to use the new APIs.
### Migration from Delta Lake version 3.1.0 to 3.2.0
[Section titled “Migration from Delta Lake version 3.1.0 to 3.2.0”](#migration-from-delta-lake-version-310-to-320)
Following are API changes in Delta Kernel 3.2.0 that may require changes in your connector.
#### Rename `TableClient` to [`Engine`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/engine/Engine.html)
[Section titled “Rename TableClient to Engine”](#rename-tableclient-to-engine)
The `TableClient` interface has been renamed to `Engine`. This is the most significant API change in this release. The `TableClient` interface name is not exactly representing the functionality it provides. At a high level it provides capabilities such as reading Parquet files, JSON files, evaluating expressions on data and file system functionality. These are basically the heavy lift operations that Kernel depends on as a separate interface to allow the connectors to substitute their own custom implementation of the same functionality (e.g. custom Parquet reader). Essentially, these functionalities are the core of the `engine` functionalities. By renaming to `Engine`, we are representing the interface functionality with a proper name that is easy to understand.
The `DefaultTableClient` has been renamed to [`DefaultEngine`](https://delta-io.github.io/delta/snapshot/kernel-defaults/java/io/delta/kernel/defaults/engine/DefaultEngine.html).
#### [`Table.forPath(Engine engine, String tablePath)`](https://delta-io.github.io/delta/snapshot/kernel-api/java/io/delta/kernel/Table.html#forPath-io.delta.kernel.engine.Engine-java.lang.String-) behavior change
[Section titled “Table.forPath(Engine engine, String tablePath) behavior change”](#tableforpathengine-engine-string-tablepath-behavior-change)
Earlier when a non-existent table path is passed, the API used to throw `TableNotFoundException`. Now it doesn’t throw the exception. Instead, it returns a `Table` object. When trying to get a `Snapshot` from the table object it throws the `TableNotFoundException`.
#### [`FileSystemClient.resolvePath`](https://delta-io.github.io/delta/snapshot/kernel-defaults/java/io/delta/kernel/defaults/engine/DefaultFileSystemClient.html#resolvePath-java.lang.String-) behavior change
[Section titled “FileSystemClient.resolvePath behavior change”](#filesystemclientresolvepath-behavior-change)
Earlier when a non-existent path is passed, the API used to throw `FileNotFoundException`. Now it doesn’t throw the exception. It still resolves the given path into a fully qualified path.
# Delta Kernel Rust
> Learn how to build connectors to read and write Delta tables using Delta Kernel Rust.
Work In Progress
# Other connectors
## Apache Druid
[Section titled “Apache Druid”](#apache-druid)
This [connector](https://druid.apache.org/docs/latest/development/extensions-contrib/delta-lake/) allows [Apache Druid](https://druid.apache.org/) to read from Delta Lake.
## Apache Pulsar
[Section titled “Apache Pulsar”](#apache-pulsar)
This [connector](https://github.com/streamnative/pulsar-io-lakehouse/blob/master/docs/delta-lake-demo.md) allows [Apache Pulsar](https://pulsar.apache.org/) to read from and write to Delta Lake.
## ClickHouse
[Section titled “ClickHouse”](#clickhouse)
[ClickHouse](https://clickhouse.com/) is a column-oriented database that allows users to run SQL queries on Delta Lake tables. This [connector](https://clickhouse.com/docs/engines/table-engines/integrations/deltalake) provides a read-only integration with existing Delta Lake tables in Amazon S3.
## Dagster
[Section titled “Dagster”](#dagster)
Use the [Delta Lake IO Manager](https://delta-io.github.io/delta-rs/integrations/delta-lake-dagster/) to read from and write to Delta Lake tables in your [Dagster](https://dagster.io/) orchestration pipelines.
## FINOS Legend
[Section titled “FINOS Legend”](#finos-legend)
An [extension](https://github.com/finos/legend-community-delta/blob/main/README.md) to the [FINOS](https://landscape.finos.org/) Legend framework for Apache Spark™ / Delta Lake based environment, combining best of open data standards with open source technologies. This connector allows Trino to read from and write to Delta Lake.
## Hopsworks
[Section titled “Hopsworks”](#hopsworks)
This [connectors](https://docs.hopsworks.ai/latest/user_guides/fs/feature_group/create/#batch-write-api) allows [Hopsworks Feature Store](https://www.hopsworks.ai/dictionary/feature-store) store, manage, and serve feature data in Delta Lake.
## Apache Hive
[Section titled “Apache Hive”](#apache-hive)
This integration enables reading Delta tables from Apache Hive. For details on installing the integration, see the [Delta Lake repository](https://github.com/delta-io/delta/tree/master/connectors/hive).
## Kafka Delta Ingest
[Section titled “Kafka Delta Ingest”](#kafka-delta-ingest)
This [project](https://github.com/delta-io/kafka-delta-ingest) builds a highly efficient daemon for streaming data through Apache Kafka into Delta Lake.
## SQL Delta Import
[Section titled “SQL Delta Import”](#sql-delta-import)
This [utility](https://github.com/delta-io/delta/blob/master/connectors/sql-delta-import/readme.md) is for importing data from a JDBC source into a Delta Lake table.
## StarRocks
[Section titled “StarRocks”](#starrocks)
[StarRocks](https://www.starrocks.io/), a Linux Foundation project, is a next-generation sub-second MPP OLAP database for full analytics scenarios, including multi-dimensional analytics, real-time analytics, and ad-hoc queries. StarRocks has the [ability to read](https://docs.starrocks.io/docs/introduction/StarRocks_intro/) from Delta Lake.
# Presto connector
> Learn how to set up an integration to enable you to read Delta tables from Presto.
Since Presto [version 0.269](https://prestodb.io/docs/0.269/release/release-0.269.html#delta-lake-connector-changes), Presto natively supports reading Delta Lake tables. For details on using the native Delta Lake connector, see [Delta Lake Connector - Presto](https://prestodb.io/docs/current/connector/deltalake.html). For Presto versions lower than [0.269](https://prestodb.io/docs/0.269/release/release-0.269.html#delta-lake-connector-changes), you can use the manifest-based approach detailed in [Presto, Trino, and Athena to Delta Lake integration using manifests](/presto-integration/).
# Delta Lake resources
> Learn about resources for understanding Delta Lake.
## Blog posts and talks
[Section titled “Blog posts and talks”](#blog-posts-and-talks)
[Delta Lake blog posts](https://delta.io/blog)
[Delta Lake tutorials](https://delta.io/learn/tutorials/)
[Delta Lake videos](https://delta.io/learn/videos/)
## VLDB 2020 paper
[Section titled “VLDB 2020 paper”](#vldb-2020-paper)
[Delta Lake: High-Performance ACID Table Storage over Cloud Object Stores](https://databricks.com/wp-content/uploads/2020/08/p975-armbrust.pdf)
## Examples
[Section titled “Examples”](#examples)
The Delta Lake GitHub repository has [Scala and Python examples](https://github.com/delta-io/delta/tree/master/examples/).
## Delta Lake transaction log specification
[Section titled “Delta Lake transaction log specification”](#delta-lake-transaction-log-specification)
The Delta Lake transaction log has a well-defined open protocol that can be used by any system to read the log. See [Delta Transaction Log Protocol](https://github.com/delta-io/delta/blob/master/PROTOCOL.md).
# Use row tracking for Delta tables
> Learn how Delta Lake row tracking allows tracking how rows change across table versions.
Row tracking allows Delta Lake to track row-level lineage in a Delta Lake table. When enabled on a Delta Lake table, row tracking adds two new metadata fields to the table:
* **Row IDs** provide rows with an identifier that is unique within the table. A row keeps the same ID whenever it is modified using a `MERGE` or `UPDATE` statement.
* **Row commit versions** record the last version of the table in which the row was modified. A row is assigned a new version whenever it is modified using a `MERGE` or `UPDATE` statement.
Note
This feature is available in Delta Lake 3.2.0 and above. Enabling this feature on existing non-empty tables is available in Delta Lake 3.3.0 and above.
## Enable row tracking
[Section titled “Enable row tracking”](#enable-row-tracking)
Caution
Tables created with row tracking enabled have the row tracking Delta Lake table feature enabled at creation and use Delta Lake writer version 7. Table protocol versions cannot be downgraded, and tables with row tracking enabled are not writeable by Delta Lake clients that do not support all enabled Delta Lake writer protocol table features. See [How does Delta Lake manage feature compatibility?](/versioning/).
You must explicitly enable row tracking using one of the following methods:
* **New table**: Set the table property `delta.enableRowTracking = true` in the `CREATE TABLE` command.
- SQL
```sql
-- Create an empty table
CREATE TABLE student (id INT, name STRING, age INT)
TBLPROPERTIES ('delta.enableRowTracking' = 'true');
-- Using a CTAS statement
CREATE TABLE course_new
TBLPROPERTIES ('delta.enableRowTracking' = 'true')
AS SELECT * FROM course_old;
-- Using a LIKE statement to copy configuration
CREATE TABLE graduate LIKE student;
-- Using a CLONE statement to copy configuration
CREATE TABLE graduate CLONE student;
```
* **Existing table**: Available from Delta 3.3 and above, set the table property `'delta.enableRowTracking' = 'true'` in the `ALTER TABLE` command.
- SQL
```sql
ALTER TABLE grade SET TBLPROPERTIES ('delta.enableRowTracking' = 'true');
```
* **All new tables**: Set the configuration `spark.databricks.delta.properties.defaults.enableRowTracking = true` for the current session in the `SET` command.
- SQL
```sql
SET spark.databricks.delta.properties.defaults.enableRowTracking = true;
```
- Python
```python
spark.conf.set("spark.databricks.delta.properties.defaults.enableRowTracking", True)
```
- Scala
```scala
spark.conf.set("spark.databricks.delta.properties.defaults.enableRowTracking", true)
```
Caution
Because cloning a Delta Lake table creates a separate history, the row ids and row commit versions on cloned tables do not match that of the original table.
Caution
Enabling row tracking on existing table will automatically assign row ids and row commit versions to all existing rows in the table. This process may cause multiple new versions of the table to be created and may take a long time.
### Row tracking storage
[Section titled “Row tracking storage”](#row-tracking-storage)
Enabling row tracking may increase the size of the table. Delta Lake stores row tracking metadata fields in hidden metadata columns in the data files. Some operations, such as insert-only operations do not use these hidden columns and instead track the row ids and row commit versions using metadata in the Delta Lake log. Data reorganization operations such as `OPTIMIZE` and `REORG` cause the row ids and row commit versions to be tracked using the hidden metadata column, even when they were stored using metadata.
## Read row tracking metadata fields
[Section titled “Read row tracking metadata fields”](#read-row-tracking-metadata-fields)
Row tracking adds the following metadata fields that can be accessed when reading a table:
| Column name | Type | Values |
| ------------------------------ | ---- | ---------------------------------------------------------------- |
| `_metadata.row_id` | Long | The unique identifier of the row. |
| `_metadata.row_commit_version` | Long | The table version at which the row was last inserted or updated. |
The row ids and row commit versions metadata fields are not automatically included when reading the table. Instead, these metadata fields must be manually selected from the hidden `_metadata` column which is available for all tables in Apache Spark.
* SQL
```sql
SELECT _metadata.row_id, _metadata.row_commit_version, * FROM table_name;
```
* Python
```python
spark.read.table("table_name") \
.select("_metadata.row_id", "_metadata.row_commit_version", "*")
```
* Scala
```scala
spark.read.table("table_name")
.select("_metadata.row_id", "_metadata.row_commit_version", "*")
```
## Disable row tracking
[Section titled “Disable row tracking”](#disable-row-tracking)
Row tracking can be disabled to reduce the storage overhead of the metadata fields. After disabling row tracking the metadata fields remain available, but all rows always get assigned a new id and commit version whenever they are touched by an operation.
* SQL
```sql
ALTER TABLE table_name SET TBLPROPERTIES (delta.enableRowTracking = false);
```
* Python
```python
spark.sql("ALTER TABLE table_name SET TBLPROPERTIES (delta.enableRowTracking = false)")
```
* Scala
```scala
spark.sql("ALTER TABLE table_name SET TBLPROPERTIES (delta.enableRowTracking = false)")
```
Caution
Disabling row tracking does not remove the corresponding table feature and does not downgrade the table protocol version.
## Limitations
[Section titled “Limitations”](#limitations)
The following limitations exist:
* The row ids and row commit versions metadata fields cannot be accessed while reading the [Change data feed](/delta-change-data-feed/).
* Once the Row Tracking feature is added to the table it cannot be removed without recreating the table.
# Read Delta Sharing Tables
> Learn how to perform reads on Delta Sharing tables.
[Delta Sharing](https://delta.io/sharing/) is an open protocol for secure real-time exchange of large datasets, which enables organizations to share data in real time regardless of which computing platforms they use. It is a simple REST protocol that securely grants access to part of a cloud dataset and leverages modern cloud storage systems, such as S3, ADLS, GCS or R2, to reliably transfer data.
In Delta Sharing, data provider is the one who owns the original dataset or table, and shares it with a broad range of recipients. Each table can be configured to be shared with different options (history, filtering, etc.) We will focus on consuming the shared table in this doc.
Delta Sharing data source supports most of the options provided by Apache Spark DataFrame for performing reads through [batch](/delta-batch), [streaming](/delta-streaming), or [table changes (CDF)](/delta-change-data-feed) APIs on shared tables. Delta Sharing doesn’t support writing to a shared table. Please refer to the [Delta Sharing Repo](https://github.com/delta-io/delta-sharing/blob/main/README.md) for more details. Please follow the [quick start](https://github.com/delta-io/delta-sharing?tab=readme-ov-file#quick-start) to leverage the Delta Sharing python connector to discover the shared tables.
For Delta Sharing reads on shared tables with advanced Delta Lake features such as [Deletion Vectors](/delta-deletion-vectors) and [Column Mapping](/delta-column-mapping), you need to enable integration with Apache Spark DataSourceV2 and Catalog APIs (since delta-sharing-spark 3.1) by setting the same configurations as Delta Lake when you create a new `SparkSession`. See [Configure SparkSession](/delta-batch/#configure-sparksession).
## Read a snapshot
[Section titled “Read a snapshot”](#read-a-snapshot)
After you save the [Profile File](https://github.com/delta-io/delta-sharing/blob/main/PROTOCOL.md#profile-file-format) locally and launch Spark with the connector library, you can access shared tables. A profile file is provided by the data provider to the data recipient.
* SQL
```sql
-- A table path is the profile file path followed by `#` and the fully qualified name
-- of a table (`..`).
CREATE TABLE mytable USING deltaSharing LOCATION '#..';
SELECT * FROM mytable;
```
* Python
```python
# A table path is the profile file path followed by `#` and the fully qualified name
# of a table (`..`).
table_path = "#.."
df = spark.read.format("deltaSharing").load(table_path)
```
* Scala
```scala
// A table path is the profile file path followed by `#` and the fully qualified name
// of a table (`..`).
val tablePath = "#.."
val df = spark.read.format("deltaSharing").load(tablePath)
```
* Java
```java
// A table path is the profile file path followed by `#` and the fully qualified name
// of a table (`..`).
String tablePath = "#..";
Dataset df = spark.read.format("deltaSharing").load(tablePath);
```
The DataFrame returned automatically reads the most recent snapshot of the table for any query.
Delta Sharing supports [predicate pushdown](https://github.com/delta-io/delta-sharing/blob/main/PROTOCOL.md#json-predicates-for-filtering) to efficiently fetch data from the Delta Sharing server when there are applicable predicates in the query.
## Query an older snapshot of a shared table (time travel)
[Section titled “Query an older snapshot of a shared table (time travel)”](#query-an-older-snapshot-of-a-shared-table-time-travel)
Once the data provider enables history sharing of the shared table, Delta Sharing time travel allows you to query an older snapshot of a shared table.
* SQL
```sql
SELECT * FROM mytable TIMESTAMP AS OF timestamp_expression
SELECT * FROM mytable VERSION AS OF version
```
* Python
```python
spark.read.format("deltaSharing").option("timestampAsOf", timestamp_string).load(tablePath)
spark.read.format("deltaSharing").option("versionAsOf", version).load(tablePath)
```
* Scala
```scala
spark.read.format("deltaSharing").option("timestampAsOf", timestamp_string).load(tablePath)
spark.read.format("deltaSharing").option("versionAsOf", version).load(tablePath)
```
The `timestamp_expression` and `version` share the same syntax as [Delta](/delta-batch#timestamp-and-version-syntax).
## Read Table Changes (CDF)
[Section titled “Read Table Changes (CDF)”](#read-table-changes-cdf)
Once the data provider turns on CDF on the original Delta Lake table and shares it with history through Delta Sharing, the recipient can query CDF of a Delta Sharing table similar to [CDF of a Delta table](/delta-change-data-feed).
* SQL
```sql
CREATE TABLE mytable USING deltaSharing LOCATION '#..';
-- version as ints or longs e.g. changes from version 0 to 10
SELECT * FROM table_changes('mytable', 0, 10)
-- timestamp as string formatted timestamps
SELECT * FROM table_changes('mytable', '2021-04-21 05:45:46', '2021-05-21 12:00:00')
-- providing only the startingVersion/timestamp
SELECT * FROM table_changes('mytable', 0)
```
* Python
```python
table_path = "#.."
# version as ints or longs
spark.read.format("deltaSharing") \
.option("readChangeFeed", "true") \
.option("startingVersion", 0) \
.option("endingVersion", 10) \
.load(tablePath)
# timestamps as formatted timestamp
spark.read.format("deltaSharing") \
.option("readChangeFeed", "true") \
.option("startingTimestamp", '2021-04-21 05:45:46') \
.option("endingTimestamp", '2021-05-21 12:00:00') \
.load(tablePath)
# providing only the startingVersion/timestamp
spark.read.format("deltaSharing") \
.option("readChangeFeed", "true") \
.option("startingVersion", 0) \
.load(tablePath)
```
* Scala
```scala
val tablePath = "#.."
// version as ints or longs
spark.read.format("deltaSharing")
.option("readChangeFeed", "true")
.option("startingVersion", 0)
.option("endingVersion", 10)
.load(tablePath)
// timestamps as formatted timestamp
spark.read.format("deltaSharing")
.option("readChangeFeed", "true")
.option("startingTimestamp", "2024-01-18 05:45:46")
.option("endingTimestamp", "2024-01-18 12:00:00")
.load(tablePath)
// providing only the startingVersion/timestamp
spark.read.format("deltaSharing")
.option("readChangeFeed", "true")
.option("startingVersion", 0)
.load(tablePath)
```
## Streaming
[Section titled “Streaming”](#streaming)
Delta Sharing Streaming is deeply integrated with [Spark Structured Streaming](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html) through `readStream`, and able to connect with any sink that is able to perform `writeStream`.
Once the data provider shares a table with history, the recipient can perform a streaming query on the table. When you load a Delta Sharing table as a stream source and use it in a streaming query, the query processes all of the data present in the shared table as well as any new data that arrives after the stream has started.
* Scala
```scala
val tablePath = "#.."
spark.readStream.format("deltaSharing").load(tablePath)
```
Delta Sharing Streaming supports the following functionalities in the same way as Delta Streaming: [Limit input rate](/delta-streaming/#limit-input-rate), [Ignore updates and deletes](/delta-streaming/#ignore-updates-and-deletes), [Specify initial position](/delta-streaming/#specify-initial-position)
In addition, `maxVersionsPerRpc` is provided to decide how many versions of files are requested from the server in every Delta Sharing rpc. This is to help reduce the per rpc workload and make the Delta sharing streaming job more stable, especially when many new versions have accumulated when the streaming resumes from a checkpoint. The default is 100.
Note
Trigger.AvailableNow is not supported in Delta Sharing Streaming. You can use Trigger.Once as a workaround, and at a proper frequency to catch up with the changes in the server.
## Read Advanced Delta Lake Features in Delta Sharing
[Section titled “Read Advanced Delta Lake Features in Delta Sharing”](#read-advanced-delta-lake-features-in-delta-sharing)
In order to support advanced Delta Lake features in Delta Sharing, “Delta Format Sharing” was introduced since delta-sharing-client 1.0 and delta-sharing-spark 3.1, in which the actions of a shared table are returned in Delta Lake format, allowing a Delta Lake library to read it.
Please remember to set the spark configurations mentioned in [Configure SparkSession](/delta-batch/#configure-sparksession) in order to read shared tables with Deletion Vectors and Column Mapping.
| Read Table Feature | Available since version |
| -------------------------------------------------------------------------------------------------------------- | ----------------------- |
| [Deletion Vectors](/delta-deletion-vectors) | 3.1.0 |
| [Column Mapping](/delta-column-mapping) | 3.1.0 |
| [Timestamp without Timezone](https://spark.apache.org/docs/latest/sql-ref-datatypes.html) | 3.3.0 |
| [Type widening (Preview)](/delta-type-widening) | 3.3.0 |
| [Variant Type (Preview)](https://github.com/delta-io/delta/blob/master/protocol_rfcs/accepted/variant-type.md) | 3.3.0 |
Batch queries can be performed as is, because it can automatically resolve the `responseFormat` based on the table features of the shared table. An additional option `responseFormat=delta` needs to be set for cdf and streaming queries when reading shared tables with Deletion Vectors or Column Mapping enabled.
* Scala
```scala
import org.apache.spark.sql.SparkSession
val spark = SparkSession
.builder()
.appName("...")
.master("...")
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
.getOrCreate()
val tablePath = "#.."
// Batch query
spark.read.format("deltaSharing").load(tablePath)
// CDF query
spark.read.format("deltaSharing")
.option("readChangeFeed", "true")
.option("responseFormat", "delta")
.option("startingVersion", 1)
.load(tablePath)
// Streaming query
spark.readStream.format("deltaSharing").option("responseFormat", "delta").load(tablePath)
```
# Delta Connect (aka Spark Connect Support in Delta)
> Learn about Delta Connect - Spark Connect Support in Delta.
Note
This feature is available in Delta Lake 4.0.0 and above. Please note, Delta Connect is currently in preview and not recommended for production workloads.
Delta Connect adds [Spark Connect](https://spark.apache.org/docs/latest/spark-connect-overview.html) support to Delta Lake for Apache Spark. Spark Connect is a new initiative that adds a decoupled client-server infrastructure which allows remote connectivity from Spark from everywhere. Delta Connect allows all Delta Lake operations to work in your application running as a client connected to the Spark server.
## Motivation
[Section titled “Motivation”](#motivation)
Delta Connect is expected to br0ng the same benefits as Spark Connect:
1. Upgrading to more recent versions of Spark and Delta Lake is now easier because the client interface is being completely decoupled from the server.
2. Simpler integration of Spark and Delta Lake with developer tooling. IDEs no longer have to integrate with the full Spark and Delta Lake implementation, and instead can integrate with a thin-client.
3. Support for languages other than Java/Scala and Python. Clients “merely” have to generate Protocol Buffers and therefore become simpler to implement.
4. Spark and Delta Lake will become more stable, as user code is no longer running in the same JVM as Spark’s driver.
5. Remote connectivity. Code can run anywhere now, as there is a gRPC layer between the user interface and the driver.
## How to start the Spark Server with Delta
[Section titled “How to start the Spark Server with Delta”](#how-to-start-the-spark-server-with-delta)
1. Download `spark-4.0.0-bin-hadoop3.tgz` from [Spark 4.0.0](https://archive.apache.org/dist/spark/spark-4.0.0).
2. Start the Spark Connect server with the Delta Lake Connect plugins:
```bash
sbin/start-connect-server.sh \
--packages io.delta:delta-connect-server_2.13:4.0.0,com.google.protobuf:protobuf-java:3.25.1 \
--conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
--conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog" \
--conf "spark.connect.extensions.relation.classes=org.apache.spark.sql.connect.delta.DeltaRelationPlugin" \
--conf "spark.connect.extensions.command.classes=org.apache.spark.sql.connect.delta.DeltaCommandPlugin"
```
## How to use the Python Spark Connect Client with Delta
[Section titled “How to use the Python Spark Connect Client with Delta”](#how-to-use-the-python-spark-connect-client-with-delta)
The Delta Lake Connect Python client is included in the same PyPi package as Delta Lake Spark.
1. `pip install pyspark==4.0.0`.
2. `pip install delta-spark==4.0.0`.
3. The usage is the same as Spark Connect (e.g. `./bin/pyspark --remote "sc://localhost"`). We just need to pass in a remote `SparkSession` (instead of a local one) to the `DeltaTable` API.
An example:
* Python
```python
from delta.tables import DeltaTable
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
deltaTable = DeltaTable.forName(spark, "my_table")
deltaTable.toDF().show()
deltaTable.update(
condition = "id % 2 == 0",
set = {"id": "id + 100"}
)
```
## How to use the Scala Spark Connect Client with Delta
[Section titled “How to use the Scala Spark Connect Client with Delta”](#how-to-use-the-scala-spark-connect-client-with-delta)
Make sure you are using Java 17!
```bash
./bin/spark-shell --remote "sc://localhost" --packages io.delta:delta-connect-client_2.13:4.0.0,com.google.protobuf:protobuf-java:3.25.1
```
An example:
* Scala
```scala
import io.delta.tables.DeltaTable
val deltaTable = DeltaTable.forName(spark, "my_table")
deltaTable.toDF.show()
deltaTable.updateExpr(
condition = "id % 2 == 0",
set = Map("id" -> "id + 100")
)
```
# Delta Standalone (deprecated)
> Learn how to read and write Delta tables from JVM applications without Apache Spark.
Warning!
Delta Standalone is deprecated and will be removed in a future release. We recommend using the [Delta Kernel](/delta-kernel) APIs.
The Delta Standalone library is a single-node Java library that can be used to read from and write to Delta tables. Specifically, this library provides APIs to interact with a table’s metadata in the transaction log, implementing the [Delta Transaction Log Protocol](https://github.com/delta-io/delta/blob/master/PROTOCOL.md) to achieve the transactional guarantees of the Delta Lake format. Notably, this project doesn’t depend on Apache Spark and has only a few transitive dependencies. Therefore, it can be used by any processing engine or application to access Delta tables.
## Use cases
[Section titled “Use cases”](#use-cases)
Delta Standalone is optimized for cases when you want to read and write Delta tables by using a non-Spark engine of your choice. It is a “low-level” library, and we encourage developers to contribute open-source, higher-level connectors for their desired engines that use Delta Standalone for all Delta Lake metadata interaction. You can find a Hive source connector and Flink sink/source connector in the [Delta Lake](https://github.com/delta-io/delta) repository. Additional connectors are in development.
### Caveats
[Section titled “Caveats”](#caveats)
Delta Standalone minimizes memory usage in the JVM by loading the Delta Lake transaction log incrementally, using an iterator. However, Delta Standalone runs in a single JVM, and is limited to the processing and memory capabilities of that JVM. Users must configure the JVM to avoid out of memory (OOM) issues.
Delta Standalone does provide basic APIs for reading Parquet data, but does not include APIs for writing Parquet data. Users must write out new Parquet data files themselves and then use Delta Standalone to commit those changes to the Delta table and make the new data visible to readers.
## APIs
[Section titled “APIs”](#apis)
Delta Standalone provides classes and entities to read data, query metadata, and commit to the transaction log. A few of them are highlighted here and with their key interfaces. See the [Java API docs](/api/latest/java/standalone/index.html) for the full set of classes and entities.
### DeltaLog
[Section titled “DeltaLog”](#deltalog)
[DeltaLog](/api/latest/java/io/delta/standalone/DeltaLog.html) is the main interface for programmatically interacting with the metadata in the transaction log of a Delta table.
* Instantiate a `DeltaLog` with `DeltaLog.forTable(hadoopConf, path)` and pass in the `path` of the root location of the Delta table.
* Access the current snapshot with `DeltaLog::snapshot`.
* Get the latest snapshot, including any new data files that were added to the log, with `DeltaLog::update`.
* Get the snapshot at some historical state of the log with `DeltaLog::getSnapshotForTimestampAsOf` or `DeltaLog::getSnapshotForVersionAsOf`.
* Start a new transaction to commit to the transaction log by using `DeltaLog::startTransaction`.
* Get all metadata actions without computing a full Snapshot using `DeltaLog::getChanges`.
### Snapshot
[Section titled “Snapshot”](#snapshot)
A [Snapshot](/api/latest/java/io/delta/standalone/Snapshot.html) represents the state of the table at a specific version.
* Get a list of the metadata files by using `Snapshot::getAllFiles`.
* For a memory-optimized iterator over the metadata files, use `Snapshot::scan` to get a `DeltaScan` (as described later), optionally by passing in a `predicate` for partition filtering.
* Read actual data with `Snapshot::open`, which returns an iterator over the rows of the Delta table.
### OptimisticTransaction
[Section titled “OptimisticTransaction”](#optimistictransaction)
The main class for committing a set of updates to the transaction log is [OptimisticTransaction](/api/latest/java/io/delta/standalone/OptimisticTransaction.html). During a transaction, all reads must go through the `OptimisticTransaction` instance rather than the `DeltaLog` in order to detect logical conflicts and concurrent updates.
* Read metadata files during a transaction with `OptimisticTransaction::markFilesAsRead`, which returns a `DeltaScan` of files that match the `readPredicate`.
* Commit to the transaction log with `OptimisticTransaction::commit`.
* Get the latest version committed for a given application ID (for example, for idempotency) with `OptimisticTransaction::txnVersion`. (Note that this API requires users to commit `SetTransaction` actions.)
* Update the medadata of the table upon committing with `OptimisticTransaction::updateMetadata`.
### DeltaScan
[Section titled “DeltaScan”](#deltascan)
[DeltaScan](/api/latest/java/io/delta/standalone/DeltaScan.html) is a wrapper class for the files inside a `Snapshot` that match a given `readPredicate`.
* Access the files that match the partition filter portion of the `readPredicate` with `DeltaScan::getFiles`. This returns a memory-optimized iterator over the metadata files in the table.
* To further filter the returned files on non-partition columns, get the portion of input predicate not applied with `DeltaScan::getResidualPredicate`.
## API compatibility
[Section titled “API compatibility”](#api-compatibility)
The only public APIs currently provided by Delta Standalone are in the `io.delta.standalone` package. Classes and methods in the `io.delta.standalone.internal` package are considered internal and are subject to change across minor and patch releases.
## Project setup
[Section titled “Project setup”](#project-setup)
You can add the Delta Standalone library as a dependency by using your preferred build tool. Delta Standalone depends upon the `hadoop-client` and `parquet-hadoop` packages. Example build files are listed in the following sections.
### Environment requirements
[Section titled “Environment requirements”](#environment-requirements)
* JDK 8 or above.
* Scala 2.11 or 2.12.
### Build files
[Section titled “Build files”](#build-files)
#### Maven
[Section titled “Maven”](#maven)
Replace the version of `hadoop-client` with the one you are using.
Scala 2.12:
```xml
io.delta
delta-standalone_2.12
0.5.0
org.apache.hadoop
hadoop-client
3.1.0
```
Scala 2.11:
```xml
io.delta
delta-standalone_2.11
0.5.0
org.apache.hadoop
hadoop-client
3.1.0
```
#### SBT
[Section titled “SBT”](#sbt)
Replace the version of `hadoop-client` with the one you are using.
```plaintext
libraryDependencies ++= Seq(
"io.delta" %% "delta-standalone" % "0.5.0",
"org.apache.hadoop" % "hadoop-client" % "3.1.0)
```
#### `ParquetSchemaConverter` caveat
[Section titled “ParquetSchemaConverter caveat”](#parquetschemaconverter-caveat)
Delta Standalone shades its own Parquet dependencies so that it works out-of-the-box and reduces dependency conflicts in your environment. However, if you would like to use utility class `io.delta.standalone.util.ParquetSchemaConverter`, then you must provide your own version of `org.apache.parquet:parquet-hadoop`.
### Storage configuration
[Section titled “Storage configuration”](#storage-configuration)
Delta Lake ACID guarantees are based on the atomicity and durability guarantees of the storage system. Not all storage systems provide all the necessary guarantees.
Because storage systems do not necessarily provide all of these guarantees out-of-the-box, Delta Lake transactional operations typically go through the [LogStore API](https://github.com/delta-io/delta/blob/master/storage/src/main/java/io/delta/storage/LogStore.java) instead of accessing the storage system directly. To provide the ACID guarantees for different storage systems, you may have to use different `LogStore` implementations. This section covers how to configure Delta Standalone for various storage systems. There are two categories of storage systems:
* **Storage systems with built-in support**: For some storage systems, you do not need additional configurations. Delta Standalone uses the scheme of the path (that is, `s3a` in `s3a://path`) to dynamically identify the storage system and use the corresponding `LogStore` implementation that provides the transactional guarantees. However, for S3, there are additional caveats on concurrent writes. See the [section on S3](#amazon-s3-configuration) for details.
* **Other storage systems**: The `LogStore`, similar to Apache Spark, uses the Hadoop `FileSystem` API to perform reads and writes. Delta Standalone supports concurrent reads on any storage system that provides an implementation of the `FileSystem` API. For concurrent writes with transactional guarantees, there are two cases based on the guarantees provided by the `FileSystem` implementation. If the implementation provides consistent listing and atomic renames-without-overwrite (that is, `rename(... , overwrite = false)` will either generate the target file atomically or fail if it already exists with `java.nio.file.FileAlreadyExistsException`), then the default `LogStore` implementation using renames will allow concurrent writes with guarantees. Otherwise, you must configure a custom implementation of `LogStore` by setting the following Hadoop configuration when you instantiate a `DeltaLog` with `DeltaLog.forTable(hadoopConf, path)`:
```java
delta.logStore..impl=
```
Here, `` is the scheme of the paths of your storage system. This configures Delta Standalone to dynamically use the given `LogStore` implementation only for those paths. You can have multiple such configurations for different schemes in your application, thus allowing it to simultaneously read and write from different storage systems.
Note
Before version 0.5.0, Delta Standalone supported configuring LogStores by setting `io.delta.standalone.LOG_STORE_CLASS_KEY`. This approach is now deprecated. Setting this configuration will use the configured `LogStore` for all paths, thereby disabling the dynamic scheme-based delegation.
#### Amazon S3 configuration
[Section titled “Amazon S3 configuration”](#amazon-s3-configuration)
Delta Standalone supports reads and writes to S3 in two different modes: Single-cluster and Multi-cluster.
| | Single-cluster | Multi-cluster |
| ------------- | ------------------------------------------------ | ------------------------------------------------ |
| Configuration | Comes out-of-the-box | Is experimental and requires extra configuration |
| Reads | Supports concurrent reads from multiple clusters | Supports concurrent reads from multiple clusters |
| Writes | Supports concurrent writes from a single cluster | Supports multi-cluster writes |
| Permissions | S3 credentials | S3 and DynamoDB operating permissions |
##### Single-cluster setup (default)
[Section titled “Single-cluster setup (default)”](#single-cluster-setup-default)
By default, Delta Standalone supports concurrent reads from multiple clusters. However, concurrent writes to S3 must originate from a single cluster to provide transactional guarantees. This is because S3 currently does not provide mutual exclusion, that is, there is no way to ensure that only one writer is able to create a file.
Warning!
Concurrent writes to the same Delta table from multiple Spark drivers can lead to data loss.
To use Delta Standalone with S3, you must meet the following requirements. If you are using access keys for authentication and authorization, you must configure a Hadoop Configuration specified as follows when you instantiate a `DeltaLog` with `DeltaLog.forTable(hadoopConf, path)`.
###### Requirements (S3 single-cluster)
[Section titled “Requirements (S3 single-cluster)”](#requirements-s3-single-cluster)
* S3 credentials: [IAM roles](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html) (recommended) or access keys.
* Hadoop’s [AWS connector (hadoop-aws)](https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-aws) for the version of Hadoop that Delta Standalone is compiled with.
###### Configuration (S3 single-cluster)
[Section titled “Configuration (S3 single-cluster)”](#configuration-s3-single-cluster)
1. Include `hadoop-aws` JAR in the classpath.
2. Set up S3 credentials. We recommend that you use [IAM roles](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html) for authentication and authorization. But if you want to use keys, configure your `org.apache.hadoop.conf.Configuration` with:
```java
conf.set("fs.s3a.access.key", "");
conf.set("fs.s3a.secret.key", "");
```
##### Multi-cluster setup
[Section titled “Multi-cluster setup”](#multi-cluster-setup)
Note
This support is new and experimental.
This mode supports concurrent writes to S3 from multiple clusters. Enable multi-cluster support by configuring Delta Standalone to use the correct `LogStore` implementation. This implementation uses [DynamoDB](https://aws.amazon.com/dynamodb/) to provide mutual exclusion.
Warning!
When writing from multiple clusters, all drivers must use this `LogStore` implementation and the same DynamoDB table and region. If some drivers use the default `LogStore` while others use this experimental `LogStore` then data loss can occur.
###### Requirements (S3 multi-cluster)
[Section titled “Requirements (S3 multi-cluster)”](#requirements-s3-multi-cluster)
* All of the requirements listed in the [Requirements (S3 single-cluster)](#requirements-s3-single-cluster) section
* In additon to S3 credentials, you also need DynamoDB operating permissions
###### Configuration (S3 multi-cluster)
[Section titled “Configuration (S3 multi-cluster)”](#configuration-s3-multi-cluster)
1. Create the DynamoDB table. See [Create the DynamoDB table](/delta-storage#setup-configuration-s3-multi-cluster) for more details on creating a table yourself (recommended) or having it created for you automatically.
2. Follow the configuration steps listed in [Configuration (S3 single-cluster)](#configuration-s3-single-cluster) section.
3. Include the `delta-storage-s3-dynamodb` JAR in the classpath.
4. Configure the `LogStore` implementation.
First, configure this `LogStore` implementation for the scheme `s3`. You can replicate this command for schemes `s3a` and `s3n` as well.
```java
conf.set("delta.logStore.s3.impl", "io.delta.storage.S3DynamoDBLogStore");
```
| Configuration Key | Description | Default |
| ------------------------------------------------------------- | ----------------------------------------------- | ---------------------------------- |
| io.delta.storage.S3DynamoDBLogStore.ddb.tableName | The name of the DynamoDB table to use | delta\_log |
| io.delta.storage.S3DynamoDBLogStore.ddb.region | The region to be used by the client | us-east-1 |
| io.delta.storage.S3DynamoDBLogStore.credentials.provider | The AWSCredentialsProvider\* used by the client | DefaultAWSCredentialsProviderChain |
| io.delta.storage.S3DynamoDBLogStore.provisionedThroughput.rcu | (Table-creation-only\*\*) Read Capacity Units | 5 |
| io.delta.storage.S3DynamoDBLogStore.provisionedThroughput.wcu | (Table-creation-only\*\*) Write Capacity Units | 5 |
\*For more details on AWS credential providers, see the [AWS documentation](https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html).
\*\*These configurations are only used when the given DynamoDB table doesn’t already exist and needs to be automatically created.
###### Production Configuration (S3 multi-cluster)
[Section titled “Production Configuration (S3 multi-cluster)”](#production-configuration-s3-multi-cluster)
By this point, this multi-cluster setup is fully operational. However, there is extra configuration you may do to improve performance and optimize storage when running in production. See the [Delta Lake documentation](/delta-storage#production-configuration-s3-multi-cluster) for more details.
#### Microsoft Azure configuration
[Section titled “Microsoft Azure configuration”](#microsoft-azure-configuration)
Delta Standalone supports concurrent reads and writes from multiple clusters with full transactional guarantees for various Azure storage systems. To use an Azure storage system, you must satisfy the following requirements, and configure a Hadoop Configuration as specified when you instantiate a `DeltaLog` with `DeltaLog.forTable(hadoopConf, path)`.
##### Azure Blob Storage
[Section titled “Azure Blob Storage”](#azure-blob-storage)
###### Requirements (Azure Blob storage)
[Section titled “Requirements (Azure Blob storage)”](#requirements-azure-blob-storage)
* A [shared key](https://docs.microsoft.com/rest/api/storageservices/authorize-with-shared-key) or [shared access signature (SAS)](https://docs.microsoft.com/azure/storage/common/storage-sas-overview).
* Hadoop’s [Azure Blob Storage libraries](https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-azure) for a version compatible with the Hadoop version Delta Standalone was compiled with.
* 2.9.1+ for Hadoop 2
* 3.0.1+ for Hadoop 3
###### Configuration (Azure Blob storage)
[Section titled “Configuration (Azure Blob storage)”](#configuration-azure-blob-storage)
1. Include `hadoop-azure` JAR in the classpath.
2. Set up credentials.
* For an SAS token, configure `org.apache.hadoop.conf.Configuration`:
```java
conf.set(
"fs.azure.sas...blob.core.windows.net",
"");
```
* To specify an account access key:
```java
conf.set(
"fs.azure.account.key..blob.core.windows.net",
"");
```
##### Azure Data Lake Storage Gen1
[Section titled “Azure Data Lake Storage Gen1”](#azure-data-lake-storage-gen1)
###### Requirements (ADLS Gen 1)
[Section titled “Requirements (ADLS Gen 1)”](#requirements-adls-gen-1)
* A [service principal](https://docs.microsoft.com/azure/active-directory/develop/app-objects-and-service-principals) for OAuth 2.0 access.
* Hadoop’s [Azure Data Lake Storage Gen1 libraries](https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-azure-datalake) for a version that is compatible with the Hadoop version that was used to compile Delta Standalone.
* 2.9.1+ for Hadoop 2
* 3.0.1+ for Hadoop 3
###### Configuration (ADLS Gen 1)
[Section titled “Configuration (ADLS Gen 1)”](#configuration-adls-gen-1)
1. Include `hadoop-azure-datalake` JAR in the classpath.
2. Set up Azure Data Lake Storage Gen1 credentials. Configure `org.apache.hadoop.conf.Configuration`:
```java
conf.set("dfs.adls.oauth2.access.token.provider.type", "ClientCredential");
conf.set("dfs.adls.oauth2.client.id", "");
conf.set("dfs.adls.oauth2.credential", "");
conf.set("dfs.adls.oauth2.refresh.url", "https://login.microsoftonline.com//oauth2/token");
```
##### Azure Data Lake Storage Gen2
[Section titled “Azure Data Lake Storage Gen2”](#azure-data-lake-storage-gen2)
###### Requirements (ADLS Gen 2)
[Section titled “Requirements (ADLS Gen 2)”](#requirements-adls-gen-2)
* Account created in [Azure Data Lake Storage Gen2](https://docs.microsoft.com/azure/storage/blobs/create-data-lake-storage-account).
* Service principal [created](https://docs.microsoft.com/azure/active-directory/develop/howto-create-service-principal-portal) and [assigned the Storage Blob Data Contributor role](https://docs.microsoft.com/azure/storage/blobs/assign-azure-role-data-access) for the storage account.
* Make a note of the storage-account-name, directory-id (also known as tenant-id), application-id, and password of the principal. These will be used for configuration.
* Hadoop’s [Azure Data Lake Storage Gen2 libraries](https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-azure-datalake) version 3.2+ and Delta Standalone compiled with Hadoop 3.2+.
###### Configuration (ADLS Gen 2)
[Section titled “Configuration (ADLS Gen 2)”](#configuration-adls-gen-2)
1. Include `hadoop-azure-datalake` JAR in the classpath. In addition, you may also have to include JARs for Maven artifacts `hadoop-azure` and `wildfly-openssl`.
2. Set up Azure Data Lake Storage Gen2 credentials. Configure your `org.apache.hadoop.conf.Configuration` with:
```java
conf.set("fs.azure.account.auth.type..dfs.core.windows.net", "OAuth");
conf.set("fs.azure.account.oauth.provider.type..dfs.core.windows.net", "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider");
conf.set("fs.azure.account.oauth2.client.id..dfs.core.windows.net", "");
conf.set("fs.azure.account.oauth2.client.secret..dfs.core.windows.net","");
conf.set("fs.azure.account.oauth2.client.endpoint..dfs.core.windows.net", "https://login.microsoftonline.com//oauth2/token");
```
where ``, ``, `` and `` are details of the service principal we set as requirements earlier.
#### HDFS
[Section titled “HDFS”](#hdfs)
Delta Standalone has built-in support for HDFS with full transactional guarantees on concurrent reads and writes from multiple clusters. See [Hadoop documentation](https://hadoop.apache.org/docs/stable/) for configuring credentials.
#### Google Cloud Storage
[Section titled “Google Cloud Storage”](#google-cloud-storage)
##### Requirements (GCS)
[Section titled “Requirements (GCS)”](#requirements-gcs)
* JAR of the [GCS Connector (gcs-connector)](https://search.maven.org/search?q=a:gcs-connector) Maven artifact.
* Google Cloud Storage account and credentials
##### Configuration (GCS)
[Section titled “Configuration (GCS)”](#configuration-gcs)
1. Include the JAR for `gcs-connector` in the classpath. See the [documentation](https://cloud.google.com/dataproc/docs/tutorials/gcs-connector-spark-tutorial) for details on how to configure your project with the GCS connector.
## Usage
[Section titled “Usage”](#usage)
This example shows how to use Delta Standalone to:
* Find parquet files.
* Write parquet data.
* Commit to the transaction log.
* Read from the transaction log.
* Read back the Parquet data.
Please note that this example uses a fictitious, non-Spark engine `Zappy` to write the actual parquet data, as Delta Standalone does not provide any data-writing APIs. Instead, Delta Standalone Writer lets you commit metadata to the Delta log after you’ve written your data. This is why Delta Standalone works well with so many connectors (e.g. Flink, Presto, Trino, etc.) since they provide the parquet-writing functionality instead.
### 1. SBT configuration
[Section titled “1. SBT configuration”](#1-sbt-configuration)
The following SBT project configuration is used:
```scala
// /build.sbt
scalaVersion := "2.12.8"
libraryDependencies ++= Seq(
"io.delta" %% "delta-standalone" % "0.5.0",
"org.apache.hadoop" % "hadoop-client" % "3.1.0")
```
### 2. Mock situation
[Section titled “2. Mock situation”](#2-mock-situation)
We have a Delta table `Sales` storing sales data, but have realized all the data written on November 2021 for customer `XYZ` had incorrect `total_cost` values. Thus, we need to update all those records with the correct values. We will use a fictious distributed engine `Zappy` and Delta Standalone to update our Delta table.
The sales table schema is given below.
```plaintext
Sales
|-- year: int // partition column
|-- month: int // partition column
|-- day: int // partition column
|-- customer: string
|-- sale_id: string
|-- total_cost: float
```
### 3. Starting a transaction and finding relevant files
[Section titled “3. Starting a transaction and finding relevant files”](#3-starting-a-transaction-and-finding-relevant-files)
Since we must read existing data in order to perform the desired update operation, we must use `OptimisticTransaction::markFilesAsRead` in order to automatically detect any concurrent modifications made to our read partitions. Since Delta Standalone only supports partition pruning, we must apply the residual predicate to further filter the returned files.
```java
import io.delta.standalone.DeltaLog;
import io.delta.standalone.DeltaScan;
import io.delta.standalone.OptimisticTransaction;
import io.delta.standalone.actions.AddFile;
import io.delta.standalone.data.CloseableIterator;
import io.delta.standalone.expressions.And;
import io.delta.standalone.expressions.EqualTo;
import io.delta.standalone.expressions.Literal;
DeltaLog log = DeltaLog.forTable(new Configuration(), "/data/sales");
OptimisticTransaction txn = log.startTransaction();
DeltaScan scan = txn.markFilesAsRead(
new And(
new And(
new EqualTo(schema.column("year"), Literal.of(2021)), // partition filter
new EqualTo(schema.column("month"), Literal.of(11))), // partition filter
new EqualTo(schema.column("customer"), Literal.of("XYZ")) // non-partition filter
)
);
CloseableIterator iter = scan.getFiles();
Map addFileMap = new HashMap(); // partition filtered files: year=2021, month=11
while (iter.hasNext()) {
AddFile addFile = iter.next();
addFileMap.put(addFile.getPath(), addFile);
}
iter.close();
List filteredFiles = ZappyReader.filterFiles( // fully filtered files: year=2021, month=11, customer=XYZ
addFileMap.keySet(),
toZappyExpression(scan.getResidualPredicate())
);
```
### 4. Writing updated Parquet data
[Section titled “4. Writing updated Parquet data”](#4-writing-updated-parquet-data)
Since Delta Standalone does not provide any Parquet data write APIs, we use `Zappy` to write the data.
```java
ZappyDataFrame correctedSaleIdToTotalCost = ...;
ZappyDataFrame invalidSales = ZappyReader.readParquet(filteredFiles);
ZappyDataFrame correctedSales = invalidSales.join(correctedSaleIdToTotalCost, "id");
ZappyWriteResult dataWriteResult = ZappyWritter.writeParquet("/data/sales", correctedSales);
```
The written data files from the preceding code will have a hierarchy similar to the following:
```shell
$ tree /data/sales
.
├── _delta_log
│ └── ...
│ └── 00000000000000001082.json
│ └── 00000000000000001083.json
├── year=2019
│ └── month=1
...
├── year=2020
│ └── month=1
│ └── day=1
│ └── part-00000-195768ae-bad8-4c53-b0c2-e900e0f3eaee-c000.snappy.parquet // previous
│ └── part-00001-53c3c553-f74b-4384-b9b5-7aa45bc2291b-c000.snappy.parquet // new
| ...
│ └── day=2
│ └── part-00000-b9afbcf5-b90d-4f92-97fd-a2522aa2d4f6-c000.snappy.parquet // previous
│ └── part-00001-c0569730-5008-42fa-b6cb-5a152c133fde-c000.snappy.parquet // new
| ...
```
### 5. Committing to our Delta table
[Section titled “5. Committing to our Delta table”](#5-committing-to-our-delta-table)
Now that we’ve written the correct data, we need to commit to the transaction log to add the new files, and remove the old incorrect files.
```java
import io.delta.standalone.Operation;
import io.delta.standalone.actions.RemoveFile;
import io.delta.standalone.exceptions.DeltaConcurrentModificationException;
import io.delta.standalone.types.StructType;
List