Apple Data Engineer (Junior Level) Interview Preparation Guide

Direct answer

Treat this as a controlled experiment, not a single before/after diff. Retrain with the new feature or component present and again with it absent, holding the code, data split, and hyperparameters fixed, and repeat each configuration across several random seeds so one lucky or unlucky run cannot masquerade as a real effect. Compare configurations with a paired statistical test on validation accuracy across matched seeds, and test the suspect changes together as well as individually, because two individually harmless changes can interact and produce a regression that neither one causes alone.

Structured elaboration

Controlled experiments to run. Define a small factorial design rather than a single before/after pair. If the change under investigation bundles more than one addition (say, a new feature A and a new preprocessing step B that shipped together), the minimum useful set of conditions is:

Every condition uses the identical training code, the identical data split per seed, and the identical hyperparameters; the only thing that varies between conditions is which of the suspect additions is active. A full factorial design costs $2^N$ configurations for $N$ suspect changes, which is fine for 2 to 3 changes but explodes past that. For a larger bundle, run leave-one-out first (start from the full shipped candidate and remove one change at a time) to get a first-order attribution, then escalate to a small factorial only among the handful of changes leave-one-out flags as suspicious, rather than paying for the full combinatorial grid up front.

Logging and comparing results. Every run writes to a shared experiment-tracking table keyed by: exact code commit hash, data-version identifier, the active-changes bitmask for that condition, the random seed, and the resulting validation accuracy (plus secondary metrics such as precision, recall, and calibration, so a regression that only shows up in one slice is not invisible to a single scalar). Comparison happens condition-by-condition as a table of mean and standard deviation across seeds, never as a single run versus a single run.

Controlling run-to-run variance so significance is assessable. The variance that matters here has two sources: data-split variance (which examples land in train versus validation) and initialization/shuffling variance (weight init, batch order, dropout masks). Pin the SAME seed to produce the SAME data split and initialization across every condition, so seed 3 in the baseline and seed 3 in the +A+B condition differ only in whether the suspect change is active, not in which examples they happened to see. This pairing is what lets you use a paired test (paired by seed) instead of an unpaired one: a paired test looks only at the within-seed differences, so shared noise across conditions (an unusually easy or hard split for seed 3) cancels out of the comparison instead of inflating the variance estimate. Run enough seeds, typically 5 to 10 for this kind of accuracy comparison, to get a real distribution of the paired difference rather than a single point estimate, and report the paired difference's mean, standard deviation, and a paired t-test p-value.

Reasoning about interactions instead of testing in isolation. Define the interaction term as what the combined condition's drop fails to explain from the sum of the individual conditions' drops:

where each $\Delta$ is the baseline-minus-condition accuracy drop. Fix the sign convention explicitly before reading the result, because every branch of the interpretation flips with it. A drop here is baseline minus condition, so a POSITIVE $\Delta$ means the condition is WORSE than baseline and a bigger positive number is a bigger regression. Under that convention $\Delta_{interaction}$ has three branches:

• $\Delta_{interaction} > 0$, super-additive: the two changes interact and make each other worse. The combined drop is larger than the sum of the individual drops, so damage is appearing that neither change produces on its own. Common mechanisms are the new feature and the new component encoding overlapping or collinear signal that confuses the model when both are present, a shared upstream preprocessing bug that only triggers when both paths run, or the second change altering what the model learns to rely on from the first (feature crowding).
• $\Delta_{interaction} \approx 0$, independent: the changes act independently and additively, and either one's individual ablation result explains its share of the regression.
• $\Delta_{interaction} < 0$, sub-additive: the pair together costs less than the two individual drops predicted, so the changes partly mask each other (often because they degrade the same examples, and those examples can only be got wrong once). Attributing each change its full individual drop here double-counts the damage.

Only the middle branch licenses reasoning about the two changes separately. This is exactly why testing each change in strict isolation and never testing the actual shipped combination can miss the real cause: neither isolated ablation shows the full regression, only the combined one does.

Worked example

Five seeded training runs per condition, same four conditions as above, same seed producing the same data split and initialization across all four conditions for a given seed index:

Paired difference per seed, shipped minus baseline: -0.0267, -0.0216, -0.0220, -0.0186, -0.0193. Mean paired difference $\bar d = -0.02164$, sample standard deviation of the differences $s_d = 0.00318$. The paired t-statistic with $n=5$ runs (4 degrees of freedom):

That gives a two-sided p-value of about 0.0001, far below any reasonable significance threshold (for example alpha = 0.05), so the shipped change's regression is real and not run-to-run noise.

Now the interaction check. Individual drops from baseline: $\Delta_A = 0.8514 - 0.8439 = 0.0075$, $\Delta_B = 0.8514 - 0.8488 = 0.0025$. (Both deltas come from the unrounded column means, 0.85138, 0.84392, 0.84884 and 0.82974; subtracting the four-decimal means printed in the table gives 0.0026 for $\Delta_B$ instead of 0.0025, which is display rounding rather than a disagreement.) If the two changes acted independently, the combined drop should be about their sum, $0.0075 + 0.0025 = 0.0100$. The actually observed combined drop is $\Delta_{A+B} = 0.8514 - 0.8297 = 0.0216$. The interaction term is:

Run that value back through the rule: $\Delta_{interaction} = +0.0116$, which is positive, which is the super-additive branch, so A and B interact and the combined condition is where the regression actually lives.

Stated in one consistent unit (one point here means one percentage point of accuracy, so 0.0075 is 0.75 points): A alone costs 0.75 points, B alone costs 0.25 points, so independence predicts 1.00 point for the pair, and the shipped pair actually costs 2.16 points. The extra 1.16 points is damage that belongs to the combination, not to either change. An investigation that only ran the two individual ablations and stopped there would have seen a three-quarter-point drop and a quarter-point drop, correctly concluded that neither alone explains a two-point regression, and then wrongly concluded "neither change is the problem" and shipped both, missing the actual cause, which only appears when both are active together.

