Add Protobuf support for trino-kafka #14734

nevillelyh · 2022-10-24T21:16:10Z

Description

Add Protobuf support for Kafka

This is a contribution from the Starburst Kafka Connector

Non-technical explanation

Release notes

( ) This is not user-visible or docs only and no release notes are required.
( ) Release notes are required, please propose a release note for me.
(x) Release notes are required, with the following suggested text:

# Section
* Add Protobuf message encoding/decoding support for Kafka connector

nevillelyh · 2022-10-24T21:16:59Z

Code, tests & docs are ready. Still missing product tests, will add later.

pom.xml

nevillelyh · 2022-10-30T23:47:28Z

It's still failing suite-6-non-generic after I fixed the kafka-protobuf-provider dependency in test containers. Cannot reproduce locally. Possibly a caching issue?

presto-master       | java.lang.NoClassDefFoundError: io/confluent/kafka/schemaregistry/protobuf/ProtobufSchemaProvider
presto-master       | 	at io.trino.plugin.kafka.schema.confluent.ConfluentModule$LazyLoadedProtobufSchemaProvider.create(ConfluentModule.java:186)
presto-master       | 	at com.google.common.base.Suppliers$NonSerializableMemoizingSupplier.get(Suppliers.java:183)
presto-master       | 	at io.trino.plugin.kafka.schema.confluent.ConfluentModule$LazyLoadedProtobufSchemaProvider.parseSchema(ConfluentModule.java:181)
presto-master       | 	at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.parseSchema(CachedSchemaRegistryClient.java:197)
presto-master       | 	at io.trino.plugin.kafka.schema.confluent.ClassLoaderSafeSchemaRegistryClient.parseSchema(ClassLoaderSafeSchemaRegistryClient.java:49)

michaelpearce-gain · 2022-11-04T03:12:05Z

One thing looking at this i couldnt spot where the smarts is to auto deduce if to use proto or avro when using schema registry client, where a broker may have topics with both

Schema registry holds the schematype now and like wise supports avro, and proto, so just from the schema id or from the subject latest schema you can auto deduce if its avro or proto, thus easily allowing to switch based on this which deserializing method to use (avro vs proto) so you can have a broker with both protocol types in use, and by topic it can be auto deduced which one to use

SchemaRegistryClient src = <init>

ParsedSchema parsedSchema = src.getSchemaById(0);
parsedSchema.schemaType()

I could be blind and missed it in which case if you could point me to it.

nevillelyh · 2022-11-05T04:24:40Z

Tests finally green. @Praveen2112 PTAL when you get a chance? 🙏

plugin/trino-kafka/pom.xml

plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/KafkaConnectorModule.java

Praveen2112 · 2022-11-07T07:57:33Z

plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/protobuf/ProtobufUtils.java

+                                        protoType.getValueType(),
+                                        definedMessages))
+                                .build());
+                // Handle for underscores and name


Can we add a test for this case ?

What do we want to test here, that the descriptor is constructed correctly? That's covered by TestProtobuf{Encoder,Decoder} already. Commenting out anything except .setName(field.getName()) fails the tests.

EDIT: actually setName(field.getName) is already called before the if branch and is redundant.

Praveen2112

I think we could have a single commit where we could implement the Protobuf in the decoder module and use them in Kafka connector for now.

lib/trino-record-decoder/src/main/java/io/trino/decoder/protobuf/ProtobufColumnDecoder.java

nevillelyh · 2022-11-08T14:45:25Z

@Praveen2112 rebased into 2 commits, decoder module & kafka support.

Praveen2112

LGTM Minor comments

Praveen2112 · 2022-11-14T09:17:00Z

lib/trino-record-decoder/pom.xml

+            <groupId>com.squareup.wire</groupId>
+            <artifactId>wire-schema</artifactId>
+        </dependency>
+
        <dependency>


Should we exclude few resources for duplicate classfinder to be happy ?

Looks like Guava is the only one we can exclude? Kotlin is already pinned to make enforcer happy.
https://mvnrepository.com/artifact/com.squareup.wire/wire-schema/3.2.2

Speaking of, do we want a more recent version of this? They removed Guava dependency but bumped Kotlin, which should be backwards compatible.
https://mvnrepository.com/artifact/com.squareup.wire/wire-schema/4.4.3

We could try updating to the latest version. We need to confirm if schema-registry libraries have some conflict with the latest version.

...ecord-decoder/src/main/java/io/trino/decoder/protobuf/FixedSchemaDynamicMessageProvider.java

lib/trino-record-decoder/src/main/java/io/trino/decoder/protobuf/ProtobufColumnDecoder.java

Praveen2112 · 2022-11-14T09:32:19Z

lib/trino-record-decoder/src/main/java/io/trino/decoder/protobuf/ProtobufRowDecoder.java

+        return Optional.of(columnDecoders.entrySet().stream()
+                .collect(toImmutableMap(
+                        Map.Entry::getKey,
+                        entry -> entry.getValue().decodeField(dynamicMessageProvider.parseDynamicMessage(data)))));


