[Source-mysql] : Remove mysql initial source operations

airbyte-integrations/connectors/source-mysql/metadata.yaml

@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
-  dockerImageTag: 3.3.17
+  dockerImageTag: 3.3.18
   dockerRepository: airbyte/source-mysql
   documentationUrl: https://docs.airbyte.com/integrations/sources/mysql
   githubIssueLabel: source-mysql

airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/MySqlSourceOperations.java

@@ -46,6 +46,7 @@
 import com.mysql.cj.result.Field;
 import io.airbyte.cdk.db.SourceOperations;
 import io.airbyte.cdk.db.jdbc.AbstractJdbcCompatibleSourceOperations;
+import io.airbyte.integrations.source.mysql.initialsync.CdcMetadataInjector;
 import io.airbyte.protocol.models.JsonSchemaType;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
@@ -54,6 +55,7 @@
 import java.time.LocalDateTime;
 import java.time.OffsetDateTime;
 import java.time.format.DateTimeParseException;
+import java.util.Optional;
 import java.util.Set;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -66,6 +68,28 @@ public class MySqlSourceOperations extends AbstractJdbcCompatibleSourceOperation
       FLOAT, FLOAT_UNSIGNED, DOUBLE, DOUBLE_UNSIGNED, DECIMAL, DECIMAL_UNSIGNED, DATE, DATETIME, TIMESTAMP,
       TIME, YEAR, VARCHAR, TINYTEXT, TEXT, MEDIUMTEXT, LONGTEXT);
 
+  private final Optional<CdcMetadataInjector> metadataInjector;
+
+  public MySqlSourceOperations() {
+    super();
+    this.metadataInjector = Optional.empty();
+  }
+
+  public MySqlSourceOperations(final Optional<CdcMetadataInjector> metadataInjector) {
+    super();
+    this.metadataInjector = metadataInjector;
+  }
+
+  @Override
+  public JsonNode rowToJson(final ResultSet queryContext) throws SQLException {
+    final ObjectNode jsonNode = (ObjectNode) super.rowToJson(queryContext);
+    if (!metadataInjector.isPresent()) {
+      return jsonNode;
+    }
+    metadataInjector.get().inject(jsonNode);
+    return jsonNode;
+  }
+
   /**
    * @param colIndex 1-based column index.
    */

airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/initialsync/CdcMetadataInjector.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.source.mysql.initialsync;
+
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.airbyte.integrations.source.mysql.cdc.MySqlCdcConnectorMetadataInjector;
+import io.airbyte.integrations.source.mysql.cdc.MySqlDebeziumStateUtil.MysqlDebeziumStateAttributes;
+
+public class CdcMetadataInjector {
+
+  private final String transactionTimestamp;
+  private final MysqlDebeziumStateAttributes stateAttributes;
+  private final MySqlCdcConnectorMetadataInjector metadataInjector;
+
+  public CdcMetadataInjector(final String transactionTimestamp,
+                             final MysqlDebeziumStateAttributes stateAttributes,
+                             final MySqlCdcConnectorMetadataInjector metadataInjector) {
+    this.transactionTimestamp = transactionTimestamp;
+    this.stateAttributes = stateAttributes;
+    this.metadataInjector = metadataInjector;
+  }
+
+  public void inject(final ObjectNode record) {
+    metadataInjector.addMetaDataToRowsFetchedOutsideDebezium(record, transactionTimestamp, stateAttributes);
+  }
+
+}

airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/initialsync/MySqlInitialReadUtil.java

@@ -32,6 +32,7 @@
 import io.airbyte.commons.util.AutoCloseableIterator;
 import io.airbyte.commons.util.AutoCloseableIterators;
 import io.airbyte.integrations.source.mysql.MySqlQueryUtils;
+import io.airbyte.integrations.source.mysql.MySqlSourceOperations;
 import io.airbyte.integrations.source.mysql.cdc.MySqlCdcConnectorMetadataInjector;
 import io.airbyte.integrations.source.mysql.cdc.MySqlCdcPosition;
 import io.airbyte.integrations.source.mysql.cdc.MySqlCdcProperties;
@@ -40,7 +41,6 @@
 import io.airbyte.integrations.source.mysql.cdc.MySqlCdcTargetPosition;
 import io.airbyte.integrations.source.mysql.cdc.MySqlDebeziumStateUtil;
 import io.airbyte.integrations.source.mysql.cdc.MySqlDebeziumStateUtil.MysqlDebeziumStateAttributes;
-import io.airbyte.integrations.source.mysql.initialsync.MySqlInitialLoadSourceOperations.CdcMetadataInjector;
 import io.airbyte.integrations.source.mysql.internal.models.CursorBasedStatus;
 import io.airbyte.integrations.source.mysql.internal.models.PrimaryKeyLoadStatus;
 import io.airbyte.protocol.models.CommonField;
@@ -141,8 +141,8 @@ public static List<AutoCloseableIterator<AirbyteMessage>> getCdcReadIterators(fi
               stateToBeUsed, catalog);
       final MysqlDebeziumStateAttributes stateAttributes = MySqlDebeziumStateUtil.getStateAttributesFromDB(database);
 
-      final MySqlInitialLoadSourceOperations sourceOperations =
-          new MySqlInitialLoadSourceOperations(
+      final MySqlSourceOperations sourceOperations =
+          new MySqlSourceOperations(
               Optional.of(new CdcMetadataInjector(emittedAt.toString(), stateAttributes, metadataInjector)));
       final MySqlInitialLoadHandler initialLoadHandler = new MySqlInitialLoadHandler(sourceConfig, database,
           sourceOperations,

docs/integrations/sources/mysql.md

@@ -223,6 +223,7 @@ Any database or table encoding combination of charset and collation is supported
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                         |
 |:--------|:-----------|:-----------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------------|
+| 3.3.18  | 2024-04-15 | [36919](https://github.com/airbytehq/airbyte/pull/36919)   | Refactor source operations.                                                                                                                     |
 | 3.3.17  | 2024-04-10 | [36919](https://github.com/airbytehq/airbyte/pull/36919)   | Fix a bug in conversion of null values.                                                                                                         |
 | 3.3.16  | 2024-04-05 | [36872](https://github.com/airbytehq/airbyte/pull/36872)   | Update to connector's metadat definition.                                                                                                       |
 | 3.3.15  | 2024-04-05 | [36577](https://github.com/airbytehq/airbyte/pull/36577)   | Config error will not send out system trace message                                                                                             |