CONN-10496 Creating pipe for SSv2 (#1115)

Author: sfc-gh-mbobowski

src/main/java/com/snowflake/kafka/connector/internal/streaming/v2/DefaultPipeDefinitionProvider.java

@@ -0,0 +1,20 @@
+package com.snowflake.kafka.connector.internal.streaming.v2;
+
+/** Default implementation does not perform any transformations on pipe level */
+public class DefaultPipeDefinitionProvider implements PipeDefinitionProvider {
+
+  private static final String CREATE_PIPE_IF_NOT_EXISTS_STATEMENT =
+      "CREATE PIPE IF NOT EXISTS identifier(?) AS COPY INTO %s FROM TABLE (DATA_SOURCE(TYPE =>"
+          + " 'STREAMING')) MATCH_BY_COLUMN_NAME=CASE_SENSITIVE";
+
+  private static final String CREATE_OR_REPLACE_PIPE_STATEMENT =
+      "CREATE OR REPLACE PIPE identifier(?) AS COPY INTO %s FROM TABLE (DATA_SOURCE(TYPE =>"
+          + " 'STREAMING')) MATCH_BY_COLUMN_NAME=CASE_SENSITIVE";
+
+  @Override
+  public String getPipeDefinition(String tableName, boolean recreate) {
+    String sqlTemplate =
+        recreate ? CREATE_OR_REPLACE_PIPE_STATEMENT : CREATE_PIPE_IF_NOT_EXISTS_STATEMENT;
+    return String.format(sqlTemplate, tableName);
+  }
+}

src/main/java/com/snowflake/kafka/connector/internal/streaming/v2/PipeDefinitionProvider.java

@@ -0,0 +1,6 @@
+package com.snowflake.kafka.connector.internal.streaming.v2;
+
+/** Construct CREATE PIPE sql statement */
+public interface PipeDefinitionProvider {
+  String getPipeDefinition(String tableName, boolean recreate);
+}

src/test/java/com/snowflake/kafka/connector/internal/streaming/v2/DefaultPipeDefinitionProviderTest.java

@@ -0,0 +1,35 @@
+package com.snowflake.kafka.connector.internal.streaming.v2;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.stream.Stream;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class DefaultPipeDefinitionProviderTest {
+
+  private static final PipeDefinitionProvider provider = new DefaultPipeDefinitionProvider();
+
+  @ParameterizedTest
+  @MethodSource("testParams")
+  void shouldCreatePipeDefinition(boolean recreate, String expected) {
+    // when
+    String result = provider.getPipeDefinition("foo", recreate);
+
+    // then
+    assertThat(result).isEqualTo(expected);
+  }
+
+  public static Stream<Arguments> testParams() {
+    return Stream.of(
+        Arguments.of(
+            true,
+            "CREATE OR REPLACE PIPE identifier(?) AS COPY INTO foo FROM TABLE (DATA_SOURCE(TYPE =>"
+                + " 'STREAMING')) MATCH_BY_COLUMN_NAME=CASE_SENSITIVE"),
+        Arguments.of(
+            false,
+            "CREATE PIPE IF NOT EXISTS identifier(?) AS COPY INTO foo FROM TABLE (DATA_SOURCE(TYPE"
+                + " => 'STREAMING')) MATCH_BY_COLUMN_NAME=CASE_SENSITIVE"));
+  }
+}