Loading S3 Parquet into Redshift — `JdbcTargetParquet` + `CopyIntoRS`¶

This guide covers two steps that form a producer → consumer pipeline:

JdbcTargetParquet extracts a source table to Parquet and records a reusable schema profile describing the columns.
CopyIntoRS reads that Parquet from S3 and bulk-loads it into a Redshift temp table with COPY, feeding a downstream Table step.

They're usually two separate jobs: a producer job lands the Parquet on S3, and a consumer Redshift job copies it in. The schema profile is the hand-off — the producer writes it, the consumer reads it to build a matching Redshift table.

Producer job:   Jdbcsource ─► JdbcTargetParquet ─► Push File ─► S3 (parquet + schema profile in Postgres)
Consumer job:                              CopyIntoRS ─► Table (Redshift)
                                           (reads S3 parquet using the profile)

Prerequisites

An AWS S3 connection (commonly named S3_CONNECTION) with its bucket configured.
For IAM auth, the Redshift cluster's IAM role must allow s3:ListBucket on the bucket and s3:GetObject on the objects. See Redshift setup.
The producer job must have run at least once so its schema profile exists for CopyIntoRS to browse.

Part 1 — `JdbcTargetParquet` (the producer)¶

Type JDBCTARGETPARQUET · Engine org.maestro.engine.steps.pipeline.core.JdbcTargetParquet · WPF TRANSFORM menu (Redshift + ClickHouse jobs) · requires an upstream Jdbcsource

What it does¶

Downstream of a Jdbcsource, it reads the source's staged PSV, runs a Spark projection, writes local snappy-compressed Parquet, and then:

Appends a maintenance column eltm_txid to every row — the load id, equal to $BATCH_RUN_NUMBER for that run. It rides along in the Parquet (and downstream) so a target can order/dedupe by load.
Base64-decodes the columns the source flagged during extraction, using Spark's native unbase64 (so it works regardless of the job's target vendor).
Saves a schema profile to Postgres (see below) describing the output columns.

The Parquet is written locally; a following Push File step uploads it to S3 (or Azure Blob) for the consumer job to read.

Column count

The Parquet has N source columns + 1 eltm_txid. This matters for the consumer: a Redshift COPY … FORMAT AS PARQUET requires the target table to have the same column count as the file. The schema profile records all N+1 columns so the consumer builds a matching table.

The schema profile¶

After the write, JdbcTargetParquet persists a schema profile — a platform-neutral description of the output columns — to t_sql_profile_table / t_sql_profile_column in the audit DB. It is keyed by:

Field	Value
`profile_name`	`jobName.connection.catalog.schema.table`
per column	name, canonical `java.sql.Types` id (`12\\|VARCHAR`, `-5\\|BIGINT`, …), measured length, scale

The profile includes the eltm_txid column (as BIGINT), so its column count matches the Parquet. Sizes are measured from the written Parquet (not declared metadata) and tracked grow-only across runs. See Schema profile for the storage model, functions, and roadmap.

Inspect the latest profile:

select p.profile_name, c.column_order, c.column_name, c.column_data_type,
       c.column_length, c.column_decimal_digits
  from v_sql_profile_column c
  join v_sql_profile_table  p on p.profile_id = c.profile_id
 where c.profile_id = (select max(profile_id) from t_sql_profile_table)
 order by c.column_order;

Part 2 — `CopyIntoRS` (the consumer)¶

Type COPYINTORS · Engine org.maestro.engine.steps.redshift.core.CopyIntoRS · WPF Steps/Redshift/CopyIntoRS.cs → CopyIntoRSWindow.xaml ("Copy Into Redshift")

Add it from the job-editor palette under the REDSHIFT ▸ DATA SOURCE menu. It's a source step — it has no input arrow; it produces the loaded temp table for a downstream Table step.

What it does at runtime¶

Creates a Redshift temp table from the step's columns (already cast to Redshift types).
Runs COPY <temp table> FROM 's3://<bucket>/<objectPath>' <credentials> <format>.
Exposes the temp table to the next step (typically a Table that upserts into the final table).

The bucket comes from the S3 connection (not the path), and the credentials + format are built from the fields below.

Configuring the step¶

