{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 实验 10：共享扫描（Shared Scan）\n",
    "\n",
    "## 为什么做这个实验\n",
    "\n",
    "三个查询都要扫同一张 40 万行的表。最直白的做法是各扫各的——扫三遍。可它们读的是**同一份数据**，凭什么读三遍？\n",
    "\n",
    "**共享扫描**（shared scan）的想法很直接：扫一遍，读出来的每条元组同时交给三个查询判定，各自拿走自己要的。若表有 `n` 条元组、同时有 `Q` 个查询，独立扫描要读 `n·Q` 次，共享扫描只读 `n` 次，**读取量降到 1/Q**。\n",
    "\n",
    "本实验要做的四件事：\n",
    "\n",
    "1. 实现四种扫描做法，从\"每查询各扫一遍\"到\"共享扫描 + 打标签\"。\n",
    "2. 用计数器实证\"读取元组次数\"确实降到 `1/Q`。\n",
    "3. 理解本实验最反直觉的现象：**在本机的 Python 实现下，共享扫描的墙钟时间未必更短，甚至更长**。这不是代码写错了，而是你测错了指标。\n",
    "4. 明白为什么真实数据库仍然要采用共享扫描——因为真实系统的瓶颈是 I/O，不是本实验里这种\"遍历一个 Python 列表\"。\n",
    "\n",
    "出处：Zukowski, Héman, Nes, Boncz, *Cooperative Scans: Dynamic Bandwidth Sharing in a DBMS*, VLDB 2007（提出协同扫描与循环扫描，让陆续到达的查询共享同一次扫描）。\n"
   ],
   "id": "b70028dc-7307-4f7d-b7fd-14fa49d78b0a"
  },
  {
   "cell_type": "markdown",
   "id": "dd35460d",
   "metadata": {},
   "source": [
    "## 运行前准备\n",
    "\n",
    "1. **依赖**：本笔记本只用标准库（`dataclasses`、`typing`、`unittest`）加 `faker`（造假数据）。缺 `faker` 时安装：\n",
    "\n",
    "```bash\n",
    "python -m pip install faker\n",
    "```\n",
    "\n",
    "   末尾\"大规模实测\"一节如果要画图，再装 `matplotlib`。\n",
    "2. **执行顺序**：从上到下逐单元运行。四种算法的定义分散在几个单元里，跳过会导致 `NameError`。\n",
    "3. **单元测试**：最后一个测试单元用了 `unittest.main(argv=[\"ignored\", \"-v\"], defaultTest=\"MyTest\", verbosity=2, exit=False)`。在 Jupyter 里必须这样传 `argv` 和 `exit=False`，否则它会去读 `sys.argv`（也就是 Jupyter 的启动参数）并报错、或直接把内核关掉。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9528e8b5-4817-46df-98d1-75bf3d3d367f",
   "metadata": {},
   "source": [
    "## 要执行的是这类查询\n",
    "\n",
    "``` sql\n",
    "SELECT * \n",
    "FROM stuff\n",
    "WHERE a=7\n",
    "```\n",
    "\n",
    "`stuff` 表上没有索引，所以数据库只能**全表扫描**（full table scan）：把每一条元组读出来，逐个检查谓词 `a = 7`。\n",
    "\n",
    "```\n",
    "元组 0   读 → 检查 → 丢弃\n",
    "元组 1   读 → 检查 → 丢弃\n",
    "元组 2   读 → 检查 → 命中 → 加入结果\n",
    "...\n",
    "元组 n-1 读 → 检查 → 丢弃\n",
    "```\n",
    "\n",
    "注意一个关键事实：**读取元组的次数等于 `n`，与谓词命中多少无关**。命中 0 条也是读 `n` 次，命中 `n` 条也是读 `n` 次。这个\"与选择性无关\"的性质，正是共享扫描的立论基础。\n",
    "\n",
    "### 选择性（selectivity）\n",
    "\n",
    "选择性 σ 是一个无量纲的比例，取值 `[0, 1]`：\n",
    "\n",
    "```\n",
    "σ_q = |结果_q| / n\n",
    "```\n",
    "\n",
    "含义：查询 `q` 命中的元组占全表的比例，`σ_q` 越小谓词越\"挑剔\"。本实验末尾的大规模实测里，`a` 与 `b` 都在 `[0, 100]` 上均匀取值（101 个可能值），所以 `a = 7` 的期望选择性是 `1/101 ≈ 0.0099`。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "05088ba1-38d6-493d-a6d6-176152f56a4e",
   "metadata": {},
   "source": [
    "## 生成假数据\n",
    "\n",
    "`number_of_tuples = 40`，每条元组有两个整数属性 `a`、`b`。规模小是为了让你**一眼看清结果**——打印出来的列表可以逐条核对。\n",
    "\n",
    "- `fake.seed_instance(42)` 固定随机种子，保证每次运行数据相同、结果可复现。\n",
    "- `Stuff` 是一个 dataclass，两个字段都是 `int`。\n",
    "- `fake.pyint(max_value=20)` 让 `a` 取 `[0, 20]`，`fake.pyint(max_value=42)` 让 `b` 取 `[0, 42]`。两个属性的值域不同，后面构造查询时要注意别选超出值域的常量（否则结果为空列表，而且**不会报错**）。\n",
    "\n",
    "末尾有一节把规模放大到 400,000 条，那时才看得出读取次数的差距。\n"
   ]
  },
  {
   "cell_type": "code",
   "id": "e8cb3442-39a2-41ff-823a-b5f7ef3f07ba",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2024-11-12T08:01:41.661641Z",
     "start_time": "2024-11-12T08:01:41.610838Z"
    }
   },
   "source": [
    "from faker import Faker\n",
    "\n",
    "fake = Faker()\n",
    "fake.seed_instance(42)\n",
    "\n",
    "from dataclasses import dataclass\n",
    "\n",
    "number_of_tuples = 40\n",
    "\n",
    "\n",
    "@dataclass\n",
    "class Stuff:\n",
    "    a: int\n",
    "    b: int\n",
    "\n",
    "\n",
    "# create some fake data mimicking tuples in a relation:\n",
    "data: list[Stuff] = [\n",
    "    Stuff(fake.pyint(max_value=20), fake.pyint(max_value=42))\n",
    "    for f in range(number_of_tuples)\n",
    "]\n",
    "\n",
    "data[:10]"
   ],
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Stuff(a=20, b=7),\n",
       " Stuff(a=0, b=17),\n",
       " Stuff(a=7, b=14),\n",
       " Stuff(a=4, b=6),\n",
       " Stuff(a=17, b=5),\n",
       " Stuff(a=18, b=27),\n",
       " Stuff(a=1, b=1),\n",
       " Stuff(a=2, b=13),\n",
       " Stuff(a=7, b=32),\n",
       " Stuff(a=19, b=1)]"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "execution_count": 1
  },
  {
   "cell_type": "markdown",
   "id": "e8ca52a6-73fa-4324-98aa-ca8f48207ae5",
   "metadata": {},
   "source": [
    "## 算法 1：每查询各扫一遍（independent scan）\n",
    "\n",
    "```python\n",
    "def scan(data, attribute, constant):\n",
    "    res = []\n",
    "    for el in data:\n",
    "        if el.__dict__[attribute] == constant:\n",
    "            res.append(el)\n",
    "    return res\n",
    "```\n",
    "\n",
    "这就是最朴素的全表扫描：遍历一次 `data`，把谓词成立的元组收进结果。\n",
    "\n",
    "两个细节：\n",
    "\n",
    "- `el.__dict__[attribute]` 是**动态属性查找**：属性名以字符串形式传进来，运行时才去字典里查。写死成 `el.a` 会快得多，但那样就没法用同一个函数处理不同属性了。**这个动态查找是本实验后续性能现象的关键**，先记住它。\n",
    "- 运行后应该打印出若干条 `Stuff(a=7, b=...)`。命中几条取决于随机数据；如果一条都没有，说明 `a=7` 在 `a ∈ [0, 20]` 的 40 条数据里没出现——换个常量或把数据量调大即可。\n"
   ]
  },
  {
   "cell_type": "code",
   "id": "1fe469e7-6987-4fd1-b795-2554b7a68d08",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2024-11-12T08:01:41.669574Z",
     "start_time": "2024-11-12T08:01:41.667413Z"
    }
   },
   "source": [
    "def scan(data: list[Stuff], attribute, constant) -> list[Stuff]:\n",
    "    \"\"\"Scans a list of Stuff-objects and returns all objects where the attribute is equal to the constant.\n",
    "    :param data: the list of Stuff-objects\n",
    "    :param attribute: the attribute to check\n",
    "    :param constant: the constant to check against\n",
    "    :return: a list of Stuff-objects\n",
    "    \"\"\"\n",
    "\n",
    "    res: list[Stuff] = list[Stuff]()\n",
    "    el: Stuff\n",
    "    for el in data:\n",
    "        if el.__dict__[attribute] == constant:\n",
    "            res.append(el)\n",
    "    return res\n",
    "\n",
    "\n",
    "scan(data, \"a\", 7)"
   ],
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Stuff(a=7, b=14), Stuff(a=7, b=32), Stuff(a=7, b=18)]"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "execution_count": 2
  },
  {
   "cell_type": "markdown",
   "id": "b27249fc-f49e-48ea-9031-1c29316771c5",
   "metadata": {},
   "source": [
    "## 如果查询不止一个呢\n",
    "\n",
    "``` sql\n",
    "SELECT * FROM stuff WHERE a=7\n",
    "SELECT * FROM stuff WHERE b=32\n",
    "SELECT * FROM stuff WHERE b=1\n",
    "```\n",
    "\n",
    "三个查询都在扫**同一张表**。按算法 1 的做法，要扫三遍，读取元组次数是 `3n`。\n",
    "\n",
    "表有 40 万行时，`3n = 1,200,000` 次。可这三遍读的是完全一样的数据。能不能只读一遍，把每条元组同时交给三个谓词判定？这就是下面三个算法要做的。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "13176996-2dc5-4282-bbca-0c1e025ada11",
   "metadata": {},
   "source": [
    "## 算法 2：每个查询各调一次 scan()\n",
    "\n",
    "这个单元把三个查询写成字典，然后循环调用 `scan()`：\n",
    "\n",
    "```python\n",
    "queries_dict = {\n",
    "    \"q1\": {\"attribute\": \"a\", \"constant\": 7},\n",
    "    \"q2\": {\"attribute\": \"b\", \"constant\": 32},\n",
    "    \"q3\": {\"attribute\": \"b\", \"constant\": 1},\n",
    "}\n",
    "```\n",
    "\n",
    "**算法 2 只是把算法 1 重复了三遍**——它仍然是三个独立的扫描，读取量还是 `3n`。它的价值在于给出了一个**基准结果**：后面两种共享扫描必须用同样的 `queries_dict` 跑出**完全相同**的结果，才能说明改写是等价的。\n",
    "\n",
    "运行后打印三行（`q1`、`q2`、`q3` 各自的结果列表）。请留意每行的条数，后面要和共享扫描的结果逐条对上。\n"
   ]
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "### 为什么查询要用字典表示\n",
    "\n",
    "因为谓词要在运行时才能确定。若把查询写死成 `el.a == 7`，就没法循环处理 `q1`/`q2`/`q3` 三个不同的谓词了。字典把\"属性名\"和\"常量\"都变成数据，算法 3、4 才能用一个统一的循环遍历所有查询。\n",
    "\n",
    "代价就是前面说的**动态查找**：`el.__dict__[entry[\"attribute\"]]` 要比 `el.a` 多查两次字典。\n"
   ],
   "id": "92fed3f2f25c2e3"
  },
  {
   "cell_type": "code",
   "id": "1c0099bc-c6fa-4e9d-a73f-15a226887c04",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2024-11-12T08:01:41.732002Z",
     "start_time": "2024-11-12T08:01:41.729881Z"
    }
   },
   "source": [
    "from typing import Dict\n",
    "\n",
    "# a dictionary with the queries we want to run:\n",
    "queries_dict: Dict[str, Dict[str, str | int]] = {\n",
    "    \"q1\": {\"attribute\": \"a\", \"constant\": 7},\n",
    "    \"q2\": {\"attribute\": \"b\", \"constant\": 32},\n",
    "    \"q3\": {\"attribute\": \"b\", \"constant\": 1},\n",
    "}\n",
    "\n",
    "key: str\n",
    "value: Dict[str, str]\n",
    "\n",
    "# run each query with a separate call to scan() for each query:\n",
    "for key, value in queries_dict.items():\n",
    "    print(key, scan(data, value[\"attribute\"], value[\"constant\"]))"
   ],
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "q1 [Stuff(a=7, b=14), Stuff(a=7, b=32), Stuff(a=7, b=18)]\n",
      "q2 [Stuff(a=7, b=32)]\n",
      "q3 [Stuff(a=1, b=1), Stuff(a=19, b=1)]\n"
     ]
    }
   ],
   "execution_count": 3
  },
  {
   "cell_type": "markdown",
   "id": "1d4a5adc-25d2-41c6-a3f2-25ef7f7cff11",
   "metadata": {},
   "source": [
    "## 算法 3：共享扫描 + 每个查询一份结果列表\n",
    "\n",
    "```python\n",
    "res = {k: [] for k in queries_dict}      # 每个查询一份结果列表\n",
    "for el in data:                          # 整表只扫一遍\n",
    "    for query, entry in queries_dict.items():\n",
    "        if el.__dict__[entry[\"attribute\"]] == entry[\"constant\"]:\n",
    "            res[query].append(el)\n",
    "```\n",
    "\n",
    "结构上的变化只有一处：**两层 `for` 的顺序调换了**。算法 2 是\"外层遍历查询、内层遍历表\"；算法 3 是\"外层遍历表、内层遍历查询\"。\n",
    "\n",
    "就这一处调换，读取元组次数从 `n·Q` 降到 `n`：表只被遍历了一次。\n",
    "\n",
    "- **结果条数完全不变。** `res[\"q1\"]` 与算法 2 里 `scan(data, \"a\", 7)` 的结果应逐条相同——因为它们判的是同一个谓词，只是遍历顺序不同。\n",
    "- **谓词求值次数一点没省。** 每条元组仍要对 `Q` 个谓词各判一次，共 `n·Q` 次。共享扫描省的是**数据读取**，不是**计算**。这一点非常重要，它解释了为什么共享扫描在 CPU 主导的场景里不快。\n",
    "- 一条元组若同时命中两个查询，会被**存两份**（分别进两个结果列表）。\n"
   ]
  },
  {
   "cell_type": "code",
   "id": "68407bdb-35bf-448e-8bf8-3d56e1c9c900",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2024-11-12T08:01:41.740029Z",
     "start_time": "2024-11-12T08:01:41.737537Z"
    }
   },
   "source": [
    "def shared_scan(\n",
    "    data: list[Stuff], queries_dict: Dict[str, Dict[str, str]]\n",
    ") -> dict[str, list[Stuff]]:\n",
    "    \"\"\"Shared scan with result lists. Scans a list of Stuff-objects and returns a dictionary with the query as key and for each\n",
    "    query the list of results as value.\"\"\"\n",
    "\n",
    "    # create a result dictionary with one result list per incoming query:\n",
    "    res: dict[str, list[Stuff]] = {k: [] for k in queries_dict}\n",
    "\n",
    "    # loop over all tuples:\n",
    "    el: Stuff\n",
    "    for el in data:\n",
    "        query: str\n",
    "        entry: Dict[str, str]\n",
    "        # check for each query...\n",
    "        for query, entry in queries_dict.items():\n",
    "            # whether this tuple is part of the result:\n",
    "            if el.__dict__[entry[\"attribute\"]] == entry[\"constant\"]:\n",
    "                # append to dictionary:\n",
    "                res[query].append(el)\n",
    "\n",
    "    return res\n",
    "\n",
    "\n",
    "shared_scan(data, queries_dict)"
   ],
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'q1': [Stuff(a=7, b=14), Stuff(a=7, b=32), Stuff(a=7, b=18)],\n",
       " 'q2': [Stuff(a=7, b=32)],\n",
       " 'q3': [Stuff(a=1, b=1), Stuff(a=19, b=1)]}"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "execution_count": 4
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "## 算法 4：共享扫描 + 单个列表 + 打标签（tagging）\n",
    "\n",
    "算法 3 里，同时命中两个查询的元组被存了两份。算法 4 改成**只存一份**，用一个标签列表标明它属于哪些查询：\n",
    "\n",
    "```python\n",
    "@dataclass\n",
    "class Result:\n",
    "    result: Stuff                       # 元组本身\n",
    "    relevant_for_queries: list[str]     # 标签：属于哪些查询\n",
    "```\n",
    "\n",
    "核心循环与算法 3 几乎一样，只是把\"往对应查询的列表里 append\"换成\"往标签列表里 append 查询名\"，最后**只有至少命中一个查询的元组才被保留**。\n",
    "\n",
    "### 两种存法的取舍\n",
    "\n",
    "记 `S = Σ_q |结果_q|`（分列表版本的条目总数，含重复），`U = |∪_q 结果_q|`（打标签版本的条目总数，去重）：\n",
    "\n",
    "- 查询之间结果**重叠大**时，`U` 远小于 `S`，打标签版本省内存。极端情况：三个查询完全相同，`U = S/3`。\n",
    "- 查询之间结果**几乎不相交**时，`U ≈ S`，打标签版本还要多付一份标签列表的开销，反而更占。\n",
    "\n",
    "本实验的 `q1` 作用在 `a` 上、`q2`/`q3` 作用在 `b` 上，结果几乎不相交，**所以这里看不出打标签的内存优势**（末尾的大规模实测会给出具体数字）。想看到优势，把三个查询改成同一个谓词再试。\n",
    "\n",
    "### `join_result_list_with_bit_lists()` 是干什么的\n",
    "\n",
    "它把打标签的结果还原成 `{查询: 结果列表}` 的字典，好让单元测试能和算法 2、3 直接比较。**它只做格式转换，不是查询处理的一部分。**\n"
   ],
   "id": "894918a2c0ff42a9"
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2024-11-12T08:01:41.751835Z",
     "start_time": "2024-11-12T08:01:41.748624Z"
    }
   },
   "cell_type": "code",
   "source": [
    "@dataclass\n",
    "class Result:\n",
    "    \"\"\"A result class that holds the result of a query and a list of query names (tags) indicating whether the tuple is relevant for the query (part of the query result).\"\"\"\n",
    "\n",
    "    result: Stuff\n",
    "    relevant_for_queries: list[str]\n",
    "\n",
    "\n",
    "def shared_scan_with_tagging(\n",
    "    data: list[Stuff], queries_dict: Dict[str, Dict[str, str]]\n",
    ") -> list[Result]:\n",
    "    \"\"\"Shared scan using tagging. Creates a list of Result-objects, each element is tagged with a list of query names indicating\n",
    "    whether the tuple is relevant for the query.\"\"\"\n",
    "\n",
    "    res = list[Result]()\n",
    "\n",
    "    # loop over all tuples:\n",
    "    el: Stuff\n",
    "    for el in data:\n",
    "        # a white list of query names indicating whether the tuple is relevant for the query:\n",
    "        relevant_for_queries: list[str] = list[str]()\n",
    "\n",
    "        # check predicate for each query...\n",
    "        query: str\n",
    "        entry: Dict[str, str]\n",
    "        for query, entry in queries_dict.items():\n",
    "            # check whether this tuple is part of the result and append True or False to bit list:\n",
    "            if el.__dict__[entry[\"attribute\"]] == entry[\"constant\"]:\n",
    "                relevant_for_queries.append(query)\n",
    "\n",
    "        # only append anything to the result list if at least one query has the el as its result:\n",
    "        if len(relevant_for_queries) > 0:\n",
    "            res.append(Result(el, relevant_for_queries))\n",
    "\n",
    "    return res\n",
    "\n",
    "\n",
    "def join_result_list_with_bit_lists(\n",
    "    tagged_result: list[Result],\n",
    ") -> dict[str, list[Stuff]]:\n",
    "    \"\"\"Builds a dictionary with the query as key and the list of results as value.\n",
    "    This is required to be able to compare the results of the different algorithms in the unit tests.\n",
    "    :param result: the list of Result-objects\n",
    "    :return: a dictionary with the query as key and the list of results as value\n",
    "    \"\"\"\n",
    "\n",
    "    result: dict[str, list[Stuff]] = dict[str, list[Stuff]]()\n",
    "    item: Result\n",
    "    for item in tagged_result:\n",
    "        query: str\n",
    "        for query in item.relevant_for_queries:\n",
    "            result.setdefault(query, list[Stuff]()).append(item.result)\n",
    "\n",
    "    return result\n",
    "\n",
    "\n",
    "# run the algorithm\n",
    "print(\"Result of shared scan with tagging:\")\n",
    "result: list[Result] = shared_scan_with_tagging(data, queries_dict)\n",
    "for r in result:\n",
    "    print(r)\n",
    "\n",
    "# convert the result to a dictionary with the query as key and the list of results as value (as above):\n",
    "print(\"\\nconverted dict:\")\n",
    "result_as_dict: dict[str, list[object]] = join_result_list_with_bit_lists(result)\n",
    "\n",
    "import pprint\n",
    "\n",
    "pp = pprint.PrettyPrinter(indent=4)\n",
    "pp.pprint(result_as_dict)"
   ],
   "id": "4dc2c7cdd092463e",
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Result of shared scan with tagging:\n",
      "Result(result=Stuff(a=7, b=14), relevant_for_queries=['q1'])\n",
      "Result(result=Stuff(a=1, b=1), relevant_for_queries=['q3'])\n",
      "Result(result=Stuff(a=7, b=32), relevant_for_queries=['q1', 'q2'])\n",
      "Result(result=Stuff(a=19, b=1), relevant_for_queries=['q3'])\n",
      "Result(result=Stuff(a=7, b=18), relevant_for_queries=['q1'])\n",
      "\n",
      "converted dict:\n",
      "{   'q1': [Stuff(a=7, b=14), Stuff(a=7, b=32), Stuff(a=7, b=18)],\n",
      "    'q2': [Stuff(a=7, b=32)],\n",
      "    'q3': [Stuff(a=1, b=1), Stuff(a=19, b=1)]}\n"
     ]
    }
   ],
   "execution_count": 5
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "## 单元测试：三种算法必须给出相同结果\n",
    "\n",
    "测试做的事：随机造 2000 条数据、随机生成 30 个查询，然后跑三种算法，用 `assertEqual` 比对结果字典。重复 10 轮。\n",
    "\n",
    "三个要看懂的点：\n",
    "\n",
    "1. **为什么能直接 `assertEqual` 而不先排序？** 因为三种算法产生结果列表的**顺序恰好一致**：外层都按 `queries_dict` 的键序，内层都按 `data` 的顺序 append。所以可以直接比。如果你改写了算法（比如并行执行），顺序就可能不同，那时必须先按某个规范序排序再比——关系是无序集合，只有内容相同才算等价。\n",
    "2. **这是差异测试**（differential testing）：同一件事用两条不同路径做，结果必须一致。它是验证\"改写等价\"最有效的手段。\n",
    "3. **`unittest.main` 的参数**：`argv=[\"ignored\", \"-v\"]` 阻止它去读 Jupyter 的 `sys.argv`，`exit=False` 防止它跑完把内核关掉，`defaultTest=\"MyTest\"` 只跑这一个测试类。\n",
    "\n",
    "运行后应看到 `Ran 1 test ... OK`。\n"
   ],
   "id": "8e21b5e4d8c45410"
  },
  {
   "cell_type": "code",
   "id": "87f97945-8130-4962-90ad-bcb2413650e1",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2024-11-12T08:01:41.881570Z",
     "start_time": "2024-11-12T08:01:41.757469Z"
    }
   },
   "source": [
    "import unittest\n",
    "\n",
    "\n",
    "class MyTest(unittest.TestCase):\n",
    "\n",
    "    # the following function will be called before every test-function:\n",
    "\n",
    "    def create_data(self, number_of_tuples: int):\n",
    "        self.data: list[Stuff] = [\n",
    "            Stuff(fake.pyint(max_value=20), fake.pyint(max_value=20))\n",
    "            for f in range(number_of_tuples)\n",
    "        ]\n",
    "\n",
    "    def test_scans_end_to_end(self):\n",
    "        i: int\n",
    "        for i in range(10):\n",
    "            self.create_data(2000)\n",
    "\n",
    "            # a dictionary with queries\n",
    "            _queries_dict: dict[str, dict[str, str | int]] = {\n",
    "                \"q\"\n",
    "                + str(i): {\n",
    "                    \"attribute\": \"a\" if fake.pybool() else \"b\",\n",
    "                    \"constant\": fake.pyint(max_value=20),\n",
    "                }\n",
    "                for i in range(30)\n",
    "            }\n",
    "            # compute the result for all these queries by calling the scan()-method for each query independently:\n",
    "            res1_dict: dict[str, list[Stuff]] = {\n",
    "                query: scan(self.data, entry[\"attribute\"], entry[\"constant\"])\n",
    "                for query, entry in _queries_dict.items()\n",
    "            }\n",
    "\n",
    "            # single call to the shared_scan-method:\n",
    "            res2_dict: dict[str, list[Stuff]] = shared_scan(self.data, _queries_dict)\n",
    "\n",
    "            # single call to the shared_scan-method with tagging:\n",
    "            res3_dict: dict[str, list[Stuff]] = join_result_list_with_bit_lists(\n",
    "                shared_scan_with_tagging(self.data, _queries_dict)\n",
    "            )\n",
    "\n",
    "            # end to end comparison of the results:\n",
    "            self.assertEqual(res1_dict, res2_dict)\n",
    "            self.assertEqual(res1_dict, res3_dict)\n",
    "\n",
    "\n",
    "# only execute a specific Test-class:\n",
    "unittest.main(argv=[\"ignored\", \"-v\"], defaultTest=\"MyTest\", verbosity=2, exit=False)"
   ],
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "test_scans_end_to_end (__main__.MyTest.test_scans_end_to_end) ... ok\n",
      "\n",
      "----------------------------------------------------------------------\n",
      "Ran 1 test in 0.118s\n",
      "\n",
      "OK\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<unittest.main.TestProgram at 0x1079c5f10>"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "execution_count": 6
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2024-11-12T08:01:41.941304Z",
     "start_time": "2024-11-12T08:01:41.939932Z"
    }
   },
   "cell_type": "code",
   "source": "",
   "id": "aad13bcbe291173a",
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "id": "19d58d0e",
   "metadata": {},
   "source": [
    "## 大规模实测：读取元组次数降到 1/Q\n",
    "\n",
    "前面 40 条数据只够看清逻辑。要看出\"读取量降到 1/Q\"，得把规模做大。下面这段代码可以直接粘进**新建的代码单元**运行（只用标准库，不需要额外装包）：\n",
    "\n",
    "```python\n",
    "import random, time\n",
    "from dataclasses import dataclass\n",
    "from typing import Dict, List\n",
    "\n",
    "n = 400_000\n",
    "\n",
    "@dataclass\n",
    "class Stuff:\n",
    "    a: int\n",
    "    b: int\n",
    "\n",
    "random.seed(42)\n",
    "data = [Stuff(random.randint(0, 100), random.randint(0, 100)) for _ in range(n)]\n",
    "\n",
    "tuples_read = 0\n",
    "def reset_counter():\n",
    "    global tuples_read\n",
    "    tuples_read = 0\n",
    "\n",
    "queries_dict = {\"q1\": {\"attribute\": \"a\", \"constant\": 7},\n",
    "                \"q2\": {\"attribute\": \"b\", \"constant\": 32},\n",
    "                \"q3\": {\"attribute\": \"b\", \"constant\": 1}}\n",
    "\n",
    "def scan(data, attribute, constant):\n",
    "    global tuples_read\n",
    "    res = []\n",
    "    for el in data:\n",
    "        tuples_read += 1                      # 每读一条元组计一次\n",
    "        if el.__dict__[attribute] == constant:\n",
    "            res.append(el)\n",
    "    return res\n",
    "\n",
    "def shared_scan(data, queries_dict):\n",
    "    global tuples_read\n",
    "    res = {k: [] for k in queries_dict}\n",
    "    for el in data:\n",
    "        tuples_read += 1                      # 整表只扫一遍\n",
    "        for key, entry in queries_dict.items():\n",
    "            if el.__dict__[entry[\"attribute\"]] == entry[\"constant\"]:\n",
    "                res[key].append(el)\n",
    "    return res\n",
    "\n",
    "reset_counter()\n",
    "r1 = {k: scan(data, e[\"attribute\"], e[\"constant\"]) for k, e in queries_dict.items()}\n",
    "print(\"独立扫描 读取:\", tuples_read, \" 各查询条数:\", {k: len(v) for k, v in r1.items()})\n",
    "\n",
    "reset_counter()\n",
    "r2 = shared_scan(data, queries_dict)\n",
    "print(\"共享扫描 读取:\", tuples_read, \" 各查询条数:\", {k: len(v) for k, v in r2.items()})\n",
    "```\n",
    "\n",
    "### 实测数字（本案例库，n = 400,000，Q = 3）\n",
    "\n",
    "| 做法 | 读取元组次数 | 各查询结果条数 |\n",
    "| --- | ---: | --- |\n",
    "| 独立扫描 | **1,200,000** | `[4006, 3992, 3927]` |\n",
    "| 共享扫描（分列表） | **400,000** | `[4006, 3992, 3927]` |\n",
    "| 共享扫描（打标签） | **400,000** | 同上 |\n",
    "\n",
    "- `1,200,000 = 400,000 × 3`，正好是 `n·Q`；共享扫描 `400,000 = n`。比值恰好等于 `Q = 3`。\n",
    "- 三个查询的期望命中数是 `400,000 / 101 ≈ 3,960`（`a`、`b` 各有 101 个可能取值）。实测 `[4006, 3992, 3927]` 都在期望值附近，偏差来自随机抽样。换随机种子会得到不同的数字（本案例库另一个 `random.seed(42)` 版本给出 `[3982, 3979, 3940]`）——**以你自己的运行为准**，重要的是三种算法给出的结果彼此一致。\n",
    "- 打标签版本的条目数（并集大小）与 `Σ` 之差，就是\"被多个查询共同命中的元组\"的重复计数。本案例库实测两者相差约 76 条；理论估算 `400,000 × 2 × (1/101)² ≈ 78`（`q2` 与 `q3` 都作用在 `b` 上且互斥，只有 `q1` 与 `q2`/`q3` 之间可能重叠）。\n",
    "\n",
    "### 必须诚实说明：墙钟时间反而更长\n",
    "\n",
    "用秒表测这两种做法，你会看到**共享扫描不比独立扫描快，甚至更慢**（本案例库实测：共享/独立的耗时比值约 1.46，需自行实测确认）。原因有三条：\n",
    "\n",
    "1. 共享扫描的内层是 `el.__dict__[entry[\"attribute\"]] == entry[\"constant\"]`，每条元组每个查询要做三次字典查找；独立扫描里写死的属性访问加字面常量要快得多。\n",
    "2. 多一层 `for` 循环，每元组都要建一次迭代器并解包。\n",
    "3. **最根本的**：本实验的\"读取一条元组\"只是遍历 Python 列表取一个元素，成本几十纳秒；而真实数据库读一条元组要付页面调度、磁盘/网络传输、元组解析，成本高几个数量级。省下的\"读取\"在 Python 里根本不值钱，抵不过多出来的动态查找。\n",
    "\n",
    "**所以本实验正确的观测指标是\"读取元组次数\"，不是秒表。** 想看到共享扫描在墙钟上反超，就给\"读一条元组\"人为加一点开销（模拟 I/O），例如每条元组加 0.2 μs，比值会掉到 0.5 附近；继续加大，比值趋近理论下界 `1/Q`。具体反超点在哪个数值上，需自行实测确认。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0ba99fc8",
   "metadata": {},
   "source": [
    "## 思考题与详解\n",
    "\n",
    "**1. 本实验（n = 400,000）三个查询的选择性各是多少？用实测条数算出来，并与理论期望比较。**\n",
    "\n",
    "<details><summary>答案要点</summary>\n",
    "\n",
    "定义 `σ_q = |结果_q| / n`。用实测条数 `[4006, 3992, 3927]`：\n",
    "\n",
    "```\n",
    "σ(q1) = 4006 / 400000 = 0.0100150\n",
    "σ(q2) = 3992 / 400000 = 0.0099800\n",
    "σ(q3) = 3927 / 400000 = 0.0098175\n",
    "```\n",
    "\n",
    "理论期望：`a`、`b` 都在 `[0, 100]` 上均匀取值，共 101 个可能值，所以 `σ = 1/101 ≈ 0.0099010`，期望命中数 `400,000 / 101 ≈ 3,960`。\n",
    "\n",
    "三组实测都在期望值附近，偏差来自随机抽样（标准差约 `√(n·σ·(1−σ)) ≈ 63`，实测偏差在这个量级内）。\n",
    "\n",
    "</details>\n",
    "\n",
    "**2. 为什么共享扫描把读取元组次数降到 1/Q，但谓词求值次数一点没省？请用公式说明。**\n",
    "\n",
    "<details><summary>答案要点</summary>\n",
    "\n",
    "**读取元组次数**\n",
    "\n",
    "- 独立扫描：外层 `Q` 个查询，每个内层 `n` 次 → `n·Q`。\n",
    "- 共享扫描：只有一层 `n` 次的循环 → `n`。\n",
    "\n",
    "比值 `n / (n·Q) = 1/Q`。\n",
    "\n",
    "**谓词求值次数**\n",
    "\n",
    "- 独立扫描：每条元组对每个查询判一次 → `n·Q`。\n",
    "- 共享扫描：每条元组在内层 `for q in queries` 里对每个查询判一次 → `n·Q`。\n",
    "\n",
    "**完全一样。** 因为\"要判断这条元组属于哪些查询\"这个工作量与\"分几遍扫\"无关：`Q` 个查询 × `n` 条元组 = `n·Q` 次判断，这是下界（除非用索引，或利用谓词之间的蕴含关系做剪枝）。\n",
    "\n",
    "一句话：共享扫描省的是**数据读取**，不是**计算**。理解这一点，就理解了为什么它在 CPU 主导的场景（本实验的 Python 实现）里不快。\n",
    "\n",
    "</details>\n",
    "\n",
    "**3. 打标签版本（算法 4）在什么情况下比分列表版本（算法 3）更省内存？什么情况下更费？给出判据。**\n",
    "\n",
    "<details><summary>答案要点</summary>\n",
    "\n",
    "记 `S = Σ_q |结果_q|`（分列表版本条目总数，含重复），`U = |∪_q 结果_q|`（打标签版本条目总数，去重），每条标签的开销为 `c`（一个字符串引用的列表）。\n",
    "\n",
    "- 分列表版本占用 ≈ `S`。\n",
    "- 打标签版本占用 ≈ `U + U·c`。\n",
    "\n",
    "**判据** `U·(1 + c) < S`，即重叠度 `S/U > 1 + c` 时打标签更省。\n",
    "\n",
    "- 重叠大（`U << S`）：打标签明显更省。极端情况三个查询完全相同，`U = S/3`，省 `2/3`。\n",
    "- 几乎不相交（`U ≈ S`）：打标签还要多付 `U·c`，反而更占。\n",
    "\n",
    "本实验实测 `S/U ≈ 1.006`（相差约 76 条），远小于 `1 + c`，**所以本实验里打标签版本几乎没有内存优势**——因为三个查询分别作用在 `a` 和 `b` 两个不同属性上，结果几乎不相交。\n",
    "\n",
    "想看到优势，把三个查询改成同一个谓词（如都是 `a=7`），此时 `S = 3 × |结果|`、`U = |结果|`，比值 3.0。\n",
    "\n",
    "</details>\n",
    "\n",
    "**4. 本实验测出共享扫描的墙钟时间不比独立扫描短。请列出至少三条具体原因。**\n",
    "\n",
    "<details><summary>答案要点</summary>\n",
    "\n",
    "1. **动态属性查找。** 共享扫描内层是 `el.__dict__[entry[\"attribute\"]] == entry[\"constant\"]`，每条元组每个查询三次字典查找；独立扫描里是直接属性访问加字面常量。\n",
    "2. **多一层循环。** 共享扫描是嵌套两层 `for`，内层 `for key, entry in queries_dict.items()` 每元组要建一次迭代器并做元组解包。\n",
    "3. **\"读取\"本身太便宜。** 本实验读一条元组就是遍历 Python 列表取一个元素，约几十纳秒；省下的读取抵不过多出的查找。\n",
    "4. **打标签版本还要分配对象。** 每条命中元组要新建一个 `Result` 对象和一个标签列表，多两次对象分配。\n",
    "\n",
    "验证第 1 点可以用 `timeit` 比较 `el.a == 7`、`el.__dict__[\"a\"] == 7`、完整动态查找三者，看 ns/次 的倍数关系（具体数字需自行实测确认）。\n",
    "\n",
    "</details>\n",
    "\n",
    "**5. 既然墙钟上没赢，为什么真实数据库仍然采用共享扫描？请结合代价公式说明。**\n",
    "\n",
    "<details><summary>答案要点</summary>\n",
    "\n",
    "简化的扫描代价模型：\n",
    "\n",
    "```\n",
    "Cost(扫描) = P × (t_seek + t_transfer) + n × t_cpu\n",
    "```\n",
    "\n",
    "符号：`P` 是扫描的页面数；`t_seek` 是单次寻道 + 旋转延迟（机械盘毫秒级、SSD 微秒级）；`t_transfer` 是传输一页的时间；`n` 是元组条数；`t_cpu` 是每元组的 CPU 处理时间（纳秒到微秒级）。`t_seek` 比 `t_cpu` 大 3 到 6 个数量级，所以只要 `P` 够大，第一项完全主导。\n",
    "\n",
    "两种做法：\n",
    "\n",
    "```\n",
    "Cost(独立) = Q·P·(t_seek + t_transfer) + n·Q·t_cpu\n",
    "Cost(共享) = 1·P·(t_seek + t_transfer) + n·Q·t_cpu\n",
    "```\n",
    "\n",
    "相减：\n",
    "\n",
    "```\n",
    "Cost(独立) − Cost(共享) = (Q − 1)·P·(t_seek + t_transfer) > 0\n",
    "```\n",
    "\n",
    "**第二项（CPU）完全相同，被抵消了。** 只要 `P > 0`，共享扫描一定不亏；`P` 越大越赚。\n",
    "\n",
    "本实验之所以看不出优势，是因为本实验的\"读取\"里没有 `(t_seek + t_transfer)` 这一项——数据已经在内存里了。给\"读一条元组\"加上模拟 I/O 开销后，就能看到共享扫描反超。\n",
    "\n",
    "补充：有人会说\"真实数据库有缓冲池，第二遍扫描会命中缓存\"。这不成立，因为表往往远大于缓冲池，第二遍时第一遍的页面早被淘汰了；即便命中，元组解析的 CPU 开销也不小。共享扫描提供的是**确定性**的 `1/Q` 收益，不是靠运气的缓存命中。\n",
    "\n",
    "</details>\n",
    "\n",
    "**6. 如果三个查询的选择性分别是 0.5（各命中一半），共享扫描还划算吗？请就\"读取量\"和\"结果内存\"两个维度分别讨论。**\n",
    "\n",
    "<details><summary>答案要点</summary>\n",
    "\n",
    "设 `n = 1,000,000`，`Q = 3`，三个查询各命中一半，且结果相互独立随机。\n",
    "\n",
    "**读取量维度**：独立扫描 `n·Q = 3,000,000`；共享扫描 `n = 1,000,000`。**仍然降到 1/3，与选择性无关**——选择性只影响\"输出多少\"，不影响\"要读多少\"。共享扫描在读取维度上永远划算。\n",
    "\n",
    "**结果内存维度**：分列表版本 `S = 3 × 500,000 = 1,500,000` 条；打标签版本 `U = 1,000,000 × (1 − (1−0.5)³) = 875,000` 条。比值 `S/U ≈ 1.71`，此时打标签版本明显更省。\n",
    "\n",
    "结论：**选择性越高（结果越大、重叠越多），打标签版本的内存优势越明显。**\n",
    "\n",
    "但要警惕另一个后果：选择性高意味着结果总量接近 `n`，此时真正的瓶颈变成\"结果太大放不下\"，共享扫描省下的 I/O 收益会被结果物化的开销吃掉。真实系统在这种情况下会考虑流水线执行（不物化结果）而不是共享扫描。\n",
    "\n",
    "</details>\n",
    "\n",
    "**7. 如果这 Q 个查询不是同时到达，而是在扫描过程中陆续到达（比如扫到第 100,000 条时才来第 4 个查询），共享扫描该怎么处理？这带来什么新问题？**\n",
    "\n",
    "<details><summary>答案要点</summary>\n",
    "\n",
    "这就是 Zukowski et al.（VLDB 2007）所说的**协同扫描 / 循环扫描**（cooperative scan, circular scan）问题。三种处理策略：\n",
    "\n",
    "| 策略 | 做法 | 代价 | 问题 |\n",
    "| --- | --- | --- | --- |\n",
    "| 不加入 | 新查询等下一轮扫描 | 延迟高 | 实现简单，但要等很久 |\n",
    "| 立即加入 | 从当前位置开始参与，只拿到后半段结果 | 结果不完整 | 必须补扫前面跳过的那一段 |\n",
    "| 回绕补扫 | 扫到表尾后回头补扫跳过的部分 | 多读一段 | 要记住起点，实现复杂 |\n",
    "\n",
    "**循环扫描**的做法是：把表看成一个环，扫描指针一直往前走不回头；新查询从当前指针位置加入，扫到表尾后**回绕**补上它加入之前的那一段，补完就退出。这样所有查询共享同一次连续扫描，磁盘带宽被充分利用。\n",
    "\n",
    "带来的新问题：\n",
    "\n",
    "1. **结果不完整**最难处理：调用者拿到的是部分结果，需要补扫机制或显式声明\"只返回部分结果\"来保证语义正确。\n",
    "2. **谓词集合动态变化**：查询可以中途加入或退出，代码生成（实验 09）就不能在扫描开始前一次性完成，需要支持增量重编译或回退到解释执行。\n",
    "3. **公平性**：晚到的查询只享受了部分共享，如何计费与调度是调度器层面的问题。\n",
    "4. **事务隔离**：不同查询可能跑在不同的快照上，能否共享扫描取决于 MVCC 版本是否一致。\n",
    "\n",
    "真实系统会估算\"新查询加入的收益 vs 补扫的代价\"，动态决定是否让它加入。\n",
    "\n",
    "</details>\n",
    "\n",
    "**8. `scan()` 用 `el.__dict__[attribute]` 做动态属性查找。改成什么样能让它更快？给出思路并说明代价。**\n",
    "\n",
    "<details><summary>答案要点</summary>\n",
    "\n",
    "思路：**把属性名在构造谓词时就固定下来**，循环里不再查字典。\n",
    "\n",
    "```python\n",
    "def make_pred(attribute, constant):\n",
    "    if attribute == \"a\":\n",
    "        return lambda el: el.a == constant\n",
    "    if attribute == \"b\":\n",
    "        return lambda el: el.b == constant\n",
    "\n",
    "preds = {k: make_pred(e[\"attribute\"], e[\"constant\"]) for k, e in queries_dict.items()}\n",
    "```\n",
    "\n",
    "属性名被\"固化\"成 `el.a` / `el.b` 这样的直接访问，`constant` 变成闭包里的局部变量。\n",
    "\n",
    "另一个更通用的写法是 `operator.attrgetter(attribute)`：它是 C 实现，只构造一次 getter，但每次取值仍是一次函数调用。\n",
    "\n",
    "**代价**：\n",
    "\n",
    "1. 要预先枚举所有可能的属性名，属性多了难维护（`attrgetter` 可以缓解）。\n",
    "2. 闭包本身仍是一次 Python 函数调用，彻底消除只能靠代码生成——把 `Q` 个谓词全部内联进同一个循环体（见实验 09）：\n",
    "\n",
    "```python\n",
    "for tup in data:\n",
    "    if tup.a == 7:   result_q1.append(tup)\n",
    "    if tup.b == 32:  result_q2.append(tup)\n",
    "    if tup.b == 1:   result_q3.append(tup)\n",
    "```\n",
    "\n",
    "3. 牺牲灵活性：schema 变更时要重建谓词。\n",
    "\n",
    "这也说明了一件事：**共享扫描与代码生成是互补的**。共享扫描解决 I/O，代码生成解决 CPU；两者结合才能在两个维度上都赢。\n",
    "\n",
    "</details>\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}