External Tables and the Lake/Warehouse Boundary

The boundary is a decision about access and ownership

A lake already holds Parquet files. Analysts need to query them, but making another managed copy is only one option. An external table gives an engine a table-shaped interface to data stored outside its ordinary managed-table lifecycle. A loaded table gives that engine control over a separate representation. Both consume compute; both need access controls, schema rules, and reliable publication. The choice depends on latency, concurrency, freshness, update requirements, and total cost, not simply whether the data is old or new.

Follow a query from catalog metadata to files, then consider when a managed copy is worthwhile and what a plain directory-backed table fails to guarantee. “External” and “transactional” are not opposites: a modern engine can expose an external table backed by an open table format. Here, a plain external table means registered partitions and ordinary Parquet files without a transactional table layer.

Managed storage can itself use object storage. “Local copy” here means an engine-managed representation, not a claim that warehouse data lives on a local disk.

What registration supplies—and what it does not

A catalog maps a name to a schema, storage location or table-metadata reference, format, and partition information. It can also hold properties, statistics, and access-policy references. Registration normally avoids copying the source rows, but the engine may cache metadata or data and write query results elsewhere. A shared catalog helps engines discover a table; compatible connectors, types, partition rules, and permissions are still required.

For an ordinary Hive-style table, ingest_date=2026-01-20/country=KR describes a partition location. The reader can derive these values without storing them in each file. The SQL below is an Athena/Hive-style illustration, not portable SQL for every warehouse, and is not executed here. Creating the table does not automatically register every existing partition in this mode. Athena CREATE TABLE

The execution identity needs access to the catalog, source objects, any required decryption keys, and query-result locations. A table grant alone need not grant object access. Conversely, users with direct storage access may bypass restrictions enforced only by one SQL engine. Shared files need a consistent governance path. Athena identity and access management

CREATE EXTERNAL TABLE raw.orders (
  order_id BIGINT, customer_id BIGINT, amount_cents BIGINT, note STRING
)
PARTITIONED BY (ingest_date STRING, country STRING)
STORED AS PARQUET
LOCATION 's3://example-lake/raw/orders/';

ALTER TABLE raw.orders ADD IF NOT EXISTS
PARTITION (ingest_date='2026-01-20', country='KR')
LOCATION 's3://example-lake/raw/orders/ingest_date=2026-01-20/country=KR/';

The example uses ISO date strings for the partition key, matching the Python fixture; SQL callers must use the declared type. The database, bucket, and access configuration are assumed placeholders. One ADD PARTITION registers only that one location.

Separate pruning, bytes, and billing

A partition predicate can eliminate locations before opening data files. Projection limits the columns needed for output, filters, joins, and aggregation. Parquet statistics can also eliminate row groups even when the predicate uses a non-partition column. An ingestion-date partition may be poorly selective for an event-time query with late arrivals; do not add a date predicate that changes the intended answer merely to reduce cost. Spectrum performance guidance

The lab counts files surviving partition pruning and sums their compressed chunks for all declared query columns. It includes filter columns, but deliberately does not subtract row groups or pages. This is a candidate-byte estimate, not bytes fetched, billed bytes, or an upper bound on total I/O. Metadata reads, caching, readahead, retries, and query-result writes are outside it. EXPLAIN describes the planned scan; runtime metrics are needed for actual execution.

Bytes-based pricing makes pruning economically relevant, but compute-based pricing is not equivalent: concurrency, idle capacity, CPU work, and elapsed time also matter. Athena offers scan- and compute-based options, with storage, requests, transfer, and catalog charges considered separately. The lab’s price of 5 per decimal TB is a hypothetical linear model, excludes minimum charges and rounding, and does not make tiny queries free. Use the selected service, region, billing mode, and actual metrics for a forecast. Athena pricing

What a managed copy can buy

A managed copy can provide engine-specific organization, statistics, caching, materialized aggregates, workload management, and supported update operations. Many warehouses still use columnar scans; indexes and millisecond responses are not universal. External scans can also be fast enough for dashboards. Benchmark the same result, filters, concurrency, and cache conditions before choosing a boundary. A local SQLite example illustrates a loaded representation, not the architecture or performance of a cloud warehouse.

Count the extra storage, ingestion, recurring refresh, maintenance, and managed-query compute. A simplified monthly comparison is external cost = q × e and managed cost = F + q × m, where q is query count, e and m are per-query costs, and F is the extra monthly fixed cost of the managed path. Common baseline costs cancel only if genuinely equal. If e > m, break-even is F / (e − m); if e ≤ m and F is positive, query frequency alone produces no cost break-even. The lab uses F=120, e=0.20, m=0.05, giving 800 queries. These invented values explain the calculation; real costs need workload measurements.

