求PostgreSQL C扩展函数示例：表为参数并返回表（物化模式）

核心问题解决 #

你遇到的记录未拆分为独立列的问题，本质是没有正确返回行类型集合，而是返回了单个复合类型值。以下是完整的物化模式C扩展示例，实现从3列double的输入表生成3列double的输出表。

1. 扩展SQL定义（transform_triple--1.0.sql） #

先定义匹配的行类型和函数：

-- 定义对应输入输出的行类型
CREATE TYPE triple_double AS (
    a double precision,
    b double precision,
    c double precision
);

-- 创建物化模式的表函数
CREATE FUNCTION transform_triple(input_rows SETOF triple_double)
RETURNS SETOF triple_double
LANGUAGE C
IMMUTABLE
ROWS 1000  -- 预估返回行数，优化执行计划
AS 'MODULE_PATHNAME', 'transform_triple';

2. C扩展实现（transform_triple.c） #

#include "postgres.h"
#include "fmgr.h"
#include "funcapi.h"
#include "access/htup_details.h"
#include "utils/typcache.h"

PG_MODULE_MAGIC;

// 存储全量输入数据和迭代状态的结构体
typedef struct {
    double *col_a;
    double *col_b;
    double *col_c;
    int total_rows;
    int current_row;
} TransformState;

PG_FUNCTION_INFO_V1(transform_triple);

Datum
transform_triple(PG_FUNCTION_ARGS)
{
    FuncCallContext *funcctx;
    TransformState *state;

    // 第一次调用：读取全量输入并初始化状态（物化模式核心）
    if (SRF_IS_FIRSTCALL())
    {
        MemoryContext old_ctx = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
        funcctx = SRF_FIRSTCALL_INIT();
        
        // 初始化状态结构体
        state = palloc0(sizeof(TransformState));
        funcctx->user_fctx = state;
        
        // 动态分配数组存储输入数据（初始容量1000行）
        state->col_a = palloc(1000 * sizeof(double));
        state->col_b = palloc(1000 * sizeof(double));
        state->col_c = palloc(1000 * sizeof(double));
        state->total_rows = 0;

        // 遍历所有输入行，一次性读取全量数据
        while (!PG_ARGISNULL(0))
        {
            HeapTupleHeader input_tuple = PG_GETARG_HEAPTUPLEHEADER(0);
            
            // 提取每行的三个double列值
            state->col_a[state->total_rows] = DatumGetFloat8(GetAttributeByNum(input_tuple, 1));
            state->col_b[state->total_rows] = DatumGetFloat8(GetAttributeByNum(input_tuple, 2));
            state->col_c[state->total_rows] = DatumGetFloat8(GetAttributeByNum(input_tuple, 3));
            
            state->total_rows++;
            
            // 数组容量不足时扩容
            if (state->total_rows % 1000 == 0)
            {
                state->col_a = repalloc(state->col_a, (state->total_rows + 1000) * sizeof(double));
                state->col_b = repalloc(state->col_b, (state->total_rows + 1000) * sizeof(double));
                state->col_c = repalloc(state->col_c, (state->total_rows + 1000) * sizeof(double));
            }
            
            // 移动到下一个输入行
            PG_NEXT_ARG();
        }

        state->current_row = 0;
        MemoryContextSwitchTo(old_ctx);
    }

    // 后续调用：从缓存的状态中返回结果
    funcctx = SRF_PERCALL_SETUP();
    state = funcctx->user_fctx;

    if (state->current_row < state->total_rows)
    {
        Datum output_vals[3];
        bool output_nulls[3] = {false, false, false};
        HeapTuple output_tuple;
        Datum result_datum;

        // 这里添加你的转换逻辑，示例为每个值乘以2
        output_vals[0] = Float8GetDatum(state->col_a[state->current_row] * 2);
        output_vals[1] = Float8GetDatum(state->col_b[state->current_row] * 2);
        output_vals[2] = Float8GetDatum(state->col_c[state->current_row] * 2);

        // 构建输出行
        output_tuple = heap_form_tuple(funcctx->tuple_desc, output_vals, output_nulls);
        result_datum = HeapTupleGetDatum(output_tuple);

        state->current_row++;
        SRF_RETURN_NEXT(funcctx, result_datum);
    }

    // 所有结果返回完毕
    SRF_RETURN_DONE(funcctx);
}

3. 关键细节解释 #

• 物化模式实现：通过SRF_IS_FIRSTCALL()在第一次调用时读取所有输入行，存储到内存中，后续调用直接返回缓存结果，适合需要全局数据的转换场景。
• 多列返回：用heap_form_tuple构建符合triple_double类型的行，通过SRF_RETURN_NEXT返回行类型Datum，PostgreSQL会自动解析为独立列，而非单个复合记录。
• 内存管理：使用funcctx->multi_call_memory_ctx存储状态数据，避免每次调用重新分配，确保内存生命周期与函数执行一致。

4. 测试方法 #

-- 创建测试表
CREATE TABLE test_data (a double, b double, c double);
INSERT INTO test_data VALUES (1.5, 2.5, 3.5), (4.0, 5.0, 6.0);

-- 调用函数，直接得到拆分的三列结果
SELECT * FROM transform_triple(SELECT (a,b,c) FROM test_data);

执行后会输出：

内容的提问来源于stack exchange，提问作者Rafael Scudelari de Macedo