如何将Oracle的层级查询改写为Polars代码？

嘿，我来帮你把这段Oracle的层级查询转换成Polars代码！先帮你理清原SQL的核心逻辑：你有两个数据集——prod是产品基础信息表，存了产品ID和编码；prre是产品层级关联表，记录了子产品（detail）和父产品（master）的对应关系，应该是要构建产品的层级递归结构对吧？

第一步：在Polars中构建对应数据集 #

首先我们把原Oracle的CTE转换成Polars的DataFrame，这是后续操作的基础：

import polars as pl

# 对应原SQL中的prod表
prod = pl.DataFrame({
    "id": [10, 11, 12, 13, 14, 15, 16],
    "code": ["1008", "1582", "1583", "2023", "2025", "2030", "2222"]
})

# 对应原SQL中的prre表，你原SQL里的省略号我就按给出的部分来写了
prre = pl.DataFrame({
    "detail_product_id": [10, 12, 91, 14, 90, 11],
    "master_product_id": [90, 11, 92, 12, 93, 91]
})

第二步：实现层级递归查询（模拟Oracle CONNECT BY） #

Oracle里的层级查询一般用CONNECT BY加递归条件，我们在Polars里可以用递归循环来实现类似逻辑。比如下面这个函数，能从指定的产品ID出发，递归向上查找所有父级产品：

def get_parent_hierarchy(start_product_id):
    # 先获取起始节点的直接父级
    hierarchy = (
        prre.filter(pl.col("detail_product_id") == start_product_id)
        .join(prod, left_on="master_product_id", right_on="id", how="left")
        .select(
            pl.col("detail_product_id").alias("child_id"),
            pl.col("master_product_id").alias("parent_id"),
            pl.col("code").alias("parent_code")
        )
    )
    
    # 递归向上遍历，直到没有更多父级节点
    while not hierarchy.filter(pl.col("parent_id").is_in(prod["id"])).is_empty():
        # 查找当前层级节点的父级
        next_level = (
            hierarchy.filter(pl.col("parent_id").is_in(prod["id"]))
            .join(prre, left_on="parent_id", right_on="detail_product_id", how="inner")
            .join(prod, left_on="master_product_id", right_on="id", how="left")
            .select(
                pl.col("child_id"),
                pl.col("master_product_id").alias("parent_id"),
                pl.col("code").alias("parent_code")
            )
        )
        # 合并新层级数据
        hierarchy = pl.concat([hierarchy, next_level])
    
    # 把起始节点的基础信息和层级数据合并
    start_node = prod.filter(pl.col("id") == start_product_id).select(
        pl.col("id").alias("child_id"),
        pl.lit(None).alias("parent_id"),
        pl.col("code").alias("child_code")
    )
    final_result = start_node.join(hierarchy, on="child_id", how="left")
    return final_result

# 示例：查询ID为14的产品的完整父级层级
result = get_parent_hierarchy(14)
print(result)

如果需要向下遍历子产品，只需要调换detail_product_id和master_product_id的关联逻辑，修改过滤和连接的字段即可。

第三步：大数据量优化（用LazyFrame） #

如果你的数据量很大，推荐用Polars的懒加载（LazyFrame）来提升性能，避免一次性加载全量数据：

def get_parent_hierarchy_lazy(start_product_id):
    prod_lazy = prod.lazy()
    prre_lazy = prre.lazy()
    
    # 起始节点懒加载查询
    start_node = prod_lazy.filter(pl.col("id") == start_product_id).select(
        pl.col("id").alias("child_id"),
        pl.lit(None).alias("parent_id"),
        pl.col("code").alias("child_code")
    )
    
    # 初始层级懒加载查询
    hierarchy = (
        prre_lazy.filter(pl.col("detail_product_id") == start_product_id)
        .join(prod_lazy, left_on="master_product_id", right_on="id", how="left")
        .select(
            pl.col("detail_product_id").alias("child_id"),
            pl.col("master_product_id").alias("parent_id"),
            pl.col("code").alias("parent_code")
        )
    )
    
    # 递归遍历直到无新节点
    while True:
        next_level = (
            hierarchy.join(prre_lazy, left_on="parent_id", right_on="detail_product_id", how="inner")
            .join(prod_lazy, left_on="master_product_id", right_on="id", how="left")
            .select(
                pl.col("child_id"),
                pl.col("master_product_id").alias("parent_id"),
                pl.col("code").alias("parent_code")
            )
        )
        # 检查是否有新数据，这里先collect一小部分判断
        if next_level.head(1).collect().is_empty():
            break
        hierarchy = pl.concat([hierarchy, next_level])
    
    # 最终执行查询并返回结果
    final_result = start_node.join(hierarchy, on="child_id", how="left").collect()
    return final_result

# 调用懒加载版本
lazy_result = get_parent_hierarchy_lazy(14)
print(lazy_result)

备注：内容来源于stack exchange，提问作者lmocsi