A second copy also has a freshness contract. Define a source snapshot or watermark, publish the loaded copy after validation, and reconcile counts and business aggregates. A faster stale table does not satisfy a current-data requirement.

Registration is not a commit protocol

Some external tables enumerate registered partitions; others discover paths dynamically or calculate them from configured rules. Athena partition projection can calculate locations without adding each partition to Glue, but another engine may still use Glue’s registered metadata. A shared table definition therefore need not produce identical discovery behavior. In Athena, enabling projection makes the engine ignore registered partition entries for that table. A configured range that excludes valid data can return zero rows without an error; test boundary dates and expected deliveries. Partition projection

For ordinary registered Hive partitions, explicit ADD PARTITION, a crawler, or a repair command can add new locations. Athena MSCK REPAIR TABLE adds discovered partitions; it does not remove stale ones and a timed-out repair can leave registration incomplete. These operations do not prove that each partition’s writer finished. The lab makes the actual reader consult registered locations, demonstrates new data remaining invisible, then reruns the reader after refresh. A premature refresh exposes only 2,260 of a planned 3,000 rows. Repair semantics

For a plain file table, write to isolated immutable locations, validate files and expected counts, and publish a reference through a protocol every reader honors. A _SUCCESS marker has no effect on an unaware reader. Directory rename is not a portable atomic operation over object stores, and atomic creation of one object does not commit many objects. A manifest must pin immutable files or object versions and retain them for readers.

Updates and history require a table layer

Overwriting affected Parquet files one at a time can expose mixed results to concurrent readers. Even a saved list of paths cannot reproduce the old answer after those paths are overwritten. A snapshot requires stable file identities, a consistent committed membership, and retention of both data and metadata. The deletion exercise intentionally demonstrates the unsafe rewrite only inside temporary test data. Query invisibility does not prove removal from historical versions, replicas, or backups.

Open table formats add transaction and metadata protocols, but they do not all use the same tree or one small delete file. Iceberg describes snapshots and atomic metadata replacement, with row-level delete representations depending on format version. Delta uses a transaction log and can use deletion vectors with compatible clients; rewrites are also possible. Hudi distinguishes copy-on-write and merge-on-read tables and their query behavior. Iceberg specification Delta deletion vectors Hudi table types

Atomicity depends on a supported catalog, storage, and concurrency protocol, not merely naming a metadata pointer. Time travel lasts only while required files and metadata remain available. Schema and partition evolution are format- and engine-specific, and existing files do not magically acquire a new physical layout. Adopting a format requires reader compatibility, migration validation, maintenance, and retention decisions. It is more than changing an external-table keyword.

Make the hot/cold boundary explicit

For the subscription-company example, replay history can remain in the lake, frequently refreshed dashboard aggregates can use a managed table or materialized result, and occasional historical analysis can use external scans. A thirteen-month managed window is a hypothesis to test against measured access, not an architectural rule. Small-file compaction can reduce metadata work but is not a prerequisite for registering a table; choose file sizes from scale and query behavior.

If one view combines cold external rows and hot managed rows, use mutually exclusive predicates on the same typed time field: cold < cutoff and hot ≥ cutoff. Publish a complete hot copy before advancing the boundary. Define the timezone, treatment of late updates and deletes, freshness watermark, and reconciliation procedure. UNION ALL otherwise duplicates overlaps or hides gaps; UNION can hide exact duplicates without repairing incorrect source ownership. Null boundary values satisfy neither comparison and disappear from both sides. Reject or quarantine them explicitly, or define a separate branch; do not silently treat them as cold. The lab verifies the split on one static dataset with no null dates. It does not implement a concurrent cutover.

Record p95 latency, concurrency, freshness, billed bytes or compute, refresh cost, file counts, registration lag, and errors by workload. Reconsider the boundary when those measurements or requirements change. A dashboard is not automatically a reason to load data, and a low query count is not automatically a reason to tolerate slow or inconsistent reads.

Lab

Run the setup once and run each numbered exercise separately from that fresh setup: exercises 5 and 6 intentionally change files. Python 3.12.14, PyArrow 25.0.1, and DuckDB 1.5.5 were used. All data is synthetic and stored in a temporary directory; keep workspace alive, then call workspace.cleanup() after finishing. Amounts are integer cents in one assumed currency. Sixty days contain 5,000 rows per day. The catalog stores a schema and registered leaf directories; refresh rediscovers directories, while reads list current files inside those directories. This is a local teaching model, not Glue or a production transaction protocol.

