流式图计算引擎 GeaFlow v0.6.4 发布，支持关系型访问图数据，增量匹配优化实时处理

• 🍀GeaFlow 图存储扩展支持 paimon 数据湖（实验性功能）
• 🍀图数仓能力扩展：支持对图中的实体进行关系型访问
• 🍀统一的内存管理器支持
• 🍀RBO 规则扩展：新增 MatchEdgeLabelFilterRemoveRule 和 MatchIdFilterSimplifyRule
• 🍀支持增量匹配算子

• 支持对象存储/HDFS 分布式存储，天然适配云原生环境。因此可实现存储与计算分离，降低硬件依赖，支持弹性扩展。
• 支持主键表 LSM 合并、增量更新，满足实时数据更新需求。
• 列式存储+统计索引（Z-Order、Min-Max 等），支持高效数据裁剪与 OLAP 查询加速。

• 支持在 GeaFlow 中将用户图数据存储到 paimon 数据湖。
• 当前为实验性功能，仅支持使用本地文件系统作为 paimon 的存储后端，且暂不支持 recover 能力，暂不支持动态图数据存储。
• 通过配置`geaflow.store.paimon.options.warehouse`参数来指定存储路径，默认路径为"file:///tmp/paimon/"。
  
  

• 支持图中的点/边作为 SQL 查询的来源表，进行查询。

state 和 shuffle 默认独占10%的堆外内存， default 则占用80%。

🍀RBO 规则扩展：新增 EdgeLabel 和 IdFilter 优化规则

• Edge Label 简化：针对 Match 匹配语句后接 Where 子句对边进行过滤的查询进行执行计划简化。
• ID Filter简化：针对Match匹配语句中对点的id进行过滤的查询进行执行计划简化。
• 规则在默认情况下生效，使用示例如下：

// GQL示例1（MatchIdFilterSimplifyRule优化）
MATCH (a:user where id = 1)-[e:knows]-(b:user)
RETURN a.id as a_id, e.weight as weight, b.id as b_id


// 原执行计划
LogicalProject(a_id=[$0.id], weight=[$1.weight], b_id=[$2.id])
  LogicalGraphMatch(path=[(a:user) where =(a.id, 1) -[e:knows]-(b:user)])
    LogicalGraphScan(table=[default.g0])


// MatchIdFilterSimplifyRule优化后执行计划，vertex id转移到MatchVertex中进行过滤
LogicalProject(a_id=[$0.id], weight=[$1.weight], b_id=[$2.id])
  LogicalGraphMatch(path=[(a:user)-[e:knows]-(b:user)])
    LogicalGraphScan(table=[default.g0])


// GQL示例2（MatchEdgeLabelFilterRemoveRule优化）
MATCH (a:user where id = 1)-[e:knows]-(b:user) WHERE e.~label = 'knows' 
or e.~label = 'created'
RETURN a.id as a_id, e.weight as weight, b.id as b_id


// 原执行计划
LogicalProject(a_id=[$0.id], weight=[$1.weight], b_id=[$2.id])
  LogicalGraphMatch(path=[(a:user) where =(a.id, 1) -[e:knows]-(b:user) where OR(=($1.~label, _UTF-16LE'knows'), =($1.~label, _UTF-16LE'created')) ])
    LogicalGraphScan(table=[default.g0])


// MatchEdgeLabelFilterRemoveRule优化后执行计划，针对edge label的过滤转移到MatchEdge中进行
LogicalProject(a_id=[$0.id], weight=[$1.weight], b_id=[$2.id])
  LogicalGraphMatch(path=[(a:user) where =(a.id, 1) -[e:knows]-(b:user)])
    LogicalGraphScan(table=[default.g0])

QueryTester.build()
     .withConfig(FrameworkConfigKeys.BATCH_NUMBER_PER_CHECKPOINT.getKey(), "1")
     .withQueryPath(queryPath)
     .withConfig(DSLConfigKeys.ENABLE_INCR_TRAVERSAL.getKey(), "true")
     .withConfig(DSLConfigKeys.TABLE_SINK_SPLIT_LINE.getKey(), lineSplit)
     .execute();