connector: support protobuf map type in source

[BugenZhao]

Prior to #12126, a map field is accepted and map to struct<key, value>[] in RisingWave, which might not be expected as suggested by @neverchanje so disabled there.

JSONB might be another good candidate, which supports efficient indexing.

[xiangjinwu]

https://protobuf.dev/programming-guides/proto3/#maps
I would like to list 4 candidate solutions for map<K, V>:

• (A) struct<key K, value V>[]
• (B) jsonb
• (C) hstore
• (D) map<K, V>

I will focus on the downside of each approach:

• struct[] does not support efficient lookup natively, as discussed above. One possible workaround is to sort by key during ingestion, and provide a dedicated function/operator to do binary search.
• jsonb means everything inside would also be jsonb rather than strongly typed. Proto allows K to be integral types but jsonb requires keys to be string. Proto requires V to be the same for all kv pairs but jsonb allows some to be jsonb number and some to be jsonb string. In addition jsonb does not have good support of types likes int64 / bytea / timestamptz.
• hstore is the official PostgreSQL extension for kv pairs. However it is not generic and requires both K and V to be string.
• map<K, V> means we add a new dedicated type, because none of the options above maps perfectly. This takes more time to implement, but has the best compatibility with similar types in protobuf, avro, arrow, DuckDB, Flink, ...

[BugenZhao]

I would like to list 4 candidate solutions for map<K, V>:

Thanks for your detailed exploration! Personally, I would stand for struct[] or map<K, V>. Some ideas:

• It's inevitable to introduce a (series of?) indexing function, no matter which type it is.
• Assuming the maps ingested from external sources are not likely to be mutated later, simply sorting all entries in the struct[] and mark the sorted flag will enable efficient indexing.

[fuyufjh]

-1 for struct<key K, value V>[] because it lost the semantics of map.

map is built for mapping i.e. looking up by a key to get a value. It's important to make data type "lossless", but it's more important to keep it as original semantics.

Here, I prefer jsonb. Obviously, the idea map in protobuf is brought from semi-structured format, while in PG, semi-structure are mostly handled by JSONB

[BugenZhao]

It's okay to make it jsonb if we find most of the use cases are map<string, string>. However, it's also possible for the value to be a complex message. In this case we need another set of type mappings from protobuf to JSON, in additional to the original protobuf to RisingWave data type.

[fuyufjh]

In this case we need another set of type mappings from protobuf to JSON, in additional to the original protobuf to RisingWave data type.

It's inevitable since we have already supported Any.

[fuyufjh]

Thinking from the users' perspective, if we do really use struct<key K, value V>[] for map, at least we need to provide functions to

• lookup value by key
• check whether a key exist
• update or delete a field

Adding up together, I would rather introduce a new map data type in RW.

[xiangjinwu]

It's inevitable to introduce a (series of?) indexing function, no matter which type it is.

Putting development time aside, a native map<K, V> type is the best solution. So these additional struct[] functions could support both struct[] and map<K, V>.

As for jsonb, my biggest concern is that it makes all inner content jsonb, losing "the semantics of" type definition as well. For example, given a proto map<string, Location> mapped to jsonb, we would need to write col1 -> 'SFO' ->> 'latitude' rather than (lookup(col1, 'SFO')).latitude.

we need another set of type mappings from protobuf to JSON

This is not a big deal. Proto3 has an official definition of how proto maps to JSON. But do note that int64 maps to string.

[fuyufjh]

As for jsonb, my biggest concern is that it makes all inner content jsonb, losing "the semantics of" type definition as well. For example, given a proto map<string, Location> mapped to jsonb, we would need to write col1 -> 'SFO' ->> 'latitude' rather than (lookup(col1, 'SFO')).latitude.

Yeah, I understand this, and I guess it's also the major concern of all of us. For me, because Any has already broken this, I feel acceptable.

[xiangjinwu]

google.protobuf.Any is intentionally a dynamic type, but map could be a static type. But yes, sounds more acceptable before we have a dedicated map type.

Btw do we use the official proto3 json mapping in google.protobuf.Any? Is an int64 in any message represented by json string or json number?

[fuyufjh]

Btw do we use the official proto3 json mapping in google.protobuf.Any? Is an int64 in any message represented by json string or json number?

IIRC, Any type can only point to a message rather than primitive types.

I guess @xiangjinwu means an int64 inside a message in an Any

[BugenZhao]

My concernings for a native map type is:

• Introduction of the data type itself is simple. However, there seems to be a considerable amount of work in the parser and expressions to make it first-class, just like other semi-structured types.
• We need to carefully tune the performance, apply the knowledge on the JSONB and the List type once again.
• Such challenges would we have to face, the Protobuf seems to be the only usage. So the test coverage could be limited as well.

Thinking from the users' perspective, if we do really use struct<key K, value V>[] for map, at least we need to provide functions to

• lookup value by key

• check whether a key exist

• update or delete a field

Adding up together, I would rather introduce a new map data type in RW.

As above, it might not be that simple...

[xiangjinwu]

Had another offline discussion with @fuyufjh and we aligned on these:

• hstore is out because it is only nonnull string -> nullable string.
• jsonb may be sufficient to cover common cases. It is a popular fallback when different system have different support of complex types (array, struct, map, enum, union, ...)
  • Further simplification: value can only be a base type (types except array/struct).
  • There will be two styles: int64 as number (postgres) and int64 as string (protobuf). The former is more friendly for further processing in RisingWave. Support of the latter can be added if requested to be strongly protobuf-compatible.
