Update Delta Lake 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
• Renaming existing columns

You can make these changes explicitly using DDL or implicitly using DML.

important

An update to a Delta table schema is an operation that conflicts with all concurrent Delta write operations.

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. For recommended methods, see Production considerations for Structured Streaming.

Explicitly update schema to add columns

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

ALTER TABLE table_name ADD COLUMNS (col_name.nested_col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name], ...)

For example, if the schema before running ALTER TABLE boxes ADD COLUMNS (colB.nested STRING AFTER field1) is:

- root
| - colA
| - colB
| +-field1
| +-field2

the schema after is:

- root
| - colA
| - colB
| +-field1
| +-nested
| +-field2

note

Adding nested columns is supported only for structs. Arrays and maps are not supported.

Explicitly update schema to change column comment or ordering

SQL

ALTER TABLE table_name ALTER [COLUMN] col_name (COMMENT col_comment | FIRST | AFTER colA_name)

To change a column in a nested field, use:

SQL

ALTER TABLE table_name ALTER [COLUMN] col_name.nested_col_name (COMMENT col_comment | FIRST | AFTER colA_name)

For example, if the schema before running ALTER TABLE boxes ALTER COLUMN colB.field2 FIRST is:

- root
| - colA
| - colB
| +-field1
| +-field2

the schema after is:

- root
| - colA
| - colB
| +-field2
| +-field1

Explicitly update schema to replace columns

SQL

ALTER TABLE table_name REPLACE COLUMNS (col_name1 col_type1 [COMMENT col_comment1], ...)

For example, when running the following DDL:

SQL

ALTER TABLE boxes REPLACE COLUMNS (colC STRING, colB STRUCT<field2:STRING, nested:STRING, field1:STRING>, colA STRING)

if the schema before is:

- root
| - colA
| - colB
| +-field1
| +-field2

the schema after is:

- root
| - colC
| - colB
| +-field2
| +-nested
| +-field1
| - colA

Explicitly update schema to rename columns

note

This feature is available in Databricks Runtime 10.4 LTS and above.

To rename columns without rewriting any of the columns’ existing data, you must enable column mapping for the table. See Rename and drop columns with Delta Lake column mapping.

To rename a column:

SQL

ALTER TABLE table_name RENAME COLUMN old_col_name TO new_col_name

To rename a nested field:

SQL

ALTER TABLE table_name RENAME COLUMN col_name.old_nested_field TO new_nested_field

For example, when you run the following command:

SQL

ALTER TABLE boxes RENAME COLUMN colB.field1 TO field001

If the schema before is:

- root
| - colA
| - colB
| +-field1
| +-field2

Then the schema after is:

- root
| - colA
| - colB
| +-field001
| +-field2

See Rename and drop columns with Delta Lake column mapping.

Explicitly update schema to drop columns

note

This feature is available in Databricks Runtime 11.3 LTS and above.

To drop columns as a metadata-only operation without rewriting any data files, you must enable column mapping for the table. See Rename and drop columns with Delta Lake column mapping.

important

Dropping a column from metadata does not delete the underlying data for the column in files. To purge the dropped column data, you can use REORG TABLE to rewrite files. You can then use VACUUM to physically delete the files that contain the dropped column data.

To drop a column:

SQL

ALTER TABLE table_name DROP COLUMN col_name

To drop multiple columns:

SQL

ALTER TABLE table_name DROP COLUMNS (col_name_1, col_name_2)

Explicitly update schema to 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.

The following example shows changing a column type:

Python

(spark.read.table(...)
  .withColumn("birthDate", col("birthDate").cast("date"))
  .write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable(...)
)

The following example shows changing a column name:

Python

(spark.read.table(...)
  .withColumnRenamed("dateOfBirth", "birthDate")
  .write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable(...)
)

Enable schema evolution

You can enable schema evolution by doing one of the following:

• Set the .option("mergeSchema", "true") to a Spark DataFrame write or writeStream operation. See Enable schema evolution for writes to add new columns.
• Use MERGE WITH SCHEMA EVOLUTION syntax. See Schema evolution syntax for merge.
• Set the Spark conf spark.databricks.delta.schema.autoMerge.enabled to true for the current SparkSession.

Databricks recommends enabling schema evolution for each write operation rather than setting a Spark conf.

When you use options or syntax to enable schema evolution in a write operation, this takes precedence over the Spark conf.

note

There is no schema evolution clause for INSERT INTO statements.

Enable schema evolution for writes to add new columns

Columns that are present in the source query but missing from the target table are automatically added as part of a write transaction when schema evolution is enabled. See Enable schema evolution.

Case is preserved when appending a new column. New columns are added to the end of the table schema. If the additional columns are in a struct, they are appended to the end of the struct in the target table.

The following example demonstrates using the mergeSchema option with Auto Loader. See What is Auto Loader?.

Python

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", "<path-to-schema-location>")
  .load("<path-to-source-data>")
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", "<path-to-checkpoint>")
  .trigger(availableNow=True)
  .toTable("table_name")
)

The following example demonstrates using the mergeSchema option with a batch write operation:

Python

(spark.read
  .table(source_table)
  .write
  .option("mergeSchema", "true")
  .mode("append")
  .saveAsTable("table_name")
)

Automatic schema evolution for Delta Lake merge

Schema evolution allows users to resolve schema mismatches between the target and source table in merge. It handles the following two cases:

1. A column in the source table is not present in the target table. The new column is added to the target schema, and its values are inserted or updated using the source values.
2. A column in the target table is not present in the source table. The target schema is left unchanged; the values in the additional target column are either left unchanged (for UPDATE) or set to NULL (for INSERT).