Trade-offs and pitfalls

• Combinatorial cost. The full factorial design is cheap for 2 changes (4 configurations) and painful for 5 (32 configurations); use leave-one-out to triage first and reserve the full grid for a small suspect subset.
• Pairing discipline is what makes the significance test valid. If each condition draws its own independent data split or its own independent seed instead of a shared one, the paired test's variance-reduction assumption breaks and you need substantially more runs to reach the same statistical power.
• Statistical significance is not practical significance, and mixing units hides the difference. The p-value of about 0.0001 above says the 2.16-percentage-point drop is real, not that it is big; the same five-seed paired design would return an equally tiny p-value on a 0.05-point drop nobody would block a release over. Quote the effect size in the same unit every time (percentage points throughout this answer) so the magnitude is comparable across write-ups, and weigh it against the cost of reverting or delaying the change rather than letting the p-value stand in for size.
• Underpowered ablations look like "no effect." Five seeds is often enough for a clear regression like this one, but a small true effect can hide inside run-to-run noise with too few seeds; before concluding a change is innocent, check whether the study had enough seeds to detect a meaningfully sized effect in the first place, not just whether this particular p-value crossed a threshold.
• The common wrong turn is stopping at isolated ablations. Testing A alone and B alone and declaring victory because neither individually explains the full regression, without ever testing the shipped A+B combination, is the single most common way this kind of investigation goes wrong; interactions are common enough in real feature sets (shared preprocessing, collinear signal, feature crowding) that the combined condition belongs in the design from the start, not as an afterthought.
• A change that is genuinely a schema change to an existing feature rather than a brand-new feature follows the identical design: the schema-changed version and the pre-change version are simply the two arms of one of the paired conditions above.

Direct answer. Detect it by looking for a repeating pattern of near-identical parameterized queries in the application or database logs, one per parent row, executed in a tight loop; fix it either by loading the child rows in bulk with a single query (eager loading, batch fetching) or, at the database level, by rewriting the access pattern into one set-based query.