Can we parse it once and use the DynamicMessage for all the required columns.

Good catch!

Did we fix this ?

Pretty sure I did but must've lost it during rebase. Just fixed.

lib/trino-record-decoder/src/main/java/io/trino/decoder/protobuf/ProtobufValueProvider.java

docs/src/main/sphinx/connector/kafka.rst

plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/encoder/protobuf/ProtobufRowEncoder.java

plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/schema/confluent/ConfluentModule.java

...va/io/trino/plugin/kafka/schema/confluent/ConfluentSchemaRegistryDynamicMessageProvider.java

Praveen2112 · 2022-11-15T07:14:33Z

Let me know once the changes are applied.

nevillelyh · 2022-11-15T07:35:37Z

@Praveen2112 hey yeah I addressed most except the question regarding excisions #14734 (comment)
Thoughts?

Praveen2112

Some of the changes were not applied.

Praveen2112 · 2022-11-15T06:56:07Z

lib/trino-record-decoder/pom.xml

+            <groupId>com.squareup.wire</groupId>
+            <artifactId>wire-schema</artifactId>
+        </dependency>
+
        <dependency>


We could try updating to the latest version. We need to confirm if schema-registry libraries have some conflict with the latest version.

...ecord-decoder/src/main/java/io/trino/decoder/protobuf/FixedSchemaDynamicMessageProvider.java

pom.xml

Praveen2112 · 2022-11-15T07:12:19Z

lib/trino-record-decoder/src/main/java/io/trino/decoder/protobuf/ProtobufValueProvider.java

+        return round(micros, MAX_SHORT_PRECISION - precision);
+    }
+
+    private static Descriptors.FieldDescriptor getFieldDescriptor(DynamicMessage message, String name)


lib/trino-record-decoder/src/main/java/io/trino/decoder/protobuf/ProtobufValueProvider.java

nevillelyh · 2022-11-15T16:51:29Z

@Praveen2112 localized duplicate-finder-maven-plugin settings, fixed a few more static imports, going to leave wire-schema version as is for this change. Should be good now.

Praveen2112 · 2022-11-23T11:34:57Z

Thanks for working on this.

ignitz · 2022-11-23T16:46:28Z

Holy shit guys. That's awesome. 🎉

Giovarco · 2022-11-30T11:53:18Z

Hello! I'm on version 403 and can't find a way to handle protobuf messages (as far as I tried) and this protobuf support is nice to have! I'd like to test this. Have you documented anywhere how to use this?

findepi · 2022-12-12T11:19:13Z

...n/trino-kafka/src/main/java/io/trino/plugin/kafka/encoder/protobuf/ProtobufSchemaParser.java

