Metadata Catalog

The previous two lessons accumulated a set of assets: a central warehouse, unit-focused subsets, and unit products that each chose their own partition key. In every measurement it was already known which store the question would go to; the cost started after the address was known. In reality, if that address is not written down somewhere, it cannot be found. If the name of a unit’s product, its owner, the columns it carries, its partition key and the periods it covers are not written, finding the asset that will answer a question is itself a search job.

What this lesson measures is that search. A metadata catalog keeps a single entry per asset and the search runs over those entries; if there is no catalog, the search runs over the files of the assets. In both cases the number of entries looked at is counted. Then the catalog’s own flaw is measured: when the storage side changes, how many entries now point to a layout that no longer exists, and what a search that trusts those entries costs.

• SA61. The source, the pipeline, the storage layout and the catalog are a fiction and are defined within the lesson. No warehouse, lake, catalog product or query engine is called. Seed 20260218.
• SA62. An entry looked at is of two kinds and both are counted one by one. In a search without a catalog, the files of an asset must be opened to learn what it covers; the entry looked at is the file opened. In a search with a catalog, the entry looked at is the catalog entry scanned.
• SA63. The assets are scanned sorted by name. The search stops at the first asset that suffices for the task; the decision to pick the cheapest asset is outside the measure of this lesson.
• SA64. A catalog entry carries the layout at the moment it was written. A change on the storage side is not reflected in the entry by itself; reflecting it is a separate job.
• SA65. A search with a catalog trusts the entry but verifies: if the asset of the matching entry does not really suffice for the task, the search falls back to the search without a catalog and the two costs are added.
• SA66. The lineage measure was measured in the Data Lineage lesson of the Introduction to Data Engineering course; it is not repeated here. The object of lineage is the rows a row rests on; the object of the catalog is where an asset stands. One explains the value, the other gives the address.

With and Without a Catalog

The set consists of twelve assets. The central warehouse carries all the regions and the tariff column, but holds only the last six periods because of the retention window. The five region marts carry all the periods and do not carry the tariff column. A tariff mart collects the commercial rows from all the regions. The five unit products store all the rows of their own region, with the key each chose. Five search tasks run over this set.

# FICTIONAL. The meter readings of the measurement network, the pipeline and the storage layout
# are defined within the lesson; no warehouse, lake, catalog product or query engine is called.
SEED, M32 = 20260218, 0xFFFFFFFF


def rng(t):
    x = ((t ^ (t >> 16)) * 2246822507) & M32
    x = ((x ^ (x >> 13)) * 3266489909) & M32
    s = [(x ^ (x >> 16)) & M32]

    def next_value():
        s[0] = (s[0] * 1664525 + 1013904223) & M32
        return s[0] / 4294967296
    return next_value

# region values stay Turkish: asset names are built from them and scanned sorted by name
# dogu/bati/kuzey/guney/merkez = east/west/north/south/central
REGION = ["dogu", "bati", "kuzey", "guney", "merkez"]
TYPE = ["residential", "commercial"]
PERIODS = [f"2026-{a:02d}" for a in range(1, 13)]
SUBSCRIBERS = {}
for i in range(200):
    r = rng(SEED + 37 * i)
    SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)],
                                    "start": 10000 + int(r() * 60000)}


def generate_source():
    records = []
    for i, (sub, a) in enumerate(SUBSCRIBERS.items()):
        e = a["start"]
        for p in PERIODS:
            r = rng(SEED + 101 * i + 7 * PERIODS.index(p))
            e += 4 + int(r() * 46)
            records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"],
                            "index": e, "unit": "m3"})
    return records


def s1_screen(records):
    REQUIRED = ("subscriber", "period", "region", "index", "unit")
    keep = [k for k in records if all(a in k for a in REQUIRED)]
    return keep, len(records) - len(keep)


def s2_dedupe(records):
    seen, keep = set(), []
    for k in records:
        key = (k["subscriber"], k["period"])
        if key in seen:
            continue
        seen.add(key)
        keep.append(k)
    return keep, len(records) - len(keep)


def s3_consumption(records):
    table = {(k["subscriber"], k["period"]): k for k in records}
    out, dropped = [], 0
    for k in records:
        i = PERIODS.index(k["period"])
        if i == 0:
            continue
        prev = table.get((k["subscriber"], PERIODS[i - 1]))
        if prev is None:
            dropped += 1
            continue
        out.append(dict(k, m3=int(k["index"]) - int(prev["index"])))
    return out, dropped