Structured elaboration. N+1 happens when code fetches a list of parent records with one query and then, inside a loop over that list, issues one additional query per parent to fetch its related child records, turning what should be two queries total into N+1. It's detectable in a system you didn't build by watching the query log (or an APM tool's query trace) for a burst of structurally-identical queries differing only in a single bind parameter, immediately following a query that returned a list of that same size.

At the application/ORM (object-relational mapper) layer, the standard fixes are eager loading (telling the ORM up front to fetch the related rows via a join or a second batched query rather than lazily, one at a time) or explicit batch fetching (collecting all the parent IDs first, then issuing one query with an IN-list or a join to get every child row at once). At the database/query layer, the equivalent fix is simply writing the whole thing as one query from the start: a single JOIN, or a single query filtered by the full set of parent IDs, instead of ever looping over parent rows in application code to begin with.

Worked example. A page that loads 50 blog posts and then, for each post, separately queries for its author's name is 51 queries; eager-loading the author via a JOIN in the original post-listing query, or collecting the 50 author IDs and issuing one WHERE author_id IN (...) query, both turn it back into 2 queries total regardless of how many posts are on the page.

Trade-offs and pitfalls. Eager-loading everything unconditionally can overcorrect into fetching data the caller doesn't actually need on every request, even when only some code paths use it; batch fetching driven by the actual set of IDs needed is usually the more targeted fix, at the cost of slightly more explicit code than relying on an ORM's lazy-loading default.

Direct answer

Take the representative case this format targets: a triple-nested loop that enumerates every subarray arr[i..j] and sums it with an innermost loop, to find all pairs whose sum equals a target. That loop is $\Theta(n^3)$ time. Two staged rewrites bring it down: prefix sums remove the innermost summation loop for $O(n^2)$ overall, and a hashmap of prefix sums removes one of the two remaining nested loops for $O(n)$ expected time.

Structured elaboration

# Original: O(n^3) time
def brute_force(arr, target):
    n = len(arr)
    res = []
    for i in range(n):
        for j in range(i, n):
            s = 0
            for k in range(i, j + 1):
                s += arr[k]
            if s == target:
                res.append((i, j))
    return res

Complexity of the original: the innermost loop does $j - i + 1$ additions for each $(i, j)$ pair. Summed over all pairs:
$\sum_{i=0}^{n-1} \sum_{j=i}^{n-1} (j - i + 1) = \Theta(n^3)$
Space is $O(1)$ extra beyond the output list, which itself can hold up to $O(n^2)$ pairs in the worst case (every subarray matches).

Stage 1, prefix sums bring the inner loop to $O(1)$, overall $O(n^2)$:

def prefix_sum_on2(arr, target):
    n = len(arr)
    P = [0] * (n + 1)
    for t in range(n):
        P[t + 1] = P[t] + arr[t]
    res = []
    for i in range(n):
        for j in range(i, n):
            if P[j + 1] - P[i] == target:
                res.append((i, j))
    return res

P[j+1] - P[i] replaces the whole k loop with one subtraction, since the sum of arr[i..j] is exactly the difference of two prefix sums.

Stage 2, a hashmap of prefix sums brings the whole search to $O(n)$ expected time:

from collections import defaultdict

def hashmap_on(arr, target):
    pref = 0
    idxs = defaultdict(list)
    idxs[0].append(-1)
    res = []
    for j, val in enumerate(arr):
        pref += val
        need = pref - target
        for i in idxs.get(need, []):
            res.append((i + 1, j))
        idxs[pref].append(j)
    return res

At each j, the subarrays ending at j that sum to target are exactly those starting right after an earlier index i where P[i] == P[j+1] - target; the hashmap makes that lookup $O(1)$ expected instead of scanning all earlier i.

Worked example

Verified on CPython 3.12 with a fixed seed, all three implementations agree exactly on a 30-element array:

import random
random.seed(42)
arr = [random.randint(-5, 5) for _ in range(30)]
target = 4
assert sorted(brute_force(arr, target)) == sorted(prefix_sum_on2(arr, target)) == sorted(hashmap_on(arr, target))

This assertion passes, and all three report the same 16 matching pairs.

Complexity of the improved versions: $O(n^2)$ for the prefix-sum version (two nested loops, $O(1)$ work each); $O(n + m)$ expected for the hashmap version, where $m$ is the number of matching pairs actually reported (finding them all is inherently at least $\Omega(m)$ work, since each one has to be emitted). Space is $O(n)$ for the prefix array or the hashmap, beyond the output list.

Trade-offs & pitfalls

• Edge case: if every element is strictly positive (not merely non-negative), a two-pointer sliding window reaches $O(n)$ time with $O(1)$ extra space and correctly enumerates every matching subarray, because the running sum strictly increases as the right pointer advances and strictly decreases as the left pointer advances, so each right endpoint has exactly one valid window. Verified on CPython 3.12 against the brute-force result on a strictly-positive array, this matches exactly.
• Edge case, a genuine correctness gap worth calling out explicitly: with zeros allowed in an otherwise non-negative array, the classic sliding window under-reports. Multiple valid (i, j) pairs can share the same sum once zeros sit at a boundary, but the simple version only advances the left pointer while the sum strictly exceeds the target, so it captures one window per right endpoint, not all of them. You can confirm this by comparison: on an array containing zeros, the simple sliding window returns fewer subarrays than a brute-force check, because it stops at one window per right endpoint and never enumerates the extra zero-extended windows that share that endpoint. "Sliding window works for non-negative arrays" is a common but imprecise claim; it is correct for existence checks or for strictly positive arrays, and needs extra handling (or the hashmap approach) to enumerate all matches once zeros are possible.
• The hashmap approach's $O(n)$ claim covers finding the matches; if the number of matches $m$ is itself $\Theta(n^2)$ (a target of 0 against an array that is mostly zeros, for instance), any correct algorithm that must report every match is bounded below by $\Omega(m)$, so the total runtime in that adversarial case is still $\Theta(n^2)$, output size, not the search strategy, dominates.
• Integer overflow is not a Python concern (integers are arbitrary precision), but very large prefix sums do cost more per addition as they grow beyond machine-word size; this rarely matters in practice for typical interview-scale inputs.

Direct answer
A groupby().shift() lag implementation has five edge cases that a happy-path test (one clean, sorted, multi-row group) will never exercise: single-row groups, duplicated timestamps within a group, non-monotonic (out-of-order) input, a group present in training but absent from a given test/inference batch, and a group whose values are entirely NaN. Each needs its own assertion, because each exercises a different part of the implementation: the groupby boundary, the sort-then-shift ordering, and NaN propagation are three genuinely separate code paths.

Structured elaboration
The five edge cases each isolate a different part of groupby().shift()'s behavior: the groupby boundary (does a lag ever leak across groups), the sort-then-shift ordering (does the shift respect true time order or just row order), and NaN propagation (does an all-missing group stay missing rather than being fabricated). A sample dataset and the concrete unit tests for each case follow in the worked example below.

Worked example: sample dataset, code, and unit test inputs/expected outputs

import pandas as pd, numpy as np

def add_lag1(df, group_col="id", time_col="ts", value_col="value", sort_first=True):
    if sort_first:
        df = df.sort_values([group_col, time_col], kind="mergesort")  # stable sort: preserves tie order
    df = df.copy()
    df["lag_1"] = df.groupby(group_col)[value_col].shift(1)
    return df

sample = pd.DataFrame({
    "id":    ["A", "A", "A", "B",  "C", "C", "D",  "E",  "E"],
    "ts":    [1,   2,   3,   1,    1,   1,   1,    1,    2],   # C has a duplicated timestamp
    "value": [10,  20,  30,  100,  5,   7,   np.nan, np.nan, np.nan],
})

Actually calling add_lag1 and the non-monotonic / missing-group cases below, for real

result = add_lag1(sample)
for gid in ["A", "B", "C", "D", "E"]:
    lags = result.loc[result["id"] == gid, "lag_1"].tolist()
    print(f"{gid}: lag_1 = {lags}")

f_raw = pd.DataFrame({"id": ["F", "F", "F", "F"], "ts": [3, 1, 4, 2], "value": [300, 100, 400, 200]})
f_sorted = add_lag1(f_raw, sort_first=True)
f_unsorted = add_lag1(f_raw, sort_first=False)
row_ts2_sorted = f_sorted.loc[f_sorted["ts"] == 2, "lag_1"].iloc[0]
row_ts2_unsorted = f_unsorted.loc[f_unsorted["ts"] == 2, "lag_1"].iloc[0]
print(f"Non-monotonic group F, row ts=2: lag_1 with sort_first=True -> {row_ts2_sorted}; with sort_first=False -> {row_ts2_unsorted}")

train = sample.copy()
inference_batch = pd.DataFrame({"id": ["A", "B"], "ts": [4, 2], "value": [40, 200]})
combined = pd.concat([train, inference_batch], ignore_index=True)
combined_result = add_lag1(combined)
new_a_row = combined_result[(combined_result["id"] == "A") & (combined_result["ts"] == 4)]
new_b_row = combined_result[(combined_result["id"] == "B") & (combined_result["ts"] == 2)]
print(f"Group A new row lag_1 = {new_a_row['lag_1'].iloc[0]} (A's last known value was 30)")
print(f"Group B new row lag_1 = {new_b_row['lag_1'].iloc[0]} (B's last known value was 100, not leaked from another group)")
train_counts = {g: int((train['id'] == g).sum()) for g in ['C', 'D', 'E']}
combined_counts = {g: int((combined_result['id'] == g).sum()) for g in ['C', 'D', 'E']}
print(f"C/D/E row counts unchanged by the inference batch: {train_counts} -> {combined_counts}")

Running add_lag1(sample) produces: A gets lag_1 = [NaN, 10, 20] (each row lags the previous within the group); B (single row) gets lag_1 = [NaN]; C (duplicated ts=1 twice) gets lag_1 = [NaN, 5.0]; D (single row, value is NaN) gets lag_1 = [NaN]; E (two rows, both NaN) gets lag_1 = [NaN, NaN].

Unit tests, inputs and expected outputs

1. Single-row group (B). Input: one row, ts=1, value=100. Expected: lag_1 = NaN. This catches an implementation that tries to "borrow" a lag value from an adjacent group instead of correctly scoping the shift to the group boundary.
2. Duplicated timestamps (C). Input: two rows both at ts=1, values 5 and 7. Observed (and asserted) behavior: shift() is POSITIONAL after the sort, not time-aware, so the second row in sort order still receives lag_1 = 5.0 even though its timestamp did not actually advance. The test asserts this documented behavior explicitly (rather than silently trusting it), because a duplicated-timestamp row's lag value does not represent a genuine time step and downstream consumers need a companion flag (e.g. a has_duplicate_ts column) to know that.
3. Non-monotonic input. A group F arriving in raw order ts=[3,1,4,2], value=[300,100,400,200]. Expected: after sorting, the row at ts=2 gets lag_1 = 100 (the value at ts=1, its true temporal predecessor). With sort_first=False, the SAME row instead gets lag_1 = 400, because it lags off whatever row happened to precede it in the raw, unsorted input, a distinctly different and wrong number, not a subtly-off one, which makes this an especially easy regression to catch once tested but an especially easy bug to ship if sort_first is ever accidentally disabled or bypassed upstream.
4. Group missing from a test/inference batch. Training data contains groups A through E; an inference batch contains a NEW row for group A (ts=4, value=40) and a row for group B, but no rows at all for C, D, or E. Expected: group A's new row correctly lags off A's last known value (30); groups C, D, E simply produce no output rows (they are absent from the batch, which is not an error), and critically, no lag value leaks FROM one group INTO another just because both were concatenated into the same batch.
5. Group with only NaN values (E). Input: two rows, both value=NaN. Expected: lag_1 stays NaN for both rows. This confirms the implementation never fabricates a numeric lag (e.g. via an unintended forward-fill) for a group that was never actually observed.

Trade-offs & pitfalls
The single most common bug in a hand-rolled lag implementation is forgetting sort_first entirely, or sorting by time_col alone without including group_col in the sort key, which lets rows from different groups interleave and produces a lag that silently crosses group boundaries. The second most common bug is treating duplicated timestamps as a data-quality problem to fix upstream rather than a case the lag function must have DEFINED (even if imperfect) behavior for, since upstream data will eventually contain them regardless of how the pipeline is documented, and a lag function that raises on duplicates instead of behaving predictably turns a data-quality issue into a production outage. Finally, testing lag features purely at the unit level misses the missing-group case, which only shows up once training data and a live inference batch genuinely diverge in which groups are present, so that case belongs in an integration-style test that constructs training and inference frames independently rather than slicing them from the same dataframe.

Direct answer

A topological order (an ordering of a directed acyclic graph's, or DAG's,
nodes where every edge points from earlier to later in the order) is unique,
meaning it is the single order forced by the constraints, exactly when at
every step of Kahn's algorithm there is only one node with in-degree 0 to
choose from; if two or more nodes are ever simultaneously ready, at least one
other valid order exists. When tasks additionally carry durations, the
longest path through the DAG (the critical path) is found by extending the
same topological sweep into a dynamic-programming pass: process nodes in
topological order, and for each node compute the earliest possible finish
time as its own duration plus the maximum finish time among its
predecessors.

Structured elaboration

Extension (a): detecting a unique, forced order. Kahn's algorithm
repeatedly removes a node with in-degree 0 (all its prerequisites already
placed) and decrements its neighbors' in-degrees. At each step, track how
many nodes currently have in-degree 0. If that count is ever greater than 1,
there was a real choice at that step, so a different valid order exists, and
the order is not unique. Only if every single step has exactly one ready
node is the resulting order the single one forced by the graph's edges (this
is exactly the condition for the DAG to contain a Hamiltonian path, a path
that visits every node exactly once, following edges). This is also the
underlying idea behind deriving a unique letter order from partial
ordering constraints (as in an "alien dictionary" style problem): if the
constraints pin down a single node as next at every step, the full order is
forced; if they leave two letters mutually unconstrained relative to each
other, the order is genuinely ambiguous between them, not just
under-specified by the puzzle.

