SNOW-1057812: Support Open Rowset API Channel #731
Conversation
| @@ -114,19 +135,23 @@ private OpenChannelRequest(OpenChannelRequestBuilder builder) { | |||
| Utils.assertStringNotNullOrEmpty("channel name", builder.channelName); | |||
| Utils.assertStringNotNullOrEmpty("database name", builder.dbName); | |||
| Utils.assertStringNotNullOrEmpty("schema name", builder.schemaName); | |||
| Utils.assertStringNotNullOrEmpty("table name", builder.tableName); | |||
| Utils.assertStringsNotNullOrEmpty( | |||
There was a problem hiding this comment.
should we also assert that both table and pipe name are not set at the same time?
There was a problem hiding this comment.
Yeah this is a good edge case check to add.
There was a problem hiding this comment.
The behavior right now is that the table name will be ignored if there is a pipe name, I thought about assert if both are provided but I think it's too restricted, it's perfect valid if customer inputs both and pipe is created on that table
| @@ -247,6 +262,8 @@ public interface SnowflakeStreamingIngestChannel { | |||
| */ | |||
| InsertValidationResponse insertRow(Map<String, Object> row, @Nullable String offsetToken); | |||
|
|
|||
| InsertValidationResponse insertRow(Map<String, Object> row); | |||
There was a problem hiding this comment.
in which cases will a customers use insertRow without an offsetToken?
There was a problem hiding this comment.
Yeah shouldn't this be InsertValidationResponse insertRow(Map<String, Object> row, @Nullable String offsetToken);?
There was a problem hiding this comment.
remember that we say the offset_token will be per row for Rowset API?
| * Get the fully qualified channel name | ||
| * | ||
| * @return fully qualified name of the channel, in the format of | ||
| * dbName.schemaName.tableName.channelName |
There was a problem hiding this comment.
nit : or dbName.schemaName.pipeName.channelName
| InsertValidationResponse insertRows( | ||
| Iterable<Map<String, Object>> rows, @Nullable String offsetToken); |
There was a problem hiding this comment.
we might need to add something like
insertRows(Iterable<Map<String, Object>> rows, List offsetTokens)
for partial ingestion
There was a problem hiding this comment.
Yes, that's the one below, offset_token will be part of the row now.
| * @throws SFException | ||
| */ | ||
| /** Assert when all the Strings are either null or Empty */ | ||
| public static void assertStringsNotNullOrEmpty(String name, Collection<String> values) |
There was a problem hiding this comment.
nit: when I first saw this method, I assumed this was asserting that all the strings in the collection should be not null, maybe we can rename it to something like assertAtLeastOneStringNotNullOrEmpty to be extra clear?
| /** | ||
| * Get the type of channel, please referring to {@link | ||
| * net.snowflake.ingest.streaming.OpenChannelRequest.ChannelType} for the supported channel type | ||
| * | ||
| * @return type of the channel | ||
| */ | ||
| @Override | ||
| public OpenChannelRequest.ChannelType getType() { | ||
| return OpenChannelRequest.ChannelType.CLOUD_STORAGE; | ||
| } | ||
|
|
There was a problem hiding this comment.
Should we rename this class to SnowflakeStreamingIngestChannelCloud to be parallel with the rowset?
There was a problem hiding this comment.
Yes, I can do that in a separate PR
| return SnowflakeStreamingIngestChannelFactory.<T>builder(request.getChannelName()) | ||
| .setDBName(request.getDBName()) | ||
| .setSchemaName(request.getSchemaName()) | ||
| .setTableName(request.getTableName()) |
There was a problem hiding this comment.
we need to setPipeName as well
| public enum ChannelType { | ||
| CLOUD_STORAGE, | ||
| ROWSET_API, | ||
| } |
There was a problem hiding this comment.
I think this exists already? If you look at StreamingResource there's a check somewhere that compares against an enum.
There was a problem hiding this comment.
Yes, but the one used before was hided in the SDK, now this is a public enum in the public class
This PR contains the change to allow creation of a Rowset API channel:
Note that we're merging into a feature branch instead of master