def run_pipeline(source):
    v, _ = s1_screen(source)
    v, _ = s2_dedupe(v)
    v, _ = s3_consumption(v)
    return v


def write(records, partitioning=()):
    """Distributes the records into files by the partition key."""
    layout = {}
    for k in records:
        key = tuple(k[a] for a in partitioning)
        layout.setdefault(key, []).append(k)
    return layout


ALL_ROWS = run_pipeline(generate_source())
LAST6 = sorted({k["period"] for k in ALL_ROWS})[-6:]
FULL = ("subscriber", "period", "region", "type", "m3")
REGIONAL = ("subscriber", "period", "region", "m3")     # the tariff column is not carried
MESH_KEYS = {"dogu": ("period",), "bati": ("period", "type"), "kuzey": (),
             "guney": ("period",), "merkez": ("type",)}


def asset(name, owner, records, key, columns):
    """A data asset: its rows , partition key and the columns it carries."""
    layout = write(records, key)
    return {"name": name, "owner": owner, "key": key, "columns": columns,
            "files": len(layout), "rows": {(k["subscriber"], k["period"]) for k in records},
            "region": {k["region"] for k in records}, "type": {k["type"] for k in records},
            "period": {k["period"] for k in records}}


def setup_assets():
    """The assets accumulated by the previous two lessons , sorted by name."""
    v = [asset("central_warehouse", "central team",
               [k for k in ALL_ROWS if k["period"] in LAST6], ("period",), FULL)]
    for r in REGION:
        v.append(asset(f"mart_{r}", f"{r} reporting",
                       [k for k in ALL_ROWS if k["region"] == r], ("period",), REGIONAL))
    v.append(asset("mart_tariff", "tariff team",
                   [k for k in ALL_ROWS if k["type"] == "commercial"], ("period",), FULL))
    for r in REGION:
        v.append(asset(f"product_{r}", f"{r} unit",
                       [k for k in ALL_ROWS if k["region"] == r], MESH_KEYS[r], FULL))
    return sorted(v, key=lambda x: x["name"])


def needed(task):
    return {(k["subscriber"], k["period"]) for k in ALL_ROWS
            if all(k[a] == d for a, d in task.items())}


def covers_in_fact(v, task):
    """By looking at the asset itself: are all the requested rows here."""
    return set(task) <= set(v["columns"]) and needed(task) <= v["rows"]


def covers_by_entry(g, task):
    """By looking only at the catalog entry , without opening a file."""
    if not set(task) <= set(g["columns"]):
        return False
    if task["period"] not in g["period"] or task["region"] not in g["region"]:
        return False
    return {task["type"]} <= g["type"] if "type" in task else g["type"] == set(TYPE)


def write_catalog(assets):
    """A single entry per asset: name , owner , key , files , columns , coverage."""
    return [{a: v[a] for a in ("name", "owner", "key", "files", "columns",
                               "region", "type", "period")} for v in assets]


def search_without_catalog(assets, task):
    """The files of each asset are opened in order; entries looked at = files opened."""
    looked = 0
    for v in assets:
        looked += v["files"]
        if covers_in_fact(v, task):
            return looked, v["name"]
    return looked, None


def search_with_catalog(assets, catalog, task):
    """The entries are scanned; the asset of the matching entry is verified. If it is not
    found, the search falls back to the search without a catalog and the two costs are added."""
    by_name = {v["name"]: v for v in assets}
    for i, g in enumerate(catalog, 1):
        if not covers_by_entry(g, task):
            continue
        v = by_name.get(g["name"])
        if v is not None and covers_in_fact(v, task):
            return i, g["name"], False
        looked, name = search_without_catalog(assets, task)
        return i + looked, name, True
    return len(catalog), None, False


TASKS = [("kuzey / 2026-03 / commercial", {"region": "kuzey", "period": "2026-03", "type": "commercial"}),
         ("guney / 2026-02", {"region": "guney", "period": "2026-02"}),
         ("bati / 2026-11 / residential", {"region": "bati", "period": "2026-11", "type": "residential"}),
         ("merkez / 2026-02", {"region": "merkez", "period": "2026-02"}),
         ("kuzey / 2026-05 / residential", {"region": "kuzey", "period": "2026-05", "type": "residential"})]

ASSETS = setup_assets()
CATALOG = write_catalog(ASSETS)
print(f"assets {len(ASSETS)} , files {sum(v['files'] for v in ASSETS)} , "
      f"catalog entries {len(CATALOG)}\n")