Extension (b): critical path with durations. Process nodes in any valid
topological order (their existence is guaranteed since the graph is acyclic).
Maintain finish[v], initialized to duration[v] for every node. Walking
nodes in topological order, for each edge u -> v, relax:
finish[v] = max(finish[v], finish[u] + duration[v]). Because nodes are
processed in topological order, finish[u] is already final by the time any
edge u -> v is relaxed, the same reasoning that makes single-pass dynamic
programming (an approach that builds a solution from previously-solved
subproblems, avoiding recomputation) correct over a DAG. The overall critical
path length, also called the makespan (the total time to complete every
task, respecting all dependencies) in scheduling terminology, is
max(finish[v] for v in nodes), and the path itself is recovered by walking
backward through whichever predecessor produced each node's winning finish
value.

Why both extensions build on the same primitive. Kahn's algorithm's core
loop, repeatedly consuming in-degree-0 nodes, is the mechanism for both: for
uniqueness, you watch the width of the ready set at each step (is it ever
more than one); for critical path, you use topological order as a safe
processing sequence for a relaxation pass that assumes all predecessors are
already finalized, exactly the same guarantee topological order provides for
plain dependency scheduling.

Worked example

from collections import deque

def unique_topo_order(nodes, edges):
    adj = {n: [] for n in nodes}
    indeg = {n: 0 for n in nodes}
    for u, v in edges:
        adj[u].append(v); indeg[v] += 1
    ready = deque(sorted(n for n in nodes if indeg[n] == 0))
    order, is_unique = [], True
    while ready:
        if len(ready) > 1:
            is_unique = False
        node = ready.popleft()
        order.append(node)
        for nxt in adj[node]:
            indeg[nxt] -= 1
            if indeg[nxt] == 0:
                ready.append(nxt)
    if len(order) != len(nodes):
        raise ValueError("graph has a cycle")
    return order, is_unique