Field	What to enter
S3 Connection	An AWS S3 connection (e.g. `S3_CONNECTION`). The bucket is taken from here.
S3 Object Path	The object key relative to the bucket — e.g. `exports/ProductPhoto/2026-07-05`. No `s3://`, no bucket name. This is the same value a `Push File` step would use as its destination. It may instead be a watermark reference `$WM{<exportJob>::<pushStepId>}` (see Run-folder dedupe) — the engine resolves it to the exact folder the export's `Push File` step last wrote.
Format	`csv` · `psv` · `tsv` · `parquet`. Selecting a format re-seeds Format Options (below).
Format Options	Extra `COPY` options. For `csv/psv/tsv` this pre-fills `ignoreheader 1` + the matching `delimiter`; for `parquet` the box is cleared and disabled (Redshift ignores options for Parquet).
Authentication Mode	IAM (uses the connection's `iam_role`) or KEY (uses `aws_access_key_id` / `aws_secret_access_key`).
Columns	The Redshift-typed columns to create + load. Use Browse Profile to populate them.

Populating columns with Browse Profile¶

Click Browse Profile…, search by profile name, and pick the profile the producer job wrote (e.g. JDBC2PRQ.ADVENTUREWORKS.AdventureWorks2022.Production.ProductPhoto). The step loads every column — including eltm_txid — already cast to Redshift types. This is what keeps the temp table's column count matching the Parquet.

Re-browse after the producer changes

The columns are saved into the job when you browse. If the producer's schema changes (new column, or a first run that populates the profile), re-run the producer job to refresh the profile, then re-open CopyIntoRS, Browse Profile again, and Save.

Re-browse only for a change to the set of columns (added / removed / renamed). A column that merely grew wider than its saved width no longer needs a re-browse — the engine sizes the load to the data automatically (see Automatic column sizing).

The COPY it generates¶

-- IAM auth, parquet:
copy "<schema>"."TMP_..." from 's3://<bucket>/<objectPath>'
     iam_role '<role from connection>' FORMAT AS PARQUET;

-- KEY auth, csv/psv/tsv (with your Format Options appended):
copy "<schema>"."TMP_..." from 's3://<bucket>/<objectPath>'
     credentials 'aws_access_key_id=...;aws_secret_access_key=...'
     FORMAT AS CSV
     ignoreheader 1
     delimiter '|';

$bucket.name, $iam.role, and the access/secret keys are resolved from the S3 connection at runtime.

Automatic column sizing¶

CopyIntoRS sizes string columns to the data at load time, so an under-stated width in the saved profile can never fail the COPY:

The transient temp table is created with every varchar/char column at Redshift's maximum width (varchar(65532)), so COPY … FORMAT AS PARQUET cannot overflow — regardless of what the saved profile said.
Immediately after the COPY, the step measures each string column's actual maximum width (max(octet_length(...))) in one pass and narrows the temp table to that measured width.

The downstream Table step derives the persistent table's schema from that right-sized temp, so the target table lands at the measured width — not the max. The result: a load that survives a stale/narrow profile and a persistent table sized to the real data (no varchar(65532) bloat, which would inflate Redshift's query-planning memory). It is best-effort — if the measurement can't run, the column stays at max width and the load still succeeds.

Engine build

Automatic column sizing is in engine 17.0.4.446 and later. On older builds an under-sized profile width surfaces as Spectrum Scan Error 15007 — see Troubleshooting.

Part 3 — Generate whole pipelines in bulk¶

Building the producer and consumer by hand for many tables is tedious. Two wizards on the JobEditor canvas (right-click ▸ Generate Pipeline, Redshift jobs only) build fully-configured, runnable pipelines for a whole list of tables at once:

Menu item	Builds, per selected table	Nodes
Export Jdbc To S3 Parquet	`Jdbcsource ─► JdbcTargetParquet ─► Push File`	3
Import S3 Parquet To Redshift	`CopyIntoRS ─► Table`	2

So 5 tables → 15 export nodes, or 10 import nodes. Each generated step is filled in and immediately valid (no dialogs to open) — nodes self-add and auto-connect on the canvas.

Export Jdbc To S3 Parquet¶

Field	What it does
Source Connection + Browse Schema…	The source JDBC connection and its catalog/schema.
Tables	Multi-select (with search / Select All). One 3-node row is generated per table.
S3 Connection	The AWS S3 connection the `Push File` steps use.
Path Template	The destination key. Generate-time tokens `$DATABASE` `$SCHEMA` `$TABLE` are substituted now; runtime `$`-vars (`$CURRENT_DATE`, `$BATCH_RUN_NUMBER`, …) are left for the engine to resolve each run. Default: `pipeline/$DATABASE/$TABLE/$CURRENT_DATE/$BATCH_RUN_NUMBER/`.
Dedupe re-runs	Enables the `Push File` step watermark so each run's folder is recorded — see below. On by default.

