高性能企业级数据底座
国产自研 SQL 数据引擎

不是「能跑简单查询」的玩具, 而是把完整的分析型能力、跨方言兼容与内存鲁棒性集于一身。

与传统关系库、大数据 / 流批引擎、列存分析库全面对照。

注:Hive/Spark/Flink 为大数据 / 流批引擎, 与本引擎(嵌入式 OLAP)定位不同, 此处仅就 SQL 分析语法与查询特性维度对照。

完整的分析型 SQL 语法收入囊中, 且几乎所有「不支持」都是带改写提示的明确报错, 而非静默错误。

不是浅层字符串截取, 而是深入数据处理内核 —— 功能完整度与内存效率兼得。

一个引擎, 听懂四种「方言」 —— 天然的 SQL 迁移友好层,MySQL / Oracle / PG / SQL Server 的存量 SQL 大量可直接运行。

铁律(2026-07-13 拍板): 各库互不冲突的语法/函数 —— MySQL 的 SUBSTRING_INDEX、PG 的 SPLIT_PART、Oracle 的 NVL —— 一律兼容并包、同时全支持, 与方言开关无关; 只有下面这些真语义冲突的极少数点, 才在求值处读 ctx.session.dialect() 分流。默认 MySQL 8(历史行为零迁移), 可用 -Dyoungsdata.sql.dialect=mysql|postgresql|oracle 或 SessionContext.withDialect(...) 切换。

└─ 表中只是真冲突点; 除此以外的跨库函数与语法始终全部可用, 方言开关不影响它们。

-- 同一段跨库函数, 三方言下始终可用(互不冲突, 兼容并包)
SELECT NVL(name, '匿名'),                    -- Oracle 写法
       IFNULL(phone, '-'),                   -- MySQL 写法
       SPLIT_PART(email, '@', 2),            -- PostgreSQL 写法
       SUBSTRING_INDEX(path, '/', -1)        -- MySQL 写法
FROM   users;
-- 仅真冲突点跟随方言: 下句 1/0 在 MySQL 返 NULL, 在 PG/Oracle 报错
--   java -Dyoungsdata.sql.dialect=postgresql ...
SELECT amount / NULLIF(qty, 0) FROM orders;

不是「只读分析玩具」—— 内置完整的事务写入引擎: 跨表原子提交、断电不丢、崩溃后自动恢复到一致状态; 外加唯一/联合索引、双方言 UPSERT、变更订阅(CDC)与时间点恢复(PITR),分析与在线写入一体。

└─ 对比的是「吞吐优先 / 异步持久」这一档: YoungsData 异步窗口 ≤10ms, 比 MongoDB 约 100ms、MySQL 放宽档约 1s 紧一个数量级; 要绝对零丢失随时切 commit()(与各库最严档同级)。

SqlEngine eng = SqlEngine.builder(dir).transactional().build();   // 开事务档
SqlQuery  sql = eng.sql();

// ① 声明式(首选): 正常返回自动 commit(同步落盘), 抛异常自动回滚, 冲突自动重试
sql.transact(txn -> {
    txn.executeUpdate("UPDATE account SET bal = bal - 100 WHERE id = ?", 1);
    txn.executeUpdate("UPDATE account SET bal = bal + 100 WHERE id = ?", 2);
});   // 跨两表原子; 高并发下总额精确守恒、无扣穿

// ② 命令式: try-with-resources, 忘 commit 自动回滚, 事务内 SELECT 读己写
try (SqlTxn txn = sql.begin()) {
    txn.executeInsert("INSERT INTO audit (id, note) VALUES (10, 'transfer')");
    Object[] mine = txn.one("SELECT note FROM audit WHERE id = 10");  // 立即可见
    txn.commit();        // 同步落盘: 返回即断电不丢
}

// 同步 vs 异步 —— 按业务对"丢失容忍度"选一档
txn.commit();        // 零丢失: 返回即已落盘, 转账/账务等零容忍场景
txn.commitAsync();   // 高吞吐: 返回快, 进程被杀零丢失, 断电最多丢约 10ms(埋点/流水)

