Schema evolution is an essential aspect of data management, and Hudi supports schema evolution on write out-of-the-box, and experimental support for schema evolution on read. This page will discuss the schema evolution support in Hudi.

Schema Evolution on Write

Hudi supports backwards-compatible schema evolution scenarios out of the box, such as adding a nullable field or promoting a field’s datatype.

We recommend employing this approach as much as possible. This is a practical and efficient way to evolve schemas, proven at large-scale data lakes at companies like Uber, Walmart, and LinkedIn. It is also implemented at scale by vendors like Confluent for streaming data. Given the continuous nature of streaming data, there are no boundaries to define a schema change that can be incompatible with the previous schema (e.g., renaming a column).

Furthermore, the evolved schema is queryable across high-performance engines like Presto and Spark SQL without additional overhead for column ID translations or type reconciliations. The following table summarizes the schema changes compatible with different Hudi table types.

The incoming schema will automatically have missing columns added with null values from the table schema. For this we need to enable the following config hoodie.write.set.null.for.missing.columns, otherwise the pipeline will fail.

Schema Change COW MOR Remarks
Add a new nullable column at root level at the end Yes Yes Yes means that a write with evolved schema succeeds and a read following the write succeeds to read entire dataset.
Add a new nullable column to inner struct (at the end) Yes Yes
Add a new complex type field with default (map and array) Yes Yes
Add a new nullable column and change the ordering of fields No No Write succeeds but read fails if the write with evolved schema updated only some of the base files but not all. Currently, Hudi does not maintain a schema registry with history of changes across base files. Nevertheless, if the upsert touched all base files then the read will succeed.
Add a custom nullable Hudi meta column, e.g. _hoodie_meta_col Yes Yes
Promote datatype for a field at root level Yes Yes
Promote datatype for a nested field Yes Yes
Promote datatype for a complex type (value of map or array) Yes Yes
Add a new non-nullable column at root level at the end No No In case of MOR table with Spark data source, write succeeds but read fails. As a workaround, you can make the field nullable.
Add a new non-nullable column to inner struct (at the end) No No
Demote datatype for a field at root level No No
Demote datatype for a nested field No No
Demote datatype for a complex type (value of map or array) No No

Type Promotions

This chart shows what the table schema will be when an incoming column type has changed (X means that it is not allowed):

Incoming Schema ↓ \ Table Schema → int long float double string bytes
int int long float double string X
long long long float double string X
float float float float double string X
double double double double double string X
string string string string string string bytes
bytes X X X X string bytes

Schema Evolution on read

There are often scenarios where it’s desirable to have the ability to evolve the schema more flexibly. For example,

  1. Columns (including nested columns) can be added, deleted, modified, and moved.
  2. Renaming of columns (including nested columns).
  3. Add, delete, or perform operations on nested columns of the Array type.

Hudi has experimental support for allowing backward incompatible schema evolution scenarios on write while resolving it during read time. To enable this feature, hoodie.schema.on.read.enable=true needs to be set on the writer config (Datasource) or table property (SQL).

Hudi versions > 0.11 and Spark versions > 3.1.x, and 3.2.1 are required. For Spark 3.2.1 and above, spark.sql.catalog.spark_catalog must also be set. If schema on read is enabled, it cannot be disabled again since the table would have accepted such schema changes already.

Adding Columns

  1. -- add columns
  2. ALTER TABLE tableName ADD COLUMNS(col_spec[, col_spec ...])

Column specification consists of five field, next to each other.

Parameter Description
col_name name of the new column. To add sub-column col1 to a nested map type column member map>, set this field to member.value.col1
col_type type of the new column.
nullable whether or not the new column allows null values. (optional)
comment comment of the new column. (optional)
col_position The position where the new column is added. The value can be FIRST or AFTER origin_col. If it is set to FIRST, the new column will be added before the first column of the table. If it is set to AFTER origin_col, the new column will be added after the original column. FIRST can be used only when new sub-columns are added to nested columns and not in top-level columns. There are no restrictions on the usage of AFTER.

Examples

  1. ALTER TABLE h0 ADD COLUMNS(ext0 string);
  2. ALTER TABLE h0 ADD COLUMNS(new_col int not null comment 'add new column' AFTER col1);
  3. ALTER TABLE complex_table ADD COLUMNS(col_struct.col_name string comment 'add new column to a struct col' AFTER col_from_col_struct);

Altering Columns

