A WellView-to-warehouse copy looks like a small script: list the tables, read each one with pandas, write it with to_sql, schedule it nightly. That script has four ways to fail, and a test run against a dev copy shows you none of them. Each of the four has come up on a production WellView load we’ve worked on. The lab below reproduces them with numbers you can rerun.
Peloton puts WellView at “over 70% of the industry,” and says its “open database architecture means custom queries and direct integrations are also supported.”[1]
There’s no published WellView schema we could use, so the lab tables are made up and shaped like the problem: 4,000 well headers, a 2,000,000-row cost table and an empty stimulation table. Source and warehouse were both PostgreSQL 17.11, with Python 3.12.15, pandas 3.1.0, SQLAlchemy 2.1.6 and psycopg2 2.9.13, plus pandas 2.2.3 and 1.5.3 for comparison, run on October 28, 2026. If your WellView database is on SQL Server, reflection hands you SQL Server types such as uniqueidentifier, datetime2 and nvarchar(max), and you’ll need your own map from those to the warehouse’s types.
Size the tables before you write the loop
A WellView copy runs to a few hundred tables. In one copy we’ve seen, a single daily report cost table runs past a million rows while the well headers number in the thousands. Get every table’s size before you design the load. On Postgres the catalog estimate costs nothing:
select c.relname as table_name, c.reltuples::bigint as approx_rows
from pg_class c join pg_namespace n on n.oid = c.relnamespace
where c.relkind = 'r' and n.nspname = 'public'
order by c.reltuples desc;
table_name | approx_rows
-------------+-------------
report_cost | 2000042
well_header | 4000
stim_stage | 0
analyze refreshes the estimate, and it’s only an estimate: the cost table really has 2,000,000 rows, not 2,000,042. Give your top few tables a task each and batch the long tail into the rest.
Copy every table verbatim into a raw layer and model later, the bronze layer from the medallion pattern. You can’t guess which table your drilling engineers will want next quarter, and a verbatim copy keeps WellView the system of record for what was measured. Rate of penetration, cost per lateral foot and the rest of your drilling metrics get built on top of it later. Tying them to production and accounting is a well identity problem, which the master well table post covers.
Create tables from the source catalog, not the first chunk
Here’s the loop you’d write first:
with SRC.connect() as src:
for i, chunk in enumerate(pd.read_sql_query("select * from stim_stage", src, chunksize=10_000)):
chunk.to_sql("stim_stage", DST, if_exists="replace" if i == 0 else "append", index=False)
Point it at a table with zero rows. pandas hands you one empty chunk, and to_sql builds your target from it. The source columns were bigint, text, integer, numeric(10,2) and timestamptz, and the warehouse got this:
Table "public.stim_stage"
Column | Type | Collation | Nullable | Default
--------------+------+-----------+----------+---------
idrec | text | | |
idwell | text | | |
stage_number | text | | |
top_depth_ft | text | | |
pumped_at | text | | |
An empty DataFrame has nothing to infer types from, so every column is object, and to_sql writes object as text. The first night your source has rows, the replace drops that table and rebuilds it with numbers and timestamps. Your downstream views get a table whose types changed overnight. Nothing errors.
A LIMIT 0 probe only helps if you take the types from the catalog or the cursor. Run through pandas, select * from stim_stage limit 0 came back all object and built the same text table. Reflect the source with SQLAlchemy and create your target before any rows move:
from sqlalchemy import MetaData, Table
def create_like_source(name, SRC, DST):
src_table = Table(name, MetaData(), autoload_with=SRC)
src_table.to_metadata(MetaData()).create(DST)
Column | Type | Collation | Nullable | Default
--------------+--------------------------+-----------+----------+---------
idrec | bigint | | not null |
idwell | text | | not null |
stage_number | integer | | not null |
top_depth_ft | numeric(10,2) | | |
pumped_at | timestamp with time zone | | |
Indexes:
"stim_stage_pkey" PRIMARY KEY, btree (idrec)
Append your chunks into that table and never let to_sql create one, whether the source has zero rows tonight or two million next month.
Stream the read, then cap the container
We measured peak resident memory reading the 2,000,000-row cost table three ways, then ran the same reads in a container capped at 512 MB:
| Read | Chunks | pandas 3.1.0 | pandas 2.2.3 | 512 MB limit |
|---|---|---|---|---|
read_sql_query(sql, src) |
1 | 1,604 MB | 1,604 MB | killed, exit 137 |
read_sql_query(sql, src, chunksize=10_000) |
200 | 544 MB | 544 MB | killed, exit 137 |
same, connection with stream_results=True |
200 | 110 MB | 112 MB | finished, 109 MB |
Look at the middle row. You asked for 10,000 rows at a time and still lost the container, because the driver had already pulled all 2,000,000 into memory. The pandas docs warn about exactly this: chunksize “typically does not reduce peak memory usage, as most drivers buffer the full result set unless a server-side cursor is used.”[2] SQLAlchemy’s docs note that PostgreSQL and MySQL drivers “generally use client side cursors by default.”[3]
Ask for a server-side cursor:
with SRC.connect().execution_options(stream_results=True) as src:
for chunk in pd.read_sql_query(sql, src, chunksize=10_000):
...
Then keep a memory limit on your loader’s container, sized from the streamed peak with room to spare. We used 512 MB against the 109 MB peak, which the streamed read never touched and both buffered reads blew straight through. If someone later drops stream_results or a table triples, the loader dies with exit code 137 in its own container, and your Airflow worker keeps running every other task.
Run replace loads one at a time first
Eight workers each replacing their own table, 50 rounds apiece, ran 400 replaces in the lab without an error. Three workers replacing the same table failed 2 times in 90:
IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "pg_type_typname_nsp_index"
A replace is a drop, a create and an insert, and because Postgres creates a row type alongside every table,[4] two sessions creating a table with the same name at the same moment both try to claim one entry in the catalog’s unique index of type names, and one of them loses.
Don’t read the clean 400 as permission to go wide. That was eight small tables on an idle server. Start your load serial, go to two concurrent tasks for a week of nightly runs, then try three, and make sure no table has two writers before each step. If you keep each loader’s table list in code, a dozen lines will tell you:
from collections import defaultdict
writers = defaultdict(list)
for loader, tables in LOADERS.items():
for table in tables:
writers[table].append(loader)
for table, loaders in sorted(writers.items()):
if len(loaders) > 1:
print(f"{table}: {', '.join(loaders)}")
report_cost: nightly_group_a, old_pipeline
stim_stage: nightly_group_b, old_pipeline
well_header: nightly_group_a, nightly_group_b
Include every pipeline that reads the source, not just the one you’re changing. An old pipeline left paused after a rewrite turns into a second writer the day someone unpauses it.
Gate downstream reads on a completion marker
Replacing a table in place locks out your readers until the load commits. A reader polling every 20 ms got 4 queries through during a 3.5-second reload of a 300,000-row table, and waited 3.4 seconds for one of them. Load into a second table and swap it in:
from sqlalchemy import text
frame.to_sql(f"{table}__next", DST, if_exists="replace", index=False, chunksize=20_000)
with DST.begin() as c:
c.execute(text(f"drop table {table}"))
c.execute(text(f"alter table {table}__next rename to {table}"))
Postgres runs schema changes inside the transaction, so the drop and the rename commit together or not at all.[5] The drop still takes an exclusive lock for the moment the swap commits, and to_sql builds __next without your keys or indexes, so create __next with create_like_source first and append into it. The same reader got 106 queries through that load, never waited more than 0.1 seconds, and saw all 300,000 rows every time.
That’s one table. In an 8.9-second load that swapped two tables back to back, with the same 20 ms reader checking both, 61 of 135 reads saw one table new and the other old. Stretch that across a few hundred tables loading in groups, and a report that runs mid-load joins tonight’s jobs to last night’s costs.
End your load with a done task that depends on every group, and hang everything downstream off done, never off a single group. If the downstream work is another Airflow DAG (directed acyclic graph), dataset scheduling lets it wait for done without a sensor or a guessed start time.
Troubleshooting
A loader container exits with code 137 and no Python traceback. The kernel killed it for going over its memory limit. In the lab a chunksize=10_000 read of 2,000,000 rows did exactly that under 512 MB, because psycopg2 buffered the whole result first. Add stream_results=True to the connection.
duplicate key value violates unique constraint "pg\_type\_typname\_nsp\_index" from to\_sql. Two sessions created a table with the same name at the same moment. Run the writer check above across every pipeline that touches the target schema.
A table that should be numeric is all text. Its first load came from an empty result. Drop it, recreate it from the source’s reflected definition, reload.
Deadlocks right after you raise a replace load’s concurrency. We hit this on a production WellView load at three concurrent tasks replacing tables in one schema. We never confirmed the cause, and the lab’s separate-table runs were clean, so treat this as a workaround: set it back to one task, rule out shared tables with the writer check, and raise it one step at a time.
Reproduce it
Dockerfile:
FROM python:3.12.15-slim
ARG PANDAS=3.1.0
RUN pip install --no-cache-dir "pandas==${PANDAS}" "SQLAlchemy==2.1.6" "psycopg2-binary==2.9.13"
WORKDIR /lab
COPY *.py ./
compose.yaml:
services:
src:
image: postgres:17.11
environment: { POSTGRES_PASSWORD: lab, POSTGRES_DB: wv }
volumes: [ "./init:/docker-entrypoint-initdb.d:ro" ]
dst:
image: postgres:17.11
environment: { POSTGRES_PASSWORD: lab, POSTGRES_DB: dw }
lab:
build: .
depends_on: [ src, dst ]
command: [ "sleep", "infinity" ]
lab-pandas2:
build: { context: ., args: { PANDAS: "2.2.3" } }
depends_on: [ src, dst ]
command: [ "sleep", "infinity" ]
init/01-seed.sql:
CREATE TABLE well_header (
idwell text PRIMARY KEY,
well_name text NOT NULL,
spud_date date
);
INSERT INTO well_header
SELECT md5(g::text), 'WELL ' || g, date '2015-01-01' + (g % 3000)
FROM generate_series(1, 4000) g;
CREATE TABLE report_cost (
idrec bigint PRIMARY KEY,
idwell text NOT NULL,
report_date date NOT NULL,
cost_code text NOT NULL,
description text,
amount numeric(12,2) NOT NULL
);
INSERT INTO report_cost
SELECT g, md5((g % 4000 + 1)::text), date '2020-01-01' + (g % 2000),
'CC-' || (g % 400), 'daily cost line item ' || g || ' rental and services', (g % 100000) / 7.0
FROM generate_series(1, 2000000) g;
CREATE TABLE stim_stage (
idrec bigint PRIMARY KEY,
idwell text NOT NULL,
stage_number integer NOT NULL,
top_depth_ft numeric(10,2),
pumped_at timestamptz
);
common.py:
from sqlalchemy import create_engine
SRC = create_engine("postgresql+psycopg2://postgres:lab@src:5432/wv")
DST = create_engine("postgresql+psycopg2://postgres:lab@dst:5432/dw")
empty_table.py:
"""What a chunked copy does with a source table that has zero rows."""
import pandas as pd
from sqlalchemy import MetaData, Table, inspect, text
from common import SRC, DST
print("pandas", pd.__version__)
def target_columns(name):
insp = inspect(DST)
if not insp.has_table(name):
return "table does not exist"
return [(c["name"], str(c["type"])) for c in insp.get_columns(name)]
with DST.begin() as c:
for t in ("stim_stage", "stim_stage_probe", "stim_stage_reflect"):
c.execute(text(f"drop table if exists {t}"))
# 1. naive chunked copy
chunks = 0
with SRC.connect() as src:
for i, chunk in enumerate(pd.read_sql_query("select * from stim_stage", src, chunksize=10_000)):
chunks += 1
chunk.to_sql("stim_stage", DST, if_exists="replace" if i == 0 else "append", index=False)
print("naive: chunks yielded =", chunks)
print("naive: target =", target_columns("stim_stage"))
# 2. LIMIT 0 probe through pandas, then append chunks
with SRC.connect() as src:
probe = pd.read_sql_query("select * from stim_stage limit 0", src)
print("probe dtypes:", dict(probe.dtypes.astype(str)))
probe.to_sql("stim_stage_probe", DST, if_exists="replace", index=False)
print("probe: target =", target_columns("stim_stage_probe"))
# 3. reflect the source table and create it with the source's own types
src_table = Table("stim_stage", MetaData(), autoload_with=SRC)
dst_table = src_table.to_metadata(MetaData(), name="stim_stage_reflect")
dst_table.create(DST)
print("reflect: target =", target_columns("stim_stage_reflect"))
memory.py:
"""Peak memory of three ways to read a 2,000,000-row table with pandas."""
import resource
import sys
import pandas as pd
from common import SRC
mode = sys.argv[1]
sql = "select * from report_cost"
rows = chunks = 0
if mode == "whole":
with SRC.connect() as src:
rows = len(pd.read_sql_query(sql, src)); chunks = 1
elif mode == "chunksize":
with SRC.connect() as src:
for chunk in pd.read_sql_query(sql, src, chunksize=10_000):
rows += len(chunk); chunks += 1
elif mode == "stream":
with SRC.connect().execution_options(stream_results=True) as src:
for chunk in pd.read_sql_query(sql, src, chunksize=10_000):
rows += len(chunk); chunks += 1
peak_mb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024
print(f"{mode:9s} rows={rows} chunks={chunks} peak_rss_mb={peak_mb:.0f}")
replace_race.py:
"""Several workers running to_sql(if_exists='replace') at once against one schema."""
import sys
from multiprocessing import Pool
import pandas as pd
from common import DST
frame = pd.DataFrame({"id": range(5_000), "v": ["x" * 20] * 5_000})
def worker(args):
table, rounds = args
errors = []
for _ in range(rounds):
try:
frame.to_sql(table, DST, if_exists="replace", index=False)
except Exception as e:
errors.append(f"{type(e).__name__}: {str(e).splitlines()[0]}")
return errors
if __name__ == "__main__":
mode, workers, rounds = sys.argv[1], int(sys.argv[2]), int(sys.argv[3])
tables = ["shared_t"] * workers if mode == "same" else [f"t_{i}" for i in range(workers)]
with Pool(workers) as p:
results = p.map(worker, [(t, rounds) for t in tables])
errs = [e for r in results for e in r]
print(f"mode={mode} workers={workers} rounds={rounds} attempts={workers*rounds} errors={len(errs)}")
for e in sorted(set(errs)):
print(f" {errs.count(e):4d} x {e[:220]}")
reader_during_load.py:
"""What a downstream reader sees while tables reload: replace in place, build-then-swap, and a two-table load."""
import sys
import threading
import time
import pandas as pd
from sqlalchemy import text
from common import DST
N = 300_000
def frame(generation):
return pd.DataFrame({"id": range(N), "generation": generation})
def replace_in_place(table, generation):
frame(generation).to_sql(table, DST, if_exists="replace", index=False, chunksize=20_000)
def build_then_swap(table, generation):
frame(generation).to_sql(f"{table}__next", DST, if_exists="replace", index=False, chunksize=20_000)
with DST.begin() as c:
c.execute(text(f"drop table {table}"))
c.execute(text(f"alter table {table}__next rename to {table}"))
def watch(query, run_load):
seen, done = [], threading.Event()
def reader():
while not done.is_set():
t0 = time.monotonic()
with DST.connect() as c:
row = c.execute(text(query)).one()
seen.append((time.monotonic() - t0, tuple(row)))
time.sleep(0.02)
t = threading.Thread(target=reader); t.start()
t0 = time.monotonic(); run_load(); load_s = time.monotonic() - t0
done.set(); t.join()
return load_s, seen
mode = sys.argv[1]
for table in ("wv_job", "wv_job_cost"):
replace_in_place(table, 1)
if mode in ("in_place", "swap"):
load = replace_in_place if mode == "in_place" else build_then_swap
load_s, seen = watch("select count(*) from wv_job", lambda: load("wv_job", 2))
waits = [w for w, _ in seen]
print(f"{mode:8s} load_s={load_s:.1f} reads={len(seen)} longest_read_wait_s={max(waits):.1f}"
f" counts={sorted({r[0] for _, r in seen})}")
elif mode == "two_tables":
q = "select (select max(generation) from wv_job), (select max(generation) from wv_job_cost)"
def load_both():
build_then_swap("wv_job", 2)
build_then_swap("wv_job_cost", 2)
load_s, seen = watch(q, load_both)
mixed = [r for _, r in seen if r[0] != r[1]]
print(f"two_tables load_s={load_s:.1f} reads={len(seen)} mixed_generation_reads={len(mixed)}"
f" example={mixed[0] if mixed else None}")
shared_tables.py, the writer check, with an example table list:
"""List every table that more than one loader writes."""
from collections import defaultdict
LOADERS = {
"nightly_group_a": ["well_header", "report_cost"],
"nightly_group_b": ["stim_stage", "well_header"],
"old_pipeline": ["report_cost", "stim_stage"],
}
writers = defaultdict(list)
for loader, tables in LOADERS.items():
for table in tables:
writers[table].append(loader)
shared = {t: w for t, w in sorted(writers.items()) if len(w) > 1}
for table, loaders in shared.items():
print(f"{table}: {', '.join(loaders)}")
raise SystemExit(1 if shared else 0)
drill.sh runs sections 1 to 6 from a clean start:
#!/usr/bin/env bash
# Usage: bash drill.sh 2>&1 | tee drill.log
set -eu
C="docker compose -p wvlab"
echo "== clean start"
$C down -v
$C up -d --build
for i in $(seq 1 100); do $C exec -T src psql -U postgres -d wv -Atc "select count(*) from report_cost" 2>/dev/null | grep -q 2000000 \
&& $C exec -T dst pg_isready -h 127.0.0.1 -q && break; sleep 3; done
echo "== versions"
$C exec -T src psql -U postgres -d wv -Atc "select version()"
$C exec -T lab python -c "import sys, pandas, sqlalchemy, psycopg2; print(sys.version.split()[0], 'pandas', pandas.__version__, 'SQLAlchemy', sqlalchemy.__version__, 'psycopg2', psycopg2.__version__.split()[0])"
$C exec -T lab-pandas2 python -c "import pandas; print('pandas', pandas.__version__)"
$C exec -T src psql -U postgres -d wv -Atc "select 'well_header', count(*) from well\_header union all select 'report\_cost', count(*) from report_cost union all select 'stim_stage', count(*) from stim_stage"
echo "== 1. empty source table"
$C exec -T lab python empty_table.py
$C exec -T lab-pandas2 python empty_table.py
$C exec -T dst psql -U postgres -d dw -c '\d stim_stage' -c '\d stim_stage_reflect'
echo "== 2. peak memory reading 2,000,000 rows"
for img in lab lab-pandas2; do for m in whole chunksize stream; do $C exec -T $img python memory.py $m; done; done
echo "== 3. concurrent to_sql(if_exists='replace')"
$C exec -T lab python replace_race.py different 8 50
$C exec -T lab python replace_race.py same 3 30
$C exec -T lab python replace_race.py same 1 30
echo "== 4. readers during a reload"
for m in in_place swap two_tables; do $C exec -T lab python reader_during_load.py $m; done
echo "== 5. 512 MB container memory limit"
for m in whole chunksize stream; do
rc=0; docker run --rm --network wvlab_default --memory=512m --memory-swap=512m wvlab-lab python memory.py $m || rc=$?
echo "$m exit code: $rc"
done
echo "== 6. table sizes from the catalog"
$C exec -T src psql -U postgres -d wv -c "analyze" -c "select c.relname as table_name, c.reltuples::bigint as approx_rows from pg_class c join pg_namespace n on n.oid = c.relnamespace where c.relkind = 'r' and n.nspname = 'public' order by c.reltuples desc"
echo "== drill done"
The pandas 1.5.3 comparison ran from its own image against the same stack, Dockerfile.pandas15:
FROM python:3.11.13-slim
RUN pip install --no-cache-dir "pandas==1.5.3" "numpy==1.26.4" "SQLAlchemy==1.4.54" "psycopg2-binary==2.9.13"
WORKDIR /lab
COPY *.py ./
docker build -f Dockerfile.pandas15 -t wvlab-pandas15 .
docker run --rm --network wvlab_default wvlab-pandas15 python empty_table.py
The writer check needs no database: docker run --rm -v "$PWD:/lab" -w /lab python:3.12.15-slim python shared_tables.py. It exits 1 when it finds a shared table.
Every figure above is from one logged run of these scripts. Memory figures are ru_maxrss for one Python process in Docker Engine 28.5.1 on Docker Desktop, Windows host.
Peloton, “WellView” product page and FAQ (retrieved October 28, 2026). https://www.peloton.com/products/well-data-lifecycle/wellview ↩︎
pandas 3.1.0 documentation, “pandas.read_sql_query,” parameter
chunksize(retrieved October 28, 2026). https://pandas.pydata.org/docs/reference/api/pandas.read_sql_query.html ↩︎SQLAlchemy 2.0 documentation, “Using Server Side Cursors (a.k.a. stream results)” (retrieved October 28, 2026). https://docs.sqlalchemy.org/en/20/core/connections.html#using-server-side-cursors-a-k-a-stream-results ↩︎
PostgreSQL Global Development Group, “CREATE TABLE,” PostgreSQL 17 Documentation (retrieved October 28, 2026). https://www.postgresql.org/docs/17/sql-createtable.html ↩︎
PostgreSQL wiki, “Transactional DDL in PostgreSQL: A Competitive Analysis” (retrieved October 28, 2026). https://wiki.postgresql.org/wiki/Transactional_DDL_in_PostgreSQL:_A_Competitive_Analysis ↩︎