-- UPSERT 双方言 + 唯一 / 普通索引
INSERT INTO up (id, n, v) VALUES (1, 77, 'd') ON CONFLICT (id) DO UPDATE SET n = EXCLUDED.n;  -- PG
INSERT INTO up (id, n, v) VALUES (1,  5, 'x') ON DUPLICATE KEY UPDATE n = VALUES(n);          -- MySQL
CREATE UNIQUE INDEX uq_email ON users (email);
CREATE INDEX        ix_city  ON users (city, level);   -- 多列联合

一套「装配入口 SqlEngine + 双门面(SQL 文本 SqlQuery / Fluent no-SQL Db+D)+ 预编译 Prepared* + 三态消费(list / forEach / cursor)+ RPC 电报 spec」的完整编程式 API —— 每个入口都是一行一签名、可直接抄用。

try (SqlEngine engine = SqlEngine.builder(new File("/data/mydb")).build()) {
    SqlQuery sql = engine.sql();
    sql.executeDdl("CREATE TABLE orders (id BIGINT PRIMARY KEY, customer VARCHAR(64), amount DECIMAL(18,2))");
    sql.executeInsert("INSERT INTO orders (id, customer, amount) VALUES (1, 'Alice', 12.50)");
    Object[]       one  = sql.one ("SELECT customer FROM orders WHERE id = 1");
    List<Object[]> rows = sql.list("SELECT id, amount FROM orders WHERE amount > 1000");
}  // close() 自动 flush 落盘 + 关 reader/writer 池

PreparedSqlQuery pq = sql.prepare("SELECT id FROM orders WHERE amount > ? AND status = ? LIMIT ?");
List<Object[]> r1 = pq.list(5000L, "PAID", 10L);
List<Object[]> r2 = pq.list(8000L, "PAID", 20L);          // 零 plan 开销

// 真 batch: 一批共享 1 次 borrow + flush, 大批量远快于循环 execute
PreparedInsertSqlQuery ins = sql.prepareInsert("INSERT INTO orders (id, customer, amount) VALUES (?, ?, ?)");
ins.executeBatch(List.of(new Object[]{1L,"A",10.0}, new Object[]{2L,"B",20.0}));

// 拉取式游标: 最省内存, EOF 自动 close
try (Cursor c = sql.cursor("SELECT id, amount FROM orders WHERE amount > 100")) {
    Object[] row;
    while ((row = c.next()) != null) process(row);
}

Db db = engine.db();
db.createTable("orders").column("id", Db.Type.BIGINT).pk()
  .column("customer", Db.Type.VARCHAR, 64).column("amount", Db.Type.DECIMAL, 18, 2).execute();

List<Object[]> r = db.select(col("o.id"), col("c.name"), sum(col("o.amount")).as("total"))
        .from("orders", "o")
        .join("customers", "c").on(eq(col("o.cust_id"), col("c.id")))   // JOIN 必须 .on/.using 收尾
        .where(gt("o.amount", 0L))
        .groupBy(col("o.id"), col("c.name"))
        .having(gt(col("total"), val(1000)))
        .orderBy(desc("total")).limit(20).list();

// RPC 电报: 客户端组装 → byte[] → 服务端还原执行
QuerySpec spec = QuerySpec.select(col("id"), col("amount")).from("orders").where(gt("amount", 1000L)).limit(10);
byte[] wire = SpecCodec.DEFAULT.encode(spec);            // 上线
List<Object[]> back = db.fromSpec(wire).list();          // 服务端 Db.fromSpec 直达车

