Quarantine Data Sources¶

Every data source in your workspace has an associated quarantine data source that stores data that doesn't fit the schema. If you send rows that don't match the data source schema, they're automatically sent to the quarantine table so that the ingest process doesn't fail.

By convention, quarantine data sources follow the naming pattern {datasource_name}_quarantine. You can review quarantined rows at any time or perform operations on them using Pipes. This is a useful source of information when fixing issues in the origin source or applying changes during ingest.

Data in quarantine data sources is retained for 1 month, after which it's automatically deleted.

Review quarantined data¶

To check your quarantine data sources, run the tb sql command. For example:

tb sql "select * from <datasource_name>_quarantine limit 10"

A sample output of the tb sql command is the following:

──────────────────────────────────────────────────────────────────
c__error_column: ['abslevel']
c__error: ["value '' on column 'abslevel' is not Float32"]
c__import_id: 01JKQPWT8GVXAN5GJ1VBD4XM27
day: 2014-07-30
station: Embassament de Siurana (Cornudella de Montsant)
volume: 11.57
insertion_date: 2025-02-10 10:36:20
──────────────────────────────────────────────────────────────────

The quarantine data source schema contains the columns of the original row and the following columns with information about the issues that caused the quarantine:

• c__error_column Array(String) contains an array of all the columns that contain an invalid value.
• c__error Array(String) contains an array of all the errors that caused the ingestion to fail and led to the row being stored in the Quarantine Data Source. Use this column with c__error_column to identify which columns have problems and their specific errors.
• c__import_id Nullable(String) contains the job's identifier in case the column was imported through a job.
• insertion_date (DateTime) contains the timestamp in which the ingestion was done.

Fixing quarantined data example¶

Using the Electric Vehicle Population Data example:

tb datasource create --url "https://data.wa.gov/api/views/f6w7-q2d2/rows.csv?accessType=DOWNLOAD" --name rows
tb build
tb datasource append rows "https://data.wa.gov/api/views/f6w7-q2d2/rows.csv?accessType=DOWNLOAD"

You should get the following quarantine error: Error appending fixtures for 'rows': There was an error with file contents: 564 rows in quarantine.

Inspecting the rows_quarantine data source:

tb sql "SELECT DISTINCT c__error FROM rows_quarantine"

# ────────────────────────────────────────────────────────────────────────────────────────────
# c__error: ["value '' on column 'postal_code' is not Int64", "value '' on column 'legislative_district' is not Int16", # "value '' on column 'c_2020_census_tract' is not Int64"]
# ────────────────────────────────────────────────────────────────────────────────────────────
# c__error: ["value '' on column 'electric_range' is not Int32", "value '' on column 'base_msrp' is not Int64"]
# ────────────────────────────────────────────────────────────────────────────────────────────
# c__error: ["value '' on column 'legislative_district' is not Int16"]
# ────────────────────────────────────────────────────────────────────────────────────────────

The problem is that some columns should be Nullable or have a DEFAULT value. Add a DEFAULT value of 0 for them.

Edit the datasources/rows.datasource file.

DESCRIPTION >
    Generated from https://data.wa.gov/api/views/f6w7-q2d2/rows.csv?accessType=DOWNLOAD

SCHEMA >
    `vin__1_10_` String,
    `county` String,
    `city` String,
    `state` String,
    `postal_code` Int64 DEFAULT 0,
    `model_year` Int32,
    `make` String,
    `model` String,
    `electric_vehicle_type` String,
    `clean_alternative_fuel_vehicle__cafv__eligibility` String,
    `electric_range` Int32 DEFAULT 0,
    `base_msrp` Int64 DEFAULT 0,
    `legislative_district` Int16 DEFAULT 0,
    `dol_vehicle_id` Int64,
    `vehicle_location` String,
    `electric_utility` String,
    `c_2020_census_tract` Int64 DEFAULT 0

The dev server rebuilds the edited resources.

tb build
tb datasource append rows "https://data.wa.gov/api/views/f6w7-q2d2/rows.csv?accessType=DOWNLOAD"

No errors now, you're good to continue developing.

Recover data from quarantine¶

If you have data in quarantine, the first thing to do is understand what caused the error and fix it. See Fixing quarantined data example.

Once the schema issues are fixed, you can recover the quarantined data back to your main data source. The approach involves deploying the schema fix and a Copy Pipe that reads from quarantine at the same time, then triggering the copy to recover the data.

Step 1: Fix the Data Source schema¶

Edit the Data Source definition so the quarantined rows become valid. Most quarantine issues come down to a wrong data type or a missing column, which you fix by changing the type, making the column nullable, adding a DEFAULT value, or adding and dropping columns.

Make only the minimum changes required to fix the schema, and leave anything unrelated for a later deployment.

Continuing the previous example, datasources/rows.datasource now has DEFAULT 0 on the columns that were quarantining rows. See Evolve data sources for how Tinybird applies each type of change.

Step 2: Add a Copy Pipe that reads from quarantine¶

Add a Copy Pipe that selects the quarantined rows, casts them to the fixed schema, and appends them to the Data Source. Because quarantine columns are Nullable(String), cast every column to its target type, and don't select the c__error, c__error_column, c__import_id, and insertion_date columns.

The Copy Pipe has no schedule, so it only runs when you trigger it.

DESCRIPTION Recovers quarantined rows into the rows data source

NODE recover
SQL >
    SELECT
        coalesce(vin__1_10_, '') AS vin__1_10_,
        coalesce(county, '') AS county,
        coalesce(city, '') AS city,
        coalesce(state, '') AS state,
        coalesce(toInt64OrNull(postal_code), 0) AS postal_code,
        coalesce(toInt32OrNull(model_year), 0) AS model_year,
        coalesce(make, '') AS make,
        coalesce(model, '') AS model,
        coalesce(electric_vehicle_type, '') AS electric_vehicle_type,
        coalesce(clean_alternative_fuel_vehicle__cafv__eligibility, '') AS clean_alternative_fuel_vehicle__cafv__eligibility,
        coalesce(toInt32OrNull(electric_range), 0) AS electric_range,
        coalesce(toInt64OrNull(base_msrp), 0) AS base_msrp,
        coalesce(toInt16OrNull(legislative_district), 0) AS legislative_district,
        coalesce(toInt64OrNull(dol_vehicle_id), 0) AS dol_vehicle_id,
        coalesce(vehicle_location, '') AS vehicle_location,
        coalesce(electric_utility, '') AS electric_utility,
        coalesce(toInt64OrNull(c_2020_census_tract), 0) AS c_2020_census_tract
    FROM rows_quarantine

TYPE COPY
TARGET_DATASOURCE rows
COPY_SCHEDULE @on-demand
COPY_MODE append

Step 3: Deploy the schema fix and the Copy Pipe together¶

Validate the deployment first:

tb --cloud deploy --check

Then deploy. The schema fix and the Copy Pipe go live in the same deployment:

tb --cloud deploy

Step 4: Trigger the copy¶

Run the Copy Pipe and wait for it to finish:

tb --cloud copy run recover_rows_quarantine --wait

Check that the rows landed in the Data Source:

tb --cloud sql "SELECT count() FROM rows"

For large recoveries, run the copy on a dedicated instance to avoid competing with your production workloads:

tb --cloud copy run recover_rows_quarantine --wait --on-demand-compute

See on-demand compute for copy jobs.

Step 5: Remove the Copy Pipe¶

Once the rows are recovered, delete the Copy Pipe and deploy again:

tb --cloud deploy

This deployment doesn't change the Data Source schema, so it doesn't affect the Quarantine Data Source or its remaining rows.