COPY INTO and Auto Loader: Incremental File Loading into Unity Catalog Tables (Databricks Data Engineer Associate)

COPY INTO is a SQL command that loads files from a cloud storage path or Unity Catalog volume into a Delta table and skips every file it has already loaded into that table. Because of that file tracking, the command is retriable and idempotent: a rerun, whether scheduled or accidental, simply picks up the files that arrived since the last run and appends their rows; rows from earlier loads are left as they are.

COPY INTO main.bronze.orders
FROM '/Volumes/main/raw/orders/'
FILEFORMAT = CSV
FORMAT_OPTIONS ('header' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true');

Three details of the file tracking decide most scenarios:

• History lives with the target table. The record of loaded files belongs to the Delta table being written. A second, new table loading the same path starts with an empty history and loads every file on its first run; nothing is shared between tables and nothing is keyed on the source path or the table's creation time.
• A modified file is still skipped. If a file with an already-loaded name is overwritten with new content, COPY INTO does not load it again; a newer modification time changes nothing.
• Deleting rows does not reset the history. Truncating the table or deleting rows leaves the record of loaded files in place, so the next normal run inserts zero rows.

COPY INTO suits SQL users loading thousands of files on a schedule. Databricks now marks it as a legacy path in places and recommends streaming tables over read_files for new SQL pipelines (see the lesson on ingestion patterns), but the exam guide still lists COPY INTO, and its behaviour is what this lesson teaches.

To load files that COPY INTO has already loaded, add COPY_OPTIONS ('force' = 'true'): it disables idempotency and loads every selected file regardless of history. That is the only switch that reloads; mergeSchema, inferSchema, parser modes and VALIDATE do not change which files count as loaded.

Because force ignores history on every run, it belongs in a one-off statement, not in a scheduled job. Left in a nightly job, it reloads the whole source folder each night and the table fills with repeated rows; removing the option restores skipping. Parser modes deal with unparsable records, not with duplicates, so they are no defence against a stray force.

Correcting a file that was replaced at the source

When a supplier overwrites an already-loaded file with corrected data under the same name, the normal job skips it. The clean fix has two steps: delete the old rows for that file (a column populated from _metadata.file_path at load time identifies them), then reload only that file with FILES = ('<name>') plus force. Scope the force to the one file: force on the whole path reloads everything under it, and the stale rows still have to be removed explicitly.

COPY INTO reads every file under the source path unless you narrow it with one of two mutually exclusive clauses: PATTERN, a glob over file names, or FILES, an explicit list of file names. A statement cannot use both.

Neither clause chooses columns. Picking and renaming columns is the job of a SELECT in the FROM clause (next section), and the optional BY POSITION target clause maps headerless CSV columns to table columns by position; it does not filter files either.

The FROM clause of COPY INTO can be a SELECT over the source path, so simple shaping happens as the files load: choosing a subset of columns, renaming with aliases, casting types and adding file metadata.

COPY INTO main.bronze.orders
FROM (
  SELECT cust_no AS customer_id,
         order_id,
         CAST(order_ts AS TIMESTAMP) AS order_ts,
         _metadata.file_path AS source_file
  FROM '/Volumes/main/raw/orders/'
)
FILEFORMAT = CSV
FORMAT_OPTIONS ('header' = 'true');

• File metadata: the hidden _metadata column exposes fields such as file_path, file_name and file_modification_time. It replaces input_file_name(), which is deprecated and not supported on Unity Catalog standard access mode.
• Aggregation: global aggregates (for example a single SUM over everything loaded) are allowed, but GROUP BY is not supported. When the business needs totals per group, load the raw rows with COPY INTO, which keeps exactly-once file tracking, and aggregate in a separate query or materialized view over the bronze table. Options such as force or mergeSchema do not change what the source query may contain.

COPY INTO has one option name, mergeSchema, in two places with two different jobs, and a VALIDATE clause that checks a load without writing it.

A new source field becomes a table column automatically only with the write-side setting. Since Databricks Runtime 11.3 LTS the target can also be a schemaless placeholder, created with CREATE TABLE IF NOT EXISTS main.raw.vendor_feed; and no column list. COPY INTO then derives the columns from the data, which needs both settings when files differ: read-side mergeSchema to union the columns from every file, and write-side mergeSchema to let the empty table take that schema.

VALIDATE: check without loading

VALIDATE (Databricks Runtime 10.4 LTS and above) runs the load's checks and returns a preview of rows, but writes nothing. It checks that the data parses, that its schema matches the table or could evolve to it, and that NOT NULL and CHECK constraints hold. VALIDATE ALL (the default when no count is given) checks all the data; VALIDATE 10 ROWS checks only that many rows, so a bad row further in can be missed. A validation run commits nothing to the table, so it records no files as loaded: the real run afterwards, without VALIDATE and without force, loads every file. Parser modes such as FAILFAST only change how a real, writing load treats malformed records.

COPY INTO into a Unity Catalog table reads its source through one of three routes, and each needs a different authorisation. Databricks recommends volumes.

COPY INTO main.audit.findings
FROM 's3://auditor-share/batch-07/' WITH (
  CREDENTIAL (AWS_ACCESS_KEY = '...', AWS_SECRET_KEY = '...',
              AWS_SESSION_TOKEN = '...')
)
FILEFORMAT = JSON;

On Azure the inline credential is AZURE_SAS_TOKEN. SELECT is a table and view privilege and does not apply to volumes. Legacy workarounds such as DBFS mounts or keys in a cluster's Spark configuration sit outside Unity Catalog governance and expose the credentials to other users; an external volume is not an option either when no external location exists, because a volume must live inside one. The target table needs its own privileges as usual.

Auto Loader is the cloudFiles Structured Streaming source: it discovers new files in cloud storage or a volume and processes each one once, tracking them in the stream's checkpoint (checkpoints are covered in the ingestion patterns lesson). It can infer the table's schema and evolve it as files change.

(spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", "/Volumes/main/raw/_meta/orders")
  .option("cloudFiles.inferColumnTypes", "true")
  .option("cloudFiles.schemaHints", "unit_price DECIMAL(10,2)")
  .load("/Volumes/main/raw/orders/")
  .writeStream
  .option("checkpointLocation", "/Volumes/main/raw/_meta/orders_ckpt")
  .option("mergeSchema", "true")
  .trigger(availableNow=True)
  .toTable("main.bronze.orders"))

• cloudFiles.schemaLocation is what enables inference and evolution: Auto Loader writes each schema version to a _schemas folder under it and reads it back on restart instead of inferring from scratch.
• Strings by default: for JSON, CSV and XML, inference types every column (including nested JSON fields) as STRING, to avoid type mismatches as files change. Set cloudFiles.inferColumnTypes to true to detect types. Parquet and Avro always use the types encoded in the files.
• Sampling: inference reads a sample of the first files (by default up to 50 GB or 1,000 files). It does not retype columns later as more data arrives.
• cloudFiles.schemaHints takes SQL schema syntax for a few named columns and fixes their types while everything else is still inferred, for example a decimal where early files held only whole numbers. Passing a full schema with .schema() instead turns inference off for every column.

cloudFiles.schemaEvolutionMode decides what Auto Loader does when files contain fields the schema does not have. Its default depends on whether the schema is inferred or provided.

The addNewColumns failure is by design: production jobs recover with task retries on the Lakeflow Jobs task (Lakeflow Spark Declarative Pipelines restart automatically), and the Delta write needs mergeSchema so the new column can land. Never "fix" it with a fresh checkpoint, which makes every existing file look new. addNewColumns never changes the type of an existing column; a preview mode, addNewColumnsWithTypeWidening, also widens types. When columns may change only after human review but ingestion must not stop or lose data, rescue is the fit.

What lands in _rescued_data

When Auto Loader infers a schema it adds a _rescued_data column by default. It captures, as a JSON blob that includes the source file path:

• fields missing from the schema;
• values whose type does not match (text in a numeric column, which would otherwise arrive as null);
• fields whose name differs only by case (CustomerID against customerId), which are rescued rather than added as a new column.

With a provided schema, rescue is off until you set the rescuedDataColumn option; that is how silently dropped values (new fields in none mode, or type mismatches) are caught without changing the table's columns. The rescued column cannot hold corrupt records that are not valid JSON at all; those are what _corrupt_record and the parser modes deal with. With rescue enabled, a type mismatch is no longer malformed, so FAILFAST does not fail on it and DROPMALFORMED does not drop it, but a cut-off, unparsable line still fails a FAILFAST stream.

Auto Loader finds new files in one of two ways. Directory listing is the default: it lists the input path and needs only read access to the storage. File notification mode reacts to storage events instead of listing, which keeps discovery cheap when directories hold millions of files.

When legacy notifications hit the per-bucket limit or the cloud team refuses per-stream queues, the answer is managed file events on the external location. A notification queue (cloudFiles.queueUrl) belongs to one stream: each message is consumed by a single reader, so a queue cannot be shared between streams. cloudFiles.backfillInterval adds periodic full listings to catch files that notifications missed; incremental listing (cloudFiles.useIncrementalListing) is deprecated; and cloudFiles.maxFilesPerTrigger changes batch size, not listing cost. A Lakeflow Jobs file arrival trigger is a different thing: it starts job runs when files land, rather than being how a stream discovers files.

To run Auto Loader as a scheduled batch, use trigger(availableNow=True): the run processes every file that arrived before the query started, in as many micro-batches as the rate limits require, then stops so the job cluster can terminate.

The default rate limit is 1,000 files per micro-batch (cloudFiles.maxFilesPerTrigger; cloudFiles.maxBytesPerTrigger caps bytes). A backlog of 2,500 files is therefore processed in several micro-batches in one run. A file that lands after the run started is not part of that run, but it is not marked as processed either: the next run picks it up as new.

File-handling options

Choosing between COPY INTO, Auto Loader and streaming tables by volume and frequency, and triggered versus continuous scheduling in general, are covered in the lesson on ingestion patterns and choosing an ingestion method.