wal2json中numeric类型转字符串问题及还原方法咨询

Hey there, let's break down what's going on with your numeric column showing up as a weird string like "Ao5s" in Kafka, and how to get it back to a proper numeric value.

First: Why is this happening? #

The short answer is that wal2json defaults to serializing PostgreSQL numeric types as base64-encoded binary data, not human-readable strings or JSON numbers.

PostgreSQL's numeric type is arbitrary-precision, meaning it can hold numbers with far more precision than JSON's native number type (which has limits similar to Java's double). To avoid losing any precision during replication, wal2json doesn't convert numeric values directly to JSON numbers. Instead, it takes the raw binary representation of the numeric value from PostgreSQL's WAL (Write-Ahead Log), encodes it as base64, and sends that string to Kafka. That's why you're seeing values like "Ao5s"—it's just base64 for the binary numeric data.

Fix 1: Configure wal2json to Output Numeric as a Readable String #

The easiest solution is to tweak wal2json's settings to serialize numeric values directly as string representations of the number (e.g., "1675.32" instead of base64). Here's how:

Option A: Set the option when creating the replication slot #

If you're creating the logical replication slot manually, pass the numeric_output_mode=string parameter:

SELECT pg_create_logical_replication_slot(
    'your_debezium_slot_name', 
    'wal2json', 
    true, 
    '{"numeric_output_mode": "string"}'
);

Option B: Add the setting to your Debezium Connector config #

If you're using Debezium to manage the replication slot, add this line to your connector's properties:

plugin.options=numeric_output_mode=string

This tells Debezium to pass the numeric_output_mode setting to wal2json, so all future numeric changes will be sent as plain string numbers that your Java consumer can easily parse into BigDecimal or other numeric types.

Fix 2: Decode the Existing Base64 String Back to Numeric #

If you already have messages in Kafka with the base64-encoded numeric values, you can decode them in your Java consumer by parsing PostgreSQL's binary numeric format. Here's a simplified implementation to get you started:

Step-by-Step Java Decoding Example #

import java.math.BigDecimal;
import java.util.Base64;

public class PostgresNumericDecoder {

    public static BigDecimal decodeFromBase64(String base64Encoded) {
        byte[] rawBytes = Base64.getDecoder().decode(base64Encoded);
        
        // Parse PostgreSQL numeric header fields
        int totalLength = bytesToInt(rawBytes, 0, 4);
        int weight = bytesToShort(rawBytes, 4, 2);
        int sign = rawBytes[6] & 0xFF; // 0 = positive, 1 = negative
        int decimalScale = rawBytes[7] & 0xFF;
        
        // Build the full number string from 1000-digit groups
        StringBuilder numberBuilder = new StringBuilder();
        if (sign == 1) {
            numberBuilder.append('-');
        }
        
        int groupCount = (totalLength - 8) / 2;
        boolean isFirstGroup = true;
        
        for (int i = 0; i < groupCount; i++) {
            int groupStart = 8 + (i * 2);
            int groupValue = bytesToShort(rawBytes, groupStart, 2);
            
            if (isFirstGroup) {
                numberBuilder.append(groupValue);
                isFirstGroup = false;
            } else {
                // Pad with leading zeros to maintain 4-digit groups
                numberBuilder.append(String.format("%04d", groupValue));
            }
        }
        
        // Insert the decimal point based on the scale
        if (decimalScale > 0) {
            int integerPartLength = numberBuilder.length() - (sign == 1 ? 1 : 0);
            int decimalPosition = integerPartLength - decimalScale;
            
            if (decimalPosition <= 0) {
                // Handle cases where decimal is before the first digit (e.g., 0.00123)
                numberBuilder.insert(sign == 1 ? 1 : 0, "0.");
                numberBuilder.insert(sign == 1 ? 3 : 2, "0".repeat(-decimalPosition));
            } else {
                numberBuilder.insert(decimalPosition + (sign == 1 ? 1 : 0), '.');
            }
        }
        
        return new BigDecimal(numberBuilder.toString());
    }

    // Helper to convert byte ranges to int/short
    private static int bytesToInt(byte[] bytes, int offset, int length) {
        int result = 0;
        for (int i = offset; i < offset + length; i++) {
            result = (result << 8) | (bytes[i] & 0xFF);
        }
        return result;
    }

    private static int bytesToShort(byte[] bytes, int offset, int length) {
        return bytesToInt(bytes, offset, length);
    }
}

Notes on the Decoder #

• PostgreSQL's numeric binary format uses 1000-digit groups (each stored as 2 bytes) to represent large numbers. The weight field indicates the exponent of the highest group (e.g., a weight of 1 means the first group is multiplied by 1000^1). The simplified example above assumes the weight is aligned correctly for most common cases, but for edge cases (like extremely large/small numbers), you may need to adjust the logic to account for the weight value.
• Always test this decoder with your actual data to ensure precision is preserved.

Final Recommendation #

If you can, go with Fix 1—configuring wal2json to output numeric as a string is far simpler and avoids the need for custom decoding logic. Only use Fix 2 if you can't reprocess the existing Kafka messages.

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