linkedin / linkedin/goavro

Add support for custom logical type conversions to `TextualFromNative`

Open
#253 9 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
Go
Stars
1.1k
Forks
232
PR merge metrics
No merged PRs in 30d

Description

While trying to replicate the Snowflake Kafka Connector functionality using [Benthos](https://benthos.dev/), I noticed that in Java they are [adding](https://github.com/snowflakedb/snowflake-kafka-connector/blob/9eaecdd625e13c708188dd6fad589dcc2b360903/src/main/java/com/snowflake/kafka/connector/records/SnowflakeAvroConverter.java#L199) `genericData.addLogicalTypeConversion(new Conversions.DecimalConversion())` in addition to the implicit conversions when converting AVRO to JSON. I **believe** this is done such that a `bytes` type with logical type `decimal` can be transformed in a way which preserves some information about the scale and precision. For example, the following Java code:

```java
// > export JAVA_HOME=/usr/local/opt/openjdk
// > export PATH="${JAVA_HOME}/bin:$PATH"
// > java --version
// openjdk 18.0.1 2022-04-19
// OpenJDK Runtime Environment Homebrew (build 18.0.1+0)
// OpenJDK 64-Bit Server VM Homebrew (build 18.0.1+0, mixed mode, sharing)
// > java -cp avro_1.11/avro-1.11.0.jar:avro_1.11/jackson-core-2.12.5.jar:avro_1.11/jackson-annotations-2.12.5.jar:avro_1.11/jackson-databind-2.12.5.jar:avro_1.11/slf4j-api-1.7.32.jar Main.java

import java.math.BigInteger;
import java.util.Arrays;
import java.io.*;
import org.apache.avro.Schema;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.*;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.Conversions;
import org.apache.avro.file.DataFileReader;
import java.util.HexFormat;

public class Main {
static byte[] fromJsonToAvro(String json, Schema schema) throws Exception {
InputStream input = new ByteArrayInputStream(json.getBytes());
DataInputStream din = new DataInputStream(input);

Decoder decoder = DecoderFactory.get().jsonDecoder(schema, din);

DatumReader reader = new GenericDatumReader(schema);
Object datum = reader.read(null, decoder);

GenericDatumWriter w = new GenericDatumWriter(schema);
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();

Encoder e = EncoderFactory.get().binaryEncoder(outputStream, null);

w.write(datum, e);
e.flush();

return outputStream.toByteArray();
}

static String schemaJSON = """
{
"type": "record",
"name": "bytesdecimal",
"fields": [
{
"default": null,
"name": "pos_0_33333333",
"type": [
"null",
{
"logicalType": "decimal",
"precision": 16,
"scale": 2,
"type": "bytes"
}
]
}
]
}
""";

static String inputJSON = """
{
"pos_0_33333333": {
"bytes": "!"
}
}
""";

public static void main(String[] args) {
try {
Schema schema = new Schema.Parser().parse(schemaJSON);

byte[] avroByteArray = fromJsonToAvro(inputJSON, schema);

final GenericData genericData = new GenericData();
genericData.addLogicalTypeConversion(new Conversions.DecimalConversion());
DatumReader reader = new GenericDatumReader(schema, schema, genericData);

Decoder decoder = DecoderFactory.get().binaryDecoder(avroByteArray, null);
GenericRecord record = reader.read(null, decoder);
System.out.println(record);
}
catch (Exception e) {
System.out.println(e);
}
}
}
```

outputs `{"pos_0_33333333": 0.33}`, while this Go code:

```go
package main

import (
"testing"

"github.com/linkedin/goavro/v2"
"github.com/stretchr/testify/require"
)

func TestLogicalTypeToJSON(t *testing.T) {
schema := `
{
"type": "record",
"name": "bytesdecimal",
"fields": [
{
"default": null,
"name": "pos_0_33333333",
"type": [
"null",
{
"logicalType": "decimal",
"precision": 16,
"scale": 2,
"type": "bytes"
}
]
}
]
}`

jsonString := `
{
"pos_0_33333333": "!"
}`

codec, err := goavro.NewCodecForStandardJSON(schema)
require.NoError(t, err)

native, _, err := codec.NativeFromTextual([]byte(jsonString))
require.NoError(t, err)

bs, err := codec.TextualFromNative(nil, native)
require.NoError(t, err)

require.Equal(t, `{"pos_0_33333333":{"bytes.decimal":"!"}}`, string(bs))
}
```

outputs `{"pos_0_33333333":{"bytes.decimal":"!"}}`. While this might be more intuitive, a consumer loses any information on precision and scale. It also exhibits the issue raised in #252 regarding `bytes.decimal` vs `bytes`.

Note: There are [other converters](https://github.com/search?l=Java&q=addLogicalTypeConversion&type=Code) which can be added in Java, but this Snowflake connector only uses this `DecimalConversion`.

Would there be any interest to support this functionality? I'm happy to put in the work to write the new code with unit tests and whatever is needed for it, but I'd like to know what your thoughts are on this first and maybe how you'd expect the API to be extended for this purpose. Maybe adding an extra method on the codec, like `AddLogicalTypeConversion`?

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start by tracing the codec paths named in the report, especially TextualFromNative and NativeFromTextual, and compare them with the provided TestLogicalTypeToJSON example. Define the extension point for custom logical type conversions and add tests showing that decimal logical types preserve their scale and precision without regressing the bytes.decimal behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
go
Domain
data-engineering
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.