--- title: Quarantine Data Sources meta: description: Quarantine data sources store data that doesn't fit the schema. --- # 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: ```shell tb sql "select * from _quarantine limit 10" ``` A sample output of the `tb sql` command is the following: ```text ────────────────────────────────────────────────────────────────── 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: ```shell 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: ```shell 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. ```text {% title="datasources/rows.datasource" %} 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. ```shell 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](#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](/forward/guides/evolve-data-source) 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. {% tabs initial="Tinybird CLI" %} {% tab label="Tinybird CLI" %} ```tinybird {% title="pipes/recover_rows_quarantine.pipe" %} 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 ``` {% /tab %} {% tab label="TypeScript SDK" %} ```typescript {% title="tinybird.ts" %} import { defineCopyPipe, node } from "@tinybirdco/sdk"; export const recoverRowsQuarantine = defineCopyPipe("recover_rows_quarantine", { description: "Recovers quarantined rows into the rows Data Source", datasource: rows, mode: "append", nodes: [ node({ name: "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 `, }), ], }); ``` {% /tab %} {% tab label="Python SDK" %} ```python {% title="tinybird.py" %} from tinybird_sdk import define_copy_pipe, node recover_rows_quarantine = define_copy_pipe("recover_rows_quarantine", { "description": "Recovers quarantined rows into the rows Data Source", "datasource": rows, "copy_mode": "append", "nodes": [ node({ "name": "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 """, }), ], }) ``` {% /tab %} {% /tabs %} ### Step 3: Deploy the schema fix and the Copy Pipe together Validate the deployment first: ```shell tb --cloud deploy --check ``` Then deploy. The schema fix and the Copy Pipe go live in the same deployment: ```shell tb --cloud deploy ``` {% callout type="caution" %} Wait until the deployment is live before you run the copy. Changes applied with `ALTER` aren't available in staging deployments, so a copy triggered before promotion doesn't write to the fixed schema. `tb deploy` promotes the deployment for you. {% /callout %} ### Step 4: Trigger the copy Run the Copy Pipe and wait for it to finish: ```shell tb --cloud copy run recover_rows_quarantine --wait ``` Check that the rows landed in the Data Source: ```shell tb --cloud sql "SELECT count() FROM rows" ``` For large recoveries, run the copy on a dedicated instance to avoid competing with your production workloads: ```shell tb --cloud copy run recover_rows_quarantine --wait --on-demand-compute ``` See [on-demand compute for copy jobs](/forward/core-concepts/copy-pipes#on-demand-compute-for-copy-jobs). ### Step 5: Remove the Copy Pipe Once the rows are recovered, delete the Copy Pipe and deploy again: ```shell 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.