高级 SQL：从应用访问到递归分析

CREATE TABLE store (
    store_id       bigint PRIMARY KEY,
    store_name     text NOT NULL,
    parent_id      bigint REFERENCES store(store_id),
    store_type     text NOT NULL
                   CHECK (store_type IN ('总部', '区域', '城市', '门店'))
);
 
CREATE TABLE customer (
    customer_id    bigint PRIMARY KEY,
    customer_name  text NOT NULL,
    member_level   text NOT NULL DEFAULT '普通'
);
 
CREATE TABLE product (
    product_id     bigint PRIMARY KEY,
    product_name   text NOT NULL,
    category       text NOT NULL,
    unit_price     numeric(12, 2) NOT NULL CHECK (unit_price >= 0)
);
 
CREATE TABLE inventory (
    store_id       bigint REFERENCES store(store_id),
    product_id     bigint REFERENCES product(product_id),
    quantity       integer NOT NULL CHECK (quantity >= 0),
    reorder_level  integer NOT NULL DEFAULT 10 CHECK (reorder_level >= 0),
    PRIMARY KEY (store_id, product_id)
);
 
CREATE TABLE sales_order (
    order_id       bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    store_id       bigint NOT NULL REFERENCES store(store_id),
    customer_id    bigint NOT NULL REFERENCES customer(customer_id),
    status         text NOT NULL DEFAULT '待支付',
    ordered_at     timestamptz NOT NULL DEFAULT current_timestamp,
    total_amount   numeric(14, 2) NOT NULL DEFAULT 0
);
 
CREATE TABLE order_item (
    order_id       bigint REFERENCES sales_order(order_id) ON DELETE CASCADE,
    product_id     bigint REFERENCES product(product_id),
    quantity       integer NOT NULL CHECK (quantity > 0),
    unit_price     numeric(12, 2) NOT NULL CHECK (unit_price >= 0),
    PRIMARY KEY (order_id, product_id)
);
 
CREATE TABLE inventory_audit (
    audit_id       bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    store_id       bigint NOT NULL,
    product_id     bigint NOT NULL,
    old_quantity   integer NOT NULL,
    new_quantity   integer NOT NULL,
    changed_at     timestamptz NOT NULL DEFAULT current_timestamp,
    changed_by     text NOT NULL DEFAULT current_user
);
 
CREATE TABLE restock_task (
    store_id       bigint NOT NULL,
    product_id     bigint NOT NULL,
    requested_qty  integer NOT NULL CHECK (requested_qty > 0),
    task_status    text NOT NULL DEFAULT '待处理',
    created_at     timestamptz NOT NULL DEFAULT current_timestamp
);
 
CREATE TABLE daily_sales (
    sale_date      date NOT NULL,
    store_id       bigint NOT NULL REFERENCES store(store_id),
    category       text NOT NULL,
    amount         numeric(14, 2) NOT NULL,
    PRIMARY KEY (sale_date, store_id, category)
);

import java.math.BigDecimal;
import java.sql.*;
import java.time.LocalDate;
 
public final class OrderReader {
    private final String url;
    private final String user;
    private final String password;
 
