dlt will evolve the schema at the destination by following the structure and data types of the extracted data. There are several modes
that you can use to control this automatic schema evolution, from the default modes where all changes to the schema are accepted to
a frozen schema that does not change at all.

Consider this example:

@dlt.resource(schema_contract={"tables": "evolve", "columns": "freeze"})
def items():
    ...

This resource will allow new tables (both nested tables and tables with dynamic names) to be created, but will throw an exception if data is extracted for an existing table which contains a new column.

Setting up the contract#

You can control the following schema entities:

• tables - the contract is applied when a new table is created
• columns - the contract is applied when a new column is created on an existing table
• data_type - the contract is applied when a data type property of an existing column changes. This includes variant columns (where data cannot be coerced into the existing type) as well as explicit changes to data_type, nullable, precision, scale, or timezone on a column that is already complete.

You can use contract modes to tell dlt how to apply the contract for a particular entity:

• evolve: No constraints on schema changes.
• freeze: This will raise an exception if data is encountered that does not fit the existing schema, so no data will be loaded to the destination.
• discard_row: This will discard any extracted row if it does not adhere to the existing schema, and this row will not be loaded to the destination.
• discard_value: This will discard data in an extracted row that does not adhere to the existing schema, and the row will be loaded without this data.

Passing the schema_contract argument#

The schema_contract exists on the dlt.source decorator as a default for all resources in that source and on the
dlt.resource decorator as a directive for the individual resource - and as a consequence - on all tables created by this resource.
Additionally, it exists on the pipeline.run() method, which will override all existing settings.

The schema_contract argument accepts two forms:

1. full: a mapping of schema entities to contract modes
2. shorthand: a contract mode (string) that will be applied to all schema entities.

For example, setting schema_contract to freeze will expand to the full form:

{"tables": "freeze", "columns": "freeze", "data_type": "freeze"}

You can change the contract on the source instance via the schema_contract property. For resource, you can use apply_hints.

Nuances of contract modes#

1. Contracts are applied after names of tables and columns are normalized.
2. A contract defined on a resource is applied to all root tables and nested tables created by that resource.
3. discard_row works on the table level. For example, if you have two tables in a nested relationship, i.e., users and users__addresses, and the contract is violated in the users__addresses table, the row of that table is discarded while the parent row in the users table will be loaded.

Use Pydantic models for data validation#

Pydantic models can be used to define table schemas and validate incoming data. You can use any model you already have. dlt will internally synthesize (if necessary) new models that conform to the schema contract on the resource.

Just passing a model in the column argument of the dlt.resource sets a schema contract that conforms to the default Pydantic behavior:

{
  "tables": "evolve",
  "columns": "discard_value",
  "data_type": "freeze"
}

New tables are allowed, extra fields are ignored, and invalid data raises an exception.

If you pass a schema contract explicitly, the following happens to schema entities:

1. tables do not impact the Pydantic models.
2. columns modes are mapped into the extra modes of Pydantic (see below). dlt will apply this setting recursively if models contain other models.
3. data_type supports the following modes for Pydantic: evolve will synthesize a lenient model that allows for any data type. This may result in variant columns upstream.
   freeze will re-raise ValidationException. discard_row will remove the non-validating data items.
   discard_value is not currently supported. We may eventually do that in Pydantic v2.

dlt maps column contract modes into the extra fields settings as follows.

Note that this works in two directions. If you use a model with such a setting explicitly configured, dlt sets the column contract mode accordingly. This also avoids synthesizing modified models.

discard_row requires additional handling when a ValidationError is raised.

Authoritative Pydantic models#

:::caution Work in progress
Authoritative Pydantic models are under active development. Behavior below may change in future releases.
:::

By default, columns derived from a Pydantic model are subject to schema contract checks just like any other columns. If you want the model to be treated as the authoritative source of truth — so its columns bypass columns and data_type contract enforcement — set is_authoritative_model in DltConfig:

from typing import ClassVar

from pydantic import BaseModel

from dlt.common.libs.pydantic import DltConfig

class MyModel(BaseModel):
    dlt_config: ClassVar[DltConfig] = {"is_authoritative_model": True}

    class Config:
        extra = "forbid"

    id: int
    name: str

With is_authoritative_model=True, if you later add an email field to MyModel, the new column will be accepted on existing tables even though extra=forbid maps to columns=freeze. The freeze contract will still reject any extra fields in the data that are not defined in the model.

Without is_authoritative_model (the default), adding a new field to the model while columns=freeze is active will raise a DataValidationError — the same as adding a column from any other data source.

Validating event streams with Pydantic#

When a single resource produces items of different types (e.g., an event stream), you can use a Pydantic discriminated union wrapped in a RootModel to validate and dispatch them. Each variant model defines its own columns, and dlt resolves the correct variant per item using the discriminator field — without re-validating.

Define variant models with a shared Literal discriminator field and combine them in a RootModel:

from typing import ClassVar, Literal, Union
from typing_extensions import Annotated
from pydantic import BaseModel, Field, RootModel
from dlt.common.libs.pydantic import DltConfig

class EventBase(BaseModel):
    kind: str
    id: int

class ClickEvent(EventBase):
    kind: Literal["click"]
    element_id: str

class PurchaseEvent(EventBase):
    kind: Literal["purchase"]
    amount: float

EventUnion = Annotated[
    Union[ClickEvent, PurchaseEvent],
    Field(discriminator="kind"),
]