// 纯 SELECT(+ 可选 WHERE/LIMIT)可走直通路径: 跳过装箱拆箱, 每行不再新建 Object[]
if (sql.canRunRow(q)) {
    List<Order> os = sql.row().list(
        r -> new Order((Long) r.get(0), (String) r.get(1)),   // r 是 RowRec, get(i) 取值
        q);
}
// 预编译 + 直通 + 映射, OLTP 小查询最省; 直通视图同样有 list / forEach / first / one / batch
PreparedSqlQuery pq = sql.prepare("SELECT id, name FROM users WHERE city = ?");
List<Object[]> rows = pq.row().list("hangzhou");

└─ RowRec.get(i) 按表字段序取值; 需要按列名 / 类型化取值时用通用 Object[] 路径。计数 / 存在性用 count(sql) / exists(sql)。

以「参数化预编译从结构上根除 SQL 注入」为核心, 叠加写入值强校验、严格类型规则、扫描审计与内存预算守护, 形成一条 fail-loud 的纵深防线 —— 脏数据不入库、错答不返回、恶意查询打不爆内存。

// ✅ 安全: ? 只当"值"比较, 永不改变 SQL 结构
PreparedSqlQuery pq = sql.prepare("SELECT id, amount FROM orders WHERE status = ? AND amount > ?");
List<Object[]> rows = pq.list(userInput, threshold);   // userInput 即使是  x' OR '1'='1  也只是一个字符串常量

// ✗ 危险: 把用户输入拼进 SQL 文本 —— 结构可被篡改(应改用上面的 ? 绑定)
// String sql = "SELECT ... WHERE status = '" + userInput + "'";

// 参数个数不匹配 / 未绑定 → 立即报错, 而非静默当 NULL 返回空集(比报错更危险)
pq.list(1);   // 抛异常: 参数数量不匹配(SQL 有 2 个 ?, 传入 1)

// 类型 / 审计 / 内存, 都是可选开的会话级开关(零开销)
SessionContext s = SessionContext.defaults()
        .withCastMode(CastMode.NULL_ON_ERROR)   // 批量导入: CAST 失败返 NULL 而非抛错
        .withAuditSingleScan(true)              // 断言: 同表被重复扫描即报错
        .withScanObserver(t -> metrics.mark(t)) // 监控: 上报每个物理表扫描次数
        .withSpillBudgetBytes(512L << 20);      // 内存预算: 超限清晰报错而非 OOM

-- 账号与授权: MySQL 8 风格 DCL, 8 个权限动词 × 全局 / 库 / 表三级作用域(ADMIN 仅全局); 本组语句均需管理员身份执行
CREATE USER etl IDENTIFIED BY 'Str0ng!Pass';
GRANT SELECT, INSERT ON sales.* TO etl;   -- 表 / 库级拦截默认观察模式(违规只记审计日志), 确认无误后由服务端开关升为强制
SHOW GRANTS FOR etl;

-- 来源访问规则: 「账号 × 网段」有序规则链(对齐 PostgreSQL pg_hba 模型), SQL 动态增删、即时生效、重启保留、恒强制
CREATE ACCESS RULE r_etl       ALLOW  etl FROM '10.20.3.0/24' PRIORITY 10;   -- etl 仅允许从专属网段登录
CREATE ACCESS RULE r_etl_close REJECT etl FROM ALL            PRIORITY 20;   -- etl 其余来源一律拒(其他账号不受影响)
CREATE ACCESS RULE r_default   REJECT ALL FROM ALL            PRIORITY 9999; -- 收口: 未命中任何规则默认放行, 「名单外全拒」须显式加这条
SHOW ACCESS RULES;             -- 输出可回放语句, 行序即求值序(按 PRIORITY 升序首条命中)
DROP ACCESS RULE r_etl_close;  -- 删除即刻生效, 只影响新连接; 本机 loopback 恒放行防误配自锁

不靠「分布式堆机器」, 而把单机的每一滴性能榨干、把内存鲁棒性做到极致。

读取路径相对值(对照同机「手写极致优化直读」基准):比值越低越快,<1 即代表反超手写直读。

注:即便以 out-of-core 著称的 DuckDB, 官方文档至今仍专设 OOM 排查指南(部分聚合 / PIVOT / 多阻塞算子组合场景);YoungsData 将全部物化算子纳入弹性内存自适应, 无豁免算子。

