Decompress Kinesis message if needed #6442

guyco33 · 2020-12-26T19:46:17Z

Producer might push compressed messages to save network bandwidth and consumers should transparently uncompress the message before decoding it

Praveen2112 · 2020-12-28T05:41:49Z

presto-kinesis/src/main/java/io/prestosql/plugin/kinesis/KinesisRecordSet.java

+                return data;
+            }
+            int magic = data[0] & 0xff | ((data[1] << 8) & 0xff00);
+            if (magic == GZIPInputStream.GZIP_MAGIC) {


Does this support only GZIP or ?

How about wrapping this logic in a RowDecoder which decompresses the row and passes it to the underlying decoder.

My idea was to limit it now to Kinesis, since other consumer libraries like Kafka and Redis might have an option to uncompress it.

Praveen2112

Minor nits

Praveen2112 · 2021-01-04T07:08:08Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

@@ -220,7 +232,7 @@ public boolean advanceNextPosition()
        {
            if (shardIterator == null && getRecordsRequest == null) {
                getIterator(); // first shard iterator
-                log.debug("Starting read.  Retrieved first shard iterator from AWS Kinesis.");
+                log.debug("(%s:%s) Starting read.  Retrieved first shard iterator from AWS Kinesis.", split.getStreamName(), split.getShardId());


nit : Can we capture these changes as a separate commit ?

Praveen2112 · 2021-01-04T07:09:16Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

-            Optional<Map<DecoderColumnHandle, FieldValueProvider>> decodedValue = messageDecoder.decodeRow(messageData);
+            Optional<Map<DecoderColumnHandle, FieldValueProvider>> decodedValue;
+            if (isGZipped(messageData)) {
+                try {


How about using try-with-resources ?

Praveen2112 · 2021-01-04T07:10:07Z

plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestRecordAccess.java

@@ -97,7 +100,21 @@ private void createDummyMessages(String streamName, int count)
        mockClient.putRecords(putRecordsRequest);
    }

-    private void createJsonMessages(String streamName, int count, int idStart)
+    private byte[] compressMessage(byte[] data)


nit : can we move it below createJsonMessages

Praveen2112 · 2021-01-04T07:12:20Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

+            return false;
+        }
+        int magic = data[0] & 0xff | ((data[1] << 8) & 0xff00);
+        return magic == GZIPInputStream.GZIP_MAGIC;


static import ?

kokosing

Is it is sad that we don't have end-to-end tests for Kinesis. Please create ticket for that.

Can kinesis message be compressed with something else?

kokosing · 2021-01-07T08:38:16Z

plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestRecordAccess.java

+    private byte[] compressMessage(byte[] data)
+    {
+        try {
+            ByteArrayOutputStream byteOS = new ByteArrayOutputStream();


use try-with-resources for byteOS and gzipOS

kokosing · 2021-01-07T08:39:19Z

plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestRecordAccess.java

@@ -97,7 +100,7 @@ private void createDummyMessages(String streamName, int count)
        mockClient.putRecords(putRecordsRequest);
    }

-    private void createJsonMessages(String streamName, int count, int idStart)
+    private void createJsonMessages(String streamName, int count, int idStart, boolean compress)


can we have utility methods below the usage (below test)? Please reorder methods in separate commit.

kokosing · 2021-01-07T08:40:02Z

plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestRecordAccess.java

@@ -160,7 +182,8 @@ public void testStreamHasData()
    public void testJsonStream()
    {
        // Simple case: add a few specific items, query object and internal fields:
-        createJsonMessages(jsonStreamName, 4, 100);
+        createJsonMessages(jsonStreamName, 2, 100, false);
+        createJsonMessages(jsonStreamName, 2, 102, true);


Let's make test parametric. You can use io.trino.testing.DataProviders#trueFalse.

kokosing · 2021-01-07T08:40:28Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

+            Optional<Map<DecoderColumnHandle, FieldValueProvider>> decodedValue;
+            if (isGZipped(messageData)) {
+                try (
+                        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(messageData);


use try-with-resources for streams.

kokosing · 2021-01-07T08:40:42Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

+                    decodedValue = messageDecoder.decodeRow(gZIPInputStream.readAllBytes());
+                }
+                catch (IOException e) {
+                    throw new RuntimeException(e);


PrestoException or better UncheckedIOException

kokosing · 2021-01-07T08:40:55Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

-            Optional<Map<DecoderColumnHandle, FieldValueProvider>> decodedValue = messageDecoder.decodeRow(messageData);
+            Optional<Map<DecoderColumnHandle, FieldValueProvider>> decodedValue;
+            if (isGZipped(messageData)) {
+                try (


Please extract a method

guyco33 · 2021-01-07T08:56:39Z

Can kinesis message be compressed with something else?

Compression is done by the message producer so that actually the message can by anything

kokosing

Please squash first and last commits together.

kokosing · 2021-01-14T08:29:02Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

-            log.info("Closing cursor - read complete.  Total read: %d batches %d messages, processed: %d messages and %d bytes.",
-                    batchesRead, messagesRead, totalMessages, totalBytes);
+            log.info("(%s:%s) Closing cursor - read complete.  Total read: %d batches %d messages, processed: %d messages and %d bytes.",
+                    split.getStreamName(), split.getShardId(), batchesRead, messagesRead, totalMessages, totalBytes);


please put each argument in separate line

kokosing · 2021-01-14T08:29:53Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisTableHandle.java

+    /**
+     * Compression codec to use for decompressing message. Can be null.
+     */
+    private final String compressionCodec;


Please enum.

kokosing · 2021-01-14T08:30:03Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisTableHandle.java

@@ -47,17 +49,24 @@

    private final String messageDataFormat;

+    /**
+     * Compression codec to use for decompressing message. Can be null.


This is obvious, I would remove this comment.

kokosing · 2021-01-14T08:30:20Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisTableHandle.java

@@ -84,6 +93,13 @@ public String getMessageDataFormat()
        return messageDataFormat;
    }

+    @Nullable


Please return Optional

kokosing · 2021-01-16T19:54:34Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisCompressionCodec.java

+
+public enum KinesisCompressionCodec
+{
+    GZIP


add a coma at the end GZIP,

kokosing · 2021-01-16T19:55:33Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

@@ -171,6 +179,15 @@ public RecordCursor cursor()
        return new KinesisRecordCursor();
    }

+    public static boolean isGZipped(byte[] data)


make it private, and move below the usage

kokosing · 2021-01-16T19:56:37Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

+                return messageDecoder.decodeRow(messageData);
+            }
+            if (split.getCompressionCodec().get().equals(KinesisCompressionCodec.GZIP)) {
+                if (isGZipped(messageData)) {


getCompressionCodec() == GZIP then I guess we should fail if !isGZipped(messageData) as this is unexpected.

Maybe we could have KinesisCompressionCodec.AUTOMATIC, wdyt?

I guess we should fail if !isGZipped(messageData) as this is unexpected.

It's expected for cases that both compressed and uncompressed messages exist in stream at the same time.
For example: during a deployment rollout, for compression based on message size or for cases that some producers didn't implement the compression yet.

Maybe we could have KinesisCompressionCodec.AUTOMATIC

That would be the ultimate option. I would expect it to decompress any codec, but it's not needed for now, so I would leave it for further implementation

It's expected for cases that both compressed and uncompressed messages exist in stream at the same time.

This is the case I would like to be handled by KinesisCompressionCodec.AUTOMATIC

kokosing · 2021-01-16T20:00:28Z

plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestRecordAccess.java

@@ -119,6 +127,20 @@ private void createJsonMessages(String streamName, int count, int idStart)
        mockClient.putRecords(putRecordsRequest);
    }

+    private byte[] compressMessage(byte[] data)


kokosing · 2021-01-16T20:00:58Z

plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestRecordAccess.java

@@ -119,6 +127,20 @@ private void createJsonMessages(String streamName, int count, int idStart)
        mockClient.putRecords(putRecordsRequest);
    }

+    private byte[] compressMessage(byte[] data)
+    {
+        try (


I would not not wrap line here.

Use:

try (ByteArrayOutputStream byteOS = new ByteArrayOutputStream()) {

Same above.

plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestRecordAccess.java

kokosing · 2021-01-19T09:17:47Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

+                return messageDecoder.decodeRow(messageData);
+            }
+            if (split.getCompressionCodec().get() == GZIP || split.getCompressionCodec().get() == AUTOMATIC) {
+                if (isGZipped(messageData)) {


Please invert the condition, so there will be less code nested, and corner case will be handled first.

The idea was to skip testing the GZIP header if it's not required

kokosing · 2021-01-19T09:20:16Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

+            if (!split.getCompressionCodec().isPresent()) {
+                return messageDecoder.decodeRow(messageData);
+            }
+            if (split.getCompressionCodec().get() == GZIP || split.getCompressionCodec().get() == AUTOMATIC) {


Extract variable for split.getCompressionCodec().get(). Or even introduce a method to enum canUseGzip() which would return true for AUTOMATIC and GZIP

kokosing · 2021-01-19T09:23:16Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

+                    throw new TrinoException(GENERIC_INTERNAL_ERROR, format("Failed to decompress %s message", split.getCompressionCodec().get()));
+                }
+            }
+            if (split.getCompressionCodec().get() == AUTOMATIC) {


We could add UNCOMPRESSED (default value) then we could merge this block with the one at top of this method.

kokosing

% minor comments

kokosing · 2021-01-19T13:38:36Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

+        {
+            KinesisCompressionCodec kinesisCompressionCodec = split.getCompressionCodec();
+            if (isGZipped(messageData)) {
+                if (canUseGzip(kinesisCompressionCodec)) {


please invert the condition

kokosing · 2021-01-19T13:39:30Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisStreamFieldGroup.java

@@ -42,6 +47,12 @@ public String getDataFormat()
        return dataFormat;
    }

+    @JsonProperty


I am not sure if this going to work. You would need to add round trip json serialization test to verify this. Maybe simply remove this annotation?

I've just remove the annotation

kokosing · 2021-01-19T13:41:26Z

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java

@@ -290,6 +300,28 @@ private void getKinesisRecords()
            messagesRead += kinesisRecords.size();
        }

+        private Optional<Map<DecoderColumnHandle, FieldValueProvider>> decodeMessage(byte[] messageData)


please move this method under the usage

kokosing · 2021-01-19T20:08:35Z

Automation failures:

2021-01-19T17:14:25.6499061Z [ERROR] Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.8.0:compile (default-compile) on project trino-kinesis: Compilation failure: Compilation failure: 
2021-01-19T17:14:25.6501167Z [ERROR] /home/runner/work/trino/trino/plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java:[388,76] [FormatString] format specifier '%c' is not compatible with the given flag(s):  
2021-01-19T17:14:25.6502536Z [ERROR]     (see https://errorprone.info/bugpattern/FormatString)
2021-01-19T17:14:25.6504263Z [ERROR] /home/runner/work/trino/trino/plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java:[402,68] [FormatString] illegal format conversion: 'java.lang.Object' cannot be formatted using '%o'
2021-01-19T17:14:25.6506332Z [ERROR]     (see https://errorprone.info/bugpattern/FormatString)
2021-01-19T17:14:25.6507263Z [ERROR] -> [Help 1]
2021-01-19T17:14:25.6507682Z [ERROR]

kokosing · 2021-01-21T10:41:52Z

Merged, thanks!

cla-bot bot added the cla-signed label Dec 26, 2020

Praveen2112 reviewed Dec 28, 2020

View reviewed changes

guyco33 force-pushed the uncompress_kinesis_message branch 4 times, most recently from cf92bd3 to 50776ad Compare January 3, 2021 14:15

guyco33 changed the title ~~Uncompress Kinesis message if needed~~ Decompress Kinesis message if needed Jan 3, 2021

Praveen2112 reviewed Jan 4, 2021

View reviewed changes

Praveen2112 requested a review from kokosing January 4, 2021 07:13

guyco33 force-pushed the uncompress_kinesis_message branch from 50776ad to edbab1a Compare January 4, 2021 08:08

kokosing reviewed Jan 7, 2021

View reviewed changes

kokosing reviewed Jan 14, 2021

View reviewed changes

guyco33 force-pushed the uncompress_kinesis_message branch from 8cd54af to 122cfd3 Compare January 16, 2021 12:09

kokosing reviewed Jan 16, 2021

View reviewed changes

plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestRecordAccess.java Show resolved Hide resolved

guyco33 force-pushed the uncompress_kinesis_message branch 2 times, most recently from a051ccf to f40a1f4 Compare January 19, 2021 09:11

kokosing reviewed Jan 19, 2021

View reviewed changes

guyco33 force-pushed the uncompress_kinesis_message branch from f40a1f4 to c8ac6f4 Compare January 19, 2021 12:59

kokosing approved these changes Jan 19, 2021

View reviewed changes

guyco33 force-pushed the uncompress_kinesis_message branch from c8ac6f4 to 58098e2 Compare January 19, 2021 16:25

guyco33 added 2 commits January 20, 2021 02:21

Decompress Kinesis message if needed

82f9148

Add stream and shard names to log

3223a03

guyco33 force-pushed the uncompress_kinesis_message branch from 58098e2 to 3223a03 Compare January 20, 2021 00:22

kokosing merged commit 5f68daf into trinodb:master Jan 21, 2021

kokosing added this to the 352 milestone Jan 21, 2021

kokosing mentioned this pull request Jan 21, 2021

Release notes for 352 #6502

Closed

10 tasks

Decompress Kinesis message if needed #6442

Decompress Kinesis message if needed #6442

Conversation

guyco33 commented Dec 26, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

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

kokosing 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

kokosing Jan 7, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guyco33 commented Jan 7, 2021 • edited

kokosing 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kokosing Jan 16, 2021 • edited

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

kokosing 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

kokosing commented Jan 19, 2021

kokosing commented Jan 21, 2021

kokosing Jan 7, 2021 •

edited

guyco33 commented Jan 7, 2021 •

edited

kokosing Jan 16, 2021 •

edited