def critical_path(nodes, edges, duration):
    order, _ = unique_topo_order(nodes, edges)
    adj = {n: [] for n in nodes}
    for u, v in edges:
        adj[u].append(v)
    finish = {n: duration[n] for n in nodes}
    pred = {n: None for n in nodes}
    for u in order:
        for v in adj[u]:
            candidate = finish[u] + duration[v]
            if candidate > finish[v]:
                finish[v] = candidate
                pred[v] = u
    end_node = max(finish, key=lambda n: finish[n])
    path, cur = [], end_node
    while cur is not None:
        path.append(cur); cur = pred[cur]
    path.reverse()
    return finish, finish[end_node], path

# A strict chain: only one order is ever possible.
print(unique_topo_order(["A", "B", "C", "D"], [("A", "B"), ("B", "C"), ("C", "D")]))

# A diamond: B and C are both ready after A, no edge orders them relative
# to each other, so the order is not forced.
print(unique_topo_order(["A", "B", "C", "D"], [("A", "B"), ("A", "C"), ("B", "D"), ("C", "D")]))

# Critical path over a small build pipeline.
nodes = ["compile", "unit_test", "lint", "package", "deploy"]
edges = [("compile", "unit_test"), ("compile", "lint"),
         ("unit_test", "package"), ("lint", "package"), ("package", "deploy")]
duration = {"compile": 5, "unit_test": 8, "lint": 2, "package": 3, "deploy": 4}
print(critical_path(nodes, edges, duration))

Output (verified by running this exact code):

(['A', 'B', 'C', 'D'], True)
(['A', 'B', 'C', 'D'], False)
({'compile': 5, 'unit_test': 13, 'lint': 7, 'package': 16, 'deploy': 20}, 20, ['compile', 'unit_test', 'package', 'deploy'])

For the pipeline: compile finishes at 5; unit_test (depends only on
compile) finishes at 5 + 8 = 13; lint finishes at 5 + 2 = 7;
package depends on both, so it takes the later of the two,
max(13, 7) + 3 = 16; deploy finishes at 16 + 4 = 20. The critical path,
compile -> unit_test -> package -> deploy, is the chain whose durations sum
to that final 20, and it is lint's slack (7 versus the 13 it would need to
match unit_test) that keeps it off the critical path despite finishing
earlier in absolute terms.

Complexity

