🎉 Source MSSQL: add option to disable initial snapshot and change snapshot isolation level (#12759)

sivankumar86 · web-flow · commit 90b757b22909 · 2022-05-25T07:58:39.000-07:00
* MSSQL CDC feature to capture only changes option added

* MSSQL CDC feature to capture only changes option added

* added option to disable snapshot 

snapshot isolation setting is hardcode and I modified as user option

* recommitting

* docker version change

docker version change

* resolve conflict

* review 1 added

* review 1 added

* removed extra space

removed extra space
diff --git a/airbyte-integrations/connectors/source-mssql/Dockerfile b/airbyte-integrations/connectors/source-mssql/Dockerfile
@@ -16,5 +16,5 @@ ENV APPLICATION source-mssql
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.3.22
+LABEL io.airbyte.version=0.3.23
 LABEL io.airbyte.name=airbyte/source-mssql
diff --git a/airbyte-integrations/connectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlCdcProperties.java b/airbyte-integrations/connectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlCdcProperties.java
@@ -4,23 +4,16 @@
 
 package io.airbyte.integrations.source.mssql;
 
+import com.fasterxml.jackson.databind.JsonNode;
+
 import java.util.Properties;
 
 public class MssqlCdcProperties {
 
-  static Properties getDebeziumProperties() {
+  static Properties getDebeziumProperties(final JsonNode config) {
     final Properties props = new Properties();
     props.setProperty("connector.class", "io.debezium.connector.sqlserver.SqlServerConnector");
 
-    // snapshot config
-    // https://debezium.io/documentation/reference/1.4/connectors/sqlserver.html#sqlserver-property-snapshot-mode
-    props.setProperty("snapshot.mode", "initial");
-    // https://docs.microsoft.com/en-us/sql/t-sql/statements/set-transaction-isolation-level-transact-sql?view=sql-server-ver15
-    // https://debezium.io/documentation/reference/1.4/connectors/sqlserver.html#sqlserver-property-snapshot-isolation-mode
-    // we set this to avoid preventing other (non-Airbyte) transactions from updating table rows while
-    // we snapshot
-    props.setProperty("snapshot.isolation.mode", "snapshot");
-
     // https://debezium.io/documentation/reference/1.4/connectors/sqlserver.html#sqlserver-property-include-schema-changes
     props.setProperty("include.schema.changes", "false");
     // https://debezium.io/documentation/reference/1.4/connectors/sqlserver.html#sqlserver-property-provide-transaction-metadata
@@ -29,6 +22,31 @@ static Properties getDebeziumProperties() {
     props.setProperty("converters", "mssql_converter");
     props.setProperty("mssql_converter.type", "io.airbyte.integrations.debezium.internals.MSSQLConverter");
 
+      final JsonNode replication_config = config.get("replication_method");
+      if(replication_config.hasNonNull("replication_method")) {
+        final JsonNode cdcMethod = config.get("replication_method");
+        if(cdcMethod.hasNonNull("is_snapshot_disabled") &&
+                cdcMethod.get("is_snapshot_disabled").asBoolean()) {
+          props.setProperty("snapshot.isolation.mode", "read_committed");
+        } else {
+          // https://docs.microsoft.com/en-us/sql/t-sql/statements/set-transaction-isolation-level-transact-sql?view=sql-server-ver15
+          // https://debezium.io/documentation/reference/1.4/connectors/sqlserver.html#sqlserver-property-snapshot-isolation-mode
+          // we set this to avoid preventing other (non-Airbyte) transactions from updating table rows while
+          // we snapshot
+          props.setProperty("snapshot.isolation.mode", "snapshot");
+        }
+        if(cdcMethod.hasNonNull("is_cdc_only") &&
+                cdcMethod.get("is_cdc_only").asBoolean()) {
+          props.setProperty("snapshot.mode", "schema_only");
+        } else {
+          // snapshot config
+          // https://debezium.io/documentation/reference/1.4/connectors/sqlserver.html#sqlserver-property-snapshot-mode
+          props.setProperty("snapshot.mode", "initial");
+        }
+      } else {
+        props.setProperty("snapshot.isolation.mode", "snapshot");
+        props.setProperty("snapshot.mode", "initial");
+      }
     return props;
   }
 
diff --git a/airbyte-integrations/connectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlSource.java b/airbyte-integrations/connectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlSource.java
@@ -38,11 +38,7 @@
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.time.Instant;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
 import java.util.stream.Stream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -317,26 +313,30 @@ protected void assertSqlServerAgentRunning(final JdbcDatabase database) throws S
 
   protected void assertSnapshotIsolationAllowed(final JsonNode config, final JdbcDatabase database)
       throws SQLException {
-    final List<JsonNode> queryResponse = database.queryJsons(connection -> {
-      final String sql = "SELECT name, snapshot_isolation_state FROM sys.databases WHERE name = ?";
-      final PreparedStatement ps = connection.prepareStatement(sql);
-      ps.setString(1, config.get("database").asText());
-      LOGGER.info(String.format(
-          "Checking that snapshot isolation is enabled on database '%s' using the query: '%s'",
-          config.get("database").asText(), sql));
-      return ps;
-    }, sourceOperations::rowToJson);
+    final JsonNode  replication_config =config.get("replication_method");
+    if(!replication_config.hasNonNull("is_snapshot_disabled") ||
+            !replication_config.get("is_snapshot_disabled").asBoolean()) {
+      final List<JsonNode> queryResponse = database.queryJsons(connection -> {
+        final String sql = "SELECT name, snapshot_isolation_state FROM sys.databases WHERE name = ?";
+        final PreparedStatement ps = connection.prepareStatement(sql);
+        ps.setString(1, config.get("database").asText());
+        LOGGER.info(String.format(
+                "Checking that snapshot isolation is enabled on database '%s' using the query: '%s'",
+                config.get("database").asText(), sql));
+        return ps;
+      }, sourceOperations::rowToJson);
 
-    if (queryResponse.size() < 1) {
-      throw new RuntimeException(String.format(
-          "Couldn't find '%s' in sys.databases table. Please check the spelling and that the user has relevant permissions (see docs).",
-          config.get("database").asText()));
-    }
-    if (queryResponse.get(0).get("snapshot_isolation_state").asInt() != 1) {
-      throw new RuntimeException(String.format(
-          "Detected that snapshot isolation is not enabled for database '%s'. MSSQL CDC relies on snapshot isolation. "
-              + "Please check the documentation on how to enable snapshot isolation on MS SQL Server.",
-          config.get("database").asText()));
+      if (queryResponse.size() < 1) {
+        throw new RuntimeException(String.format(
+                "Couldn't find '%s' in sys.databases table. Please check the spelling and that the user has relevant permissions (see docs).",
+                config.get("database").asText()));
+      }
+      if (queryResponse.get(0).get("snapshot_isolation_state").asInt() != 1) {
+        throw new RuntimeException(String.format(
+                "Detected that snapshot isolation is not enabled for database '%s'. MSSQL CDC relies on snapshot isolation. "
+                        + "Please check the documentation on how to enable snapshot isolation on MS SQL Server.",
+                config.get("database").asText()));
+      }
     }
   }
 
@@ -350,9 +350,10 @@ public List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(
     final JsonNode sourceConfig = database.getSourceConfig();
     if (isCdc(sourceConfig) && shouldUseCDC(catalog)) {
       LOGGER.info("using CDC: {}", true);
+      Properties props=MssqlCdcProperties.getDebeziumProperties(sourceConfig);
       final AirbyteDebeziumHandler handler = new AirbyteDebeziumHandler(sourceConfig,
           MssqlCdcTargetPosition.getTargetPosition(database, sourceConfig.get("database").asText()),
-          MssqlCdcProperties.getDebeziumProperties(), catalog, true);
+              props, catalog, true);
       return handler.getIncrementalIterators(
           new MssqlCdcSavedInfoFetcher(stateManager.getCdcStateManager().getCdcState()),
           new MssqlCdcStateHandler(stateManager), new MssqlCdcConnectorMetadataInjector(),
@@ -364,9 +365,18 @@ public List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(
   }
 
   private static boolean isCdc(final JsonNode config) {
-    return config.hasNonNull("replication_method")
-        && ReplicationMethod.valueOf(config.get("replication_method").asText())
-            .equals(ReplicationMethod.CDC);
+    if(config.hasNonNull("replication_method")){
+      final JsonNode replication_config = config.get("replication_method");
+      if(replication_config.hasNonNull("replication_method")){
+        return ReplicationMethod.valueOf(replication_config.get("replication_method").asText())
+                .equals(ReplicationMethod.CDC);
+      }else{
+        return ReplicationMethod.valueOf(replication_config.asText())
+                .equals(ReplicationMethod.CDC);
+      }
+    } else {
+      return false;
+    }
   }
 
   private static boolean shouldUseCDC(final ConfiguredAirbyteCatalog catalog) {
@@ -408,6 +418,7 @@ private static AirbyteStream addCdcMetadataColumns(final AirbyteStream stream) {
     return stream;
   }
 
+
   private void readSsl(final JsonNode sslMethod, final List<String> additionalParameters) {
     final JsonNode config = sslMethod.get("ssl_method");
     switch (config.get("ssl_method").asText()) {
diff --git a/airbyte-integrations/connectors/source-mssql/src/main/resources/spec.json b/airbyte-integrations/connectors/source-mssql/src/main/resources/spec.json
@@ -105,12 +105,54 @@
         ]
       },
       "replication_method": {
-        "type": "string",
+        "type": "object",
         "title": "Replication Method",
-        "description": "The replication method used for extracting data from the database. STANDARD replication requires no setup on the DB side but will not be able to represent deletions incrementally. CDC uses {TBC} to detect inserts, updates, and deletes. This needs to be configured on the source database itself.",
-        "default": "STANDARD",
-        "enum": ["STANDARD", "CDC"],
-        "order": 8
+        "description": "The replication method used for extracting data from the database.  ",
+        "order": 8,
+        "oneOf": [
+          {
+            "title": "STANDARD",
+            "additionalProperties": false,
+            "description": "STANDARD replication requires no setup on the DB side but will not be able to represent deletions incrementally.",
+            "properties": {
+              "replication_method": {
+                "type": "string",
+                "const": "STANDARD",
+                "enum": ["STANDARD"],
+                "default": "STANDARD",
+                "order": 0
+              }
+            }
+          },
+          {
+            "title": "CDC",
+            "additionalProperties": false,
+            "description": "CDC uses {TBC} to detect inserts, updates, and deletes. This needs to be configured on the source database itself.",
+            "properties": {
+              "replication_method": {
+                "type": "string",
+                "const": "CDC",
+                "enum": ["CDC"],
+                "default": "CDC",
+                "order":0
+              },
+              "is_snapshot_disabled": {
+                "title": "DisableSnapshot",
+                "type": "boolean",
+                "description": "validate the snapshot enable in the database. true would set snapshot.isolation.mode to snapshot otherwise, it would be ",
+                "default": false,
+                "order": 1
+              },
+              "is_cdc_only": {
+                "title": "cdcOnly",
+                "type": "boolean",
+                "default": false,
+                "description": "true would set snapshot mode to schema_only otherwise, it would set to initial. Refer, https://debezium.io/documentation/reference/stable/connectors/sqlserver.html",
+                "order": 2
+              }
+            }
+          }
+          ]
       }
     }
   }
diff --git a/airbyte-integrations/connectors/source-mssql/src/test/java/io/airbyte/integrations/source/mssql/CdcMssqlSourceTest.java b/airbyte-integrations/connectors/source-mssql/src/test/java/io/airbyte/integrations/source/mssql/CdcMssqlSourceTest.java
@@ -21,6 +21,7 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.string.Strings;
 import io.airbyte.db.Database;
@@ -38,6 +39,7 @@
 import io.airbyte.protocol.models.AirbyteStream;
 import io.debezium.connector.sqlserver.Lsn;
 import java.sql.SQLException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -81,7 +83,6 @@ private void init() {
 
     dbName = Strings.addRandomSuffix("db", "_", 10).toLowerCase();
     source = new MssqlSource();
-
     config = Jsons.jsonNode(ImmutableMap.builder()
         .put("host", container.getHost())
         .put("port", container.getFirstMappedPort())
@@ -268,6 +269,23 @@ void testAssertSnapshotIsolationAllowed() {
     assertThrows(RuntimeException.class, () -> source.assertSnapshotIsolationAllowed(config, testJdbcDatabase));
   }
 
+  @Test
+  void testAssertSnapshotIsolationDisabled() {
+    //disabled the snapshot
+   JsonNode replication_config = Jsons.jsonNode(ImmutableMap.builder()
+            .put("replication_method", "CDC")
+            .put("is_cdc_only", "false")
+            .put("is_snapshot_disabled", "true")
+            .build());
+    Jsons.replaceNestedValue(config, Arrays.asList(new String[]{"replication_method"}),replication_config);
+    // snapshot isolation enabled by setup so assert check passes
+    assertDoesNotThrow(() -> source.assertSnapshotIsolationAllowed(config, testJdbcDatabase));
+    // now disable snapshot isolation and assert that check fails
+    switchSnapshotIsolation(false, dbName);
+    // snapshot isolation disabled and snapshot validation is disabled so assert check passes
+    assertDoesNotThrow(() -> source.assertSnapshotIsolationAllowed(config, testJdbcDatabase));
+  }
+
   // Ensure the CDC check operations are included when CDC is enabled
   // todo: make this better by checking the returned checkOperations from source.getCheckOperations
   @Test
diff --git a/docs/integrations/sources/mssql.md b/docs/integrations/sources/mssql.md
@@ -63,7 +63,7 @@ Please read the [CDC docs](../../understanding-airbyte/cdc.md) for an overview o
 * There are some critical issues regarding certain datatypes. Please find detailed info in [this Github issue](https://github.com/airbytehq/airbyte/issues/4542).
 * CDC is only available for SQL Server 2016 Service Pack 1 \(SP1\) and later.
 * _db\_owner_ \(or higher\) permissions are required to perform the [neccessary setup](mssql.md#setting-up-cdc-for-mssql) for CDC.
-* You must enable [snapshot isolation mode](https://docs.microsoft.com/en-us/dotnet/framework/data/adonet/sql/snapshot-isolation-in-sql-server) on the database\(s\) you want to sync. This is used for retrieving an initial snapshot without locking tables.
+* You must enable [snapshot isolation mode](https://docs.microsoft.com/en-us/dotnet/framework/data/adonet/sql/snapshot-isolation-in-sql-server) on the database\(s\) you want to sync. This is used for retrieving an initial snapshot without locking tables. You can disable snapshot isolation and initial load in connector setting if required.
 * On Linux, CDC is not supported on versions earlier than SQL Server 2017 CU18 \(SQL Server 2019 is supported\).
 * Change data capture cannot be enabled on tables with a clustered columnstore index. \(It can be enabled on tables with a _non-clustered_ columnstore index\).
 * The SQL Server CDC feature processes changes that occur in user-created tables only. You cannot enable CDC on the SQL Server master database.
