PostgreSQL Apache AGE扩展中访问边属性值失败及函数无返回问题

背景 #

在PostgreSQL中使用Apache AGE扩展时，已创建两条Flight类型的边：

SELECT * FROM cypher('age_graph', $$ 
    MATCH (a1:Airport),(a2:Airport) WHERE a1.iata_code='ANC' AND a2.iata_code='SEA' CREATE (a1)-[e:Flight{flight_number: 'N407DA', departure_time: 55200, arrival_time: 64800, distance: 1448}]->(a2) RETURN e 
$$) as (e agtype);

SELECT * FROM cypher('age_graph', $$ 
    MATCH (a1:Airport),(a2:Airport) WHERE a1.iata_code='SEA' AND a2.iata_code='LAX' CREATE (a1)-[e:Flight{flight_number: 'N405KJ', departure_time: 63000, arrival_time: 70000, distance: 1448}]->(a2) RETURN e 
$$) as (e agtype);

随后编写了plpython3u函数paircpath：

CREATE OR REPLACE FUNCTION age_graph.paircpath(
    origin text,
    edge text,
    destination text
)
RETURNS TABLE(
    city1 ag_catalog.agtype, 
    airport1 ag_catalog.agtype, 
    flight ag_catalog.agtype, 
    airport2 ag_catalog.agtype, 
    city2 ag_catalog.agtype
) 
LANGUAGE 'plpython3u'
COST 100
VOLATILE PARALLEL UNSAFE
ROWS 1000
AS $function$
    import age
    import psycopg2

    def paircpath(origin, edge, destination):
        conn = psycopg2.connect(host="localhost", port="5432", dbname="postgres", user="postgres", password="13711992")
        with conn.cursor() as cursor:
            try:
                cursor.execute("SET search_path = ag_catalog, public, age_graph;")
                cursor.execute("LOAD 'age';")
                cursor.execute("GRANT USAGE ON SCHEMA ag_catalog TO postgres;")

                query = f"""SELECT * FROM cypher('age_graph', $$ 
                    MATCH (a)-[:LocatedAt*]->(c:City {{name: '{origin}'}})
                    MATCH (a:Airport)-[e:{edge}]->(b:Airport) 
                    MATCH (b)-[:LocatedAt*]->(c1:City {{name: '{destination}'}}) 
                    RETURN c, a, e, b, c1 
                $$) AS (city1 agtype, airport1 agtype, flight agtype, airport2 agtype, city2 agtype); """

                cursor.execute(query, ('age_graph',))
                paths = cursor.fetchall()

                for row in paths:
                    edges = row[2]
                    is_continuous = True
                    for i in range(len(edges) - 1):
                        arrival_time = edges[i].properties('arrival_time').value
                        departure_time = edges[i].properties('departure_time').value
                        next_departure_time = edges[i+1].properties('departure_time').value
                        if arrival_time < next_departure_time:
                            is_continuous = False
                            break
                    if is_continuous:
                        yield(row[0], row[1], row[2], row[3], row[4])


            except Exception as ex:
                print(type(ex), ex)

    for result in paircpath(origin, edge, destination):
        yield result

$function$;

调用该函数时PostgreSQL未返回任何结果，推测是无法正确访问边的属性值，尝试多种方法仍未解决，需要指导如何正确访问边属性值并修复函数问题。

问题分析与修复方案 #

1. 核心问题总结 #

• 属性访问方式错误：PL/Python中不能用.properties('xxx').value直接获取AGE边属性，需转换为Python字典操作。
• Cypher查询逻辑错误：原查询仅匹配单条Flight边，无法获取多段中转路径，且后续代码错误地尝试遍历单条边的列表。
• 时间判断逻辑倒置：原代码将到达时间早于出发时间判定为不连续，这与实际逻辑相反。
• 冗余连接操作：PL/Python函数无需重新建立数据库连接，使用plpy模块即可执行查询。

2. 具体修复步骤 #

步骤1：修正Cypher查询，匹配多段路径

修改查询以匹配从起点城市到终点城市的完整航班路径，并返回所有路径中的边：

query = f"""SELECT * FROM cypher('age_graph', $$ 
    MATCH path = (c:City {{name: '{origin}'}})<-[:LocatedAt*]-(a)-[{edge}*]->(b)-[:LocatedAt*]->(c1:City {{name: '{destination}'}})
    RETURN c, a, relationships(path), b, c1
$$) AS (city1 agtype, airport1 agtype, flights agtype, airport2 agtype, city2 agtype); """

步骤2：修正属性访问方式

使用age.convert()将AG类型转换为Python字典，直接通过键名获取属性：

from age import convert

flight_list = convert(row['flights'])
prev_arrival = flight_list[i]['arrival_time']
next_departure = flight_list[i+1]['departure_time']

步骤3：修复时间判断逻辑

正确逻辑应为：前序航班的到达时间小于等于后续航班的出发时间，才判定为连续：

if prev_arrival > next_departure:
    is_continuous = False
    break

步骤4：替换为plpy执行查询

移除冗余的psycopg2连接，使用PL/Python标准的plpy模块执行查询：

result_set = plpy.execute(query)

3. 修复后的完整函数 #

CREATE OR REPLACE FUNCTION age_graph.paircpath(
    origin text,
    edge text,
    destination text
)
RETURNS TABLE(
    city1 ag_catalog.agtype, 
    airport1 ag_catalog.agtype, 
    flights ag_catalog.agtype, 
    airport2 ag_catalog.agtype, 
    city2 ag_catalog.agtype
) 
LANGUAGE 'plpython3u'
COST 100
VOLATILE PARALLEL UNSAFE
ROWS 1000
AS $function$
    import plpy
    from age import convert

    def paircpath(origin, edge, destination):
        try:
            query = f"""SELECT * FROM cypher('age_graph', $$ 
                MATCH path = (c:City {{name: '{origin}'}})<-[:LocatedAt*]-(a)-[{edge}*]->(b)-[:LocatedAt*]->(c1:City {{name: '{destination}'}})
                RETURN c, a, relationships(path), b, c1
            $$) AS (city1 agtype, airport1 agtype, flights agtype, airport2 agtype, city2 agtype); """
            
            result_set = plpy.execute(query)
            
            for row in result_set:
                flight_list = convert(row['flights'])
                is_continuous = True
                
                for i in range(len(flight_list) - 1):
                    prev_arrival = flight_list[i]['arrival_time']
                    next_departure = flight_list[i+1]['departure_time']
                    if prev_arrival > next_departure:
                        is_continuous = False
                        break
                
                if is_continuous:
                    yield (row['city1'], row['airport1'], row['flights'], row['airport2'], row['city2'])
                    
        except Exception as ex:
            plpy.error(f"Function failed: {type(ex).__name__} - {str(ex)}")

    for res in paircpath(origin, edge, destination):
        yield res

$function$;

4. 额外注意事项 #

• 确保Airport与City节点之间存在LocatedAt边，否则Cypher查询无法匹配路径。
• 原代码中cursor.execute(query, ('age_graph',))属于冗余操作，且参数传递不匹配，会导致执行错误。

内容的提问来源于stack exchange，提问作者Jaber