复杂报表、跨源融合、多结果集合并 —— 一条 SQL 搞定, 无需在应用层来回搬运数据。

一套自研引擎内核, 既能单机极致、又有自研的分布式数据处理逻辑。业务只需描述「表是什么、数据在哪」, 引擎负责 SQL 全栈 —— 按数据规模自由伸缩, 小到嵌入边缘设备, 大到分布式计算集群节点。

SQL 表达不了的业务逻辑(脱敏加密、用户轨迹、自定义统计……), 写个普通 Java 方法就能变成 SQL 函数、在查询里直接调。三种类型全覆盖, 每种都能一次返回多列 —— 简单到「一个注解 + 一个方法」。

传统数据库(MySQL / PostgreSQL 等)给不了这样的体验:存储过程语言受限, 能跑任意 Java 逻辑、引第三方库、还能热替换的方案在工程上几乎不存在, 自定义聚合更无从谈起。

YoungsData 用纯 Java 提供三种自定义函数 —— 单行(SCALAR)逐行变换、分组(GROUP)对整组数据 holistic 计算、流式(STREAMING)逐行累加, 三种都能一次产出多列结果;而对标 Spark / Flink, 我们在「怎么写、怎么配、怎么改」上简单一个量级 —— 尤其分组 / 流式聚合, 它们要继承 UDAF / AggregateFunction、实现一大堆方法和 schema, 我们只要一个方法。

而真正把这套 UDF 推上量级的, 是每种函数都能接一个 UdfState s —— 一个 put / get 的状态对象(就是下面例 ③ 里那个 stats,get 还能直接强转类型)。它不止在行与行、组与组之间保存中间量,更能在 SQL 执行时接住你从查询里传入的自定义参数 —— 漏斗的步骤定义、路径的规则、留存的时间窗、各种阈值……函数里一句 s.get(...) 取出来即用。于是同一个 UDF 不再写死, 而是按查询参数化:SQL 负责把海量数据筛出来,UdfState 负责带着「参数 + 状态」边扫边算。

三种类型 · 易用度对比

注:对比的是「自定义函数」的写法与二次开发体验, 非整体引擎定位。

一眼看懂:三种类型, 各省多少代码

import org.apache.spark.sql.*;
import org.apache.spark.sql.types.*;

StructType out = new StructType()
        .add("cipher", DataTypes.StringType)
        .add("masked", DataTypes.StringType);

UDF1<String, Row> enc = phone -> {
    if (phone == null) return null; // null 也得自己判
    return RowFactory.create(
            aesEncrypt(phone),                          // 密文
            phone.substring(0, 3) + "****" + phone.substring(7)); // 掩码
};
spark.udf().register("encrypt", enc, out);
// 调用还得 .getField 拆列:
// SELECT encrypt(phone).cipher, encrypt(phone).masked …

@Udf(name = "ENCRYPT",
        type = UdfKind.SCALAR,
        returns = { @Col(name = "cipher", type = UdfType.STRING),
                @Col(name = "masked", type = UdfType.STRING) })
public static Object encrypt(UdfRow row) {
    String phone = (String) row.value();
    if (phone == null) { return null; }

    return new Object[] {
            aesEncrypt(phone),                            // 密文
            phone.substring(0, 3) + "****" + phone.substring(7) // 138****5678
    };
}

case class Buf(path: String, n: Int)
case class Out(path: String, stops: Int)

object UserPath extends Aggregator[Row, Buf, Out] {
    def zero = Buf("", 0)
    def reduce(b: Buf, r: Row) = {              // 自己拼缓冲
        val p = r.getString(0)
        if (p == null) b                            // null 也得自己跳
        else {
            val sep = if (b.n > 0) " → " else ""
            Buf(b.path + sep + p, b.n + 1)
        }
    }
    def merge(a: Buf, b: Buf) =                  // 还要写 merge
        Buf(a.path + " → " + b.path, a.n + b.n)
    def finish(b: Buf) = Out(b.path, b.n)
    def bufferEncoder = Encoders.product[Buf]    // 还要两个 encoder
    def outputEncoder = Encoders.product[Out]
}
spark.udf.register("user_path", functions.udaf(UserPath))