+        if (fieldDescriptor.getMessageType().getFullName().equals(TIMESTAMP_TYPE_NAME)) {
+            return createTimestampType(6);


Per https://developers.google.com/protocol-buffers/docs/reference/google.protobuf#google.protobuf.Timestamp

A Timestamp represents a point in time independent of any time zone or calendar, represented as seconds and fractions of seconds at nanosecond resolution in UTC Epoch time

I.e. it's an instant.
We don't have an instant type (#2273), so we represent point-in-time data as "timestamp with time zone".

Would you consider changing the mapping to reveal the point-in-time semantics of the data?

Also, the type is nano-second precision. Why do we map it to microsecond precision?

cc @mx123

These are direct ports from SEP, maybe @Praveen2112 remembers why?
Changing the semantic or precision might break back-compat, we might have to support both local/instant x all precisions combo?

Changing the semantic or precision might break back-compat,

Of course. Every change, including bug fixes, can break something.

we might have to support both local/instant x all precisions combo?

not sure why we would want that.

cla-bot bot added the cla-signed label Oct 24, 2022

github-actions bot added the docs label Oct 24, 2022

nevillelyh requested a review from Praveen2112 October 25, 2022 12:42

nevillelyh force-pushed the neville/kafka-protobuf branch from fb36d7f to eb77237 Compare October 26, 2022 23:49

nevillelyh commented Oct 26, 2022

View reviewed changes

pom.xml Show resolved Hide resolved

github-actions bot added the tests:hive label Oct 27, 2022

nevillelyh marked this pull request as ready for review October 27, 2022 04:28

nevillelyh mentioned this pull request Oct 27, 2022

Flaky Parquet tests by ParquetMemoryManagerRuntimeException in Hive connector #13633

Closed

nevillelyh force-pushed the neville/kafka-protobuf branch 7 times, most recently from da33362 to 424ef36 Compare October 30, 2022 21:13

nevillelyh force-pushed the neville/kafka-protobuf branch 2 times, most recently from 3595513 to 6f85b81 Compare November 2, 2022 16:24

nevillelyh force-pushed the neville/kafka-protobuf branch 3 times, most recently from 1e828f6 to 2649add Compare November 4, 2022 21:58

Praveen2112 reviewed Nov 7, 2022

View reviewed changes

nevillelyh force-pushed the neville/kafka-protobuf branch 2 times, most recently from d662a51 to 0914de4 Compare November 8, 2022 02:18

Praveen2112 reviewed Nov 8, 2022

View reviewed changes

lib/trino-record-decoder/src/main/java/io/trino/decoder/protobuf/ProtobufColumnDecoder.java Outdated Show resolved Hide resolved

nevillelyh force-pushed the neville/kafka-protobuf branch from 0914de4 to 9d54d97 Compare November 8, 2022 14:43

Praveen2112 approved these changes Nov 14, 2022

View reviewed changes

nevillelyh mentioned this pull request Nov 14, 2022

Add TINYINT and SMALLINT support for Kafka Protobuf encoder #15012

Open

nevillelyh force-pushed the neville/kafka-protobuf branch 2 times, most recently from 182cf78 to 1f92c36 Compare November 15, 2022 01:57

Praveen2112 reviewed Nov 15, 2022

View reviewed changes

nevillelyh force-pushed the neville/kafka-protobuf branch 3 times, most recently from 8e825c2 to 3db448d Compare November 15, 2022 16:48

nevillelyh force-pushed the neville/kafka-protobuf branch 2 times, most recently from 18729a3 to c46bb5a Compare November 23, 2022 01:24

nevillelyh added 2 commits November 22, 2022 20:46

Add Protobuf decoder

b798bde

Add Protobuf support for trino-kafka

5c40f44

nevillelyh force-pushed the neville/kafka-protobuf branch from c46bb5a to 5c40f44 Compare November 23, 2022 01:46

Praveen2112 merged commit 55f306b into trinodb:master Nov 23, 2022

github-actions bot added this to the 404 milestone Nov 23, 2022

nevillelyh deleted the neville/kafka-protobuf branch November 23, 2022 15:25

colebow mentioned this pull request Nov 28, 2022

Add Trino 405 release notes #15139

Merged

findepi reviewed Dec 12, 2022

View reviewed changes

hashhar mentioned this pull request May 15, 2023

Kafka confluent schema registry error on startup on v415 #17299

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add Protobuf support for trino-kafka #14734

Add Protobuf support for trino-kafka #14734

nevillelyh commented Oct 24, 2022 •

edited

Loading

nevillelyh commented Oct 24, 2022

nevillelyh commented Oct 30, 2022

michaelpearce-gain commented Nov 4, 2022 •

edited

Loading

nevillelyh commented Nov 5, 2022

Praveen2112 Nov 7, 2022

nevillelyh Nov 7, 2022 •

edited

Loading

Praveen2112 left a comment

nevillelyh commented Nov 8, 2022

Praveen2112 left a comment

Praveen2112 Nov 14, 2022

nevillelyh Nov 14, 2022

Praveen2112 Nov 15, 2022

Praveen2112 Nov 14, 2022

nevillelyh Nov 14, 2022

Praveen2112 Nov 21, 2022

nevillelyh Nov 23, 2022

Praveen2112 commented Nov 15, 2022

nevillelyh commented Nov 15, 2022

Praveen2112 left a comment

Praveen2112 Nov 15, 2022

Praveen2112 Nov 15, 2022

nevillelyh commented Nov 15, 2022 •

edited

Loading

Praveen2112 commented Nov 23, 2022

ignitz commented Nov 23, 2022

Giovarco commented Nov 30, 2022 •

edited

Loading

findepi Dec 12, 2022

findepi Dec 12, 2022

nevillelyh Dec 13, 2022

findepi Dec 14, 2022

		if (fieldDescriptor.getMessageType().getFullName().equals(TIMESTAMP_TYPE_NAME)) {
		return createTimestampType(6);

Add Protobuf support for trino-kafka #14734

Add Protobuf support for trino-kafka #14734

Conversation

nevillelyh commented Oct 24, 2022 • edited Loading

Description

Non-technical explanation

Release notes

nevillelyh commented Oct 24, 2022

nevillelyh commented Oct 30, 2022

michaelpearce-gain commented Nov 4, 2022 • edited Loading

nevillelyh commented Nov 5, 2022

Choose a reason for hiding this comment

nevillelyh Nov 7, 2022 • edited Loading

Choose a reason for hiding this comment

Praveen2112 left a comment

Choose a reason for hiding this comment

nevillelyh commented Nov 8, 2022

Praveen2112 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Praveen2112 commented Nov 15, 2022

nevillelyh commented Nov 15, 2022

Praveen2112 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nevillelyh commented Nov 15, 2022 • edited Loading

Praveen2112 commented Nov 23, 2022

ignitz commented Nov 23, 2022

Giovarco commented Nov 30, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nevillelyh commented Oct 24, 2022 •

edited

Loading

michaelpearce-gain commented Nov 4, 2022 •

edited

Loading

nevillelyh Nov 7, 2022 •

edited

Loading

nevillelyh commented Nov 15, 2022 •

edited

Loading

Giovarco commented Nov 30, 2022 •

edited

Loading