数据库流处理后端数据工程【免费下载链接】risingwaveEvent streaming platform for agentic AI. Continuously ingest, transform, and serve event streams in real time, at scale.项目地址https://gitcode.com/gh_mirrors/ri/risingwave点击查看免费下载src/README.md是 RisingWave 仓库中面向开发者的核心入口文档它一方面以 crate 为粒度梳理了src/目录下各计算引擎与基础设施的代码组织方式另一方面给出了两条阅读宏密集型 Rust 代码库的实用路径用cargo expand输出展开后的代码或借助rust-analyzer在编辑器内就地展开宏。读完本文你将能够快速定位 RisingWave 各核心模块流式计算、批处理、前端、存储、元数据等在仓库中的位置并掌握commit_meta!这类声明式宏与#[aggregate]这类过程宏的展开阅读方法从而更高效地进入源码调试与二次开发。一、src/目录十大 crate 的职责划分RisingWave 的绝大多数组件都用 Rust 开发并拆分到多个独立 crate 中。src/README.md给出的划分如下#crate职责1config存放各服务器frontend / compute / meta / compactor 等的默认配置2prost存放由 protobuf 生成的 Rust 代码包括 gRPC 服务定义与消息定义3stream流式计算引擎Stream Compute Engine4batch批处理计算引擎负责对物化视图执行查询5frontendSQL 查询规划器Planner与调度器Scheduler6storage云原生存储引擎State Store7meta元数据服务Meta Service8utils若干相互独立的工具 crate用于简化开发文档注明未来成熟后会发布到 crates.io9cmd/cmd_allcmd包含所有独立二进制cmd_all包含一体化二进制risingwave10risedevtool面向开发者的强大工具集详见 src/risedevtool/README.md与文档一一对应当前仓库src/下确实存在config、prost、stream、batch、frontend、storage、meta、utils、cmd、cmd_all、risedevtool等目录此外还包括common跨模块共享的基础库、expr表达式与 UDF 实现、connector上下游连接器、ctlrisectl 控制工具、error错误码体系、object_store、rpc_client、sqlparser、workspace-hack等辅助 crate。各核心引擎的深入入口文档为stream、storage、meta三个模块分别指向了更详细的专题文档Stream Engine / State Store / Meta Service并建议阅读 src/risedevtool/README.md 了解开发工具链。当前仓库内虽然没有保留这三篇专题文档的原文件但你可以在以下目录中直接阅读对应源码流式计算引擎src/stream聚合、窗口、join、物化视图维护等算子均在此实现云原生存储引擎src/storage包括 Hummock 层次化 LSM 存储、版本管理、compaction 等元数据服务src/meta涵盖 catalog、barrier、hummock manager、备份恢复等模块RPC 服务开发规范可参考 src/docs/development/how-to-write-a-rpc-service.md。二、为什么要专门学习读宏RisingWave 在开发中大量使用宏来简化代码声明式宏macro_rules!负责批量生成重复的样板代码过程宏#[derive]/#[proc_macro_attribute]则承担用声明式写法生成 trait 实现的任务。直接阅读宏定义也能理解其行为但对读者而言更高效的方式是把宏展开后的代码打出来再读。文档给出了两条互补的路径命令行展开使用cargo expand将宏展开后的完整代码输出到文件再在编辑器中对照阅读编辑器内展开在rust-analyzer中开启宏展开能力交互式地查看任意宏的展开结果见 rust-analyzer 官方文档的 Expand Macro Recursively 一节文档明确指出这是交互式理解代码的更优选择。下面结合仓库源码逐一复现文档中的两个展开案例。三、示例 1声明式宏commit_meta!()文档给出的第一个案例位于元数据服务的 catalog 管理逻辑中在meta::manager::catalog::CatalogManager::finish_create_table_procedure内部调用了commit_meta!()宏。展开步骤# 将 risingwave_meta crate 的代码展开到 meta.rs cargo expand -p risingwave_meta meta.rs然后在编辑器中打开meta.rs搜索函数调用finish_create_table_procedure将原始代码与meta.rs中的展开代码逐行对比。展开后看到的实质逻辑文档给出了展开结果的核心部分后续的 instrumentation 统计代码可以忽略async { tables.apply_to_txn(mut trx)?; self.env.meta_store().txn(trx).await?; tables.commit(); MetaResult::Ok(()) }.instrument(/* ... */)这段展开代码揭示了commit_meta!的本质把一个元数据变更集合tables先通过apply_to_txn应用到事务上再以单次事务提交到元数据存储成功后调用commit使内存中的缓存生效。整个流程被包装在一个async块中并附带instrument埋点以便追踪。需要说明的是随着代码演进commit_meta!宏在当前仓库的src/meta中已不在原位置但把一批变更原子地写入 MetaStore 事务这一模式在src/meta中仍然随处可见。例如 src/meta/src/hummock/manager/utils.rs 中定义了commit_multi_var!、commit_multi_var_with_provided_txn!等同类声明式宏src/meta/model/src/lib.rs 中还有面向所有元数据实体database、table、schema、source、sink、fragment、hummock 版本等 40 余种实体批量生成代码的for_all_meta_model_entities!宏。理解了commit_meta!的展开模式就能举一反三地读懂这些宏族。四、示例 2过程宏#[aggregate]第二个案例是#[aggregate]过程宏它负责把以函数式写法描述聚合逻辑的 Rust 函数自动转换为实现AggregateFunctiontrait 的聚合算子并注册到全局聚合函数注册表。展开步骤# 将 risingwave_expr crate 的代码展开到 expr.rs cargo expand -p risingwave_expr expr.rscargo expand会把各模块合并成一个按模块名命名空间隔离的大文件因此以string_agg为例在生成的expr.rs中搜索mod string_agg即可定位。string_agg 的原始定义在仓库中string_agg的真实定义位于 src/expr/impl/src/aggregate/string_agg.rs#[aggregate(string_agg(varchar, varchar) - varchar)] fn string_agg( state: OptionBoxstr, value: Optionstr, delimiter: Optionstr, ) - OptionBoxstr { let Some(value) value else { return state }; let Some(state) state else { return Some(value.into()); }; let mut state String::from(state); state delimiter.unwrap_or(); state value; Some(state.into()) }可以看到开发者只需以状态 输入值的函数式签名描述聚合逻辑state是当前累计状态value与delimiter是本次输入遇到NULL值时直接跳过首次输入时以该值初始化状态后续则把分隔符与值追加到状态字符串末尾。展开后生成的代码做了什么文档指出展开后的string_agg模块中会出现extern fn string_agg_varchar_varchar_varchar即以参数类型命名的导出函数。阅读展开代码可以确认#[aggregate]实际做了三件事注册将聚合函数注册到全局聚合注册表签名对象注册的内容是一个AggFuncSig聚合函数签名对象可在risingwave_expr中查看其定义生成实现生成一个匿名结构体实现AggregateFunctiontrait——这正是流式聚合算子、批处理聚合算子共同依赖的核心 trait。该宏的底层实现在 src/expr/macro/src/lib.rs#[proc_macro_attribute] pub fn aggregate把属性字符串如string_agg(varchar, varchar) - varchar解析为FunctionAttr把被修饰的函数解析为AggregateFnOrImpl然后调用generate_aggregate_descriptor生成注册代码与匿名结构体最终拼接回原有函数定义。与之配套的还有#[build_aggregate]用于已手动实现 trait 的场景与#[function]、#[build_function]普通标量函数等过程宏它们共享同一套FunctionAttr展开框架。从展开到运行聚合算子的落地展开生成的匿名结构体实现AggregateFunctiontrait而 trait 本身定义在 src/expr/core/src/aggregate/mod.rs其衍生实现还包括scalar_wrapper标量包装等辅助类型。聚合算子的实际消费方则位于批处理与流式两侧批处理侧src/batch/executors/src/executor/aggregation/下的hash_agg.rs、sort_agg.rs、distinct.rs、filter.rs、orderby.rs、projection.rs等执行器流式侧src/stream中的聚合相关算子。这意味着当你通过cargo expand读懂一个#[aggregate]函数的展开代码后可以继续沿着AggregateFunctiontrait 的调用链追踪到真正的执行器形成声明 → 展开 → 注册 → 执行的完整认知闭环。五、宏阅读实战小结综合两个示例阅读 RisingWave 宏代码的推荐流程可以总结为定位宏调用点在源码中搜索宏名如commit_meta!、#[aggregate]找到被修饰的函数或调用位置命令式展开对所在 crate 执行cargo expand -p crate out.rs生成合并后的展开文件按模块/函数名定位在展开文件中搜索原始函数名或mod name展开后的代码按模块命名空间组织对照原始代码逐行比对原始写法的声明意图与展开代码的实际行为忽略 instrumentation 等无关噪音交互式替代在编辑器中配置rust-analyzer的 Expand Macro Recursively 功能可省去生成文件的步骤适合频繁切换宏调用点时使用。cargo expand需要cargo-expand子命令可通过cargo install cargo-expand安装而rust-analyzer的宏展开属于编辑器内建能力无需额外工具链。两种方式互为补充前者适合批量输出、整体检索与离线分析后者适合逐点跟进、快速验证对单个宏的理解。掌握了这套方法论你在阅读 RisingWave 的src/meta、src/expr、src/frontend等宏密集模块时就能把黑盒的宏还原成白盒的代码让源码研究事半功倍。赞分享数据库流处理后端数据工程【免费下载链接】risingwaveEvent streaming platform for agentic AI. Continuously ingest, transform, and serve event streams in real time, at scale.项目地址https://gitcode.com/gh_mirrors/ri/risingwave点击查看免费下载相关推荐Notepad-- 插件开发完整教程快速写出第一个插件并跑通构建Notepad 插件开发完整教程快速写出第一个插件并跑通构建 Notepad 是一款支持 Windows、Linux、Mac 的跨平台文本编辑器。它的 C桌面应用CherryPy静态文件服务快速部署CSS、JS和图片资源的完整指南CherryPy静态文件服务快速部署CSS、JS和图片资源的完整指南 CherryPy是一个Pythonic的面向对象HTTP框架它提供了简单而强大的静态文后端Web框架rust-clippy宏展开检查深入处理复杂Rust宏代码rust clippy宏展开检查深入处理复杂Rust宏代码 宏展开检查的必要性与挑战 Rust宏系统Macro System作为元编程Metaprogr静态分析代码质量开发工具上一篇DataHub Elasticsearch 数据源接入指南索引元数据、Schema 字段类型与 Index Template 的完整摄取方案下一篇为什么安全专家都在用SlackPirate5大核心优势揭秘创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考