Both extensions run in $O(V + E)$ time (one pass over nodes and edges each,
identical to plain Kahn's algorithm) and $O(V + E)$ space for the adjacency
list, in-degree map, and the finish/predecessor maps.

Edge cases

• A graph with a cycle has no topological order at all; both functions above
  detect this by checking that the produced order includes every node, and
  raise rather than silently returning a partial or wrong result.
• A single isolated node with no edges has a trivially unique order (a
  ready-set size of exactly 1 throughout).
• Multiple disconnected components: the uniqueness check still works
  correctly, since a ready set of size greater than 1 anywhere (including
  across components) correctly signals a non-forced order.
• Ties in finish time for the critical path's end node: max with a key
  picks one such node; if multiple nodes tie for the longest finish time,
  any one of them defines a valid critical path length, though the specific
  path returned depends on iteration order if you need a canonical choice.

Trade-offs & pitfalls

• A related but different question, absorbed into this same topological-order
  family, is enumerating all valid topological orders rather than deciding
  whether there is exactly one; that requires backtracking through every
  choice point (every step where more than one node is ready) and, in the
  worst case, the number of valid orders is exponential in the number of
  independent nodes, so it is only tractable with a hard cap on how many
  orders to produce, a fundamentally different cost profile from the
  linear-time uniqueness check and critical-path computation above.
• Do not conflate "the order is not unique" with "the graph is invalid" or
  "the scheduling problem has no solution"; non-uniqueness just means the
  constraints under-specify a total order, which is the common case, not an
  error condition.
• For critical path, a frequent bug is initializing finish[v] to 0 instead
  of duration[v], which silently drops the cost of source nodes (nodes with
  no predecessors) from every downstream calculation.

Direct answer: Compute this in two separate stages because a "7-session moving average" and a window over raw events are aggregating at two different grains that cannot be collapsed into one window function call. Stage 1 collapses each session down to a single number (its average event value) with a plain GROUP BY. Stage 2 treats those per-session averages as the new rows and applies a ROWS BETWEEN 6 PRECEDING AND CURRENT ROW window, ordered by session start, partitioned by user, to average the last 7 of them. A single window function over the raw events cannot do this because its frame boundary (ROWS, RANGE, or GROUPS, the third and least common frame unit, which counts by peer-group: rows tied on the same ORDER BY value count as one step instead of one row each) always counts something about the current row's grain, rows, values, or peer-groups, and none of those map onto "the last 7 sessions" when the current grain is individual events, not sessions.

Structured elaboration

WITH session_avg AS (
  -- Stage 1: collapse events to one row per session
  SELECT
    s.session_id,
    s.user_id,
    s.session_start,
    AVG(e.event_value) AS avg_event_value_per_session
  FROM events e
  JOIN sessions s ON e.session_id = s.session_id
  GROUP BY s.session_id, s.user_id, s.session_start
)
SELECT
  user_id, session_id, session_start, avg_event_value_per_session,
  -- Stage 2: window over the session-level rows, not the event-level rows
  ROUND(AVG(avg_event_value_per_session) OVER (
    PARTITION BY user_id
    ORDER BY session_start, session_id
    ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
  ), 4) AS moving_avg_last_7_sessions
FROM session_avg
ORDER BY user_id, session_start, session_id;

Why two stages, concretely: ROWS BETWEEN 6 PRECEDING AND CURRENT ROW counts physical rows in whatever the current FROM clause produces. If that FROM clause is events JOIN sessions, one row per event, then "6 preceding rows" means 6 preceding events, which has nothing to do with 7 sessions unless every session happens to contain exactly one event. A session with 50 events and a session with 2 events contribute wildly different numbers of rows to that frame, so a naive single-stage window silently computes a moving average weighted by how chatty each session was, not an equal-weighted average of 7 session averages, a different, and wrong, metric for the stated question. Collapsing to one row per session in Stage 1 is what makes "7 rows" and "7 sessions" the same thing by the time Stage 2's window runs.

ORDER BY session_start, session_id in both the GROUP BY and the window's ORDER BY matters for the same reason ties matter in any window function: if two sessions for the same user share an identical session_start timestamp, session_id is the deterministic tiebreaker that keeps the window's row order (and therefore which 7 sessions are "the last 7") reproducible across re-runs.

Worked example

Executed against a small sessions(session_id, user_id, session_start) and events(event_id, session_id, event_value) pair: user 1 has 3 sessions. Session 1 (2 events: 10, 20) averages 15. Session 2 (2 events: 30, 50) averages 40. Session 3 (2 events: 5, 15) averages 10. The two-stage query correctly produced a moving average of 15 after session 1 (only one session so far), 27.5 after session 2 ((15+40)/2), and 21.6667 after session 3 ((15+40+10)/3), each is the plain average of all sessions seen so far since fewer than 7 sessions exist yet.

Running a naive single-stage window directly over the joined events/sessions rows, AVG(event_value) OVER (PARTITION BY user_id ORDER BY session_start, session_id ROWS BETWEEN 6 PRECEDING AND CURRENT ROW), on the exact same underlying data produced a different sequence: 10, then 15, then 20, then 27.5, then 23, then 21.6667, one number per event, six numbers total instead of three. None of those six numbers is the "moving average of session averages" the question asked for; the window is silently averaging raw event values in event order, giving sessions with more events more influence over the result purely because they contributed more rows to the frame. This is the concrete, executed proof that the single-stage approach computes a genuinely different, incorrect metric, not just a less elegant version of the right one.

Key points

• A window function's frame counts rows (or a value-range, or peer-groups) at whatever grain the query's FROM clause currently produces; changing the intended grain requires changing what rows exist first, which is what the Stage 1 GROUP BY does.
• The bug in the single-stage attempt is not a syntax error, the query runs and returns plausible-looking numbers, which is what makes it a dangerous trap rather than an obvious one.
• ROUND is cosmetic; the moving average itself works identically without it, useful to strip out when double-checking exact values against hand-computed expectations.

Complexity

Stage 1 is a straightforward aggregation: a scan of events joined to sessions, grouped by session, O(n) for n events plus the cost of the join. Stage 2 sorts the (much smaller) session-level rows by user_id, session_start, O(m log m) for m sessions, then computes the moving average in a single pass with an incrementally maintained running sum over a fixed 7-row frame, O(m). Because m (sessions) is typically orders of magnitude smaller than n (events), Stage 2 is cheap relative to Stage 1 regardless of table size.

Edge cases

• A user with fewer than 7 sessions total (as in the worked example): the window simply averages however many sessions exist so far, ROWS BETWEEN 6 PRECEDING AND CURRENT ROW never demands exactly 7 rows to exist, it caps at 7 but is content with fewer.
• A session with zero events: the JOIN in Stage 1 drops it entirely (an inner join), so it never contributes a row to session_avg; switching to a LEFT JOIN and deciding whether a sessionless average should be NULL or 0 is a deliberate choice, not a default to leave implicit.
• Two sessions sharing an identical session_start: resolved deterministically by the session_id tiebreaker in ORDER BY, without it which session counts as "more recent" in the moving window is undefined and can vary between runs.

Trade-offs & pitfalls

The single-stage version is the trap most candidates walk into under interview pressure because it looks like it should work, PARTITION BY user_id ORDER BY session_start ROWS BETWEEN 6 PRECEDING AND CURRENT ROW reads exactly like "the last 7 things," and the query runs without error. The fix is not a more clever window clause, it is recognizing that the question is really "aggregate, then aggregate the aggregate," which SQL expresses as two window/group stages chained through a CTE (common table expression, a named, reusable subquery introduced with WITH), not one window clause with a cleverer frame. If the requirement changes from "last 7 sessions" to "last 7 calendar days of sessions," Stage 2's frame changes from ROWS BETWEEN 6 PRECEDING to a RANGE BETWEEN INTERVAL '7' DAY PRECEDING on session_start, the two-stage structure itself does not need to change.

Purpose limitation: collect and use personal data only for a specified, documented purpose and not repurpose it without legal basis. Data minimization: collect, store and process only the minimum data needed for that purpose, in the least-identifiable form and for the shortest retention required.

How these change ETL design (two concrete examples):

1. Ingestion schema — restrict and validate fields at source

• Instead of ingesting full event payloads, require producers to send only fields mapped to approved use-cases. Implement schema registry + enforcement (e.g., Apache Avro/Glue schema) that rejects or strips unnecessary PII at ingestion.
• Reasoning: prevents accidental collection of unused sensitive attributes and reduces attack surface; simplifies downstream compliance.

2. Storage & transformations — pseudonymize and tier by purpose

• Immediately apply deterministic hashing or tokenization to identifiers needed for joinable analytics, and persist raw identifiers in a tightly controlled vault only when legally necessary. Store hashed IDs and aggregate-only tables in the main data lake; keep detailed raw logs in an encrypted, access-audited cold store with short retention.
• Reasoning: minimization of identifiable data available to analysts, supports purpose-limited access (analytics vs. audit) and reduces risk if the primary lake is breached.

Additional practical steps: maintain a data catalog with purpose tags, enforce column-level access policies, and document retention rules so ETL jobs automatically drop or expire fields after their legitimate retention period. These design changes make compliance operational and reduce downstream frictions.

Direct answer

A consumer product at real scale typically ingests five recognizably different kinds of source: high-volume client behavioral events, infrastructure and CDN logs, lower-volume but business-critical transactional events (billing, membership), relatively small and slow-changing catalog or content metadata, and partner-reported measurement data whose format you do not control. Each has a distinct volume profile, cardinality, and downstream audience, and conflating them under one ingestion design is where a lot of real pipelines go wrong.

Structured elaboration

Client behavioral events

• Volume: the highest-volume source by a wide margin, often billions of events per day at real scale, generated continuously by every user session.
• Cardinality: very high on dimensions like user ID and session ID, moderate on event type (a bounded, known vocabulary of action names).
• Downstream consumers: personalization and recommendation systems needing near-real-time signal, plus analytics and experimentation platforms consuming it in batch.

CDN and infrastructure logs

• Volume: also very high, driven by request volume rather than user actions, and often noisier and less structured than application-level events.
• Cardinality: high on request-level dimensions (IP, URL, timestamp), but the record shape itself is usually simpler and more uniform than a rich behavioral event.
• Downstream consumers: operations and reliability teams for real-time monitoring, plus security teams for anomaly and abuse detection.

Billing and membership events

• Volume: orders of magnitude lower than behavioral events, since they correspond to discrete business transactions rather than continuous activity.
• Cardinality: lower on most dimensions, but each individual record carries much higher business stakes than a single behavioral event does.
• Downstream consumers: finance and revenue reporting, customer support (for account status lookups), and fraud detection.

Catalog or content metadata

• Volume: the lowest-volume and slowest-changing source of the group, updated on the order of the catalog's own size and change rate, not user activity.
• Cardinality: bounded by the size of the catalog itself, typically far smaller than any of the event-volume sources.
• Downstream consumers: the personalization and search systems that join it against behavioral events, plus the product surfaces that render it directly.

Partner-reported measurement

• Volume: modest and typically batch-delivered on the partner's own schedule (daily or weekly files), not continuously streamed.
• Cardinality: depends heavily on the specific partner and measurement type, but the defining trait is that its format and delivery schedule are entirely outside your control.
• Downstream consumers: business reporting, revenue reconciliation, and any feature that specifically depends on that partner's data.

Worked example

A large streaming platform's product, personalization, and operations functions draw on exactly this mix: playback and interaction events from every viewing session (the highest-volume behavioral source, feeding both real-time personalization and batch analytics), CDN delivery logs (feeding operational dashboards and anomaly detection for stream quality issues), billing and subscription events (lower volume, high business stakes, feeding revenue reporting), catalog and content metadata (title, genre, cast, availability windows, feeding search and recommendation), and partner-reported measurement from advertising or co-production partners (batch-delivered, feeding revenue-share reconciliation). A team designing ingestion for this platform that treated all five as "just events to ingest" with one uniform pipeline would badly under-serve the billing source's correctness requirements while badly over-engineering the catalog source's freshness needs, since the two have almost nothing in common except both technically being "data."

Trade-offs & pitfalls

• The biggest real mistake is applying one uniform service-level agreement (SLA) and one uniform pipeline design across all five source types; a design tuned for the highest-volume behavioral stream is usually the wrong shape for the lowest-volume, highest-stakes billing stream, and vice versa.
• Cardinality is easy to underestimate for behavioral data specifically; a naive schema or index design that works fine at prototype scale can fail badly once real user-ID and session-ID cardinality is at production volume.
• Partner-reported measurement is the source most likely to have unannounced format drift, precisely because you have the least influence over the partner's own release process; it deserves proportionally more ingestion-time validation than its modest volume alone would suggest.
• Do not assume "downstream consumer" is singular for any of these; behavioral events in particular routinely feed both a real-time system (personalization) and a batch system (analytics) with genuinely different freshness needs from the same underlying stream.

Direct answer

Running dbt transformations inside Airflow across dev, staging, and prod comes down to three coordination problems: making the same DAG definition behave differently per environment without duplicating it, keeping concurrent runs (a scheduled prod run and a developer's manual staging run, say) from stepping on each other, and sequencing dbt models correctly alongside the non-dbt tasks (file ingestion, for example) and schema migrations they actually depend on, since dbt's own internal dependency graph only covers dbt models, not the ingestion step that has to land raw data before dbt can transform it.

Structured elaboration

Environment-specific variables. Parameterize the DAG with an environment identifier (dev, staging, prod) passed at the Airflow connection or Airflow Variable level, not hardcoded into the DAG file, and have the dbt invocation select its target (dbt run --target {env}) from dbt's own profiles.yml, which maps each named target to its own connection details. One DAG definition then serves all three environments; only the resolved variables differ per run, which avoids maintaining three near-duplicate DAG files that inevitably drift apart.

Concurrent runs. Give each environment its own Airflow pool (or its own dedicated set of dbt-invoking tasks with environment-scoped concurrency limits), so a long-running prod run and a developer's ad hoc staging run never compete for the same worker slots, and, more importantly, so dbt's own internal state (its manifest, its run results written to the target schema) for one environment is never overwritten mid-run by a concurrent run against a different environment's schema.

Dependency management between dbt models and non-dbt tasks. dbt's own dependency graph (built from ref() calls between models) only covers dbt models; it has no visibility into an upstream file-ingestion task that has to land raw source data before any dbt model can read it. Model that ingestion step as its own Airflow task, upstream of the dbt run task, with an explicit Airflow dependency edge (ingest_files >> dbt_run), so Airflow's dependency graph is the one true source of truth for "must happen before," spanning both dbt and non-dbt work, rather than expecting dbt's own graph to somehow know about a task outside of it.

Schema migrations requiring coordination with data producers. A schema change on an upstream source table (a data producer team adding or renaming a column) can break a dbt model that references it. Coordinate this the same way any cross-team contract needs coordinating: a data producer proposing a schema change signals it in advance (a pull request against the dbt project's source definitions, for instance, or a shared schema-registry entry), dbt's own source freshness and schema tests catch a mismatch as a failing test run rather than a silent wrong result, and the DAG's own failure-handling routes that test failure to an alert naming the specific model and the specific upstream source, not a generic "dbt run failed."

Worked example

The DAG uses an Airflow Variable target_env resolved per run. For a scheduled prod run, target_env=prod; the dbt task executes dbt run --target prod --profiles-dir /opt/dbt, referencing a prod target block in profiles.yml that points at the production warehouse schema. For a developer's manual staging run of the same DAG, triggered with a run configuration overriding target_env=staging, the exact same DAG and the exact same dbt project run instead against the staging target block, writing to a separate staging schema, with no code change required, only the run configuration differs.

The DAG's dependency chain is ingest_raw_files >> dbt_run (which internally sequences staging models, then mart models, per dbt's own ref() graph) >> dbt_test >> notify_on_failure (only if dbt_test fails). If a data producer renames a source column without warning, dbt_test catches it as a failing schema test for the specific downstream model that referenced the old column name, and notify_on_failure pages the owning team with that model's name in the message, rather than a downstream business-intelligence dashboard silently rendering a null column with no alert at all.

Trade-offs and pitfalls

Parameterizing environments through a single DAG definition is more maintainable long-term, but it means a bug in the shared DAG logic affects all three environments at once; a genuinely risky change to the DAG itself, not the dbt models it invokes, deserves its own staged rollout discipline, the same as any other shared infrastructure change.

Environment-scoped pools prevent concurrent runs from clobbering each other's dbt state, but they do not by themselves prevent a developer's staging run from reading stale upstream data if the staging environment's own ingestion has not run recently; staging environments commonly lag production data freshness, and that gap should be visible, not silently assumed away.

Relying solely on dbt's own tests to catch upstream schema drift means the failure is only caught after the fact, when the scheduled run actually executes; a producer team that wants earlier warning needs a separate, proactive schema-change-review process, since dbt tests are a safety net for this specific pipeline, not a substitute for the upstream coordination itself.

Evaluation framework — objectives: maximize detection (PII) with acceptable cost and ops burden across semi-structured logs and Parquet at scale.

Metrics (quantitative + qualitative)

• Precision (TP / (TP+FP)): how many flagged items are true PII.
• Recall (TP / (TP+FN)): fraction of actual PII detected.
• F1 / PR-AUC: balance for thresholded ML models.
• Throughput / latency: records/sec and end-to-end scanning latency per TB (batch) and per-minute streaming.
• False-positive cost: estimated downstream cost = (manual review time * hourly_rate * FP_count) + pipeline re-processing cost.
• False-negative risk (business impact): expected cost per FN (regulatory fines, breach risk).
• Integration complexity: estimated engineering hours to integrate with existing ingestion, catalog, audit, and masking workflows.
• Operational maintenance: daily/weekly runbook tasks, model retraining frequency, rule tuning effort, monitoring surface area (alerts, drift).
• Resource cost: compute/storage cost per TB scanned.
• Explainability & auditability: ability to trace why an item flagged (important for compliance).
• Scalability & resilience: performance under multi-TB, schema evolution.

Representative test datasets

• Synthetic labeled set: 100k+ records with injected varied PII (SSNs, emails, API keys, hashed tokens, phone numbers, names) across JSON logs and nested Parquet columns; include obfuscated, truncated, and noise variants.
• Real anonymized samples: 1–10 TB sampled from production logs and Parquet (with manual labeling on stratified subset ~10k items).
• Edge cases: binary blobs, base64, concatenated fields, multi-lingual names, timestamps mixed with IDs.
• Streaming sample: high-throughput log replay to measure latency and FP spikes.

Evaluation procedure

• Run each solution on identical datasets; compute precision/recall, throughput, cost metrics.
• Measure integration effort via spike implementation (POC): time to deploy detector into dev pipeline, to emit metadata/census, and to mask/enrich.
• Run 30-day stability test: monitor drift, FP trend, maintenance events.

Success criteria (example thresholds)

• Recall ≥ 95% for high-risk PII classes (SSN, credit card); overall recall ≥ 90%
• Precision ≥ 90% to keep manual review cost low (or FP cost below business threshold)
• Throughput: can scan production daily ingest within maintenance window (e.g., ≤ 6 hours per TB or meet streaming SLA)
• Integration: POC integration ≤ 4 engineer-weeks for chosen approach
• Operational: automated monitoring + <4 manual tuning incidents/month
• Cost: total cost (compute + human review + infra) within allocated budget

Decision guidance

• Choose cloud-native DLP if: time-to-market is critical, compliance auditability needed, and budget allows—expect lower ops but possible higher cost and less customizability.
• Choose regex/custom rules if: dataset formats are stable, required PII patterns are well-defined, and you need deterministic explainability with low baseline cost—best for high precision but recall likely lower on messy/unexpected formats.
• Choose ML-based discovery if: diverse/obfuscated PII, evolving schemas, and high recall required; accept higher ops (labeling, retraining), more compute, but better adaptability and recall.

Recommend: run POC scoring by the above metrics; if ML exceeds recall significantly without prohibitive FP cost and ops overhead fits team capacity, pick ML; otherwise prefer cloud DLP for compliance-heavy use-cases or regex for narrow predictable domains.