print("asset             owner             key                 files  period  columns")
for v in ASSETS:
    print(f"{v['name']:<17} {v['owner']:<17} {str(v['key']):<19} {v['files']:>4}"
          f"   {len(v['period']):>4}   {len(v['columns']):>4}")

print("\ntask                         no catalog    catalog   asset found")
t1 = t2 = 0
for name, task in TASKS:
    b1, v1 = search_without_catalog(ASSETS, task)
    b2, v2, _ = search_with_catalog(ASSETS, CATALOG, task)
    t1, t2 = t1 + b1, t2 + b2
    print(f"{name:<28} {b1:>10}  {b2:>9}   {v1}")
print(f"{'total':<28} {t1:>10}  {t2:>9}")

assets 12 , files 119 , catalog entries 12

asset             owner             key                 files  period  columns
central_warehouse central team      ('period',)            6      6      5
mart_bati         bati reporting    ('period',)           11     11      4
mart_dogu         dogu reporting    ('period',)           11     11      4
mart_guney        guney reporting   ('period',)           11     11      4
mart_kuzey        kuzey reporting   ('period',)           11     11      4
mart_merkez       merkez reporting  ('period',)           11     11      4
mart_tariff       tariff team       ('period',)           11     11      5
product_bati      bati unit         ('period', 'type')    22     11      5
product_dogu      dogu unit         ('period',)           11     11      5
product_guney     guney unit        ('period',)           11     11      5
product_kuzey     kuzey unit        ()                     1     11      5
product_merkez    merkez unit       ('type',)              2     11      5

task                         no catalog    catalog   asset found
kuzey / 2026-03 / commercial         72          7   mart_tariff
guney / 2026-02                      39          4   mart_guney
bati / 2026-11 / residential          6          1   central_warehouse
merkez / 2026-02                     61          6   mart_merkez
kuzey / 2026-05 / residential        117         11   product_kuzey
total                               295         29

The ratio between them is tenfold: five searches look at 295 entries without a catalog and 29 with one. The source of the difference is not scale but the unit. A search without a catalog opens all the files of each asset, because there is no other way of knowing which periods or which tariffs an asset covers; a search with a catalog reads a single entry per asset. Twelve assets hold 119 files, and the catalog describes the same set in 12 entries. A catalog is not an index and does not speed up the query; what it gains for the search is the ratio between the number of assets and the number of files.

The distribution among the tasks should also be read. The cheapest search is the one for bati (west): the central warehouse carries the tariff column, covers that period, and is the first asset in the order. The most expensive is kuzey (north) in 2026-05: the warehouse drops out because of the retention window, the five region marts drop out because they do not carry the tariff column, the tariff mart drops out because it holds only commercial rows, and the search goes on until the eleventh asset — 117 files without a catalog, 11 entries with one. What determines the cost of finding an asset is not the size of the set but the place of the sought thing in the order; what the catalog does is convert that order from the unit of files to the unit of entries.

The Staleness of the Catalog

The catalog carries the layout of the moment it was written. The storage side does not stand still: a mart is removed, a unit product is repartitioned, a retention window is applied to a mart. These three changes are not reflected in the entries by themselves. The run below first counts how many entries have become wrong, then repeats the same five searches with the old catalog on the new set.

# The definitions of the previous block carry on: ASSETS, CATALOG, TASKS, asset, ALL_ROWS, LAST6, FULL,
# REGIONAL, search_without_catalog, search_with_catalog.
def layout_changed(assets):
    """Three changes on the storage side after the catalog was written."""
    new = []
    for v in assets:
        if v["name"] == "mart_tariff":                 # the asset was removed
            continue
        if v["name"] == "product_kuzey":               # repartitioned
            new.append(asset(v["name"], v["owner"],
                             [k for k in ALL_ROWS if k["region"] == "kuzey"],
                             ("period", "type"), FULL))
        elif v["name"] == "mart_guney":                # the retention window narrowed
            new.append(asset(v["name"], v["owner"],
                             [k for k in ALL_ROWS if k["region"] == "guney"
                              and k["period"] in LAST6], ("period",), REGIONAL))
        else:
            new.append(v)
    return new


