Eng 789 auto create schema feature

src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnector.java

@@ -294,7 +294,8 @@ public Config validate(Map<String, String> connectorConfigs) {
       return result;
     }
 
-    try {
+    // Disabling config validation due to ENG-789/Auto create schema feature for Append SF connector
+    /*try {
       testConnection.schemaExists(connectorConfigs.get(Utils.SF_SCHEMA));
     } catch (SnowflakeKafkaConnectorException e) {
       LOGGER.error("Validate Error msg:{}, errorCode:{}", e.getMessage(), e.getCode());
@@ -304,7 +305,7 @@ public Config validate(Map<String, String> connectorConfigs) {
         throw e;
       }
       return result;
-    }
+    }*/
 
     LOGGER.info("Validated config with no error");
     return result;

src/main/java/com/snowflake/kafka/connector/SnowflakeSinkTask.java

@@ -229,7 +229,8 @@ public void start(final Map<String, String> parsedConfig) {
             .setErrorReporter(kafkaRecordErrorReporter)
             .setSinkTaskContext(this.context)
             .build();
-
+    createSchemaIfNotExists(getConnection(),
+            parsedConfig.get(SnowflakeSinkConnectorConfig.SNOWFLAKE_SCHEMA));
     this.streamkapQueryTemplate = StreamkapQueryTemplate.buildStreamkapQueryTemplateFromConfig(parsedConfig);
 
     DYNAMIC_LOGGER.info(
@@ -238,6 +239,12 @@ public void start(final Map<String, String> parsedConfig) {
         getDurationFromStartMs(this.taskStartTime));
   }
 
+  private void createSchemaIfNotExists(SnowflakeConnectionService con, String schemaName){
+      if(!con.schemaExist(schemaName)) {
+        con.createSchema(schemaName);
+      }
+  }
+
   /**
    * stop method is invoked only once outstanding calls to other methods have completed. e.g. after
    * current put, and a final preCommit has completed.

src/main/java/com/snowflake/kafka/connector/internal/InternalUtils.java

@@ -59,6 +59,8 @@ static int resultSize(ResultSet resultSet) throws SQLException {
   static void assertNotEmpty(String name, Object value) {
     if (value == null || (value instanceof String && value.toString().isEmpty())) {
       switch (name.toLowerCase()) {
+        case "schemaname":
+          throw SnowflakeErrors.ERROR_0031.getException();
         case "tablename":
           throw SnowflakeErrors.ERROR_0005.getException();
         case "stagename":

src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionService.java

@@ -7,6 +7,13 @@
 import java.util.Map;
 
 public interface SnowflakeConnectionService {
+  /**
+   * create schema is not exists
+   *
+   * @param schemaName schema name
+   */
+  void createSchema(String schemaName);
+
   /**
    * Create a table with two variant columns: RECORD_METADATA and RECORD_CONTENT
    *
@@ -59,6 +66,14 @@ public interface SnowflakeConnectionService {
    */
   void createStage(String stageName);
 
+  /**
+   * check schema existence
+   *
+   * @param schemaName table name
+   * @return true if schema exists, false otherwise
+   */
+  boolean schemaExist(String schemaName);
+
   /**
    * check table existence
    *

src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionServiceV1.java

@@ -110,6 +110,25 @@ private static Properties mergeProxyAndConnectionProperties(
     return mergedProperties;
   }
 
+  @Override
+  public void createSchema(final String schemaName) {
+    checkConnection();
+    InternalUtils.assertNotEmpty("schemaName", schemaName);
+    String query;
+      query = "create schema if not exists identifier(?) comment = 'created by"
+              + " automatic schema creation from Snowflake Kafka Connector'";
+    try {
+      PreparedStatement stmt = conn.prepareStatement(query);
+      stmt.setString(1, schemaName);
+      stmt.execute();
+      stmt.close();
+    } catch (SQLException e) {
+      throw SnowflakeErrors.ERROR_2018.getException(e);
+    }
+
+    LOGGER.info("create schema {}", schemaName);
+  }
+
   @Override
   public void createTable(final String tableName, final boolean overwrite) {
     checkConnection();
@@ -237,6 +256,33 @@ public void createStage(final String stageName) {
     createStage(stageName, false);
   }
 
+  @Override
+  public boolean schemaExist(final String schemaName) {
+    checkConnection();
+    InternalUtils.assertNotEmpty("schemaName", schemaName);
+    String query = "desc schema identifier(?)";
+    PreparedStatement stmt = null;
+    boolean exist;
+    try {
+      stmt = conn.prepareStatement(query);
+      stmt.setString(1, schemaName);
+      stmt.execute();
+      exist = true;
+    } catch (Exception e) {
+      LOGGER.debug("schema {} doesn't exist", schemaName);
+      exist = false;
+    } finally {
+      if (stmt != null) {
+        try {
+          stmt.close();
+        } catch (SQLException e) {
+          e.printStackTrace();
+        }
+      }
+    }
+    return exist;
+  }
+
   @Override
   public boolean tableExist(final String tableName) {
     checkConnection();

src/main/java/com/snowflake/kafka/connector/internal/SnowflakeErrors.java

@@ -136,7 +136,7 @@ public enum SnowflakeErrors {
       String.format(
           "Failed to parse %s map",
           SnowflakeSinkConnectorConfig.SNOWPIPE_STREAMING_CLIENT_PROVIDER_OVERRIDE_MAP)),
-
+  ERROR_0031("0031", "Empty Schema name", "Input Schema name is empty string or null"),
   // Snowflake connection issues 1---
   ERROR_1001(
       "1001",
@@ -205,6 +205,10 @@ public enum SnowflakeErrors {
       "2017",
       "Failed to check schema evolution permission",
       "Failed to check schema evolution permission"),
+  ERROR_2018(
+          "2018",
+          "Failed to create schema",
+          "Failed to create schema on Snowflake, please check that you have permission to do so."),
   // Snowpipe related issues 3---
   ERROR_3001("3001", "Failed to ingest file", "Exception reported by Ingest SDK"),
 

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

@@ -131,6 +131,7 @@ public class DirectTopicPartitionChannel implements 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;
@@ -232,8 +233,13 @@ public DirectTopicPartitionChannel(
     /* Schematization related properties */
     this.enableSchematization =
         this.recordService.setAndGetEnableSchematizationFromConfig(sfConnectorConfig);
-
-    this.enableSchemaEvolution = this.enableSchematization && hasSchemaEvolutionPermission;
+    this.autoSchematization =
+            this.recordService.setAndGetAutoSchematizationFromConfig(sfConnectorConfig);
+    //this.enableSchemaEvolution = this.enableSchematization && hasSchemaEvolutionPermission;
+    this.enableSchemaEvolution =
+            this.enableSchematization
+                    && this.conn != null
+                    && (!autoSchematization || hasSchemaEvolutionPermission);
 
     if (isEnableChannelOffsetMigration(sfConnectorConfig)) {
       /* Channel Name format V2 is computed from connector name, topic and partition */