The scan helper handles this flat schema and requires callers to declare every filter field via filter_columns. It cannot infer that dependency list or prove it complete. Candidate bytes include output and declared filter columns across all row groups of candidate files. Partition values come from directories and have no physical chunks. No minimum billing, cache, I/O timing, or cloud traffic is simulated.

import os, random, sqlite3, tempfile
from pathlib import Path
import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.dataset as ds
import pyarrow.parquet as pq
from datetime import date, timedelta

rng = random.Random(0)
N = 300_000
first_day = date(2026, 1, 1)
days = [(first_day + timedelta(days=i // (N // 60))).isoformat() for i in range(N)]
table = pa.table({
    "order_id": list(range(1, N + 1)),
    "ingest_date": days,
    "country": [rng.choice(["US", "US", "KR", "DE"]) for _ in range(N)],
    "customer_id": [rng.randint(1, 50_000) for _ in range(N)],
    "amount_cents": [rng.randint(500, 20_000) for _ in range(N)],
    "note": [f"ref-{rng.randrange(10**9):09d}" for _ in range(N)],
})
workspace = tempfile.TemporaryDirectory(prefix="de018-")
LAKE = workspace.name
ds.write_dataset(table, os.path.join(LAKE, "orders"), format="parquet", partitioning=["ingest_date", "country"],
                 partitioning_flavor="hive", use_threads=False)

catalog = {}
partitioning = ds.partitioning(pa.schema([("ingest_date", pa.string()), ("country", pa.string())]), flavor="hive")

def discover(location):
    return sorted(str(p) for p in Path(location).rglob("*.parquet"))

def register(name, location):
    dataset = ds.dataset(location, format="parquet", partitioning=partitioning)
    catalog[name] = {"location": location, "schema": dataset.schema,
                     "known_dirs": sorted({str(Path(f).parent) for f in discover(location)})}

def refresh(name):
    entry = catalog[name]
    entry["known_dirs"] = sorted({str(Path(f).parent) for f in discover(entry["location"])})

def external_dataset(name):
    entry = catalog[name]
    visible = sorted(str(f) for d in entry["known_dirs"] for f in Path(d).glob("*.parquet"))
    return ds.dataset(visible, format="parquet", schema=entry["schema"],
                      partitioning=partitioning, partition_base_dir=entry["location"])

def scan(name, columns=None, filter=None, filter_columns=None):
    if filter is not None and filter_columns is None:
        raise ValueError("declare every field used by the filter")
    dataset = external_dataset(name)
    wanted = set(dataset.schema.names if columns is None else columns) | set(filter_columns or [])
    unknown = wanted - set(dataset.schema.names)
    if unknown:
        raise ValueError(f"unknown columns: {sorted(unknown)}")
    fragments = list(dataset.get_fragments(filter=filter))
    candidate_bytes = 0
    for fragment in fragments:
        meta = fragment.metadata
        for g in range(meta.num_row_groups):
            for c in range(meta.num_columns):
                chunk = meta.row_group(g).column(c)
                if chunk.path_in_schema in wanted:
                    candidate_bytes += chunk.total_compressed_size
    result = dataset.to_table(columns=columns, filter=filter)
    return {"rows": result.num_rows, "candidate_files": len(fragments), "candidate_chunk_bytes": candidate_bytes}

register("orders", os.path.join(LAKE, "orders"))
files = discover(os.path.join(LAKE, "orders"))
print("rows:", N, "files:", len(files), "partition directories:", len(catalog["orders"]["known_dirs"]))
print("dates:", min(days), max(days))
# rows: 300000 files: 180 partition directories: 180
# dates: 2026-01-01 2026-03-01

1. Partition candidates and a hypothetical price.

one_day = pc.field("ingest_date") == "2026-01-20"
queries = [("one day", one_day, ["ingest_date"]),
           ("day and KR", one_day & (pc.field("country") == "KR"), ["ingest_date", "country"]),
           ("whole", None, []),
           ("customer", pc.field("customer_id") == 4471, ["customer_id"])]
for label, predicate, deps in queries:
    result = scan("orders", filter=predicate, filter_columns=deps)
    print(label, "rows:", result["rows"], "candidate files:", result["candidate_files"],
          "candidate MB:", round(result["candidate_chunk_bytes"] / 1e6, 2))
whole = scan("orders")["candidate_chunk_bytes"]
day_bytes = scan("orders", filter=one_day, filter_columns=["ingest_date"])["candidate_chunk_bytes"]
price_per_tb, table_tb = 5.0, 2.0
print(f"hypothetical 2 TB: whole {table_tb * price_per_tb:.2f}; one day {table_tb * price_per_tb * day_bytes / whole:.2f}")
# one day rows: 5000 candidate files: 3 candidate MB: 0.14
# day and KR rows: 1236 candidate files: 1 candidate MB: 0.03
# whole rows: 300000 candidate files: 180 candidate MB: 8.26
# customer rows: 7 candidate files: 180 candidate MB: 8.26
# hypothetical 2 TB: whole 10.00; one day 0.17

2. Projection includes filter dependencies.

one_day = pc.field("ingest_date") == "2026-01-20"
for label, columns, predicate, deps in [
    ("all", None, None, []),
    ("amount", ["amount_cents"], None, []),
    ("one day amount", ["amount_cents"], one_day, ["ingest_date"]),
    ("customer amount", ["amount_cents"], pc.field("customer_id") == 4471, ["customer_id"])]:
    result = scan("orders", columns, predicate, deps)
    print(label, "candidate MB:", round(result["candidate_chunk_bytes"] / 1e6, 3), "rows:", result["rows"])
whole = scan("orders")["candidate_chunk_bytes"]
selected = scan("orders", ["amount_cents"], one_day, ["ingest_date"])["candidate_chunk_bytes"]
print("whole / one-day-amount candidate bytes:", round(whole / selected, 1))
# all candidate MB: 8.258 rows: 300000
# amount candidate MB: 1.621 rows: 300000
# one day amount candidate MB: 0.027 rows: 5000
# customer amount candidate MB: 3.345 rows: 7
# whole / one-day-amount candidate bytes: 303.8

3. Small files: verify the same aggregate.

small_root = os.path.join(LAKE, "orders_small")
for f in files:
    part = pq.ParquetFile(f).read()
    target_dir = os.path.join(small_root, os.path.relpath(os.path.dirname(f), os.path.join(LAKE, "orders")))
    os.makedirs(target_dir, exist_ok=True)
    for n, offset in enumerate(range(0, part.num_rows, 200)):
        pq.write_table(part.slice(offset, 200), os.path.join(target_dir, f"part-{n:03d}.parquet"))
register("orders_small", small_root)
expected = pc.sum(table["amount_cents"]).as_py()
for name in ("orders", "orders_small"):
    estimate = scan(name, ["amount_cents"])
    values = external_dataset(name).to_table(columns=["amount_cents"])
    print(name, "files:", estimate["candidate_files"], "candidate MB:", round(estimate["candidate_chunk_bytes"] / 1e6, 2),
          "sum correct:", pc.sum(values["amount_cents"]).as_py() == expected)
# orders files: 180 candidate MB: 1.62 sum correct: True
# orders_small files: 1613 candidate MB: 1.78 sum correct: True

4. A managed copy, an exact boundary, and break-even.

last_day = date.fromisoformat(max(days))
cutoff = (last_day - timedelta(days=6)).isoformat()
hot = table.filter(pc.field("ingest_date") >= cutoff)
with sqlite3.connect(":memory:") as wh:
    wh.execute("CREATE TABLE orders_hot (order_id INTEGER, ingest_date TEXT, country TEXT, customer_id INTEGER, amount_cents INTEGER)")
    wh.executemany("INSERT INTO orders_hot VALUES (?, ?, ?, ?, ?)",
                   zip(*[hot.to_pylist() for c in ("order_id", "ingest_date", "country", "customer_id", "amount_cents")]))
    wh.execute("CREATE INDEX ix_day ON orders_hot (ingest_date)")
    local_sum = wh.execute("SELECT sum(amount_cents) FROM orders_hot WHERE ingest_date = ?", ("2026-02-27",)).fetchone()[0]
external = external_dataset("orders").to_table(columns=["amount_cents"], filter=pc.field("ingest_date") == "2026-02-27")
print("hot rows:", hot.num_rows, "from:", cutoff, "through:", last_day)
print("same daily sum:", local_sum == pc.sum(external["amount_cents"]).as_py())

cold = table.filter(pc.field("ingest_date") < cutoff)
combined = pa.concat_tables([cold, hot]).sort_by("order_id")
print("disjoint cold/hot union:", combined.equals(table.sort_by("order_id")))
fixed_monthly, external_per_query, managed_per_query = 120.0, 0.20, 0.05
break_even = fixed_monthly / (external_per_query - managed_per_query)
print(f"hypothetical monthly break-even: {break_even:.0f} queries")
for queries in (100, 1000):
    print(f"{queries} queries: external {queries * external_per_query:.2f}; managed {fixed_monthly + queries * managed_per_query:.2f}")
# hot rows: 35000 from: 2026-02-23 through: 2026-03-01
# same daily sum: True
# disjoint cold/hot union: True
# hypothetical monthly break-even: 800 queries
# 100 queries: external 20.00; managed 125.00
# 1000 queries: external 200.00; managed 170.00

5. Missing registration and premature visibility.

new_day = table.slice(0, 5000).set_column(1, "ingest_date", pa.array(["2026-03-05"] * 5000))
new_day = new_day.set_column(0, "order_id", pa.array(range(N + 1, N + 5001)))
ds.write_dataset(new_day, os.path.join(LAKE, "orders"), format="parquet", partitioning=partitioning, use_threads=False, existing_data_behavior="overwrite_or_ignore", basename_template="new-{i}.parquet")
print("before refresh:", external_dataset("orders").count_rows(), "directory discovery:", ds.dataset(os.path.join(LAKE, "orders"), format="parquet", partitioning=partitioning).count_rows())
refresh("orders")
print("after refresh:", external_dataset("orders").count_rows())
partial_day = os.path.join(LAKE, "orders", "ingest_date=2026-03-06")
half = table.slice(5000, 3000).set_column(1, "ingest_date", pa.array(["2026-03-06"] * 3000))
half = half.set_column(0, "order_id", pa.array(range(N + 5001, N + 8001)))
for country in ("US", "KR"):
    part = half.filter(pc.field("country") == country).drop(["ingest_date", "country"])
    target = os.path.join(partial_day, f"country={country}")
    os.makedirs(target, exist_ok=True)
    pq.write_table(part, os.path.join(target, "part-0.parquet"))
refresh("orders")
visible = external_dataset("orders").to_table(filter=pc.field("ingest_date") == "2026-03-06")
print("premature refresh exposes:", visible.num_rows, "of expected:", half.num_rows)
# before refresh: 300000 directory discovery: 305000
# after refresh: 305000
# premature refresh exposes: 2260 of expected: 3000

6. An unsafe rewrite also destroys path-based history.

victim = 4471
pinned_paths = list(files)
def victim_count(paths):
    result = ds.dataset(paths, format="parquet").to_table(columns=["customer_id"])
    return pc.sum(pc.cast(pc.fill_null(pc.equal(result["customer_id"], victim), False), pa.int64())).as_py()
old_count = victim_count(pinned_paths)
touched, rewritten_bytes = 0, 0
for f in files:
    part = pq.ParquetFile(f).read()
    matches = pc.fill_null(pc.equal(part["customer_id"], victim), False)
    if pc.any(matches).as_py():
        pq.write_table(part.filter(pc.invert(matches)), f)
        touched += 1
        rewritten_bytes += os.path.getsize(f)
print("files read:", len(files), "rewritten:", touched, "output MB:", round(rewritten_bytes / 1e6, 2))
print("same pinned paths, customer rows before/after:", old_count, victim_count(pinned_paths))
print("remaining rows:", external_dataset("orders").count_rows())
# files read: 180 rewritten: 6 output MB: 0.28
# same pinned paths, customer rows before/after: 7 0
# remaining rows: 299993

7. Run the external query in DuckDB.

import duckdb

con = duckdb.connect()
try:
    pattern = os.path.join(LAKE, "orders", "*", "*", "*.parquet")
    con.read_parquet(pattern, hive_partitioning=True).create_view("orders_external")
    result = con.execute("SELECT count(*), min(ingest_date), max(ingest_date) FROM orders_external").fetchone()
    print("SQL rows and dates:", result)
    sql_sum = con.execute("SELECT sum(amount_cents) FROM orders_external WHERE ingest_date = DATE '2026-01-20'").fetchone()[0]
    expected = pc.sum(table.filter(pc.field("ingest_date") == "2026-01-20")["amount_cents"]).as_py()
    print("SQL sum correct:", sql_sum == expected)
    plan = con.execute("EXPLAIN SELECT sum(amount_cents) FROM orders_external WHERE ingest_date = DATE '2026-01-20'").fetchone()[1]
    import re
    pruning = re.search(r"Scanning Files:\s*(\d+)/(\d+)", plan)
    if pruning is None:
        raise RuntimeError("inspect EXPLAIN: expected file-pruning diagnostic missing")
    print("EXPLAIN planned files:", pruning.group(1), "of", pruning.group(2))
finally:
    con.close()
# SQL rows and dates: (300000, datetime.date(2026, 1, 1), datetime.date(2026, 3, 1))
# SQL sum correct: True
# EXPLAIN planned files: 3 of 180

8. Write the boundary decision. Choose one dataset and state its latency, concurrency, freshness, access pattern, total-cost assumptions, publication protocol, and owner. Explain how you will detect a missing partition or a stale managed copy.