SNOW-938038: Support AVRO Logical Types #722

sfc-gh-tzhang · 2023-10-11T23:33:46Z

There're 10 AVRO logical types listed in https://avro.apache.org/docs/1.11.0/spec.html#Logical+Types and we're able to support 4 of them, the mapping is as below:

date -> DATE
time-mills -> TIME(6)
timestamp-mills -> TIMESTAMP_NTZ(6)
decimal -> VARCHAR (We can't do NUMBER because we could have precision bigger than 36)

We can't do for the rest of 6 types because that's not supported by ConnectSchema, see code for more detail. We need to find another way to support other logical types or any sources (like Debezium).

sfc-gh-tzhang · 2023-10-11T23:36:11Z

src/main/java/com/snowflake/kafka/connector/records/RecordService.java

@@ -87,7 +87,7 @@ public class RecordService {
          });

  public static final ThreadLocal<SimpleDateFormat> TIME_FORMAT =
-      ThreadLocal.withInitial(() -> new SimpleDateFormat("HH:mm:ss.SSSZ"));
+      ThreadLocal.withInitial(() -> new SimpleDateFormat("HH:mm:ss.SSSXXX"));


This needs to be updated for Snowpipe Streaming but I don't see a good way to distinguish Snowpipe VS Snowpipe Streaming in RecordService, happy to hear any suggestions

simplest way to do it would be in TopicPartitionChannel ctor. since TopicPartitionChannel is only used in Snowpipe Streaming, just set a private field in RecordService to true. this will not change anything in snowpipe. Something like

public void isIngestionMethodSnowpipeStreaming(final boolean isSnowpipeStreaming) { this.isSnowpipeStreaming = isSnowpipeStreaming; }

I thought about that, but convertToJson is a static function :)

Basically we need to add a boolean to convertToJson and update all the references, I'm debating on whether this is really needed. The difference between Z and XXX is that there will be a colon with XXX in the timezone, for example, Z = 02:24:00.000+0000 and XXX = 02:24:00.000+00:00. In fact, it will fail if you try to insert 02:24:00.000+0000 to a TIME column so this is technically a bug. I'm not sure why there is no complains from customer, probably because not many people is using logical types.

out of scope of this PR, but RecordService does have a lot of overlap between snowpipe and streaming, might be useful to refactor or split it into two files for simplicity

Basically we need to add a boolean to convertToJson and update all the references, I'm debating on whether this is really needed. The difference between Z and XXX is that there will be a colon with XXX in the timezone, for example, Z = 02:24:00.000+0000 and XXX = 02:24:00.000+00:00. In fact, it will fail if you try to insert 02:24:00.000+0000 to a TIME column so this is technically a bug. I'm not sure why there is no complains from customer, probably because not many people is using logical types.

Thanks for detailed explanation.
I would be careful to make changes in snowpipe code.
Most folks havent complained because we dont have schematization in snowpipe. they have base table and then more tables on top of that. This change will break their downstream pipeline right?