@Udf(name = "USER_PATH",
        type = UdfKind.GROUP,
        returns = { @Col(name = "stops", type = UdfType.INT),
                @Col(name = "path", type = UdfType.STRING) })
public static Object userPath(Iterator<UdfRow> rows) {
    int stops = 0;
    StringBuilder path = new StringBuilder();
    while (rows.hasNext()) {
        String place = (String) rows.next().value();
        if (place == null) { continue; }

        if (stops++ > 0) { path.append(" → "); }
        path.append(place);
    }
    return new Object[] {
            stops,            // 经停数
            path.toString()   // 路径
    };
}
// 结果: stops = 3, path = "公司 → 食堂 → 健身房"

public static class Acc {
    double min = Double.MAX_VALUE, max = -Double.MAX_VALUE;
    double sum = 0;
    long cnt = 0;
}

public class StreamStats extends AggregateFunction<Row, Acc> {
    public Acc createAccumulator() { return new Acc(); }

    public void accumulate(Acc a, Double v) {
        if (v == null) return;             // null 也得自己判
        a.min = Math.min(a.min, v);
        a.max = Math.max(a.max, v);
        a.sum += v;
        a.cnt++;
    }

    public Row getValue(Acc a) {
        if (a.cnt == 0) return null;        // 空组也得自己处理
        return Row.of(a.min, a.max, a.sum / a.cnt);
    }

    public TypeInformation<Row> getResultType() {
        // 还得手定义 3 列类型
        return Types.ROW(Types.DOUBLE, Types.DOUBLE, Types.DOUBLE);
    }
}

tableEnv.createTemporarySystemFunction("stream_stats", StreamStats.class);

@Udf(name = "STREAM_STATS",
        type = UdfKind.STREAMING,
        returns = { @Col(name = "min_v", type = UdfType.DOUBLE),
                @Col(name = "max_v", type = UdfType.DOUBLE),
                @Col(name = "avg_v", type = UdfType.DOUBLE) })
// s = 状态袋:put / get(get 可强转)。既存逐行中间量,也能接住 SQL 查询时传入的参数
public static Object stats(UdfState s, UdfRow row, boolean end) {
    if (end) {                       // 组末:产出一次
        long n = s.getLong("cnt", 0);
        if (n == 0) { return null; }     // 空组 → 各列 NULL

        return new Object[] {
                s.getDouble("min", 0),        // min
                s.getDouble("max", 0),        // max
                s.getDouble("sum", 0) / n      // avg
        };
    }

    Double v = (Double) row.value(); // 逐行:只累加
    if (v == null) { return null; }

    s.put("min", Math.min(s.getDouble("min", v), v));
    s.put("max", Math.max(s.getDouble("max", v), v));
    s.put("sum", s.getDouble("sum", 0) + v);
    s.put("cnt", s.getLong("cnt", 0) + 1);
    return null;
}

从嵌入式分析到 AI 生成 SQL 的执行底座，YoungsData 覆盖七大典型落地场景。

内置 140+ 函数,11 大类全覆盖。输入关键词即时检索, 签名 + 说明一目了然 —— 开发者与 AI 即学即用。同义词(如 NVL / IFNULL)均可直接书写。

窗口帧支持 ROWS / RANGE / GROUPS 三种 + 命名窗口(WINDOW w AS …)+ IGNORE NULLS + QUALIFY 过滤窗口结果。

同一个业务需求, 传统数据库 / 大数据引擎写起来又长又绕,YoungsData 一条简洁 SQL 搞定 —— 下面都是真实可运行的写法。

重点导读 · 最建议先看的 4 个硬核例子