Columns are fetched per table and the Parquet transit columns are typed in the SparkSQL dialect (matching how JdbcTargetParquet writes the file).

Import S3 Parquet To Redshift¶

Field	What it does
Export Job + Load Pipelines	Pick the export job that produced the Parquet. The wizard parses its XML and lists every table it pushes, with the S3 path.
Tables	Multi-select the tables to ingest.
S3 Connection + IAM / Access Key	The connection + credential mode for the `COPY`.
Target DB / Schema	Database defaults to `$SYSTEM_DEFAULT_DATABASE` — the engine resolves it to the run connection's current database, so the workflow migrates across connections without a hardcoded DB. Only the schema is required.
Create target table	`CREATE IF NOT EXISTS` the target (it persists across runs). Uncheck to load into an existing table.
Use First Column as Key	Marks the first column as the key and turns the `Table` step into upsert mode (delete-by-key + insert) so re-runs merge instead of duplicate.

Column types are derived to match the Parquet physical types Spark wrote (e.g. smallint→integer, decimal→double precision, text→varchar(<measured length>)), because Redshift COPY … FORMAT AS PARQUET is strict about type matching.

Step-by-step¶

A. Generate the export pipeline

Open (or create) a Redshift job in the Job Editor.
Right-click the canvas ▸ Generate Pipeline ▸ Export Jdbc To S3 Parquet.
Pick the Source Connection, then Browse Schema… to set the catalog/schema (or type them).
Click Load Tables, then filter / multi-select the tables (or Select All).
Pick the S3 Connection. Adjust the Path Template if you like — keep the trailing $BATCH_RUN_NUMBER so re-runs get their own folder.
Leave Dedupe re-runs checked, then click Generate. One Jdbcsource ─► JdbcTargetParquet ─► Push File row appears per table.
Save / Deploy the job and run it. Each table's Parquet lands under …/<run>/, and the Push File step records that folder to its watermark (<exportJob>::<stepId>).

B. Generate the import pipeline

Open (or create) another Redshift job.
Right-click the canvas ▸ Generate Pipeline ▸ Import S3 Parquet To Redshift.
Choose the Export Job from step A and click Load Pipelines — the wizard parses it and lists each table with its S3 path.
Multi-select the tables to ingest.
Pick the S3 Connection + IAM / Access Key. Leave Target Database as $SYSTEM_DEFAULT_DATABASE (migratable); set the Target Schema (required).
Leave Create target table checked. Check Use First Column as Key if you want re-runs to upsert (merge on the first column) instead of append.
Click Generate. One CopyIntoRS ─► Table row appears per table; each CopyIntoRS's S3 Object Path is the reference $WM{<exportJob>::<stepId>}.
Save / Deploy and run. Every CopyIntoRS resolves the watermark to the export's latest run folder and COPYs exactly that; the Table step then upserts (or appends) into the final table.

Re-running

Re-run the export whenever the source changes — it writes a new …/<run>/ folder and repoints the watermark. The import always follows the watermark to that latest folder, so it never double-loads earlier runs. With Use First Column as Key, re-running the import merges on the key (idempotent — the target row count stays constant across re-runs) rather than duplicating.

Run-folder dedupe (watermark handoff)¶

Problem. If the export runs more than once for the same $CURRENT_DATE, its Parquet piles up in the same date folder. A consumer that COPYs the whole prefix then double-counts every re-run.

Solution. Give each run its own folder and hand the exact folder to the importer:

The export path ends with $BATCH_RUN_NUMBER (default template), so run 5115 writes pipeline/…/2026-07-07/5115/, run 5116 writes …/5116/, etc. — never colliding.
With Dedupe re-runs on, each Push File step (isWatermarkEnabled) records the exact folder it wrote on success in t_step_watermark, keyed by JOB_NAME::STEP_ID.
The generated CopyIntoRS S3 Object Path is a reference token $WM{<exportJob>::<pushStepId>}. At import time the engine reads that step's watermark and COPYs exactly that folder — the export's latest successful run, regardless of the import's own $BATCH_RUN_NUMBER.

Export run 5116  →  writes   pipeline/AdventureWorks2022/Address/2026-07-07/5116/
                    records  RS_DEMO_JDBC_2_S3::8 = pipeline/AdventureWorks2022/Address/2026-07-07/5116/