    public OrderReader(String url, String user, String password) {
        this.url = url;
        this.user = user;

import psycopg
 
def load_orders(dsn: str, store_id: int):
    with psycopg.connect(dsn) as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                SELECT order_id, ordered_at, total_amount
                FROM sales_order
                WHERE store_id = %s
                ORDER BY ordered_at, order_id
                """,
                (store_id,),
            )
            return list(cursor)

String unsafe =
    "SELECT customer_id FROM customer WHERE customer_name = '" + input + "'";
 
String safe =
    "SELECT customer_id FROM customer WHERE customer_name = ?";
try (PreparedStatement statement = connection.prepareStatement(safe)) {
    statement.setString(1, input);
    try (ResultSet result = statement.executeQuery()) {
        // input 始终按一个字符串值参与比较
    }
}

connection.setAutoCommit(false);
try {
    insertOrder(connection, order);
    insertItems(connection, order.items());
    int changed = decreaseInventory(connection, order.items());
    if (changed != order.items().size()) {
        throw new SQLException("部分商品库存不足");
    }
    connection.commit();
} catch

CREATE OR REPLACE FUNCTION store_revenue(
    p_store_id bigint,
    p_from      date,
    p_to        date
)
RETURNS numeric(14, 2)
LANGUAGE sql
STABLE
AS $$
    SELECT COALESCE(SUM(ds.amount), 0)::numeric(14, 2)

CREATE OR REPLACE FUNCTION low_stock_products(p_store_id bigint)
RETURNS TABLE (
    product_id    bigint,
    product_name  text,
    quantity      integer,
    shortage      integer
)
LANGUAGE sql
STABLE
AS $$
    SELECT p.product_id,
           p.product_name,
           i.quantity,

CREATE OR REPLACE PROCEDURE create_restock_tasks(p_store_id bigint)
LANGUAGE plpgsql
AS $$
DECLARE
    item record;
    v_created integer := 0;
BEGIN
    FOR item IN
        SELECT product_id,
               reorder_level - quantity + 10 AS requested_qty
        FROM inventory
        WHERE store_id = p_store_id
          AND

CREATE OR REPLACE FUNCTION audit_inventory_change()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
    IF OLD.quantity IS DISTINCT FROM NEW.quantity THEN
        INSERT INTO inventory_audit(
            store_id, product_id, old_quantity, new_quantity
        )
        VALUES (
            NEW.store_id, NEW.product_id, OLD.quantity

CREATE OR REPLACE FUNCTION enqueue_restock_from_batch()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
    INSERT INTO restock_task(store_id, product_id, requested_qty)
    SELECT n.store_id,
           n.product_id,
           n.reorder_level - n.quantity + 10
    FROM new_rows AS n
    WHERE n.quantity <=

WITH RECURSIVE store_tree AS (
    -- 锚点：选定根节点
    SELECT s.store_id,
           s.store_name,
           s.parent_id,
           0 AS depth,
           ARRAY[s.store_id] AS path
    FROM store AS s
    WHERE s.store_id = 1
 
    UNION ALL
 
    -- 递归项：把当前节点与其直接子节点连接
    SELECT

CREATE TABLE delivery_route (
    from_store_id bigint REFERENCES store(store_id),
    to_store_id   bigint REFERENCES store(store_id),
    PRIMARY KEY (from_store_id, to_store_id),
    CHECK (from_store_id <> to_store_id)
);
 
WITH RECURSIVE reachable(from_id, to_id) AS (
    SELECT from_store_id, to_store_id
    FROM delivery_route
 
    UNION
 
    SELECT r.from_id, d.to_store_id

SELECT store_id,
       category,
       sale_date,
       SUM(amount) AS revenue
FROM daily_sales
GROUP BY ROLLUP (store_id, category, sale_date)
ORDER BY store_id NULLS LAST,
         category NULLS LAST,
         sale_date NULLS LAST;

SELECT store_id,
       category,
       sale_date,
       SUM(amount) AS revenue
FROM daily_sales
GROUP BY GROUPING SETS (
    (store_id, category),
    (sale_date, category),
    ()
);

SELECT CASE WHEN GROUPING(store_id) = 1
            THEN '全部门店' ELSE store_id::text END AS store_scope,
       CASE WHEN GROUPING(category) = 1
            THEN '全部品类' ELSE category END AS category_scope,
       SUM(amount) AS revenue
FROM daily_sales
GROUP BY ROLLUP (store_id, category);

SELECT store_id,
       SUM(amount) FILTER (WHERE category = '食品')   AS food_revenue,
       SUM(amount) FILTER (WHERE category = '日用品') AS daily_revenue,
       SUM(amount) FILTER (WHERE category = '服饰')   AS clothing_revenue
FROM daily_sales
GROUP BY store_id
ORDER BY store_id;

WITH store_month AS (
    SELECT store_id,
           date_trunc('month', sale_date)::date AS month_start,
           SUM(amount) AS revenue
    FROM daily_sales
    GROUP BY store_id, date_trunc('month', sale_date)::date
), ranked AS (
    SELECT store_id,
           month_start,
           revenue,
           RANK() OVER (
               PARTITION BY month_start
               ORDER BY

SELECT sale_date,
       store_id,
       category,
       amount,
       AVG(amount) OVER (
           PARTITION BY store_id, category
           ORDER BY sale_date
           ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
       ) AS moving_avg_3_rows,
       SUM(amount) OVER (
           PARTITION BY store_id, category
           ORDER BY sale_date
           ROWS BETWEEN UNBOUNDED PRECEDING

SELECT sale_date,
       store_id,
       SUM(amount) AS day_revenue,
       AVG(SUM(amount)) OVER (
           PARTITION BY store_id
           ORDER BY sale_date
           RANGE BETWEEN INTERVAL '6 days' PRECEDING AND CURRENT ROW
       ) AS moving_avg_7_days
FROM daily_sales
GROUP BY sale_date, store_id
ORDER BY store_id, sale_date;