如果用户只看这一页的一小部分, 最该点开的就是下面这四段。它们不是“少写几行 SQL”那么简单, 而是直接证明 YoungsData 已经把 Oracle 级行模式识别做成了真正可用的嵌入式能力。

SELECT
    JSON_VALUE(payload, '$.user.name') AS user_name,
    JSON_VALUE(payload, '$.user.vip')  AS vip,
    CAST(JSON_VALUE(payload, '$.order.amount') AS DECIMAL(18, 2)) AS amount
FROM events
WHERE JSON_VALUE(payload, '$.status')   = 'paid'
  AND JSON_VALUE(payload, '$.user.vip') = 'true';

SELECT
    payload ->> '$.user.name' AS user_name,   -- 按路径直读,免 JSON_VALUE/CAST
    payload ->> '$.user.vip' AS vip,
    payload ->> '$.order.amount' AS amount
FROM events
WHERE
    payload ->> '$.status' = 'paid'
    AND payload ->> '$.user.vip' = 'true';

SELECT dept, name, salary
FROM (
    SELECT dept, name, salary,
        ROW_NUMBER() OVER (PARTITION BY dept ORDER BY salary DESC) AS rn
    FROM emp
) t
WHERE rn <= 3;

SELECT
    dept, name, salary
FROM
    emp
QUALIFY
    ROW_NUMBER() OVER (PARTITION BY dept
                       ORDER BY salary DESC) <= 3;

SELECT product,
    SUM(CASE WHEN quarter='Q1' THEN amount END) AS Q1,
    SUM(CASE WHEN quarter='Q2' THEN amount END) AS Q2,
    SUM(CASE WHEN quarter='Q3' THEN amount END) AS Q3,
    SUM(CASE WHEN quarter='Q4' THEN amount END) AS Q4
FROM sales GROUP BY product;

SELECT *
FROM sales
PIVOT (
    SUM(amount) FOR quarter IN ('Q1','Q2','Q3','Q4')   -- 新增季度只改这一行
);

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

val w = Window.partitionBy("dept")
    .orderBy(col("salary").desc)
df.withColumn("rn", row_number().over(w))
    .filter(col("rn") <= 3)
    .select("dept", "name", "salary")
    .show()

SELECT dept, name, salary
FROM emp
QUALIFY ROW_NUMBER() OVER (PARTITION BY dept
                           ORDER BY salary DESC) <= 3;

// Spark
df.withColumn("day", date_trunc("day", col("ts")))
    .groupBy("day", "product")
    .agg(sum("amount").alias("revenue"))
    .orderBy("day")
    .show()
// Flink DataStream 还需 WindowAssigner /
// Watermark / KeyedProcessFunction…

SELECT
    DATE_TRUNC('day', ts) AS stat_day,   -- 'hour'/'month' 粒度同理
    product,
    SUM(amount) AS revenue
FROM metrics
GROUP BY
    DATE_TRUNC('day', ts),
    product
ORDER BY stat_day;

-- 每步先取「最早发生时间」,再层层 JOIN 校验先后
SELECT v.user_id, v.ts AS view_t, c.ts AS cart_t, p.ts AS pay_t
FROM (SELECT user_id, MIN(ts) AS ts
      FROM events WHERE event = 'view' GROUP BY user_id) v
JOIN (SELECT user_id, MIN(ts) AS ts
      FROM events WHERE event = 'cart' GROUP BY user_id) c
    ON c.user_id = v.user_id AND c.ts > v.ts
JOIN (SELECT user_id, MIN(ts) AS ts
      FROM events WHERE event = 'pay' GROUP BY user_id) p
    ON p.user_id = c.user_id AND p.ts > c.ts;
-- 三层自关联扫三遍表;每加一步漏斗,多一层 JOIN;
-- 且 MIN(ts) 藏语义陷阱:最早加购早于最早浏览时,
-- 真实走完路径的用户会被漏掉

