HiveGPT AI 学习
登录

S2 第 12 讲：REST Catalog——服务端 CAS 与凭证下发 ​

Apache Paimon 源码学习

作者 X老师（DaemonforY），Paimon 2.0 / master 源码，按 CC BY-NC-SA 4.0 发布。配套实验和代码在 GitHub。

Yui和Kai在REST关口提交快照并领取临时凭证。
Yui和Kai在REST关口提交快照并领取临时凭证。AI 生成配图

TL;DR：第 5 讲说过，文件系统 Catalog 靠“谁先把 snapshot-(N+1) 写成功”来决定提交顺序，对象存储上 rename 不可靠，只能再配一把外部锁。REST Catalog 换了思路：数据文件、manifest 仍由客户端直接写到存储，但“最新快照是哪个”“这次提交成没成功”交给 HTTP 服务端裁决。提交请求带着“我是基于哪个快照做的”（base 快照 uuid），服务端只在它等于当前最新快照时才接受——这就是服务端 CAS，不再需要锁。请求里还带着表的 uuid，删表重建后旧作业的提交会被拒绝；开启数据凭证后，客户端按表向服务端领临时凭证，引擎不再持有长期存储密钥。表本身仍是前面 11 讲的那个 FileStoreTable，换掉的只是几个钩子。

实验：在进程里起一个 REST 服务端 ​

Paimon 的测试代码里有一个参考服务端 RESTCatalogServer（基于 okhttp 的 MockWebServer），Maven Central 上的 paimon-core-2.0.0-tests.jar 里就有。labs 新增的 RestLab 在进程内启动它，客户端用 metastore = rest 连上去，并打印服务端收到的每一个请求。四个场景：

场景做法
flowFlink SQL 经 REST Catalog 建表、写两次、查询
cas直接调用 commitSnapshot，带着过期的 base 快照 uuid 去提交
uuid拿着旧表对象，删表、同名重建之后再提交
token服务端下发 data-token.enabled = true，看客户端读数据前做了什么

这是一个测试用的 mock 服务端，不是生产实现（阿里云 DLF 等是另外的服务端）；本讲只用它观察客户端协议和参考实现的裁决逻辑。

一、一条 INSERT，变成了哪些请求 ​

一次插入被拆成配置、建表、提交等REST请求。
一次插入被拆成配置、建表、提交等REST请求。AI 生成配图

请求序列

客户端配置（uri 是本地端口，日志里用 ${uri} 代替）：

sql
CREATE CATALOG rest WITH ('type' = 'paimon', 'metastore' = 'rest', 'uri' = '${uri}',
  'warehouse' = 'lab14_wh', 'token.provider' = 'bear', 'token' = 'lab14-token');

服务端收到的请求（节选自 rest-2.0.0-flink2.2.log，去掉了查询参数）：

[REST] CREATE CATALOG：3 个请求
         GET /v1/config
         GET /v1/paimon/databases/default
         POST /v1/paimon/databases
[REST] CREATE TABLE：1 个请求
         POST /v1/paimon/databases/db1/tables
[REST] 第一次 INSERT：5 个请求
         GET /v1/paimon/databases/db1/tables/orders
         GET /v1/paimon/databases/db1/tables/orders/snapshot
         GET /v1/paimon/databases/db1/tables/orders/snapshot
         POST /v1/paimon/databases/db1/tables/orders/commit
         GET /v1/paimon/databases/db1/tables/orders/snapshot
[REST] 第二次 INSERT：4 个请求
         GET /v1/paimon/databases/db1/tables/orders/snapshot
         GET /v1/paimon/databases/db1/tables/orders/snapshot
         POST /v1/paimon/databases/db1/tables/orders/commit
         GET /v1/paimon/databases/db1/tables/orders/snapshot

几点：

  • 第一个请求永远是 GET /v1/config：客户端先向服务端要配置。URL 里的 paimon 前缀（prefix）、默认表参数、是否开数据凭证，都可以由服务端下发，和客户端配置合并（教程 11 章第 2 节）。
  • GET …/snapshot：问服务端“这张表最新的快照是哪个”，代替了文件系统 Catalog 的列目录、读 LATEST 提示文件。SnapshotManager.latestSnapshot()（第 201~213 行）里，有 snapshotLoader 时就问它，否则才走文件系统。
  • POST …/commit：提交交给服务端。数据文件、manifest 在这之前已经由客户端自己写到存储里了——请求里没有一条是“上传数据”。

