Add Kafka Json encoder #4477

charlesjmorgan · 2020-07-16T19:49:23Z

Add JsonRowEncoder and JsonRowEncoderFactory
Add test case in io.prestosql.plugin.kafka.TestKafkaIntegrationSmokeTest#testRoundTripAllFormats

Might it be better to have separate formatters for the date/time types rather than methods in JsonRowEncoder?

closes #3980

presto-kafka/src/main/java/io/prestosql/plugin/kafka/encoder/json/JsonRowEncoder.java

presto-kafka/src/test/java/io/prestosql/plugin/kafka/TestKafkaIntegrationSmokeTest.java

presto-kafka/src/test/java/io/prestosql/plugin/kafka/encoder/json/TestJsonDateTime.java

presto-kafka/src/test/java/io/prestosql/plugin/kafka/TestKafkaIntegrationSmokeTest.java

presto-kafka/src/test/resources/write_test/custom_date_time.json

presto-kafka/src/main/java/io/prestosql/plugin/kafka/encoder/json/JsonRowEncoder.java

presto-kafka/src/test/java/io/prestosql/plugin/kafka/TestKafkaIntegrationSmokeTest.java

charlesjmorgan · 2020-07-17T21:26:19Z

Implemented all the changes in the resolved review comments, will make it through the rest on Monday. I changed the date/time formatting to precompile a list of functions to format the values, lmk what you think of this.

losipiuk

Thanks. I left some comments but is not full review. Structurally it looks much better. Timestamp handling is hard as I honestly am not sure what is the expected behaviour here.

presto-kafka/src/main/java/io/prestosql/plugin/kafka/encoder/json/JsonRowEncoder.java

...kafka/src/main/java/io/prestosql/plugin/kafka/encoder/json/format/DefaultFormatFunction.java

presto-kafka/src/main/java/io/prestosql/plugin/kafka/encoder/json/JsonRowEncoder.java

presto-kafka/src/test/java/io/prestosql/plugin/kafka/TestKafkaIntegrationSmokeTest.java

presto-kafka/src/test/java/io/prestosql/plugin/kafka/encoder/json/TestJsonDateTime.java

...c/main/java/io/prestosql/plugin/kafka/encoder/json/format/CustomDateTimeFormatFunctions.java

presto-kafka/src/test/java/io/prestosql/plugin/kafka/TestKafkaIntegrationSmokeTest.java

...c/main/java/io/prestosql/plugin/kafka/encoder/json/format/CustomDateTimeFormatFunctions.java

charlesjmorgan · 2020-07-23T19:45:28Z

Changes:

Restructure Json Date/Time formatters according to the suggestion made by @losipiuk
Format functions return a parameterized Function
Add (better) unit tests for format functions
Restructure Json Date/Time round trip tests in TestKafkaIntegrationSmokeTest for readability

charlesjmorgan · 2020-07-27T18:00:34Z

I think I made the suggested legacy timestamp changes. The legacy format functions definitely need a close look at tho

aalbu

Legacy semantics imply that that temporal types are interpreted in the session’s time zone. See for example the Javadoc on TimestampType. It seems to me that throughout these changes, the UTC timezone is used instead.

...o-kafka/src/main/java/io/prestosql/plugin/kafka/encoder/json/format/JsonFormatFunctions.java

...c/main/java/io/prestosql/plugin/kafka/encoder/json/format/CustomDateTimeFormatFunctions.java

...o-kafka/src/main/java/io/prestosql/plugin/kafka/encoder/json/format/JsonFormatFunctions.java

aalbu

Mostly stylistic suggestions.

aalbu · 2020-07-28T21:07:07Z

presto-kafka/src/main/java/io/prestosql/plugin/kafka/encoder/json/JsonRowEncoder.java