Import (any run) →  COPY from $WM{RS_DEMO_JDBC_2_S3::8}  →  resolves to  …/5116/   (only the latest)

Why JOB_NAME::STEP_ID?

Step IDs are stable and unique within a job (names can be renamed or duplicated), so a cross-job reference by ID is unambiguous. The producer watermark uses a separate key (JOB::STEP_ID) from the file-scanner incremental watermark (getWatermarkName()), so the two never interfere.

Run the export first

A $WM{…} reference with no recorded watermark yet fails clearly ("Run the source export job first") rather than guessing a folder — so an import can never silently read the wrong run.

Inspect the recorded folders:

select watermark_name, watermark_value
  from public.t_step_watermark
 where watermark_name like '<EXPORT_JOB>::%'
 order by watermark_name;

End-to-end walkthrough¶

Producer job (job type ClickHouse or Redshift):

Jdbcsource — pick the source connection + table.
JdbcTargetParquet — connect the source into it. It writes Parquet + the schema profile.
Push File — connect it after the Parquet step; set the S3 connection + destination key (e.g. exports/ProductPhoto/). Run the job. The Parquet lands on S3; the profile lands in Postgres.

Consumer job (job type Redshift):

CopyIntoRS — set the S3 connection, the object path matching where the producer pushed, the format (parquet), IAM or KEY, then Browse Profile and pick the producer's profile → Save.
Table — connect CopyIntoRS into it, set the target schema/table, map the columns (Reset to auto-map), and choose the write mode (e.g. upsert/CDC). Run the job.

On success the temp table is created, the Parquet is COPY'd in, and the Table step upserts into the final Redshift table.

Troubleshooting¶

Symptom	Cause & fix
`COPY` fails "Unmatched number of columns between table and file. Table columns: N, Data columns: N+1"	The profile has fewer columns than the Parquet (historically it omitted `eltm_txid`). Re-run the producer to regenerate the profile, then re-Browse Profile in CopyIntoRS and Save. (Engine now records `eltm_txid` in the profile.)
`COPY` fails "Spectrum Scan Error … code 15007 … length of the data column … is longer than the length defined in the table"	A saved profile width is narrower than the actual Parquet data. Fixed in engine 17.0.4.446+ — the load auto-sizes to the data (Automatic column sizing). On older builds, re-run the producer to refresh the profile then re-Browse Profile and Save, or upgrade the engine.
Target INSERT fails "column \"eltm_txid\" of relation … does not exist"	The target table pre-existed without the column. The `Table` step's metadata-assertion self-heals it (ALTER ADD) — make sure the engine is current, then re-run.
`eltm_txid` loads as NULL	Producer wrote the column header but not its value. Fixed in the engine (the extract now emits `$BATCH_RUN_NUMBER`); re-run the producer to rewrite the Parquet.
Deploy/Save fails "Not authorized, only job/mapping owners can perform this operation"	You're not logged in as the job's owner. Sign in as the owner, or Save As a new job name (you become its owner).
`COPY` fails on S3 `ListBucket`/`GetObject` denied	Fix the IAM role's S3 policy (grant `s3:ListBucket` on the bucket + `s3:GetObject` on the objects) — not the engine CLI creds. See Redshift setup.

Source pointers¶

source/root-engine-core/.../steps/pipeline/core/JdbcTargetParquet.java — Parquet write + saveSchemaProfile.
source/root-engine-core/.../steps/redshift/core/CopyIntoRS.java — temp table + COPY + $WM{} resolution.
source/root-engine-core/.../steps/common/global/PushFile.java — records the run folder to its watermark.
source/root-engine-core/.../core/parts/BaseStep.java — getStepRefKey() (JOB::STEP_ID), setStepWatermarkString, readWatermarkByName.
source/root-win-client/ELTMAESTRO/Steps/Redshift/CopyIntoRS*.{cs,xaml} — the WPF step + dialog.
source/root-win-client/ELTMAESTRO/GeneralWindow/PipelineGenerator.cs — the bulk export/import generators.
source/root-win-client/ELTMAESTRO/GeneralWindow/Generate{Pipeline,Ingestion}Window.xaml{,.cs} — the two wizards.
Related: Schema profile, Redshift setup, Step reference.

Loading S3 Parquet into Redshift — JdbcTargetParquet + CopyIntoRS¶

Part 1 — JdbcTargetParquet (the producer)¶