SELECT user_id, view_t, pay_t
FROM events
MATCH_RECOGNIZE (
    PARTITION BY user_id                   -- 每个用户独立匹配
    ORDER BY ts                            -- 事件按时间排成序列
    MEASURES FIRST(viewed.ts) AS view_t,   -- 取首次浏览时间
             LAST(paid.ts)    AS pay_t     -- 取支付时间
    PATTERN (viewed carted+ paid)          -- 漏斗:浏览 → 加购(≥1次) → 支付
    DEFINE viewed AS event = 'view',       -- 每步的判定条件
           carted AS event = 'cart',
           paid   AS event = 'pay'
);

Pattern<Order, ?> p = Pattern
    .<Order>begin("down")
    .where(new IterativeCondition<Order>() {
        public boolean filter(Order o, Context<Order> ctx)
                throws Exception {
            for (Order prev : ctx.getEventsForPattern("down"))
                if (o.amount >= prev.amount) return false;
            return true; // 递减才算命中
        }
    })
    .times(3).consecutive();

CEP.pattern(stream.keyBy(o -> o.userId), p)
    .select(/* 匹配结果还要自己拆装 */ …);
// 打包 → 提交集群作业;改规则,再来一遍

SELECT
    user_id, first_day, last_day
FROM orders
MATCH_RECOGNIZE (
    PARTITION BY user_id                          -- 逐用户检测
    ORDER BY order_date                           -- 按下单时间排序
    MEASURES
        FIRST(decline.order_date) AS first_day,   -- 首笔下跌日
        LAST(decline.order_date)  AS last_day     -- 第三笔下跌日
    PATTERN (decline{3})                          -- 连续 3 笔递减
    DEFINE
        decline AS amount < PREV(amount)          -- 比上一笔金额小
);

以 100, 90, 80, 70, 60 这组金额为例:

想找出所有“连续 3 笔递减”序列时,
不是只找到 (90,80,70),
还必须继续找到重叠的 (80,70,60)。

传统 SQL / 代码流通常要:
1. 先取第一段
2. 手工回退起点一行
3. 再跑下一段
4. 自己保证边界不重不漏

规则一复杂,实现立刻失控

SELECT
    user_id, start_amt, end_amt
FROM orders
MATCH_RECOGNIZE (
    PARTITION BY user_id
    ORDER BY order_date
    MEASURES
        FIRST(DOWN.amount) AS start_amt,   -- 第 1 笔递减金额
        LAST(DOWN.amount)  AS end_amt      -- 第 3 笔递减金额
    AFTER MATCH SKIP TO NEXT ROW           -- 命中后从下一行继续,允许重叠
    PATTERN (DOWN{3})                      -- 连续 3 笔递减
    DEFINE
        DOWN AS amount < PREV(amount)      -- 当前金额小于上一笔
);

真实业务里经常不是“找到了就完了”,
而是还要继续回答:

1. 哪些行是起点?
2. 哪些行是中间过渡?
3. 哪些行只是参与匹配但不应该出现在最终结果里?

很多系统只能告诉你“命中了一个模式”,
但无法把匹配角色和输出裁剪一起做好。

SELECT order_date, amount, cls
FROM orders
MATCH_RECOGNIZE (
    PARTITION BY user_id
    ORDER BY order_date
    MEASURES CLASSIFIER() AS cls             -- 返回当前行在模式中的角色名
    ALL ROWS PER MATCH                       -- 逐行输出匹配结果
    PATTERN (A {- B+ -} C)                   -- B 参与匹配,但从最终输出排除
    DEFINE A AS amount > 100,                -- 起始高值
           B AS amount BETWEEN 50 AND 100,   -- 中间过渡段
           C AS amount < 50                  -- 终止低值
);

真实业务里要找每一轮“从峰值跌下去,
又涨回峰值”的回撤事件时,用户真正想看的
通常只有两个信号点:

1. 见顶日
2. 修复确认日

中间连续下跌的过程日通常不想报出来,
否则结果会非常脏。