CHANGED = layout_changed(ASSETS)
by_name = {v["name"]: v for v in CHANGED}
print("stale entry    registered key / files / period   actual key / files / period")
stale = 0
for g in CATALOG:
    v = by_name.get(g["name"])
    if v is not None and all(g[a] == v[a] for a in ("key", "files", "period")):
        continue
    stale += 1
    actual = ("no such asset" if v is None
              else f"{str(v['key']):<18} {v['files']:>3}  {len(v['period']):>3}")
    print(f"{g['name']:<14} {str(g['key']):<18} {g['files']:>3}  {len(g['period']):>3}"
          f"   {actual}")
print(f"\nof {len(CATALOG)} entries , {stale} now point to a layout that no longer exists.")

print("\ntask                         no catalog    catalog  fallback   asset found")
t1 = t2 = 0
for name, task in TASKS:
    b1, v1 = search_without_catalog(CHANGED, task)
    b2, v2, fell = search_with_catalog(CHANGED, CATALOG, task)
    t1, t2 = t1 + b1, t2 + b2
    print(f"{name:<28} {b1:>10}  {b2:>9}  {'yes' if fell else 'no':>8}   {v2}")
print(f"{'total':<28} {t1:>10}  {t2:>9}")

stale entry    registered key / files / period   actual key / files / period
mart_guney     ('period',)         11   11   ('period',)          6    6
mart_tariff    ('period',)         11   11   no such asset
product_kuzey  ()                   1   11   ('period', 'type')  22   11

of 12 entries , 3 now point to a layout that no longer exists.

task                         no catalog    catalog  fallback   asset found
kuzey / 2026-03 / commercial        122        129       yes   product_kuzey
guney / 2026-02                     100        104       yes   product_guney
bati / 2026-11 / residential          6          1        no   central_warehouse
merkez / 2026-02                     56          6        no   mart_merkez
kuzey / 2026-05 / residential        122         11        no   product_kuzey
total                               406        251

The three changes made three of the twelve entries wrong, and the three are not of the same kind. The product_kuzey entry points to an asset that still exists; only the registered key and file count are wrong, 1 instead of 22. This entry does not break the search, it breaks the cost estimate: a party that knows it will go to that asset thinks it will touch one file. The mart_guney entry promises eleven periods and the asset carries six. The mart_tariff entry points to an asset that does not exist at all. That a catalog entry is green does not mean that the layout was measured, and there are three separate degrees of wrongness: a missed estimate, a missed coverage, a complete absence.

The last table shows what this does to the search. The catalog still gains in total, 251 against 406. But in two rows the gain turns negative. The search for kuzey in 2026-03 scans seven entries, goes to the dead entry, cannot find the asset, and falls back to a full scan from the beginning: 129, though without a catalog it would have been 122. The search for guney (south) in 2026-02 scans four entries, cannot find in the asset the period the entry promised, and takes the same road: 104, against 100 without a catalog. The difference is small but its sign matters; a stale entry puts a wrong address in front of the search, and that address cannot be eliminated without trying it. The cost of the catalog is not writing the entries but refreshing the entries at the same speed as the layout.

Summary

• A metadata catalog keeps a single entry per asset: name, owner, partition key, file count, columns, and the periods and regions covered. It is not an index; it makes the search cheaper, not the query.
• On a set of twelve assets and 119 files five searches look at 295 entries without a catalog and 29 with one; the gain comes from the ratio between the number of assets and the number of files.
• What determines the cost of a search is not the size of the set but the place of the sought asset in the order: the cheapest search looks at 6 against 1, the most expensive at 117 against 11 entries.
• Three changes make three of the twelve entries wrong, in three separate degrees: a missed estimate (a file count of 22 instead of 1), a missed coverage (six periods instead of eleven) and a complete absence.
• With the stale catalog the same five searches look at 251 entries against 406, but two searches end up more expensive than without a catalog: 129 against 122 and 104 against 100.
• The cost of the catalog is not writing the entries but refreshing the entries at the same speed as the layout; an entry that is not refreshed puts a wrong address in front of the search.

Next Step

In six lessons where the data stands was separated: the place that conforms to the schema and the raw place, the layout that keeps the two together, what taking out a subset gains and multiplies, the two numbers of distributing ownership, and the cost of finding an asset in entries. All these measurements had a common silent assumption: the row always stood in the same form — subscriber, period, region, tariff and consumption side by side, in a single flat sequence. That form was never questioned. Yet what is written into a file is not a list of rows but a table shape; which fields are measurements and which are qualifiers, how many times the same qualifier is repeated, and whether the old value of a qualifier is kept when it changes over time are separate decisions. The next topic measures these decisions: the same data, the same place, a different shape.