class Event(RootModel[EventUnion]):
    dlt_config: ClassVar[DltConfig] = {"is_authoritative_model": True}

A common base class (like EventBase above) is optional. When present, dlt can derive the shared columns (kind, id) even without a specific data item.

Dispatch items to per-type tables using dlt.mark.with_table_name or a dynamic table_name function. The table names do not need to match the discriminator values:

import dlt

TABLE_MAP = {"click": "click_events", "purchase": "purchase_events"}

@dlt.resource(
    name="events",
    columns=Event,
    schema_contract={"data_type": "discard_row"},
)
def event_stream():
    for item in items:
        yield dlt.mark.with_table_name(item, TABLE_MAP[item["kind"]])

Each destination table receives only the columns defined by its variant model. For example, the click_events table gets kind, id, and element_id, while purchase_events gets kind, id, and amount.

Items not matching any variant#

When an item's discriminator value does not match any variant in the union (e.g., a "debug" event when only "click" and "purchase" are defined), the data_type contract controls what happens:

Set contracts on Arrow tables, Pandas, and Polars#

All contract settings apply to Arrow tables, Pandas or Polars DataFrames as well.

1. tables mode is the same - no matter what the data item type is.
2. columns will allow new columns, raise an exception, or modify tables/frames still in the extract step to avoid rewriting Parquet files.
3. data_type applies to variant columns and to changes in type properties (data_type, nullable, precision, scale, timezone) on existing complete columns. The contract will raise an exception, discard the column, or discard the row depending on the mode.

Here's how dlt deals with column modes:

1. evolve new columns are allowed (the table may be reordered to put them at the end).
2. discard_value the column will be deleted.
3. discard_row rows with the column present will be deleted and then the column will be deleted.
4. freeze an exception on a new column.

Get context from DataValidationError in freeze mode#

When a contract is violated in freeze mode, dlt raises a DataValidationError exception. This exception provides access to the full context and passes the evidence to the caller.
As with any other exception coming from a pipeline run, it will be re-raised via a PipelineStepFailed exception, which you should catch in an except block:

try:
  pipeline.run()
except PipelineStepFailed as pip_ex:
  if pip_ex.step == "normalize":
    if isinstance(pip_ex.__context__.__context__, DataValidationError):
      ...
  if pip_ex.step == "extract":
    if isinstance(pip_ex.__context__, DataValidationError):
      ...

DataValidationError provides the following context:

1. schema_name, table_name, and column_name provide the logical "location" at which the contract was violated.
2. schema_entity and contract_mode indicate which contract was violated.
3. table_schema contains the schema against which the contract was validated. It may be a Pydantic model or a dlt TTableSchema instance.
4. schema_contract is the full, expanded schema contract.
5. data_item is the causing data item (Python dict, arrow table, Pydantic model, or list thereof).

Contracts on new tables#

If a table is a new table that has not been created on the destination yet, dlt will allow the creation of new columns. For a single pipeline run, the column mode is changed (internally) to evolve and then reverted back to the original mode. This allows for initial schema inference to happen, and then on subsequent runs, the inferred contract will be applied to the new data.

The following tables are considered new:

1. Child tables inferred from nested data.
2. Dynamic tables created from the data during extraction.
3. Tables containing incomplete columns - columns without a data type bound to them.

For example, such a table is considered new because the column number is incomplete (defined as primary key and NOT null but no data type):

blocks:
  description: Ethereum blocks
  write_disposition: append
  columns:
    number:
      nullable: false
      primary_key: true
      name: number

Tables that are not considered new:

1. Those that already exist in the schema with at least one complete column (a column with a data_type).

Working with datasets that have manually added tables and columns on the first load#

In some cases, you might be working with datasets that have tables or columns created outside of dlt. If you are loading to a table not created by dlt for the first time, dlt will not know about this table while enforcing schema contracts. This means that if you do a load where the tables are set to evolve, all will work as planned. If you have tables set to freeze, dlt will raise an exception because it thinks you are creating a new table (which you are from dlt's perspective). You can allow evolve for one load and then switch back to freeze.

The same thing will happen if dlt knows your table, but you have manually added a column to your destination and you have columns set to freeze.

Code examples#

The below code will silently ignore new subtables, allow new columns to be added to existing tables, and raise an error if a variant of a column is discovered.

@dlt.resource(schema_contract={"tables": "discard_row", "columns": "evolve", "data_type": "freeze"})
def items():
    ...

The below code will raise an error on any encountered schema change. Note: You can always set a string which will be interpreted as though all keys are set to these values.

pipeline.run(my_source, schema_contract="freeze")

The below code defines some settings on the source which can be overwritten on the resource, which in turn can be overwritten by the global override on the run method.
Here, for all resources, variant columns are frozen and raise an error if encountered. On items, new columns are allowed, but other_items inherits the freeze setting from
the source, thus new columns are frozen there. New tables are allowed.

@dlt.resource(schema_contract={"columns": "evolve"})
def items():
    ...

@dlt.resource()
def other_items():
    ...

@dlt.source(schema_contract={"columns": "freeze", "data_type": "freeze"})
def frozen_source():
  return [items(), other_items()]

# this will use the settings defined by the decorators
pipeline.run(frozen_source())

# this will freeze the whole schema, regardless of the decorator settings
pipeline.run(frozen_source(), schema_contract="freeze")