+        if (type == DATE) {
+            return JsonFormatFunction.builder().setFormatDateFunc(getDateFormatFunction(dataFormat, formatHint)).build();
+        }
+        else if (type == TIME) {


You actually don't need all the elses.

aalbu · 2020-07-29T12:04:07Z

...c/main/java/io/prestosql/plugin/kafka/encoder/json/format/CustomDateTimeFormatFunctions.java

+    public static Function<Long, String> formatDateFunc(String formatHint)
+    {
+        try {
+            DateTimeFormatter formatter = DateTimeFormat.forPattern(formatHint).withLocale(Locale.ENGLISH).withZoneUTC();


You can use a static import (below too).

aalbu · 2020-07-29T12:08:25Z

...c/main/java/io/prestosql/plugin/kafka/encoder/json/format/CustomDateTimeFormatFunctions.java

+    {
+        try {
+            DateTimeFormatter formatter = DateTimeFormat.forPattern(formatHint).withLocale(Locale.ENGLISH).withZoneUTC();
+            return millis -> (new DateTime(millis, DateTimeZone.UTC)).toString(formatter);


Let's be consistent in the way we format. Either new DateTime(...).toString(fromatter) as here, or formatter.print(new DateTime(...)) as below.

aalbu · 2020-07-29T12:41:31Z

...to-kafka/src/main/java/io/prestosql/plugin/kafka/encoder/json/format/JsonFormatFunction.java

+
+    public static class Builder
+    {
+        private Function<Long, String> formatDateFunc = (ignored) -> { throw new RuntimeException("unsupported argument type"); };


You could define

private static final Function<Long, String> UNIMPLEMENTED = ignored -> { throw new RuntimeException("unsupported argument type"); };

and reference that:

private Function<Long, String> formatDateFunc = UNIMPLEMENTED; private Function<Long, String> formatTimeFunc = UNIMPLEMENTED; ...

aalbu · 2020-07-29T12:47:19Z

...o-kafka/src/main/java/io/prestosql/plugin/kafka/encoder/json/format/JsonFormatFunctions.java

+
+    public static long daysToEpochMillis(long value)
+    {
+        return DAYS.toMillis(value);


This is subjective, but you could just as well inline. I think DAYS.toMillis(value) reads pretty well.

aalbu · 2020-07-29T12:50:31Z

...src/main/java/io/prestosql/plugin/kafka/encoder/json/format/MillisecondsFormatFunctions.java

+
+    public static Function<Long, String> formatMillisWithTZFunc()
+    {
+        return encodedMillisWithTZ -> String.valueOf(unpackMillisUtc(encodedMillisWithTZ));


Perhaps we should not support this scenario. This does not seem a meaningful return value - without the tz info, the milliseconds don't make much sense here, right?

Sorry, that was wrong. This format implies millis since epoch.

aalbu · 2020-07-29T12:55:47Z

...afka/src/main/java/io/prestosql/plugin/kafka/encoder/json/format/RFC2822FormatFunctions.java

+                        unpackZoneKey(encodedMillisWithTZ).getZoneId()));
+    }
+
+    private static LocalDateTime localDateTimeOfEpochMillis(long epochMillis)


Perhaps this would warrant being public in JsonFormatFunctions. It's being used in ISO8601FormatFunctions, as well.

aalbu · 2020-07-29T12:56:39Z

...afka/src/main/java/io/prestosql/plugin/kafka/encoder/json/format/SecondsFormatFunctions.java

+        return millis -> String.valueOf(millisToSeconds(millis));
+    }
+
+    public static Function<Long, String> formatSecondsWithTZFunc()


Another situation in which we're throwing away tz info.

This is like unix time(), it's fine.

findepi

skimming.

i have several code style-level comments whch are generally applicable to other place as well.

i have some comemnts abut timestamp formatting semantics.
lets talk about them more.

findepi · 2020-07-29T14:18:12Z

presto-kafka/src/main/java/io/prestosql/plugin/kafka/encoder/json/JsonRowEncoder.java

+            checkArgument(isSupportedType(columnHandle.getType()), "Unsupported column type '%s' for column '%s'", columnHandle.getType(), columnHandle.getName());
+
+            if (isDateTimeType(columnHandle.getType())) {
+                checkArgument(columnHandle.getDataFormat() != null, "Unsupported or no dataFormat '%s' defined for temporal column '%s'", columnHandle.getDataFormat(), columnHandle.getName());


Suggested change

checkArgument(columnHandle.getDataFormat() != null, "Unsupported or no dataFormat '%s' defined for temporal column '%s'", columnHandle.getDataFormat(), columnHandle.getName());

checkArgument(columnHandle.getDataFormat() != null, "No dataFormat defined for temporal column '%s'", columnHandle.getName());

findepi · 2020-07-29T14:19:25Z

presto-kafka/src/main/java/io/prestosql/plugin/kafka/encoder/json/JsonRowEncoder.java

+        else if (type == TIME) {
+            return JsonFormatFunction.builder().setFormatTimeFunc(getTimeFormatFunction(dataFormat, formatHint)).build();
+        }
+        else if (type == TIME_WITH_TIME_ZONE) {


redundant else (https://github.com/prestosql/presto/#additional-ide-configuration)

findepi · 2020-07-29T14:20:20Z

...c/main/java/io/prestosql/plugin/kafka/encoder/json/format/CustomDateTimeFormatFunctions.java

+{
+    private CustomDateTimeFormatFunctions() {}
+
+    public static Function<Long, String> formatDateFunc(String formatHint)


avoid abbreviations

func -> function

findepi · 2020-07-29T14:21:52Z

...c/main/java/io/prestosql/plugin/kafka/encoder/json/format/CustomDateTimeFormatFunctions.java

+    public static Function<Long, String> formatDateFunc(String formatHint)
+    {
+        try {
+            DateTimeFormatter formatter = DateTimeFormat.forPattern(formatHint).withLocale(Locale.ENGLISH).withZoneUTC();


is the format hint documented to be Joda Time?

findepi · 2020-07-29T14:25:41Z

...c/main/java/io/prestosql/plugin/kafka/encoder/json/format/CustomDateTimeFormatFunctions.java

+            DateTimeFormatter formatter = DateTimeFormat.forPattern(formatHint).withLocale(Locale.ENGLISH).withZoneUTC();
+            return millis -> (new DateTime(millis, DateTimeZone.UTC)).toString(formatter);
+        }
+        catch (IllegalArgumentException e) {


Code is correct but I would find it more readable if the lambda was outside of try block...

findepi · 2020-07-29T14:28:48Z

...c/main/java/io/prestosql/plugin/kafka/encoder/json/format/CustomDateTimeFormatFunctions.java

+    {
+        try {
+            DateTimeFormatter formatter = DateTimeFormat.forPattern(formatHint).withLocale(Locale.ENGLISH).withZoneUTC();
+            return millis -> (new DateTime(millis, DateTimeZone.UTC)).toString(formatter);


I think it would be better to supplement the session zone here.
This would make it more similar to the non-legacy case, when we implement it.

Ie
INSERT ,.. TIMESTAMP 'some_date_time'
would be some_date_time session_zone in legacy
and some_date_time in non-legacy

and if the format hint does not contain the zone,
they would actually be the same.

findepi · 2020-07-29T14:29:25Z

...c/main/java/io/prestosql/plugin/kafka/encoder/json/format/CustomDateTimeFormatFunctions.java

+    {
+        try {
+            DateTimeFormatter formatter = DateTimeFormat.forPattern(formatHint).withLocale(Locale.ENGLISH).withZoneUTC();
+            return encodedMillisWithTZ -> formatter.print(new DateTime(


encodedMillisWithTZ -> value

findepi · 2020-07-29T14:29:33Z

...c/main/java/io/prestosql/plugin/kafka/encoder/json/format/CustomDateTimeFormatFunctions.java

+                    DateTimeZone.forID(unpackZoneKey(encodedMillisWithTZ).getId())));
+        }
+        catch (IllegalArgumentException e) {
+            throw new IllegalArgumentException(format("Invalid joda pattern '%s' passed as format hint", formatHint), e);


joda -> Joda Time

findepi · 2020-07-29T14:30:27Z

...to-kafka/src/main/java/io/prestosql/plugin/kafka/encoder/json/format/JsonFormatFunction.java

+        return formatTimeFunc.apply(value);
+    }
+
+    public String formatTimeWithTZ(long value)


withTZ -> withTimeZone

charlesjmorgan · 2020-07-30T17:12:32Z

removed temporal column support from this pr, will open another pr for that

aalbu

LGTM

findepi · 2020-07-31T06:24:04Z

Merged, thanks!

cla-bot bot added the cla-signed label Jul 16, 2020

charlesjmorgan requested review from findepi, losipiuk and aalbu July 16, 2020 19:49

charlesjmorgan force-pushed the kafka-json-encoder branch from 54fb472 to 355841e Compare July 16, 2020 20:04

aalbu reviewed Jul 17, 2020

View reviewed changes

losipiuk reviewed Jul 17, 2020

View reviewed changes

charlesjmorgan force-pushed the kafka-json-encoder branch from 163919d to b3c8fd0 Compare July 21, 2020 14:30

losipiuk reviewed Jul 21, 2020

View reviewed changes

presto-kafka/src/test/java/io/prestosql/plugin/kafka/encoder/json/TestJsonDateTime.java Outdated Show resolved Hide resolved

aalbu reviewed Jul 22, 2020

View reviewed changes

This was referenced Jul 22, 2020

Implement base functionality for kafka connector inserts #4287

Merged

Add write support for kafka connector #4230

Closed

charlesjmorgan force-pushed the kafka-json-encoder branch from b3c8fd0 to 3ddeb43 Compare July 23, 2020 19:39

charlesjmorgan requested review from losipiuk and aalbu July 23, 2020 20:40

charlesjmorgan force-pushed the kafka-json-encoder branch 3 times, most recently from 955c7f6 to 30b4b63 Compare July 27, 2020 17:59

charlesjmorgan force-pushed the kafka-json-encoder branch 2 times, most recently from fdcb87f to 33f8a82 Compare July 27, 2020 21:03

aalbu reviewed Jul 28, 2020

View reviewed changes

charlesjmorgan force-pushed the kafka-json-encoder branch from 33f8a82 to 34cb597 Compare July 28, 2020 19:19

charlesjmorgan requested a review from aalbu July 28, 2020 19:19

charlesjmorgan force-pushed the kafka-json-encoder branch from 34cb597 to a900543 Compare July 28, 2020 20:07

aalbu reviewed Jul 29, 2020

View reviewed changes

findepi reviewed Jul 29, 2020

View reviewed changes

Add Kafka Json encoder

99ad68e

charlesjmorgan force-pushed the kafka-json-encoder branch from a900543 to 99ad68e Compare July 30, 2020 17:11

charlesjmorgan requested review from findepi and aalbu July 30, 2020 17:12

aalbu reviewed Jul 30, 2020

View reviewed changes

findepi approved these changes Jul 31, 2020

View reviewed changes

findepi merged commit 85204c1 into trinodb:master Jul 31, 2020

findepi added this to the 340 milestone Jul 31, 2020

findepi mentioned this pull request Jul 31, 2020

Release notes for 340 #4527

Closed

8 tasks

charlesjmorgan deleted the kafka-json-encoder branch June 7, 2021 18:40

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add Kafka Json encoder #4477

Add Kafka Json encoder #4477

charlesjmorgan commented Jul 16, 2020 •

edited

Loading

charlesjmorgan commented Jul 17, 2020

losipiuk left a comment

charlesjmorgan commented Jul 23, 2020 •

edited

Loading

charlesjmorgan commented Jul 27, 2020 •

edited

Loading

aalbu left a comment

aalbu left a comment

aalbu Jul 28, 2020

aalbu Jul 29, 2020

aalbu Jul 29, 2020

aalbu Jul 29, 2020

aalbu Jul 29, 2020

aalbu Jul 29, 2020

aalbu Jul 29, 2020

aalbu Jul 29, 2020

aalbu Jul 29, 2020

findepi Jul 30, 2020

findepi left a comment

findepi Jul 29, 2020

findepi Jul 29, 2020

findepi Jul 29, 2020

findepi Jul 29, 2020

findepi Jul 29, 2020

findepi Jul 29, 2020

findepi Jul 29, 2020

findepi Jul 29, 2020

findepi Jul 29, 2020

charlesjmorgan commented Jul 30, 2020

aalbu left a comment

findepi commented Jul 31, 2020

	checkArgument(columnHandle.getDataFormat() != null, "Unsupported or no dataFormat '%s' defined for temporal column '%s'", columnHandle.getDataFormat(), columnHandle.getName());
	checkArgument(columnHandle.getDataFormat() != null, "No dataFormat defined for temporal column '%s'", columnHandle.getName());

Add Kafka Json encoder #4477

Add Kafka Json encoder #4477

Conversation

charlesjmorgan commented Jul 16, 2020 • edited Loading

charlesjmorgan commented Jul 17, 2020

losipiuk left a comment

Choose a reason for hiding this comment

charlesjmorgan commented Jul 23, 2020 • edited Loading

charlesjmorgan commented Jul 27, 2020 • edited Loading

aalbu left a comment

Choose a reason for hiding this comment

aalbu 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

findepi 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

charlesjmorgan commented Jul 30, 2020

aalbu left a comment

Choose a reason for hiding this comment

findepi commented Jul 31, 2020

charlesjmorgan commented Jul 16, 2020 •

edited

Loading

charlesjmorgan commented Jul 23, 2020 •

edited

Loading

charlesjmorgan commented Jul 27, 2020 •

edited

Loading