查询结果和前面的实验没有区别：1 PAID、2 CREATED，两个 APPEND 快照。（第二次 INSERT 少了 GET …/tables/orders，是因为同一个作业里表对象已经加载过；为什么 INSERT 前要连问两次 snapshot、提交后再问一次，本讲没有逐一追到调用处。）

二、表还是那张表，换的是钩子 ​

钩子

REST Catalog 加载出来的仍然是普通的 FileStoreTable，前面 11 讲的写入、合并、读取、提交冲突检测全部复用。区别在加载表时注入的 CatalogEnvironment。最关键的是谁来提交快照（CatalogEnvironment.java 第 134~147 行）：

java
if (catalogLoader != null && supportsVersionManagement) {
    snapshotCommit = new CatalogSnapshotCommit(catalogLoader.load(), identifier, uuid);
} else {
    Lock lock = Optional.ofNullable(lockFactory) ... .orElseGet(Lock::empty);
    snapshotCommit = new RenamingSnapshotCommit(snapshotManager, lock);
}

文件系统 Catalog 走 RenamingSnapshotCommit（第 5 讲：原子 rename，可选外部锁）；REST Catalog 支持版本管理，走 CatalogSnapshotCommit，它只做一件事——把提交转给 catalog（CatalogSnapshotCommit.java 第 42~51 行）：

java
return catalog.commitSnapshot(newIdentifier, uuid, baseSnapshotUuid, snapshot, statistics);

加载表时 isRestCatalog 为真，lockFactory 直接传 null（CatalogUtils.java 第 393~395 行）：REST 模式下压根没有外部锁。

钩子文件系统 CatalogREST Catalog
谁来提交快照RenamingSnapshotCommitCatalogSnapshotCommit → POST …/commit
最新快照是哪个列目录 / LATESTSnapshotLoader → GET …/snapshot
外部锁可配无
读写数据的 FileIOcatalog 配置里的长期凭证开启 data-token 时为 RESTTokenFileIO

三、服务端 CAS：base 快照对不上，就拒绝 ​

服务端用CAS校验基准快照，不匹配就拒绝提交。
服务端用CAS校验基准快照，不匹配就拒绝提交。AI 生成配图

CAS

提交请求 CommitTableRequest 有四个字段：tableId、baseSnapshotUuid、snapshot、statistics。参考服务端的裁决（RESTCatalogServer.commitSnapshot，release-2.0.0 第 2760~2786 行，节选）：

java
if (!tableId.equals(table.catalogEnvironment().uuid())) {
    throw new Catalog.TableNotExistException(identifier);
}
String currentSnapshotUuid = currentSnapshot == null ? null : currentSnapshot.snapshot().uuid();
if (!Objects.equals(currentSnapshotUuid, baseSnapshotUuid)) {
    return mockResponse(new CommitTableResponse(false), 200);
}

实验场景 cas：正常提交两次后，模拟两个“迟到”的作业直接调用 commitSnapshot（节选）：

服务端记录的最新快照 = 2，uuid = 0c139770-c894-498e-b642-70f93110724e
快照 1 的 uuid = efd2ca06-0417-4045-8c7e-10fe36092924
commitSnapshot(base = 快照 1 的 uuid) 返回 false
commitSnapshot(base = null，即“我以为表还是空的”) 返回 false
提交被拒之后，最新快照仍是 2

一个是“快照 1 之后就开始准备、现在才来提交”，一个是“以为表还是空的”，都被拒绝，最新快照没有动。客户端收到 false，处理方式和第 5 讲本地 rename 抢号失败完全一样：重新读最新快照、做冲突检测、按退避重试。顺序由服务端的一次比较决定，不再依赖存储的原子 rename，也就不需要外部锁。

这里服务端只按 base 快照 uuid 判断，不关心你改了什么——和第 5 讲一样，冲突检测（比如“要删的文件还在不在”）仍在客户端重试时做。

本讲实验没有让两个真实作业并发写同一张表：参考服务端是测试用的 mock，第一次尝试并发时它的分发逻辑在多线程下抛了 ConcurrentModificationException，所以改成了上面这种可复现的直接调用。生产服务端的并发正确性本讲没有验证。

四、删表重建之后，旧作业还能写进去吗？ ​

删表重建

场景 uuid：一个作业拿到表对象（uuid 71f26b5b-…）、写了一行，还没提交；这时有人把表删了，又建了一张同名的新表（uuid 6c7fb9c4-…）。旧作业接着提交（节选）：

[REST] 旧表对象提交：22 个请求
         GET /v1/paimon/databases/db3/tables/t/snapshot
         POST /v1/paimon/databases/db3/tables/t/commit
         …（同样的一对，共 11 次）
