Abacn commented on code in PR #39742:
URL: https://github.com/apache/beam/pull/39742#discussion_r3854029562
##########
sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/SnowflakeIO.java:
##########
@@ -231,6 +231,7 @@ public interface CsvMapper<T> extends Serializable {
*/
@FunctionalInterface
public interface UserDataMapper<T> extends Serializable {
+ @org.checkerframework.checker.nullness.qual.Nullable
Review Comment:
nit: Import the name
##########
sdks/java/io/snowflake/build.gradle:
##########
@@ -30,6 +30,8 @@ dependencies {
implementation project(path:
":sdks:java:extensions:google-cloud-platform-core")
permitUnusedDeclared project(path:
":sdks:java:extensions:google-cloud-platform-core")
implementation library.java.slf4j_api
+ implementation library.java.everit_json_schema
Review Comment:
Why do we need it now? Is it now a runtime dependency, or needed by the
expansion service now?
##########
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:
Please check, I don't see a code change?
--
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]