Syntax

  1. -- alter table ... alter column
  2. ALTER TABLE tableName ALTER [COLUMN] col_old_name TYPE column_type [COMMENT] col_comment[FIRST|AFTER] column_name

Parameter Description

Parameter Description
tableName Table name.
col_old_name Name of the column to be altered.
column_type Type of the target column.
col_comment Optional comments on the altered column.
column_name The new position to place the altered column. For example, AFTER column_name indicates that the target column is placed after column_name.

Examples

  1. --- Changing the column type
  2. ALTER TABLE table1 ALTER COLUMN a.b.c TYPE bigint
  4. --- Altering other attributes
  5. ALTER TABLE table1 ALTER COLUMN a.b.c COMMENT 'new comment'
  6. ALTER TABLE table1 ALTER COLUMN a.b.c FIRST
  7. ALTER TABLE table1 ALTER COLUMN a.b.c AFTER x
  8. ALTER TABLE table1 ALTER COLUMN a.b.c DROP NOT NULL

column type change

Source\Target long float double string decimal date int
int Y Y Y Y Y N Y
long Y Y Y Y Y N N
float N Y Y Y Y N N
double N N Y Y Y N N
decimal N N N Y Y N N
string N N N Y Y Y N
date N N N Y N Y N

Deleting Columns

Syntax

  1. -- alter table ... drop columns
  2. ALTER TABLE tableName DROP COLUMN|COLUMNS cols

Examples

  1. ALTER TABLE table1 DROP COLUMN a.b.c
  2. ALTER TABLE table1 DROP COLUMNS a.b.c, x, y

Renaming columns

Syntax

  1. -- alter table ... rename column
  2. ALTER TABLE tableName RENAME COLUMN old_columnName TO new_columnName

Examples

  1. ALTER TABLE table1 RENAME COLUMN a.b.c TO x

When using hive metastore, please disable hive.metastore.disallow.incompatible.col.type.changes if you encounter this error: The following columns have types incompatible with the existing columns in their respective positions.

Schema Evolution in Action