难点在于:
传统库就算能用 window + 自连接硬拼出结果,
也很难做到“把中间行从输出里拿掉,
但回撤幅度 / 下跌天数仍然按这些中间行计算”。

SELECT
    sym, episode, role, d, px, trough, down_days, drawdown
FROM quote
MATCH_RECOGNIZE (
    PARTITION BY sym
    ORDER BY d
    MEASURES
        MATCH_NUMBER() AS episode,           -- 第几轮回撤
        CLASSIFIER()   AS role,              -- 当前是 PEAK 还是 RECOVER
        MIN(FALL.px)    AS trough,           -- 谷底:对隐藏的下跌日统计
        COUNT(FALL.px)  AS down_days,        -- 水下天数:对隐藏行统计
        PEAK.px - MIN(FALL.px) AS drawdown   -- 最大回撤
    ALL ROWS PER MATCH
    PATTERN (PEAK {- FALL+ -} RECOVER)       -- 中间连跌段参与匹配但不输出
    DEFINE
        FALL AS px < PREV(px),               -- 连续下跌
        RECOVER AS px >= PEAK.px             -- 涨回峰值即修复
);

常见引擎的真实情况通常是:

1. 只支持最浅层的序列匹配,再深一点就没有
2. 没有 CLASSIFIER / MATCH_NUMBER / SUBSET
3. 不支持 AFTER MATCH SKIP TO NEXT ROW 重叠匹配
4. DEFINE 里做变量组聚合或普通函数,很容易直接缺失
5. Oracle 虽然强,但部署形态是重量级服务端,不是单 jar 嵌入式

也就是说:
“能写一个漏斗 Demo” 和
“把 Oracle 级高阶语义完整跑通”
中间隔着一整代实现难度

SELECT order_date, amount, match_no, cls, avg_x
FROM orders
MATCH_RECOGNIZE (
    PARTITION BY user_id
    ORDER BY order_date
    MEASURES
        MATCH_NUMBER() AS match_no,                -- 当前是第几个命中的模式
        CLASSIFIER()   AS cls,                     -- 当前行扮演的角色
        AVG(X.amount)  AS avg_x                    -- DOWN/NOISE 子集上的运行均值
    ALL ROWS PER MATCH
    AFTER MATCH SKIP TO NEXT ROW                   -- 允许重叠匹配
    PATTERN (DOWN+ NOISE* RECOVER)                 -- 下跌序列 + 噪声 + 恢复
    SUBSET X = (DOWN, NOISE)                       -- 定义聚合子集 X
    DEFINE
        DOWN AS amount < PREV(amount),             -- 继续下跌
        NOISE AS ABS(amount - PREV(amount)) < 5,   -- 小幅波动视为噪声
        RECOVER AS amount > AVG(X.amount)          -- 高于运行均值才算恢复
);

-- 下面那条 SQL,拿到各引擎试跑:
MySQL      : ERROR 1305 — FUNCTION TO_CHAR
             does not exist
PostgreSQL : ERROR — function group_concat(text)
             does not exist
SQL Server : 'NVL' is not a recognized
             built-in function name
Spark      : AnalysisException — Undefined
             function: GROUP_CONCAT

→ 换引擎 = 上千条存量 SQL 逐条改写、逐条回归,
  每一处改写都是一次出错机会

SELECT
    TO_CHAR(create_time,'YYYY-MM') AS stat_month,   -- Oracle 写法
    GROUP_CONCAT(DISTINCT region) AS regions,       -- MySQL 写法
    NVL(SUM(amount), 0) AS revenue                  -- Oracle 写法
FROM orders
GROUP BY TO_CHAR(create_time,'YYYY-MM');

-- 一条 SQL 混写两家方言,YoungsData 直接执行 ✓

一页速查:支持哪些 SQL 语法、开放哪些编程接口。函数明细见 §14 函数手册。

└─ 三种写法(SQL 文本 SqlQuery · Fluent Db+D · RPC 电报 spec)语义一致、可混用;详见 §07 编程式 API。