Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Streaming without ENABLE_SCHEMA_EVOLUTION; postgres JSON mapped to snowflake VARIANT #536

Open
wants to merge 14 commits into
base: master
Choose a base branch
from
Open

Streaming without ENABLE_SCHEMA_EVOLUTION; postgres JSON mapped to snowflake VARIANT #536

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
63bc190
Added snowflake.schematization.auto; added postgresql JSON to VARIANT…
Feb 6, 2023
c91c9b4
fix date/time conversion in SchematizationUtils; fixed junit
Feb 10, 2023
e42e169
Added support for regex matcher in snowflake.topic2table.map
Feb 10, 2023
4a68939
fix column ordering to match as much as possible the source column or…
Feb 13, 2023
1fceaef
Added snowflake.schematization.auto; added postgresql JSON to VARIANT…
Feb 6, 2023
4c03bd0
fix date/time conversion in SchematizationUtils; fixed junit
Feb 10, 2023
e6e702e
Added support for regex matcher in snowflake.topic2table.map
Feb 10, 2023
7efd254
fix column ordering to match as much as possible the source column or…
Feb 13, 2023
8369100
Merge branch 'streaming-no-auto-schematization' of https://github.com…
Feb 16, 2023
003e67e
Merge remote-tracking branch 'upstream/master' into streaming-no-auto…
Feb 28, 2023
f12e887
Merge remote-tracking branch 'upstream/master' into streaming-no-auto…
Mar 1, 2023
f463138
merge upstream/master
Mar 1, 2023
c3e093f
Merge remote-tracking branch 'upstream/master' into streaming-no-auto…
Mar 2, 2023
88ce343
fix for NPE SinkRecord.keySchema, github #582
acristu Mar 27, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnectorConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,13 @@ public class SnowflakeSinkConnectorConfig {
public static final String ENABLE_SCHEMATIZATION_CONFIG = "snowflake.enable.schematization";
public static final String ENABLE_SCHEMATIZATION_DEFAULT = "false";

public static final String SCHEMATIZATION_AUTO_CONFIG = "snowflake.schematization.auto";
public static final String SCHEMATIZATION_AUTO_DEFAULT = "true";
public static final String SCHEMATIZATION_AUTO_DISPLAY = "Use automatic schema evolution";
public static final String SCHEMATIZATION_AUTO_DOC =
"If true, use snowflake automatic schema evolution feature."
+ "NOTE: you need to grant evolve schema to " + SNOWFLAKE_USER;

// Proxy Info
private static final String PROXY_INFO = "Proxy Info";
public static final String JVM_PROXY_HOST = "jvm.proxy.host";
Expand Down
2 changes: 1 addition & 1 deletion src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -281,5 +281,5 @@ public interface SnowflakeConnectionService {
*
* @param tableName table name
*/
void createTableWithOnlyMetadataColumn(String tableName);
void createTableWithOnlyMetadataColumn(String tableName, boolean autoSchematization);
}
26 changes: 14 additions & 12 deletions src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionServiceV1.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ public void createTable(final String tableName) {
}

@Override
public void createTableWithOnlyMetadataColumn(final String tableName) {
public void createTableWithOnlyMetadataColumn(final String tableName, final boolean autoSchematization) {
checkConnection();
InternalUtils.assertNotEmpty("tableName", tableName);
String createTableQuery =
Expand All @@ -151,17 +151,19 @@ public void createTableWithOnlyMetadataColumn(final String tableName) {
throw SnowflakeErrors.ERROR_2007.getException(e);
}

// Enable schema evolution by default if the table is created by the connector
String enableSchemaEvolutionQuery =
"alter table identifier(?) set ENABLE_SCHEMA_EVOLUTION = true";
try {
PreparedStatement stmt = conn.prepareStatement(enableSchemaEvolutionQuery);
stmt.setString(1, tableName);
stmt.executeQuery();
} catch (SQLException e) {
// Skip the error given that schema evolution is still under PrPr
LOG_WARN_MSG(
"Enable schema evolution failed on table: {}, message: {}", tableName, e.getMessage());
if (autoSchematization) {
// Enable schema evolution by default if the table is created by the connector
String enableSchemaEvolutionQuery =
"alter table identifier(?) set ENABLE_SCHEMA_EVOLUTION = true";
try {
PreparedStatement stmt = conn.prepareStatement(enableSchemaEvolutionQuery);
stmt.setString(1, tableName);
stmt.executeQuery();
} catch (SQLException e) {
// Skip the error given that schema evolution is still under PrPr
LOG_WARN_MSG(
"Enable schema evolution failed on table: {}, message: {}", tableName, e.getMessage());
}
Comment on lines +154 to +166
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This won't work because we rely on the schema evolution to create the table with the correct schema, if you don't want schema evolution on the table, you should create the table yourself and it will have schema evolution turned off by default

Copy link
Author

@acristu acristu Feb 9, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thank you for the reply, the schema is evolved here: https://github.com/streamkap-com/snowflake-kafka-connector/blob/63bc190dd75692e6423d3f50b25143dafaa40d1a/src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java#L653. The main point of this draft PR is to assess if "manual schema evolution" from SchematizationUtils.evolveSchemaIfNeeded can be supported as an alternative to "automatic schema evolution". Our main concern is that that "manual schema evolution" is currently done on the error path, if the insertRow fails.

We cand contribute this "manual schema evolution" option, when setting snowflake.schematization.auto=false the connector would use the connect schema from the schema registry or embedded into the records to evolve the snowflake schema before inserting the data. But we wanted to see first if this approach is acceptable for you going forward.

Copy link
Contributor

@sfc-gh-tzhang sfc-gh-tzhang Feb 10, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Our main concern is that that "manual schema evolution" is currently done on the error path, if the insertRow fails.

This is by design, insertRow will always fail first because we just create a table with RECORD_METADATA only during connector start up

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, that should not be a problem, I mean schema changes should not happen often. The problem is that without these proposed changes https://github.com/streamkap-com/snowflake-kafka-connector/blob/63bc190dd75692e6423d3f50b25143dafaa40d1a/src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java#L264, we cannot enable schema evolution without setting ENABLE_SCHEMA_EVOLUTION. We would like to use only SchematizationUtils.evolveSchemaIfNeeded and not ENABLE_SCHEMA_EVOLUTION.

}

LOG_INFO_MSG("Created table {} with only RECORD_METADATA column", tableName);
Expand Down
9 changes: 6 additions & 3 deletions src/main/java/com/snowflake/kafka/connector/internal/streaming/SchematizationUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,7 @@ private static Map<String, String> getSchemaMapFromRecord(SinkRecord record) {
Schema schema = record.valueSchema();
if (schema != null) {
for (Field field : schema.fields()) {
schemaMap.put(field.name(), convertToSnowflakeType(field.schema().type()));
schemaMap.put(field.name(), convertToSnowflakeType(field.schema().type(), field.schema().name()));
}
}
return schemaMap;
Expand All @@ -158,7 +158,7 @@ private static String inferDataTypeFromJsonObject(JsonNode value) {
// only when the type of the value is unrecognizable for JAVA
throw SnowflakeErrors.ERROR_5021.getException("class: " + value.getClass());
}
return convertToSnowflakeType(schemaType);
return convertToSnowflakeType(schemaType, null);
}

/** Convert a json node type to kafka data type */
Expand Down Expand Up @@ -192,7 +192,7 @@ 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 semanticType) {
switch (kafkaType) {
case INT8:
return "BYTEINT";
Expand All @@ -209,6 +209,9 @@ private static String convertToSnowflakeType(Type kafkaType) {
case BOOLEAN:
return "BOOLEAN";
case STRING:
if (semanticType != null && semanticType.equals("io.debezium.data.Json")) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what's the issue of using VARCHAR in this case?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

By using varchar you'd have to parse json each time you wanted to pull out a field. With variant can use json.field. So this will skip extra processing.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see, this makes sense, I guess this is a general issue for json value since they will be all mapped to STRING, I will see what we can do here, thanks!

return "VARIANT";
}
return "VARCHAR";
case BYTES:
return "BINARY";
Expand Down
5 changes: 4 additions & 1 deletion src/main/java/com/snowflake/kafka/connector/internal/streaming/SnowflakeSinkServiceV2.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,7 @@ public class SnowflakeSinkServiceV2 implements SnowflakeSinkService {
private final String streamingIngestClientName;

private boolean enableSchematization;
private boolean autoSchematization;

/**
* Key is formulated in {@link #partitionChannelKey(String, int)} }
Expand Down Expand Up @@ -131,6 +132,8 @@ public SnowflakeSinkServiceV2(

this.enableSchematization =
this.recordService.setAndGetEnableSchematizationFromConfig(this.connectorConfig);
this.autoSchematization =
this.recordService.setAndGetAutoSchematizationFromConfig(this.connectorConfig);

this.taskId = connectorConfig.getOrDefault(Utils.TASK_ID, "-1");
this.streamingIngestClientName =
Expand Down Expand Up @@ -516,7 +519,7 @@ private void createTableIfNotExists(final String tableName) {
if (this.enableSchematization) {
// Always create the table with RECORD_METADATA only and rely on schema evolution to update
// the schema
this.conn.createTableWithOnlyMetadataColumn(tableName);
this.conn.createTableWithOnlyMetadataColumn(tableName, this.autoSchematization);
} else {
this.conn.createTable(tableName);
}
Expand Down
8 changes: 6 additions & 2 deletions src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -175,6 +175,7 @@ public class TopicPartitionChannel {

// Whether schematization has been enabled.
private final boolean enableSchematization;
private final boolean autoSchematization;

// Whether schema evolution could be done on this channel
private final boolean enableSchemaEvolution;
Expand Down Expand Up @@ -255,11 +256,14 @@ public TopicPartitionChannel(
/* Schematization related properties */
this.enableSchematization =
this.recordService.setAndGetEnableSchematizationFromConfig(sfConnectorConfig);
this.autoSchematization =
this.recordService.setAndGetAutoSchematizationFromConfig(sfConnectorConfig);
this.enableSchemaEvolution =
this.enableSchematization
&& this.conn != null
&& this.conn.hasSchemaEvolutionPermission(
tableName, sfConnectorConfig.get(SNOWFLAKE_ROLE));
&& (!autoSchematization ||
this.conn.hasSchemaEvolutionPermission(
tableName, sfConnectorConfig.get(SNOWFLAKE_ROLE)));
}

/**
Expand Down
19 changes: 19 additions & 0 deletions src/main/java/com/snowflake/kafka/connector/records/RecordService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ public class RecordService extends EnableLogging {
static final String HEADERS = "headers";

private boolean enableSchematization = false;
private boolean autoSchematization = true;

// For each task, we require a separate instance of SimpleDataFormat, since they are not
// inherently thread safe
Expand Down Expand Up @@ -120,6 +121,24 @@ public boolean setAndGetEnableSchematizationFromConfig(
return this.enableSchematization;
}

/**
* extract autoSchematization from the connector config and set the value for the recordService
*
* <p>The extracted boolean is returned for external usage.
*
* @param connectorConfig the connector config map
* @return a boolean indicating whether schematization is enabled
*/
public boolean setAndGetAutoSchematizationFromConfig(
final Map<String, String> connectorConfig) {
if (connectorConfig.containsKey(SnowflakeSinkConnectorConfig.SCHEMATIZATION_AUTO_CONFIG)) {
this.autoSchematization =
Boolean.parseBoolean(
connectorConfig.get(SnowflakeSinkConnectorConfig.SCHEMATIZATION_AUTO_CONFIG));
}
return this.autoSchematization;
}

/**
* Directly set the enableSchematization through param
*
Expand Down