Let us walk through an example to demonstrate the schema evolution support in Hudi. In the below example, we are going to add a new string field and change the datatype of a field from int to long.

  1. scala> :paste 
  2. import org.apache.hudi.QuickstartUtils._
  3. import scala.collection.JavaConversions._
  4. import org.apache.spark.sql.SaveMode._
  5. import org.apache.hudi.DataSourceReadOptions._
  6. import org.apache.hudi.DataSourceWriteOptions._
  7. import org.apache.hudi.config.HoodieWriteConfig._
  8. import org.apache.spark.sql.types._
  9. import org.apache.spark.sql.Row
  11. val tableName = "hudi_trips_cow"
  12. val basePath = "file:///tmp/hudi_trips_cow"
  13. val schema = StructType( Array(
  14.      StructField("rowId", StringType,true),
  15.      StructField("partitionId", StringType,true),
  16.      StructField("preComb", LongType,true),
  17.      StructField("name", StringType,true),
  18.      StructField("versionId", StringType,true),
  19.      StructField("intToLong", IntegerType,true)
  20. ))
  23. val data1 = Seq(Row("row_1", "part_0", 0L, "bob", "v_0", 0),
  24.                 Row("row_2", "part_0", 0L, "john", "v_0", 0),
  25.                 Row("row_3", "part_0", 0L, "tom", "v_0", 0))
  27. var dfFromData1 = spark.createDataFrame(data1, schema)
  28. dfFromData1.write.format("hudi").
  29.    options(getQuickstartWriteConfigs).
  30.    option("hoodie.datasource.write.precombine.field", "preComb").
  31.    option("hoodie.datasource.write.recordkey.field", "rowId").
  32.    option("hoodie.datasource.write.partitionpath.field", "partitionId").
  33.    option("hoodie.index.type","SIMPLE").
  34.    option("hoodie.table.name", tableName).
  35.    mode(Overwrite).
  36.    save(basePath)
  38. var tripsSnapshotDF1 = spark.read.format("hudi").load(basePath + "/*/*")
  39. tripsSnapshotDF1.createOrReplaceTempView("hudi_trips_snapshot")
  41. ctrl+D
  43. scala> spark.sql("desc hudi_trips_snapshot").show()
  44.     +--------------------+---------+-------+
  45.     |            col_name|data_type|comment|
  46.     +--------------------+---------+-------+
  47.     | _hoodie_commit_time|   string|   null|
  48.     |_hoodie_commit_seqno|   string|   null|
  49.     |  _hoodie_record_key|   string|   null|
  50.     |_hoodie_partition...|   string|   null|
  51.     |   _hoodie_file_name|   string|   null|
  52.     |               rowId|   string|   null|
  53.     |         partitionId|   string|   null|
  54.     |             preComb|   bigint|   null|
  55.     |                name|   string|   null|
  56.     |           versionId|   string|   null|
  57.     |           intToLong|      int|   null|
  58.     +--------------------+---------+-------+
  60. scala> spark.sql("select rowId, partitionId, preComb, name, versionId, intToLong from hudi_trips_snapshot").show()
  61.     +-----+-----------+-------+----+---------+---------+
  62.     |rowId|partitionId|preComb|name|versionId|intToLong|
  63.     +-----+-----------+-------+----+---------+---------+
  64.     |row_3|     part_0|      0| tom|      v_0|        0|
  65.     |row_2|     part_0|      0|john|      v_0|        0|
  66.     |row_1|     part_0|      0| bob|      v_0|        0|
  67.     +-----+-----------+-------+----+---------+---------+
  69. // In the new schema, we are going to add a String field and 
  70. // change the datatype `intToLong` field from  int to long.
  71. scala> :paste 
  72. val newSchema = StructType( Array(
  73.     StructField("rowId", StringType,true),
  74.     StructField("partitionId", StringType,true),
  75.     StructField("preComb", LongType,true),
  76.     StructField("name", StringType,true),
  77.     StructField("versionId", StringType,true),
  78.     StructField("intToLong", LongType,true),
  79.     StructField("newField", StringType,true)
  80. ))
  82. val data2 = Seq(Row("row_2", "part_0", 5L, "john", "v_3", 3L, "newField_1"),
  83.                Row("row_5", "part_0", 5L, "maroon", "v_2", 2L, "newField_1"),
  84.                Row("row_9", "part_0", 5L, "michael", "v_2", 2L, "newField_1"))
  86. var dfFromData2 = spark.createDataFrame(data2, newSchema)
  87. dfFromData2.write.format("hudi").
  88.     options(getQuickstartWriteConfigs).
  89.     option("hoodie.datasource.write.precombine.field", "preComb").
  90.     option("hoodie.datasource.write.recordkey.field", "rowId").
  91.     option("hoodie.datasource.write.partitionpath.field", "partitionId").
  92.     option("hoodie.index.type","SIMPLE").
  93.     option("hoodie.table.name", tableName).
  94.     mode(Append).
  95.     save(basePath)
  97. var tripsSnapshotDF2 = spark.read.format("hudi").load(basePath + "/*/*")
  98. tripsSnapshotDF2.createOrReplaceTempView("hudi_trips_snapshot")
  100. Ctrl + D
  102. scala> spark.sql("desc hudi_trips_snapshot").show()
  103.     +--------------------+---------+-------+
  104.     |            col_name|data_type|comment|
  105.     +--------------------+---------+-------+
  106.     | _hoodie_commit_time|   string|   null|
  107.     |_hoodie_commit_seqno|   string|   null|
  108.     |  _hoodie_record_key|   string|   null|
  109.     |_hoodie_partition...|   string|   null|
  110.     |   _hoodie_file_name|   string|   null|
  111.     |               rowId|   string|   null|
  112.     |         partitionId|   string|   null|
  113.     |             preComb|   bigint|   null|
  114.     |                name|   string|   null|
  115.     |           versionId|   string|   null|
  116.     |           intToLong|   bigint|   null|
  117.     |            newField|   string|   null|
  118.     +--------------------+---------+-------+
  121. scala> spark.sql("select rowId, partitionId, preComb, name, versionId, intToLong, newField from hudi_trips_snapshot").show()
  122.     +-----+-----------+-------+-------+---------+---------+----------+
  123.     |rowId|partitionId|preComb|   name|versionId|intToLong|  newField|
  124.     +-----+-----------+-------+-------+---------+---------+----------+
  125.     |row_3|     part_0|      0|    tom|      v_0|        0|      null|
  126.     |row_2|     part_0|      5|   john|      v_3|        3|newField_1|
  127.     |row_1|     part_0|      0|    bob|      v_0|        0|      null|
  128.     |row_5|     part_0|      5| maroon|      v_2|        2|newField_1|
  129.     |row_9|     part_0|      5|michael|      v_2|        2|newField_1|
  130.     +-----+-----------+-------+-------+---------+---------+----------+

Videos