You might have to take a giant step and add another arg in convertToJson :(
@sfc-gh-rcheng 's suggestion is the only way forward.

Added as an argument, PTAL

sfc-gh-tzhang · 2023-10-18T23:32:07Z

ping, can I get a review on this? Thanks!

sfc-gh-rcheng

lgtm after e2e tests pass. Do we need doc changes as well?

sfc-gh-rcheng · 2023-10-19T18:58:51Z

src/main/java/com/snowflake/kafka/connector/internal/streaming/SchematizationUtils.java

@@ -201,16 +213,26 @@ private static Type convertJsonNodeTypeToKafkaType(JsonNode value) {
  }

  /** Convert the kafka data type to Snowflake data type */
-  private static String convertToSnowflakeType(Type kafkaType) {
+  private static String convertToSnowflakeType(Type kafkaType, String schemaName) {
    switch (kafkaType) {
      case INT8:
        return "BYTEINT";
      case INT16:
        return "SMALLINT";
      case INT32:


nit can we add a comment explaining why we chose these 4? something like a link to the code or to this pr's description

I did add it to the PR description, do you want to take a look to see what's missing? We will definitely update our online doc to reflect this change.

pr description is great, but adding it could help with readability, but i dont feel too strongly about it, up to you

sfc-gh-rcheng · 2023-10-19T19:01:22Z

src/main/java/com/snowflake/kafka/connector/records/RecordService.java

@@ -87,7 +87,7 @@ public class RecordService {
          });

  public static final ThreadLocal<SimpleDateFormat> TIME_FORMAT =
-      ThreadLocal.withInitial(() -> new SimpleDateFormat("HH:mm:ss.SSSZ"));
+      ThreadLocal.withInitial(() -> new SimpleDateFormat("HH:mm:ss.SSSXXX"));


out of scope of this PR, but RecordService does have a lot of overlap between snowpipe and streaming, might be useful to refactor or split it into two files for simplicity

sfc-gh-japatel · 2023-10-19T19:08:25Z

src/main/java/com/snowflake/kafka/connector/internal/streaming/SchematizationUtils.java

-        schemaMap.put(field.name(), convertToSnowflakeType(field.schema().type()));
+        String snowflakeType = convertToSnowflakeType(field.schema().type(), field.schema().name());
+        LOGGER.info(
+            "Got the snowflake data type for field:{}, schema_name:{}, kafkaType:{},"


nit: to match with other fields schemaName

src/main/java/com/snowflake/kafka/connector/internal/streaming/SchematizationUtils.java

sfc-gh-japatel · 2023-10-19T19:15:37Z

src/main/java/com/snowflake/kafka/connector/internal/streaming/SchematizationUtils.java

+        if (Date.LOGICAL_NAME.equals(schemaName)) {
+          return "DATE";
+        } else if (Time.LOGICAL_NAME.equals(schemaName)) {
+          return "TIME(6)";
+        } else {
+          return "INT";


thinking out loud here: feel free to ignore.
will this be a behavior change? before this, we converted all INT32 to INT, now we are introducing new snowflake data types- which essentially mean new columns?

I also understand this is a good distinction for them and also we are still in PuPr..

Not really, not sure if you remember, but before the insert will fail because it's inserting VARCHAR into a INT column

'

sfc-gh-japatel · 2023-11-02T23:24:07Z

src/main/java/com/snowflake/kafka/connector/internal/streaming/SchematizationUtils.java

+        if (Decimal.LOGICAL_NAME.equals(schemaName)) {
+          return "VARCHAR";
+        } else {
+          return "BINARY";
+        }


it would be good to mention in description that it is not decimal to varchar but the bytes that represents decimal. Is that correct understanding?

sfc-gh-japatel

lgtm, thanks!

There're 10 AVRO logical types listed in https://avro.apache.org/docs/1.11.0/spec.html#Logical+Types and we're able to support 4 of them, the mapping is as below: date -> DATE time-mills -> TIME(6) timestamp-mills -> TIMESTAMP_NTZ(6) decimal -> VARCHAR (We can't do NUMBER because we could have precision bigger than 36) We can't do for the rest of 6 types because that's not supported by ConnectSchema, see code for more detail. We need to find another way to support other logical types or any sources (like Debezium).

sfc-gh-tzhang added 2 commits October 11, 2023 16:08

support AVRO logical types

a4a44d2

format

91b3c0d

sfc-gh-tzhang requested review from sfc-gh-japatel, sfc-gh-tjones, sfc-gh-rcheng and a team as code owners October 11, 2023 23:33

sfc-gh-tzhang commented Oct 11, 2023

View reviewed changes

sfc-gh-tzhang added 2 commits October 12, 2023 23:09

update tests

64c5d18

minor change

ef7095f

sfc-gh-rcheng approved these changes Oct 19, 2023

View reviewed changes

sfc-gh-japatel reviewed Oct 19, 2023

View reviewed changes

src/main/java/com/snowflake/kafka/connector/internal/streaming/SchematizationUtils.java Show resolved Hide resolved

sfc-gh-japatel reviewed Oct 19, 2023

View reviewed changes

sfc-gh-tzhang added 3 commits October 19, 2023 17:50

add isStreaming check

fc34e17

try again

b0544cd

'

disable test

7d776f5

sfc-gh-japatel reviewed Nov 2, 2023

View reviewed changes

sfc-gh-japatel approved these changes Nov 2, 2023

View reviewed changes

sfc-gh-tzhang merged commit 7edb215 into master Nov 3, 2023
30 checks passed

sfc-gh-tzhang deleted the tzhang-si-logical branch November 3, 2023 20:46

sfc-gh-mbobowski mentioned this pull request Aug 22, 2024

Avro decimal #897

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

SNOW-938038: Support AVRO Logical Types #722

SNOW-938038: Support AVRO Logical Types #722

sfc-gh-tzhang commented Oct 11, 2023 •

edited

Loading

sfc-gh-tzhang Oct 11, 2023

sfc-gh-japatel Oct 12, 2023 •

edited

Loading

sfc-gh-tzhang Oct 13, 2023

sfc-gh-tzhang Oct 13, 2023 •

edited

Loading

sfc-gh-rcheng Oct 19, 2023

sfc-gh-japatel Oct 19, 2023

sfc-gh-japatel Oct 19, 2023

sfc-gh-tzhang Oct 20, 2023

sfc-gh-tzhang commented Oct 18, 2023

sfc-gh-rcheng left a comment

sfc-gh-rcheng Oct 19, 2023

sfc-gh-tzhang Oct 19, 2023

sfc-gh-rcheng Nov 2, 2023

sfc-gh-rcheng Oct 19, 2023

sfc-gh-japatel Oct 19, 2023

sfc-gh-japatel Oct 19, 2023 •

edited

Loading

sfc-gh-tzhang Oct 19, 2023

sfc-gh-japatel Nov 2, 2023

sfc-gh-japatel left a comment

SNOW-938038: Support AVRO Logical Types #722

SNOW-938038: Support AVRO Logical Types #722

Conversation

sfc-gh-tzhang commented Oct 11, 2023 • edited Loading

Choose a reason for hiding this comment

sfc-gh-japatel Oct 12, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sfc-gh-tzhang Oct 13, 2023 • 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

Choose a reason for hiding this comment

sfc-gh-tzhang commented Oct 18, 2023

sfc-gh-rcheng 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

sfc-gh-japatel Oct 19, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sfc-gh-japatel left a comment

Choose a reason for hiding this comment

sfc-gh-tzhang commented Oct 11, 2023 •

edited

Loading

sfc-gh-japatel Oct 12, 2023 •

edited

Loading

sfc-gh-tzhang Oct 13, 2023 •

edited

Loading

sfc-gh-japatel Oct 19, 2023 •

edited

Loading