--- title: 'Taking the Query to the Data' source: 'https://academia.sh/en/courses/data-analysis-with-python/taking-the-query-to-the-data' course: 'Data Analysis with Python' language: en updated: '2026-10-07T04:31:11+00:00' license: 'CC BY-SA 4.0' objectives: - 'Answers the same question at four degrees of pushing and compares the number of rows and values transferred to the analysis environment' - 'Counts the items the pushed aggregation does not bring: the rows that passed the filter, the total rows in the source and the number of rows per group' - 'Measures that the gain of pushing depends on the number of groups and computes the point where the gain is zeroed' - 'Compares two totals against a threshold and writes why floating-point equality is not tested directly' --- # Taking the Query to the Data The source of the measurement network holds 3,199 rows and a single question is answered with four strategies; all four give the same number, but the number of records transferred to the analysis environment falls from 3,199 to 5. When the filter is pushed to the source 835 rows are transferred, when column selection is pushed too 1,670 values, when aggregation is pushed as well 10 values, and the gain reaches 1,919 times. The price of the gain is visibility: the pushed total does not bring the 835 rows that passed the filter, the 3,199 rows in the source or the number of rows per group, and writing the decision next to the answer raises the values transferred from 10 to 27 and lowers the ratio to 710 times. The gain also depends on the number of groups: 3,199 times with 3 groups, 426 times with 15 groups, 11 times with 835 groups, and it is zeroed when the number of groups equals the number of filtered rows. The previous lesson moved the work to several partitions and counted the cost of distribution itself. But that whole computation was built in a single direction: the data was **brought** to the analysis environment. Rows were read, split, shuffled and in the end the answer emerged. If the place where the data sits can also compute, this direction can be reversed. The measure of this lesson is a single number: **rows transferred**. How many records leave the source for the same question, and how far does this number fall when the filter and the aggregation are pushed to the source. The second number standing beside it is the item that becomes invisible in return for that gain — a pushed computation brings the answer, it does not bring the rows that produced the answer. - **SP21.** The source is fictional and holds the dataset of the previous lessons: 3,199 rows, six columns, 1,329 subscribers, three periods. It is produced with the same seed and the output is identical from run to run. **It is a model.** - **SP22.** The source is modeled as an interface that can compute where the data sits: the filter, the column selection and the aggregation can be run there. **It is a model.** - **SP23.** What is measured is the number of rows and values passing from the source to the analysis environment; how many rows the source scans inside is outside this measure. - **SP24.** For two totals to count as the same, the threshold is 0.05 m³, because readings are reported with one decimal and a difference smaller than that can come only from rounding. - **SP25.** The question is fixed: the total reading per region in the period 2024-03. This period has 835 rows and the correct answer is 17,307.2 m³. ## The Source and the Question ```python # source.py — the place where the data sits is modeled as an interface. MODEL: # in place of a real remote store, a source was written that counts every row and value # it returns for the work asked of it. The measure is the rows and values transferred to the analysis environment. SEED = 1246 REGION = ("north", "east", "center", "south", "west") PERIODS = (("2024-01", 1329), ("2024-02", 1035), ("2024-03", 835)) DENSITY = (0, 1, 2, 2, 2, 3, 4, 2, 2, 1) # center-weighted measurement network HEADER = ("subscriber_id", "meter_id", "period", "region", "reading_m3", "tier") def rng(seed: int): s = seed & 0xFFFFFFFF def next_value() -> float: nonlocal s s = (1103515245 * s + 12345) & 0xFFFFFFFF return s / 4294967296 return next_value def tier(o: float) -> str: return "0-10" if o <= 10 else "10-25" if o <= 25 else "25-40" def records() -> list[tuple]: r, out = rng(SEED), [] for period, size in PERIODS: for i in range(size): o = round(5 + r() * 32.0, 1) out.append((10000 + i, f"S-{4000 + i}", period, REGION[DENSITY[i % 10]], o, tier(o))) return out class Source: """Returns whatever is asked and counts every row and value it returns. The filter, the column selection and the aggregation can be done at the source.""" def __init__(self, rows: list): self.rows, self.rows_out, self.values_out = rows, 0, 0 def _serve(self, rows: list, width: int) -> list: self.rows_out += len(rows) self.values_out += len(rows) * width return rows def all_rows(self) -> list: return self._serve(list(self.rows), len(HEADER)) def filter_rows(self, predicate, fields=None) -> list: s = [x for x in self.rows if predicate(x)] if fields is None: return self._serve(s, len(HEADER)) ix = [HEADER.index(a) for a in fields] return self._serve([tuple(x[i] for i in ix) for x in s], len(fields)) def count(self, predicate) -> int: return self._serve([sum(1 for x in self.rows if predicate(x))], 1)[0] def sum_wide(self, predicate, group: str, measure: str) -> list: """Puts the count, the minimum and the maximum next to the total: five values.""" gi, mi = HEADER.index(group), HEADER.index(measure) t = {} for x in self.rows: if predicate(x): a = t.setdefault(x[gi], [0.0, 0, x[mi], x[mi]]) a[0], a[1] = a[0] + x[mi], a[1] + 1 a[2], a[3] = min(a[2], x[mi]), max(a[3], x[mi]) return self._serve([(g, *v) for g, v in sorted(t.items())], 5) def sum_by(self, predicate, group: str, measure: str) -> list: gi, mi = HEADER.index(group), HEADER.index(measure) t = {} for x in self.rows: if predicate(x): t[x[gi]] = t.get(x[gi], 0.0) + x[mi] return self._serve(sorted(t.items()), 2) r0 = records() TARGET = "2024-03" print(f"seed {SEED}; {len(r0)} rows, {len(HEADER)} columns, " f"{len(r0) * len(HEADER)} values in the source") print(f"question: total reading per region in the period {TARGET}; this period has " f"{sum(a for d, a in PERIODS if d == TARGET)} rows in the source") ``` ``` seed 1246; 3199 rows, 6 columns, 19194 values in the source question: total reading per region in the period 2024-03; this period has 835 rows in the source ``` The source can do three jobs: filter rows, select columns and take the total per group. All of these can also be done in the analysis environment; the difference is only in **where** they are done. The interface counts every returned row and every returned value, because what is to be measured is not the computation itself but with how many records the result of the computation is expressed. How many rows the source scans inside is outside this number, and SP23 keeps that apart. ## Four Degrees of Pushing ```python # push.py — the same question is answered with four strategies. The difference is where the # work is done: everything at the client, the filter at the source, column narrowing at the source, # and the aggregation at the source too. THRESHOLD = 0.05 # readings are reported with one decimal, a difference below that is rounding def client_sum(rows, gi: int, mi: int) -> dict: t = {} for x in rows: t[x[gi]] = t.get(x[gi], 0.0) + x[mi] return t def s1(src: Source) -> dict: # nothing is pushed return client_sum([x for x in src.all_rows() if x[2] == TARGET], 3, 4) def s2(src: Source) -> dict: # the filter is pushed return client_sum(src.filter_rows(lambda x: x[2] == TARGET), 3, 4) def s3(src: Source) -> dict: # the filter and the column selection are pushed return client_sum( src.filter_rows(lambda x: x[2] == TARGET, ["region", "reading_m3"]), 0, 1) def s4(src: Source) -> dict: # the aggregation is pushed too return dict(src.sum_by(lambda x: x[2] == TARGET, "region", "reading_m3")) G = (40, 18, 19, 10, 12, 12) fmt = lambda h: "".join(str(v).rjust(G[j]) if j else str(v).ljust(G[0]) for j, v in enumerate(h)) print(fmt(["strategy", "rows transferred", "values transferred", "ratio", "center m3", "total m3"])) base = 0 for name, f in [("nothing is pushed", s1), ("filter is pushed", s2), ("filter and column selection are pushed", s3), ("aggregation is pushed too", s4)]: src = Source(r0) t = f(src) base = base or src.values_out print(fmt([name, src.rows_out, src.values_out, f"{base / src.values_out:.0f}x", round(t["center"], 1), round(sum(t.values()), 1)])) src1, src4 = Source(r0), Source(r0) t1, t4 = s1(src1), s4(src4) diff = max(abs(t1[g] - t4[g]) for g in t1) print(f"largest difference between the region totals of the unpushed and the pushed computation " f"{diff:.1e} m3; since the threshold is {THRESHOLD} m3 they count as the same number. Because the " f"order of floating-point summation can change, equality is tested with the threshold, not with ==") ``` ``` strategy rows transferred values transferred ratio center m3 total m3 nothing is pushed 3199 19194 1x 8752.6 17307.2 filter is pushed 835 5010 4x 8752.6 17307.2 filter and column selection are pushed 835 1670 11x 8752.6 17307.2 aggregation is pushed too 5 10 1919x 8752.6 17307.2 largest difference between the region totals of the unpushed and the pushed computation 0.0e+00 m3; since the threshold is 0.05 m3 they count as the same number. Because the order of floating-point summation can change, equality is tested with the threshold, not with == ``` All four strategies give the same answer: 8,752.6 m³ in the center region and 17,307.2 m³ in total. The rows transferred fall from 3,199 to 5 and the values transferred from 19,194 to 10 — **1,919 times**. The gain is the sum of three separate pushing decisions, and each row adds one decision. Pushing the filter lowers the rows transferred from 3,199 to 835; the ratio comes directly from the selectivity. Pushing the column selection does not change the number of rows but lowers the number of values from 5,010 to 1,670, because two fields come instead of six. Pushing the aggregation, on the other hand, changes the notion of a row: what is transferred is no longer a record but a result — five groups, five rows. The last line compares the two totals. The largest difference between the region totals of the unpushed and the pushed computation is zero, but the comparison is still made with a threshold. The reason is not the result of the run but that floating-point addition is not associative: when the same numbers are added in another order the last digit can move. In what order the source adds up inside a group is outside the interface and is not written down; so equality is not tested directly, a threshold is written. ## The Price of the Gain ```python # audit.py — a pushed computation brings the result, it does not bring the decision that produced it. # First it is counted how many more values are needed to be able to write the decision next to the answer, # then how the gain of pushing melts away with the number of groups. BASE = len(r0) * len(HEADER) src = Source(r0) wide = src.sum_wide(lambda x: x[2] == TARGET, "region", "reading_m3") filtered, total_rows = src.count(lambda x: x[2] == TARGET), src.count(lambda x: True) G2 = (34, 18, 19, 10) fmt2 = lambda h: "".join(str(v).rjust(G2[j]) if j else str(v).ljust(G2[0]) for j, v in enumerate(h)) print(fmt2(["written next to the answer", "rows transferred", "values transferred", "ratio"])) print(fmt2(["total only", 5, 10, f"{BASE // 10}x"])) print(fmt2(["total, count, minimum, maximum", len(wide), 5 * len(wide), f"{BASE // (5 * len(wide))}x"])) print(fmt2(["plus filtered and total rows", src.rows_out, src.values_out, f"{BASE // src.values_out}x"])) for g, total, count, lo, hi in wide: if g in ("center", "west"): print(f"{g:8s} total {total:8.1f} m3, {count:4d} rows, average " f"{total / count:5.2f} m3, range {lo}-{hi} m3") print(f"rows that passed the filter {filtered}, total rows in the source {total_rows}; the pushed " f"total does not bring these two numbers and does not bring the number of rows per group either") # The gain of pushing depends on the number of groups: the number of result rows equals the number of groups. print(fmt2(["grouping key", "groups", "values transferred", "ratio"])) for name, field in [("tier", "tier"), ("region", "region"), ("region + tier", None), ("subscriber_id", "subscriber_id")]: src2 = Source(r0) if field is None: t = {} for x in r0: if x[2] == TARGET: t[(x[3], x[5])] = t.get((x[3], x[5]), 0.0) + x[4] src2._serve(sorted(t.items()), 3) else: src2.sum_by(lambda x: x[2] == TARGET, field, "reading_m3") print(fmt2([name, src2.rows_out, src2.values_out, f"{BASE // src2.values_out}x"])) raw = Source(r0) raw.filter_rows(lambda x: x[2] == TARGET, ["region", "reading_m3"]) print(f"the raw slice, filtered and narrowed, carries {raw.values_out} values; the aggregation " f"pushed with the subscriber key carries {2 * filtered} values") print(f"when the number of groups equals the number of filtered rows the gain of pushing is zeroed: " f"{filtered} groups, {filtered} rows") ``` ``` written next to the answer rows transferred values transferred ratio total only 5 10 1919x total, count, minimum, maximum 5 25 767x plus filtered and total rows 7 27 710x center total 8752.6 m3, 418 rows, average 20.94 m3, range 5.0-37.0 m3 west total 1789.1 m3, 83 rows, average 21.56 m3, range 5.0-37.0 m3 rows that passed the filter 835, total rows in the source 3199; the pushed total does not bring these two numbers and does not bring the number of rows per group either grouping key groups values transferred ratio tier 3 6 3199x region 5 10 1919x region + tier 15 45 426x subscriber_id 835 1670 11x the raw slice, filtered and narrowed, carries 1670 values; the aggregation pushed with the subscriber key carries 1670 values when the number of groups equals the number of filtered rows the gain of pushing is zeroed: 835 groups, 835 rows ``` The first table is the hidden default of the lesson. The pushed aggregation brings ten values, and next to those ten values no decision is written. Whether the 8,752.6 m³ of the center region came from 418 rows or from 4 rows is not in the result; that 835 rows passed the filter is not in it either, nor that there are 3,199 rows in the source. Asking for these items too raises the values transferred from 10 to 27 and lowers the gain from 1,919 times to 710 times. **The rule of the course is paid here:** a number whose decision is not written is an unmeasured number, and the price of writing that decision at this source is 17 values. The numbers themselves carry a decision too. The center region gives an average of 20.94 m³ over 418 rows and the west region an average of 21.56 m³ over 83 rows. The totals are 8,752.6 and 1,789.1 m³ and there is a fivefold difference between them; the averages are close to each other. A report that sees only the total reads the center region as "high consumption", yet the difference comes not from consumption but from the number of rows. As long as the pushed computation does not bring the number of rows, this distinction cannot be made in the report. The second table shows what the gain depends on. The rows carried by the pushed aggregation are **as many as the number of groups**; as the number of groups grows the gain melts away. The ratio is 3,199 times with a breakdown into three tiers, 1,919 times with a breakdown into five regions, 426 times with the cross of region and tier, and 11 times with a breakdown by subscriber. The last row gives exact equality: aggregation with the subscriber key carries 1,670 values, and the filtered raw slice narrowed to two columns carries 1,670 values too. **When the number of groups equals the number of filtered rows the gain of pushing is zeroed** — 835 groups, 835 rows. Past this point it is better to pull the raw slice, because the raw rows that come at the same price can answer other questions as well. ## Summary - The same question was answered at four degrees of pushing and all four gave 17,307.2 m³; the rows transferred fell from 3,199 to 5, the values transferred from 19,194 to 10 and the gain was 1,919 times. - The items separate: pushing the filter lowers the rows to 835, pushing the column selection lowers the values to 1,670, pushing the aggregation lowers the rows to 5. - The pushed total does not bring the 835 rows that passed the filter, the 3,199 rows in the source or the number of rows per group; asking for these items too raises the values transferred to 27 and lowers the ratio to 710 times. - The 8,752.6 m³ of the center region comes from 418 rows and the 1,789.1 m³ of the west from 83 rows; the averages are 20.94 and 21.56 m³ and the difference in the totals arises not from consumption but from the number of rows. - The gain of pushing depends on the number of groups: 3,199 times with 3 groups, 426 times with 15 groups, 11 times with 835 groups; when the number of groups equals the number of filtered rows the gain is zeroed. - Even if the difference between two totals comes out zero, equality is tested with a threshold, because in what order the source adds up inside a group is outside the interface. ## Next Step This topic followed a single question through five lessons: when the dataset does not fit in memory, what remains of the comfort of abstraction. The answer came in every lesson by changing the tool. Type choice shrank the column 47.9 times, chunking lowered the memory peak 32 times, the columnar layout cut the values read 6.8 times, distribution gained up to 19.7 times on a large dataset and pushing the computation to the data cut the records transferred 1,919 times. The scale problem was solved every time by changing **the layout, the partitioning or the place where the computation is done**. The only thing that did not change was the language itself. For twenty lessons the same form of expression has been used: loops, assignments, intermediate variables and computations built step by step. The next topic questions this assumption and looks at environments that express the analysis differently — it counts in how many steps the same analysis is described, how many intermediate variables the transformation chain holds and with how many layers a chart is built. The measure is the same again: what the abstraction gains and what it hides.