PostgreSQL自定义C语言聚合实现蓄水池采样的问题排查

问题1：指针地址存储/读取失败 #

你试图用sprintf把结构体指针转成字符串存入bytea，再用sscanf读回，这种做法存在两个致命问题：

• 指针地址仅在当前进程的内存上下文有效，聚合函数状态可能被序列化、跨进程传递（如并行查询），指针会完全失效。
• 字符串格式化解析指针依赖平台格式，极易出错，memcpy直接操作指针地址的逻辑也不成立——你存的是指针的字符串形式，而非二进制地址值。

解决方案：不要存储指针，直接将state_c结构体的内容序列化到bytea中，包括蓄水池的实际数据。利用PostgreSQL的内存管理函数处理序列化逻辑。

问题2：PG_ARGISNULL(0)未生效 #

你设置的INITCOND='{}'是一个非空bytea（对应空数组的bytea表示），因此PG_ARGISNULL(0)永远返回false，初始化分支无法触发。

解决方案：将INITCOND设为NULL，让初始化分支正常执行。修改聚合函数定义：

CREATE AGGREGATE reservoir_sampling_c(bigint)
(
        sfunc = res_trans_crimes_c,
        stype = bytea,
        FINALFUNC = finalize_trans_crimes_c,
        INITCOND = NULL
);

问题3：GDB调试断点无法停止 #

出现该错误的核心原因：

• C模块编译时未添加调试符号（-g选项），导致GDB无法定位函数地址。
• PostgreSQL未以调试模式启动，或断点设置在模块加载前。

解决方案：

• 编译模块时添加-g选项，若使用PGXS构建，修改Makefile：

  CFLAGS += -g
  

• 启动PostgreSQL时使用调试模式：postgres -d 1，或先通过GDB attach到postgres进程，再用函数名设置断点：break res_trans_crimes_c

修正后的完整代码 #

SQL代码 #

CREATE FUNCTION res_trans_crimes_c(bytea, bigint)
        RETURNS bytea
        AS 'MODULE_PATHNAME', 'res_trans_crimes_c'
        LANGUAGE C
        IMMUTABLE 
        PARALLEL SAFE;

CREATE FUNCTION finalize_trans_crimes_c(bytea)
        RETURNS bigint[]
        AS 'MODULE_PATHNAME','finalize_trans_crimes_c'
        LANGUAGE C
        IMMUTABLE 
        PARALLEL SAFE;

CREATE AGGREGATE reservoir_sampling_c(bigint)
(
        sfunc = res_trans_crimes_c,
        stype = bytea,
        FINALFUNC = finalize_trans_crimes_c,
        INITCOND = NULL
);

C代码 #

PG_MODULE_MAGIC;

#include <stdlib.h>
#include <postgres.h>
#include <fmgr.h>
#include <utils/array.h>
#include <utils/builtins.h>

typedef struct state_c
{
    int64 reservoir[3];  // 固定大小3的蓄水池数组
    int32 poscnt;        // 当前处理的元素总数
    int32 reservoir_size; // 蓄水池容量，固定为3
} state_c;

PG_FUNCTION_INFO_V1(res_trans_crimes_c);
Datum
res_trans_crimes_c(PG_FUNCTION_ARGS)
{
    state_c *s;
    int64 newsample = PG_GETARG_INT64(1);

    // 初始化状态
    if (PG_ARGISNULL(0))
    {
        s = palloc0(sizeof(state_c));
        s->reservoir_size = 3;
        s->poscnt = 1;
        s->reservoir[0] = newsample;
    }
    else
    {
        // 从bytea反序列化状态
        bytea *state_bytea = PG_GETARG_BYTEA_P(0);
        s = palloc0(sizeof(state_c));
        memcpy(s, VARDATA(state_bytea), sizeof(state_c));

        // 执行蓄水池采样逻辑
        if (s->poscnt <= s->reservoir_size)
        {
            s->reservoir[s->poscnt - 1] = newsample;
        }
        else
        {
            int32 pos = rand() % s->poscnt;
            if (pos < s->reservoir_size)
            {
                s->reservoir[pos] = newsample;
            }
        }
        s->poscnt++;
    }

    // 将状态序列化为bytea返回
    bytea *result = palloc(VARHDRSZ + sizeof(state_c));
    SET_VARSIZE(result, VARHDRSZ + sizeof(state_c));
    memcpy(VARDATA(result), s, sizeof(state_c));
    pfree(s);

    PG_RETURN_BYTEA_P(result);
}

PG_FUNCTION_INFO_V1(finalize_trans_crimes_c);
Datum
finalize_trans_crimes_c(PG_FUNCTION_ARGS)
{
    if (PG_ARGISNULL(0))
    {
        PG_RETURN_NULL();
    }

    bytea *state_bytea = PG_GETARG_BYTEA_P(0);
    state_c *s = palloc0(sizeof(state_c));
    memcpy(s, VARDATA(state_bytea), sizeof(state_c));

    // 构建返回的bigint数组
    Datum *elems = palloc(s->reservoir_size * sizeof(Datum));
    int i;
    for (i = 0; i < s->reservoir_size; i++)
    {
        elems[i] = Int64GetDatum(s->reservoir[i]);
    }

    ArrayType *result = construct_array(elems, s->reservoir_size, INT8OID, 8, true, 'd');
    pfree(s);
    pfree(elems);

    PG_RETURN_ARRAYTYPE_P(result);
}

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