• map may be an overkill unless we get frequent complaints about the jsonb solution above. By that time we would skip struct[] workaround and do map directly.

[fuyufjh]

the Protobuf seems to be the only usage.

So we have a new usage now #14215 😄

[fuyufjh]

I just realized that many systems introduced dedicated map type, including

• ClickHouse
• DuckDB
• Hive as well as many big-data systems compatible with Hive SQL e.g. Flink & Spark
• Materialize

Perhaps Postgres is the special one... Shall we add support for map as well?

[xiangjinwu]

So we have a new usage now #14215 😄

Concern of using jsonb in that issue for kafka headers is the missing of native bytea/bytes support. Or actually there are multiple ways to represent 8 bytes[0x61 0x30 0x00 0x20 0xf0 0x9f 0x8c 0x8a] in json as a string:

• 12-byte "YTAAIPCfjIo=" (base64)
• 16-byte "61300020f09f8c8a" (hex)
• 18-byte "\\x61300020f09f8c8a" (hex with prefix \x, pgwire text format)
• 23-byte "a0\\000 \\360\\237\\214\\212" (escape) or 11-byte "a0\\000 \ud83c\udf0a" or 11-byte "a0\\000 🌊"

It is also easy to convert from these forms into a bytea in RisingWave today:

with t(headers) as (select '{
  "a": "YTAAIPCfjIo=",
  "b": "61300020f09f8c8a",
  "c": "\\x61300020f09f8c8a",
  "d0": "a0\\000 \\360\\237\\214\\212",
  "d1": "a0\\000 \ud83c\udf0a",
  "d2": "a0\\000 🌊"
}'::jsonb) select
  octet_length(headers ->> 'a'), decode(headers ->> 'a', 'base64'),
  octet_length(headers ->> 'b'), decode(headers ->> 'b', 'hex'),
  octet_length(headers ->> 'c'), (headers ->> 'c')::bytea,            -- this format can just use cast
  octet_length(headers ->> 'd0'), decode(headers ->> 'd0', 'escape'), -- this can both decode or cast
  octet_length(headers ->> 'd0'), (headers ->> 'd0')::bytea,
  octet_length(headers ->> 'd1'), decode(headers ->> 'd1', 'escape'),
  octet_length(headers ->> 'd2'), (headers ->> 'd2')::bytea
from t;

My thoughts:

• Yes map is popular in many systems
• jsonb can also work as kafka headers to include bytes before map is ready, but it is easier to stick to one encoding rather than exposing an option to choose.
• If we anticipate users would always decode to bytea first, base64 is the most compact.
• If we anticipate users would mostly include ascii string, escape is more human readable without decoding. And it can be decoded either using a function or a cast.
• If we anticipate users would like to inspect (but not pass as function arg, which requires decoding) hex without decoding, for example integers or raw-uuid or raw-ipv6, hex shall be chosen.

[xxchan]

we need another set of type mappings from protobuf to JSON

This is not a big deal. Proto3 has an official definition of how proto maps to JSON. But do note that int64 maps to string.

To elaborate this point a bit: it's actually a big deal, and another big drawback of JSON -- It lacks native support for a lot of types, and thus leaving there many choices of encoding to JSON and decoding from JSON, which means we need to provide a lot of config options.

The encoding problem include:

• int64: Actually JSON spec doesn't accept it.
  • Some JSON libraries (serde_json, jsonbb) allow it, but some don't
  • protobuf encodes to string. RisingWave sink encodes to number (so when it overflows, it will be undefined behavior, decided by the downstream system).
• bytes: As mentioned by xiangjin above, there are many possible ways.

Different cases include:

• For decoding from external JSON into RisingWave data(source)
  • We have e.g., timestamptz.handling.mode https://docs.risingwave.com/docs/dev/supported-sources-and-formats/#general-parameters-for-supported-formats
  • It seems we don't have a doc for type mapping
  • This case is relatively simple, because it converts to a stronger type. The goal is clear
• For encoding RisingWave data into external JSON (sink)
  • We also have timestamptz.handling.mode
  • Doc for type mapping https://docs.risingwave.com/docs/dev/create-sink-kafka/#data-type-mapping---risingwave-and-debezium-json Note that it differs from protobuf json encoding.
• For encoding external data into RisingWave JSON (source, this issue):
  • A new scenario. This case is a little special. It's like sink (both do encoding), but we might want different defaults. e.g., if for ease of handling in RisingWave, we might want to choose \\x style encoding for bytes. But if for ease of handling in downstream, \\x is not suitable, because it's PG-specific.
  • Besides, introducing options for this case will become awkward: We already have timestamptz.handling.mode for decoding JSON in source, what should we have for similar thing but encoding JSON in source?
• For decoding RisingWave JSON into external data (sink): another newly raised case: feat(sink): jsonb column to avro record/map/union type #16941

---

My opinion for the last question is that we don't provide encoding option for this case, and just choose our opinionated one.

Oh, it seems xiangjin already talked about this:

jsonb can also work as kafka headers to include bytes before map is ready, but it is easier to stick to one encoding rather than exposing an option to choose.

[github-actions]

This issue has been open for 60 days with no activity.

If you think it is still relevant today, and needs to be done in the near future, you can comment to update the status, or just manually remove the no-issue-activity label.

You can also confidently close this issue as not planned to keep our backlog clean.
Don't worry if you think the issue is still valuable to continue in the future.
It's searchable and can be reopened when it's time. 😄