C语言中如何处理PostgreSQL的numeric二进制数据类型

• 基于C语言开发，目标是将libpq查询返回的二进制格式结果转换为Arrow格式。目前已通过PQftype()接口获取各列对应的数据类型OID，可完成绝大多数类型与Arrow数据类型的匹配，仅numeric类型的处理逻辑不明确。
• 执行SELECT oid, typname FROM pg_type;可确认numeric类型对应的OID为1700，但无法直接获取该类型列的精度（precision）与标度（scale）。
• 此前查阅官方文档中数值类型介绍章节未找到有效参考，不确定是否查阅位置有误。
• 核心疑问：PGresult中二进制格式的numeric值存储结构是怎样的。

numeric列精度与标度的获取方式 #

numeric是PostgreSQL的变长精度数值类型，类型本身（OID=1700）没有固定的精度、标度配置，因此仅查询pg_type系统表无法拿到这两个参数。
列定义的精度、标度存储在列属性atttypmod中，libpq提供了PQfmod()接口可直接获取结果集指定列的typmod值，计算规则如下：

• 若PQfmod()返回-1，代表该列未定义固定精度与标度
• 否则：
  • 精度 = ((typmod - 4) >> 16) & 0xFFFF
  • 标度 = (typmod - 4) & 0xFFFF

二进制格式numeric的存储结构 #

二进制返回的numeric是按网络字节序（大端）排列的16位整数序列，从低地址到高地址的结构为：

• ndigits：16位无符号整数，记录后续数位数组的元素总个数
• weight：16位有符号整数，代表小数点前的数位组权重，整数部分对应的数位组个数为weight + 1
• sign：16位无符号整数，标记数值符号：0为正数，0x4000为负数，0xC000为NaN值
• dscale：16位无符号整数，记录小数点后的十进制数位总长度
• digits[]：连续的16位无符号整数数组，每个元素存储一组0-9999的十进制数（PostgreSQL numeric采用NBASE=10000编码，每个数组元素对应4位十进制数位），按数值高位到低位顺序排列，不足4位的低位组末尾补0对齐。

以下实现可将libpq返回的二进制格式numeric转换为字符串表示，主要用于理解存储结构。实际转换为Arrow格式时，可直接解析二进制值填充Arrow Decimal数组，无需经过字符串中转，性能更高。

以精度20、标度2的数值49273.64为例，其二进制结构拆分如下：

ndigits | 00000000 | 00000011 | weight | 00000000 | 00000001 | sign | 00000000 | 00000000 | dscale | 00000000 | 00000010 | digits | 00000000 | 00000100 | 00100100 | 00111001 | 00011001 | 00000000 
对应字段值：ndigits=3, weight=1, sign=0（正数）, dscale=2, digits=[4, 9273, 6400]
拼接逻辑：整数部分取前weight+1=2个数位组 → 4、9273 → 49273；小数部分取dscale=2位，第三个数位组6400取前2位 → 64，最终结果为49273.64

转换为字符串的C语言实现：

char *getStrFromNumeric(u_int16_t *numvar){
    u_int16_t ndigits = ntohs(numvar[0]); // digits数组的16位元素总个数
    int16_t dscale = ntohs(numvar[3]);    // 小数点后的十进制数位总数
    int16_t weight = ntohs(numvar[1]) + 1;// 整数部分对应的digits元素个数
    // 分配结果内存：整数位（每个元素最多4位）+小数位+小数点+结束符，额外预留2字节缓冲
    char *result = (char *)malloc(sizeof(char) * (weight * 4 + dscale + 4));
    char *copyStr = (char *)malloc(sizeof(char) * 5);
    int strindex = 0;
    int numvarindex = 0;

    // 拼接整数部分
    while (weight > 0) {
        uint16_t group = ntohs(numvar[numvarindex + 4]);
        if (numvarindex > 0) {
            // 非首个整数组需补前导0至4位，避免类似10001被错误拼接为11
            sprintf(copyStr, "%04d", group);
        } else {
            sprintf(copyStr, "%d", group);
        }
        sprintf(result + strindex, "%s", copyStr);
        strindex += strlen(copyStr);
        numvarindex++;
        weight--;
    }

    // 处理值小于1、无整数部分的场景
    if (strindex == 0) {
        result[strindex++] = '0';
    }

    // 拼接小数部分
    if (dscale > 0) {
        result[strindex++] = '.';
        int remain = dscale;
        while (remain > 0 && numvarindex < ndigits) {
            uint16_t group = ntohs(numvar[numvarindex + 4]);
            int copy_len = remain >= 4 ? 4 : remain;
            sprintf(copyStr, "%04d", group);
            strncpy(result + strindex, copyStr, copy_len);
            strindex += copy_len;
            remain -= copy_len;
            numvarindex++;
        }
        // 数位不足时补0
        while (remain > 0) {
            result[strindex++] = '0';
            remain--;
        }
    }

    result[strindex] = '\0';
    free(copyStr);
    return result;
}

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