WARN  [FileStoreCommitImpl] Retry commit for exception.      × 11
[EXPECTED ERROR] RuntimeException: Commit failed after 9223372036854775807 millis with 10 retries, there maybe exist commit conflicts between multiple jobs.
[EXPECTED ERROR] 根因 NoSuchResourceException: Table db3.t does not exist.

提交请求里的 tableId 是旧表的 uuid，服务端发现它和现在这张同名表对不上，按“表不存在”拒绝。如果只按名字认表，旧作业就会把为旧表准备的文件提交进新表。

客户端这一侧值得多看一眼：它拿到的是一个异常，不是 false。FileStoreCommitImpl 第 1313~1323 行把提交时的异常当作“结果不确定”（网络断了也可能是这种情况），不清理文件、退避重试；11 次尝试的间隔依次是 22、37、54、103、208、355、658、1531、3093、5970 毫秒（由日志时间戳算出），符合第 5 讲的退避公式。重试到 commit.max-retries（默认 10，CoreOptions 第 982~985 行）后放弃（第 905~915 行）。结果是对的——旧作业写不进新表；代价是要等完 10 次重试才报错，日志里的报错还提示“多个作业提交冲突”，容易误导排查。根因在 Caused by 里。

（报错里的 9223372036854775807 millis 是 commit.timeout 未设置时的 Long.MAX_VALUE，不是真的等了那么久；这次从第一次重试到放弃约 12 秒。）

五、数据凭证下发 ​

凭证下发

场景 token：服务端在 GET /v1/config 的默认配置里下发 data-token.enabled = true（默认 false，RESTTokenFileIO.java 第 64~68 行），并为表登记一份临时凭证。客户端（节选自 rest-token-2.0.0-flink2.2.log）：

[REST] getTable：1 个请求
         GET /v1/paimon/databases/db4/tables/t
表的 FileIO = RESTTokenFileIO
[REST] 第一次访问数据文件：1 个请求
         GET /v1/paimon/databases/db4/tables/t/token

表的 FileIO 变成了 RESTTokenFileIO；第一次碰数据文件时，它先向服务端要这张表的临时凭证（refreshToken → loadTableToken，第 314~327 行），再用凭证构造真正访问存储的 FileIO，按凭证缓存，快过期时自动刷新（fileIO()，第 234~270 行）。好处是计算引擎不用持有长期的存储密钥，权限可以细到单表。

实验的局限：参考服务端配套的测试 FileIO 按它自己的路径约定校验凭证，在本地目录上对不上，所以这次没有真正用这份凭证读写成功，只证明了“先领凭证”这一步。

六、和其它 Catalog 比 ​

文件系统HiveREST
元数据存在哪纯文件HMS + 文件服务端（+ 文件）
提交原子性原子 rename（对象存储需锁）Hive 锁服务端 CAS
最新快照列目录 / LATEST同左问服务端
存储凭证客户端持长期密钥同左按表临时凭证
多语言要重写文件语义要 HMS 协议HTTP + OpenAPI

（Hive 一列引自教程 11 章第 8 节，本讲未实验。）

小结

  1. REST Catalog 下数据仍由客户端直接写存储，“最新快照”和“提交成没成功”交给服务端：GET …/snapshot、POST …/commit。
  2. 提交带 base 快照 uuid，服务端比较后决定接受还是返回 false，客户端照常重读、检测冲突、重试——不需要外部锁。
  3. 提交还带表 uuid，删表重建后旧作业写不进新表（客户端会先重试完 10 次再报错）。
  4. 开启 data-token 后按表领临时凭证；表本身仍是 FileStoreTable，换掉的只是提交、快照加载、FileIO 几个钩子。

第二季收官：12 讲从全景图、写入、合并、提交、读路径、删除向量、changelog，一直到 Flink 写读、Spark MERGE 和 REST Catalog。第三季我们换到 Spark 这一侧。

自测题：REST 模式下提交请求被网络重发了两次（第一次其实已经成功），会不会生成两个快照？（答案：不会。第二次请求的 base 快照 uuid 已经不是最新，服务端 CAS 返回 false；客户端重试时先做 filterCommitted（第 5 讲），按 commitUser + identifier 发现这次提交已经在快照里，直接返回成功。教程 11 章第 5 节的“两层保险”。按源码推导，未单独运行。）

代码示例在页面里运行时使用 HiveGPT 的模型接口。延伸阅读来自 JavaGuide（Apache-2.0），版权归原作者。

© HiveGPT