You must manually enable automatic schema evolution. See Enable schema evolution.

note

In Databricks Runtime 12.2 LTS and above, columns and struct fields present in the source table can be specified by name in insert or update actions. In Databricks Runtime 11.3 LTS and below, only INSERT * or UPDATE SET * actions can be used for schema evolution with merge.

In Databricks Runtime 13.3 LTS and above, you can use schema evolution with structs nested inside maps, such as map<int, struct<a: int, b: int>>.

Schema evolution syntax for merge

In Databricks Runtime 15.2 and above, you can specify schema evolution in a merge statement using SQL or Delta table APIs:

SQL

SQL

MERGE WITH SCHEMA EVOLUTION INTO target
USING source
ON source.key = target.key
WHEN MATCHED THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *
WHEN NOT MATCHED BY SOURCE THEN
  DELETE

Python

Python

from delta.tables import *

(targetTable
  .merge(sourceDF, "source.key = target.key")
  .withSchemaEvolution()
  .whenMatchedUpdateAll()
  .whenNotMatchedInsertAll()
  .whenNotMatchedBySourceDelete()
  .execute()
)

Scala

Scala

import io.delta.tables._

targetTable
  .merge(sourceDF, "source.key = target.key")
  .withSchemaEvolution()
  .whenMatched()
  .updateAll()
  .whenNotMatched()
  .insertAll()
  .whenNotMatchedBySource()
  .delete()
  .execute()

Example operations of merge with schema evolution

Here are a few examples of the effects of merge operation with and without schema evolution.

Columns	Query (in SQL)	Behavior without schema evolution (default)	Behavior with schema evolution
Target columns: key, value Source columns: key, value, new_value	SQL MERGE INTO target_table t USING source_table s ON t.key = s.key WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *	The table schema remains unchanged; only columns key, value are updated/inserted.	The table schema is changed to (key, value, new_value). Existing records with matches are updated with the value and new_value in the source. New rows are inserted with the schema (key, value, new_value).
Target columns: key, old_value Source columns: key, new_value	SQL MERGE INTO target_table t USING source_table s ON t.key = s.key WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *	UPDATE and INSERT actions throw an error because the target column old_value is not in the source.	The table schema is changed to (key, old_value, new_value). Existing records with matches are updated with the new_value in the source leaving old_value unchanged. New records are inserted with the specified key, new_value, and NULL for the old_value.
Target columns: key, old_value Source columns: key, new_value	SQL MERGE INTO target_table t USING source_table s ON t.key = s.key WHEN MATCHED THEN UPDATE SET new_value = s.new_value	UPDATE throws an error because column new_value does not exist in the target table.	The table schema is changed to (key, old_value, new_value). Existing records with matches are updated with the new_value in the source leaving old_value unchanged, and unmatched records have NULL entered for new_value. See note (1).
Target columns: key, old_value Source columns: key, new_value	SQL MERGE INTO target_table t USING source_table s ON t.key = s.key WHEN NOT MATCHED THEN INSERT (key, new_value) VALUES (s.key, s.new_value)	INSERT throws an error because column new_value does not exist in the target table.	The table schema is changed to (key, old_value, new_value). New records are inserted with the specified key, new_value, and NULL for the old_value. Existing records have NULL entered for new_value leaving old_value unchanged. See note (1).

(1) This behavior is available in Databricks Runtime 12.2 LTS and above; Databricks Runtime 11.3 LTS and below error in this condition.

Exclude columns with Delta Lake merge

In Databricks Runtime 12.2 LTS and above, you can use EXCEPT clauses in merge conditions to explicitly exclude columns. The behavior of the EXCEPT keyword varies depending on whether or not schema evolution is enabled.

With schema evolution disabled, the EXCEPT keyword applies to the list of columns in the target table and allows excluding columns from UPDATE or INSERT actions. Excluded columns are set to null.

With schema evolution enabled, the EXCEPT keyword applies to the list of columns in the source table and allows excluding columns from schema evolution. A new column in the source that is not present in the target is not added to the target schema if it is listed in the EXCEPT clause. Excluded columns that are already present in the target are set to null.

The following examples demonstrate this syntax:

Columns	Query (in SQL)	Behavior without schema evolution (default)	Behavior with schema evolution
Target columns: id, title, last_updated Source columns: id, title, review, last_updated	SQL MERGE INTO target t USING source s ON t.id = s.id WHEN MATCHED THEN UPDATE SET last_updated = current_date() WHEN NOT MATCHED THEN INSERT * EXCEPT (last_updated)	Matched rows are updated by setting the last_updated field to the current date. New rows are inserted using values for id and title. The excluded field last_updated is set to null. The field review is ignored because it is not in the target.	Matched rows are updated by setting the last_updated field to the current date. Schema is evolved to add the field review. New rows are inserted using all source fields except last_updated which is set to null.
Target columns: id, title, last_updated Source columns: id, title, review, internal_count	SQL MERGE INTO target t USING source s ON t.id = s.id WHEN MATCHED THEN UPDATE SET last_updated = current_date() WHEN NOT MATCHED THEN INSERT * EXCEPT (last_updated, internal_count)	INSERT throws an error because column internal_count does not exist in the target table.	Matched rows are updated by setting the last_updated field to the current date. The review field is added to the target table, but the internal_count field is ignored. New rows inserted have last_updated set to null.

Dealing with NullType columns in schema updates

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

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

df.write.option("overwriteSchema", "true")

important

You cannot specify overwriteSchema as true when using dynamic partition overwrite.