FLOW-7873 Autocreate table (#1241)

Author: sfc-gh-lkucharski

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

@@ -229,11 +229,6 @@ public enum SnowflakeErrors {
       "5028",
       "Failed to open Snowpipe Streaming v2 channel",
       "Failed to open Snowpipe Streaming v2 channel"),
-  ERROR_5029(
-      "5029",
-      "Destination table does not exist",
-      "Destination table does not exist. Please ensure the destination table exists in Snowflake"
-          + " before starting the connector."),
   ERROR_5030(
       "5030",
       "Channel error count threshold exceeded",

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

@@ -86,7 +86,8 @@ public void createTableWithOnlyMetadataColumn(final String tableName) {
     InternalUtils.assertNotEmpty("tableName", tableName);
     String createTableQuery =
         "create table if not exists identifier(?) (record_metadata variant comment 'created by"
-            + " automatic table creation from Snowflake Kafka Connector')";
+            + " automatic table creation from Snowflake Kafka Connector High Performance')"
+            + " enable_schema_evolution = true";
 
     try {
       PreparedStatement stmt = conn.prepareStatement(createTableQuery);

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

@@ -173,11 +173,7 @@ public void startPartitions(Collection<TopicPartition> partitions) {
     for (String topic : topics) {
       final String tableName = getTableName(topic, this.topicToTableMap);
 
-      final boolean tableExists = this.conn.tableExist(tableName);
-      if (!tableExists) {
-        throw SnowflakeErrors.ERROR_5029.getException(
-            "Table name: " + tableName, this.conn.getTelemetryClient());
-      }
+      createTableIfNotExists(tableName);
 
       // not an error, by convention we're looking for the same name as table
       final boolean pipeExists = this.conn.pipeExist(tableName);
@@ -465,7 +461,6 @@ public Optional<MetricRegistry> getMetricRegistry(String partitionChannelKey) {
   private void createTableIfNotExists(final String tableName) {
     if (this.conn.tableExist(tableName)) {
       LOGGER.info("Using existing table {}.", tableName);
-      this.conn.appendMetaColIfNotExist(tableName);
     } else {
       LOGGER.info("Creating new table {}.", tableName);
       this.conn.createTableWithOnlyMetadataColumn(tableName);

src/test/java/com/snowflake/kafka/connector/internal/streaming/CloseTopicPartitionChannelIT.java

@@ -26,7 +26,6 @@ void setUp() throws JsonProcessingException {
     topicName = TestUtils.randomTableName();
     connectorName = topicName + "_connector";
     connectCluster.kafka().createTopic(topicName, PARTITIONS_NUMBER);
-    TestUtils.getConnectionService().createTableWithMetadataColumn(topicName);
     // JVM scoped Ingest client mock
     StreamingClientManager.setIngestClientSupplier(fakeClientSupplier);
     generateKafkaMessages();

src/test/java/com/snowflake/kafka/connector/internal/streaming/SnowflakeSinkServiceV2IT.java

@@ -48,7 +48,6 @@ public class SnowflakeSinkServiceV2IT extends SnowflakeSinkServiceV2BaseIT {
   @BeforeEach
   public void setup() {
     config = TestUtils.getConnectorConfigurationForStreaming(true);
-    conn.createTableWithMetadataColumn(table);
     pipe = PipeNameProvider.buildPipeName(table);
   }
 
@@ -202,9 +201,7 @@ public void testStreamingIngestion()
   public void testStreamingIngest_multipleChannelPartitions_withMetrics()
       throws Exception { // set up telemetry service spy
     SnowflakeConnectionService connectionService = Mockito.spy(this.conn);
-    connectionService.createTableWithMetadataColumn(table);
-    SnowflakeTelemetryService telemetryService =
-        Mockito.spy((SnowflakeTelemetryService) this.conn.getTelemetryClient());
+    SnowflakeTelemetryService telemetryService = Mockito.spy(this.conn.getTelemetryClient());
     Mockito.when(connectionService.getTelemetryClient()).thenReturn(telemetryService);
 
     // opens a channel for partition 0, table and topic