Amar3tto commented on code in PR #39742: URL: https://github.com/apache/beam/pull/39742#discussion_r3798575023
########## sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/SnowflakeSchemaTransformUtils.java: ########## @@ -0,0 +1,343 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.snowflake; + +import javax.annotation.Nullable; +import org.apache.beam.sdk.io.snowflake.data.SnowflakeColumn; +import org.apache.beam.sdk.io.snowflake.data.SnowflakeDataType; +import org.apache.beam.sdk.io.snowflake.data.SnowflakeTableSchema; +import org.apache.beam.sdk.io.snowflake.data.datetime.SnowflakeTimestamp; +import org.apache.beam.sdk.io.snowflake.data.logical.SnowflakeBoolean; +import org.apache.beam.sdk.io.snowflake.data.numeric.SnowflakeDouble; +import org.apache.beam.sdk.io.snowflake.data.numeric.SnowflakeNumber; +import org.apache.beam.sdk.io.snowflake.data.text.SnowflakeBinary; +import org.apache.beam.sdk.io.snowflake.data.text.SnowflakeVarchar; +import org.apache.beam.sdk.io.snowflake.enums.CreateDisposition; +import org.apache.beam.sdk.io.snowflake.enums.StreamingLogLevel; +import org.apache.beam.sdk.io.snowflake.enums.WriteDisposition; +import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.values.Row; +import org.joda.time.Instant; + +/** Utilities shared by Snowflake schema transform providers. */ +@SuppressWarnings({ + "nullness" // TODO(https://github.com/apache/beam/issues/20497) +}) +public class SnowflakeSchemaTransformUtils { + + public static SnowflakeIO.DataSourceConfiguration createDataSourceConfiguration( Review Comment: It is shared by both the read and write providers, which have different configuration classes. Passing the individual fields keeps this utility independent of either provider-specific configuration. ########## sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/SnowflakeWriteConfiguration.java: ########## @@ -0,0 +1,226 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.snowflake; + +import static org.apache.beam.sdk.io.snowflake.SnowflakeSchemaTransformUtils.parseCreateDisposition; +import static org.apache.beam.sdk.io.snowflake.SnowflakeSchemaTransformUtils.parseStreamingLogLevel; +import static org.apache.beam.sdk.io.snowflake.SnowflakeSchemaTransformUtils.parseWriteDisposition; + +import com.google.auto.value.AutoValue; +import java.io.Serializable; +import javax.annotation.Nullable; +import org.apache.beam.sdk.schemas.AutoValueSchema; +import org.apache.beam.sdk.schemas.annotations.DefaultSchema; +import org.apache.beam.sdk.schemas.annotations.SchemaFieldDescription; + +@AutoValue +@DefaultSchema(AutoValueSchema.class) +public abstract class SnowflakeWriteConfiguration implements Serializable { + + @SchemaFieldDescription("Snowflake server name.") + public abstract String getServerName(); + + @SchemaFieldDescription( + "Snowflake username. Required for password and private key authentication.") + @Nullable + public abstract String getUsername(); + + @SchemaFieldDescription( + "Snowflake password. Mutually exclusive with OAuth token and private key.") + @Nullable + public abstract String getPassword(); + + @SchemaFieldDescription( + "Snowflake OAuth token. Mutually exclusive with password and private key.") + @Nullable + public abstract String getOauthToken(); + + @SchemaFieldDescription( + "Raw Snowflake private key. Mutually exclusive with password and OAuth token.") + @Nullable + public abstract String getPrivateKey(); + + @SchemaFieldDescription("Passphrase for the Snowflake private key.") + @Nullable + public abstract String getPrivateKeyPassphrase(); + + @SchemaFieldDescription("Snowflake database name.") + public abstract String getDatabase(); + + @SchemaFieldDescription("Snowflake schema name.") + public abstract String getSchema(); + + @SchemaFieldDescription("Snowflake warehouse name.") + @Nullable + public abstract String getWarehouse(); + + @SchemaFieldDescription("Snowflake role.") + @Nullable + public abstract String getRole(); + + @SchemaFieldDescription("Destination Snowflake table. Required for batch writes.") + @Nullable + public abstract String getTable(); + + @SchemaFieldDescription("Snowflake Snowpipe name. Required for streaming writes.") + @Nullable + public abstract String getSnowPipe(); + + @SchemaFieldDescription("GCS path used to stage CSV files. The path must end with '/'.") + public abstract String getStagingBucketName(); + + @SchemaFieldDescription("Snowflake storage integration name.") + public abstract String getStorageIntegrationName(); + + @SchemaFieldDescription( + "Table creation behavior for batch writes. " + + "Supported values are CREATE_IF_NEEDED and CREATE_NEVER.") + @Nullable + public abstract String getCreateDisposition(); + + @SchemaFieldDescription( + "Write behavior for batch writes. " + "Supported values are APPEND, TRUNCATE, and EMPTY.") + @Nullable + public abstract String getWriteDisposition(); + + @SchemaFieldDescription("Quotation mark used when writing values to staged CSV files.") + @Nullable + public abstract String getQuotationMark(); + + @SchemaFieldDescription("Maximum number of rows to stage before flushing in streaming mode.") + @Nullable + public abstract Integer getFlushRowLimit(); + + @SchemaFieldDescription( + "Maximum time in milliseconds before flushing staged rows in streaming mode.") + @Nullable + public abstract Long getFlushTimeLimitMillis(); + + @SchemaFieldDescription("Number of output shards used when staging files.") + @Nullable + public abstract Integer getShardsNumber(); + + @SchemaFieldDescription("Streaming log level. Supported values are ERROR and INFO.") + @Nullable + public abstract String getDebugMode(); + + public static Builder builder() { + return new AutoValue_SnowflakeWriteConfiguration.Builder(); + } + + public abstract Builder toBuilder(); + + void validate() { + requireNonEmpty(getServerName(), "serverName"); + requireNonEmpty(getDatabase(), "database"); + requireNonEmpty(getSchema(), "schema"); + requireNonEmpty(getStagingBucketName(), "stagingBucketName"); + requireNonEmpty(getStorageIntegrationName(), "storageIntegrationName"); + + SnowflakeSchemaTransformUtils.validateAuthentication( + getUsername(), getPassword(), getOauthToken(), getPrivateKey(), getPrivateKeyPassphrase()); + + if (!getStagingBucketName().endsWith("/")) { + throw new IllegalArgumentException("stagingBucketName must end with '/'"); + } + + String createDisposition = getCreateDisposition(); + if (createDisposition != null) { + parseCreateDisposition(createDisposition); + } + + String writeDisposition = getWriteDisposition(); + if (writeDisposition != null) { + parseWriteDisposition(writeDisposition); + } + + String debugMode = getDebugMode(); + if (debugMode != null) { + parseStreamingLogLevel(debugMode); Review Comment: Done -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
