CDAP-17482 Add option for the source configurations to ignore replicating any existing data.

Author: sagarkapare

mysql-delta-plugins/docs/mysql-cdcSource.md

@@ -73,6 +73,10 @@ to the MySQL server will be of the form 'user_name'@'%' where user_name is this
 
 **JDBC Plugin Name:** Identifier for the MySQL JDBC driver, which is the name used while uploading the MySQL JDBC driver.
 
+**Replicate Existing Data:** Whether to replicate existing data from the source database. By default, pipeline will 
+replicate the existing data from source tables. If set to false, any existing data in the source tables will be 
+ignored and only changes happening after the pipeline started will be replicated.
+
 Troubleshooting
 -----------
 If the replicator is able to start snapshotting the data, but fails when it switches over to read from the 

mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlConfig.java

@@ -57,6 +57,12 @@ public class MySqlConfig extends PluginConfig {
   @Description("Identifier for the MySQL JDBC driver, which is the name used while uploading the MySQL JDBC driver.")
   private String jdbcPluginName;
 
+  @Nullable
+  @Description("Whether to replicate existing data from the source database. By default, pipeline will replicate " +
+    "the existing data from source tables. If set to false, any existing data in the source " +
+    "tables will be ignored and only changes happening after the pipeline started will be replicated.")
+  private Boolean replicateExistingData;
+
   public MySqlConfig(String host, int port, String user, String password, int consumerID,
                      String database, @Nullable String serverTimezone) {
     this.host = host;
@@ -116,4 +122,8 @@ public Properties getConnectionProperties() {
     properties.put("serverTimezone", getServerTimezone());
     return properties;
   }
+
+  public boolean getReplicateExistingData() {
+    return replicateExistingData != null ? replicateExistingData : true;
+  }
 }

mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlEventReader.java

@@ -119,7 +119,8 @@ public void start(Offset offset) {
       .with("database.whitelist", config.getDatabase())
       .with("database.server.name", "dummy") // this is the kafka topic for hosted debezium - it doesn't matter
       .with("database.serverTimezone", config.getServerTimezone())
-      .with("table.whitelist", String.join(",", sourceTableMap.keySet()));
+      .with("table.whitelist", String.join(",", sourceTableMap.keySet()))
+      .with("snapshot.mode", config.getReplicateExistingData() ? "initial" : "schema_only");
 
     if (config.getConsumerID() != null) {
       // If not provided debezium will randomly pick integer between 5400 and 6400.
@@ -157,7 +158,7 @@ public void start(Offset offset) {
       engine = EmbeddedEngine.create()
         .using(debeziumConf)
         .notifying(new MySqlRecordConsumer(context, emitter, ddlParser, mySqlValueConverters,
-                                           new Tables(), sourceTableMap))
+                                           new Tables(), sourceTableMap, config.getReplicateExistingData()))
         .using(new NotifyingCompletionCallback(context))
         .build();
       executorService.submit(engine);

mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlRecordConsumer.java

@@ -56,16 +56,18 @@ public class MySqlRecordConsumer implements Consumer<SourceRecord> {
   private final MySqlValueConverters mySqlValueConverters;
   private final Tables tables;
   private final Map<String, SourceTable> sourceTableMap;
+  private final boolean replicateExistingData;
 
   public MySqlRecordConsumer(DeltaSourceContext context, EventEmitter emitter,
                              DdlParser ddlParser, MySqlValueConverters mySqlValueConverters,
-                             Tables tables, Map<String, SourceTable> sourceTableMap) {
+                             Tables tables, Map<String, SourceTable> sourceTableMap, boolean replicateExistingData) {
     this.context = context;
     this.emitter = emitter;
     this.ddlParser = ddlParser;
     this.mySqlValueConverters = mySqlValueConverters;
     this.tables = tables;
     this.sourceTableMap = sourceTableMap;
+    this.replicateExistingData = replicateExistingData;
   }
 
   @Override
@@ -260,7 +262,8 @@ private void handleDDL(String ddlStatement, Offset recordOffset,
           case DROP_TABLE:
             DdlParserListener.TableDroppedEvent droppedEvent = (DdlParserListener.TableDroppedEvent) event;
             sourceTable = getSourceTable(databaseName, droppedEvent.tableId().table());
-            if (shouldEmitDdlEventForOperation(readAllTables, sourceTable, DDLOperation.Type.DROP_TABLE)) {
+            if (shouldEmitDdlEventForOperation(readAllTables, sourceTable, DDLOperation.Type.DROP_TABLE) &&
+              generateDropEventOnSnapshot(isSnapshot)) {
               ddlEvent = builder.setOperation(DDLOperation.Type.DROP_TABLE)
                 .setTableName(droppedEvent.tableId().table())
                 .build();
@@ -281,7 +284,9 @@ private void handleDDL(String ddlStatement, Offset recordOffset,
             }
             break;
           case DROP_DATABASE:
-            ddlEvent = builder.setOperation(DDLOperation.Type.DROP_DATABASE).build();
+            if (generateDropEventOnSnapshot(isSnapshot)) {
+              ddlEvent = builder.setOperation(DDLOperation.Type.DROP_DATABASE).build();
+            }
             break;
           case CREATE_DATABASE:
             // due to a bug in io.debezium.relational.ddl.AbstractDdlParser#signalDropDatabase
@@ -318,6 +323,19 @@ private void handleDDL(String ddlStatement, Offset recordOffset,
     }
   }
 
+  // Mysql source during snapshotting process generates DROP TABLE and DROP Database
+  // events. If the source is configured to ignore replication of the existing data, most likely target table
+  // exists with the snapshot events and user do not want to re-do snapshotting. Do not generate the DROP
+  // (Table/Database) events in such cases. If user do not want to keep the existing target tables if any, they
+  // will have to delete those tables manually.
+  private boolean generateDropEventOnSnapshot(boolean isSnapshot) {
+    if (!isSnapshot) {
+      // if not snapshot event, generate DROP event as it is part of explicit DDL operation from user
+      return true;
+    }
+    return replicateExistingData;
+  }
+
   private boolean shouldEmitDdlEventForOperation(boolean readAllTables, SourceTable sourceTable, DDLOperation.Type op) {
     return (!sourceTableNotValid(readAllTables, sourceTable)) &&
       (!isDDLOperationBlacklisted(readAllTables, sourceTable, op));

mysql-delta-plugins/widgets/mysql-cdcSource.json

@@ -65,6 +65,22 @@
           "widget-attributes": {
             "default": "UTC"
           }
+        },
+        {
+          "name": "replicateExistingData",
+          "label": "Replicate Existing Data",
+          "widget-type": "toggle",
+          "widget-attributes": {
+            "on": {
+              "value": "true",
+              "label": "Yes"
+            },
+            "off": {
+              "value": "false",
+              "label": "No"
+            },
+            "default": "true"
+          }
         }
       ]
     }

sqlserver-delta-plugins/docs/sqlserver-cdcSource.md

@@ -70,3 +70,7 @@ Plugin Properties
 **Database:** Database to replicate data from.
 
 **JDBC Plugin Name:** Identifier for the SQLServer JDBC driver, which is the name used while uploading the SQLServer JDBC driver.
+
+**Replicate Existing Data:** Whether to replicate existing data from the source database. By default, pipeline will 
+replicate the existing data from source tables. If set to false, any existing data in the source tables will be 
+ignored and only changes happening after the pipeline started will be replicated.

sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerConfig.java

@@ -51,6 +51,12 @@ public class SqlServerConfig extends PluginConfig {
     "JDBC driver.")
   private String jdbcPluginName;
 
+  @Nullable
+  @Description("Whether to replicate existing data from the source database. By default, pipeline will replicate " +
+    "the existing data from source tables. If set to false, any existing data in the source " +
+    "tables will be ignored and only changes happening after the pipeline started will be replicated.")
+  private Boolean replicateExistingData;
+
   public SqlServerConfig(String host, int port, String user, String password,
                          String database, @Nullable String serverTimezone, String jdbcPluginName) {
     this.host = host;
@@ -93,4 +99,8 @@ public String getServerTimezone() {
   public String getJDBCPluginId() {
     return String.format("%s.%s.%s", "sqlserversource", "jbdc", jdbcPluginName);
   }
+
+  public boolean getReplicateExistingData() {
+    return replicateExistingData != null ? replicateExistingData : true;
+  }
 }

sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerEventReader.java

@@ -125,7 +125,8 @@ public void start(Offset offset) {
       .with("database.dbname", databaseName)
       .with("table.whitelist", String.join(",", sourceTableMap.keySet()))
       .with("database.server.name", "dummy") // this is the kafka topic for hosted debezium - it doesn't matter
-      .with("database.serverTimezone", config.getServerTimezone());
+      .with("database.serverTimezone", config.getServerTimezone())
+      .with("snapshot.mode", config.getReplicateExistingData() ? "initial" : "schema_only");
 
     LOG.info("Overriding sql server connector configs with arguments {}", debeziumConnectorConfigs);
     for (Map.Entry<String, String> entry: debeziumConnectorConfigs.entrySet()) {
@@ -161,7 +162,8 @@ public void start(Offset offset) {
       LOG.info("creating new EmbeddedEngine...");
       // Create the engine with this configuration ...
       engine = EmbeddedEngine.create()
-        .notifying(new SqlServerRecordConsumer(context, emitter, databaseName, ddlEventSent, sourceTableMap, offset))
+        .notifying(new SqlServerRecordConsumer(context, emitter, databaseName, ddlEventSent, sourceTableMap, offset,
+                                               config.getReplicateExistingData()))
         .using(debeziumConf)
         .using(new NotifyingCompletionCallback(context))
         .build();

sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerRecordConsumer.java

@@ -54,18 +54,20 @@ public class SqlServerRecordConsumer implements Consumer<SourceRecord> {
   // record tables that already had DDL events sent
   private final Set<String> ddlEventSent;
   private final Map<String, SourceTable> sourceTableMap;
+  private final boolean replicateExistingData;
   private Offset latestOffset;
 
 
   SqlServerRecordConsumer(DeltaSourceContext context, EventEmitter emitter, String databaseName,
                           Set<String> ddlEventSent, Map<String, SourceTable> sourceTableMap,
-    Offset latestOffset) {
+                          Offset latestOffset, boolean replicateExistingData) {
     this.context = context;
     this.emitter = emitter;
     this.databaseName = databaseName;
     this.ddlEventSent = ddlEventSent;
     this.sourceTableMap = sourceTableMap;
     this.latestOffset = latestOffset;
+    this.replicateExistingData = replicateExistingData;
   }
 
   @Override
@@ -162,11 +164,13 @@ public void accept(SourceRecord sourceRecord) {
       builder.setOffset(recordOffset);
 
       try {
-        // try to always drop the table before snapshot the schema.
-        emitter.emit(builder.setOperation(DDLOperation.Type.DROP_TABLE)
-                       .setTableName(tableName)
-                       .setSchemaName(schemaName)
-                       .build());
+        if (replicateExistingData) {
+          // try to always drop the table before snapshot the schema.
+          emitter.emit(builder.setOperation(DDLOperation.Type.DROP_TABLE)
+                         .setTableName(tableName)
+                         .setSchemaName(schemaName)
+                         .build());
+        }
 
         // try to emit create database event before create table event
         emitter.emit(builder.setOperation(DDLOperation.Type.CREATE_DATABASE)

sqlserver-delta-plugins/widgets/sqlserver-cdcSource.json

@@ -60,6 +60,22 @@
           "widget-attributes": {
             "default": "UTC"
           }
+        },
+        {
+          "name": "replicateExistingData",
+          "label": "Replicate Existing Data",
+          "widget-type": "toggle",
+          "widget-attributes": {
+            "on": {
+              "value": "true",
+              "label": "Yes"
+            },
+            "off": {
+              "value": "false",
+              "label": "No"
+            },
+            "default": "true"
+          }
         }
       ]
     }