Migrate Postgres and MySql to use new JdbcSource

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/435bb9a5-7887-4809-aa58-28c27df0d7ad.json

@@ -2,6 +2,6 @@
   "sourceDefinitionId": "435bb9a5-7887-4809-aa58-28c27df0d7ad",
   "name": "MySQL",
   "dockerRepository": "airbyte/source-mysql",
-  "dockerImageTag": "0.1.5",
+  "dockerImageTag": "0.1.6",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/mysql"
 }

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/decd338e-5647-4c0b-adf4-da0e75f5a750.json

@@ -2,6 +2,6 @@
   "sourceDefinitionId": "decd338e-5647-4c0b-adf4-da0e75f5a750",
   "name": "Postgres",
   "dockerRepository": "airbyte/source-postgres",
-  "dockerImageTag": "0.1.6",
+  "dockerImageTag": "0.1.7",
   "documentationUrl": "https://hub.docker.com/r/airbyte/source-postgres"
 }

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -26,7 +26,7 @@
 - sourceDefinitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
   name: Postgres
   dockerRepository: airbyte/source-postgres
-  dockerImageTag: 0.1.6
+  dockerImageTag: 0.1.7
   documentationUrl: https://hub.docker.com/r/airbyte/source-postgres
 - sourceDefinitionId: cd42861b-01fc-4658-a8ab-5d11d0510f01
   name: Recurly
@@ -51,7 +51,7 @@
 - sourceDefinitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
   name: MySQL
   dockerRepository: airbyte/source-mysql
-  dockerImageTag: 0.1.5
+  dockerImageTag: 0.1.6
   documentationUrl: https://docs.airbyte.io/integrations/sources/mysql
 - sourceDefinitionId: 2470e835-feaf-4db6-96f3-70fd645acc77
   name: Salesforce

airbyte-integrations/connectors/source-mysql/Dockerfile

@@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
 
 RUN tar xf ${APPLICATION}.tar --strip-components=1
 
-LABEL io.airbyte.version=0.1.5
+LABEL io.airbyte.version=0.1.6
 LABEL io.airbyte.name=airbyte/source-mysql

airbyte-integrations/connectors/source-mysql/build.gradle

@@ -2,7 +2,6 @@ plugins {
     id 'application'
     id 'airbyte-docker'
     id 'airbyte-integration-test-java'
-    // todo: needs standard source test
 }
 
 application {
@@ -16,13 +15,15 @@ dependencies {
     implementation project(':airbyte-integrations:connectors:source-jdbc')
 
     implementation 'mysql:mysql-connector-java:8.0.22'
-    testImplementation 'org.testcontainers:mysql:1.15.1'
-
     implementation 'org.apache.commons:commons-lang3:3.11'
 
-    integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-source-test')
+    testImplementation testFixtures(project(':airbyte-integrations:connectors:source-jdbc'))
+
+    testImplementation 'org.apache.commons:commons-lang3:3.11'
     testImplementation 'org.testcontainers:mysql:1.15.1'
 
+    integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-source-test')
+
     implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
     integrationTestJavaImplementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
 }

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

@@ -0,0 +1,53 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.integrations.source.mysql;
+
+import io.airbyte.db.jdbc.JdbcStreamingQueryConfiguration;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+
+public class MySqlJdbcStreamingQueryConfiguration implements JdbcStreamingQueryConfiguration {
+
+  @Override
+  public void accept(Connection connection, PreparedStatement preparedStatement) throws SQLException {
+    // This is only respected if "useCursorFetch=true" is set in the connection. See the "resultset"
+    // section the MySql docs for more details.
+    // https://dev.mysql.com/doc/connector-j/8.0/en/connector-j-reference-implementation-notes.html.
+    // When using this approach MySql creates a temporary table which may have some effect on db
+    // performance.
+    // e.g. conn = DriverManager.getConnection("jdbc:mysql://localhost/?useCursorFetch=true", "user",
+    // "s3cr3t");
+    // We set userCursorFetch in MySqlSource.
+    connection.setAutoCommit(false);
+    preparedStatement.setFetchSize(1000);
+    // If for some reason, you cannot set useCursorFetch in the connection, fall back on this
+    // implementation below. It fetches records one at a time, which while inefficient, at least does
+    // not risk OOM.
+    // connection.setAutoCommit(false);
+    // preparedStatement.setFetchSize(Integer.MIN_VALUE);
+  }
+
+}

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

@@ -29,25 +29,27 @@
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.base.IntegrationRunner;
 import io.airbyte.integrations.base.Source;
-import io.airbyte.integrations.source.jdbc.AbstractJooqSource;
-import java.util.List;
-import org.jooq.SQLDialect;
+import io.airbyte.integrations.source.jdbc.AbstractJdbcSource;
+import java.util.Set;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class MySqlSource extends AbstractJooqSource implements Source {
+public class MySqlSource extends AbstractJdbcSource implements Source {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(MySqlSource.class);
 
+  public static final String DRIVER_CLASS = "com.mysql.cj.jdbc.Driver";
+
   public MySqlSource() {
-    super("com.mysql.cj.jdbc.Driver", SQLDialect.MYSQL);
+    super(DRIVER_CLASS, new MySqlJdbcStreamingQueryConfiguration());
   }
 
   @Override
   public JsonNode toJdbcConfig(JsonNode config) {
     ImmutableMap.Builder<Object, Object> configBuilder = ImmutableMap.builder()
         .put("username", config.get("username").asText())
-        .put("jdbc_url", String.format("jdbc:mysql://%s:%s/%s",
+        // see MySqlJdbcStreamingQueryConfiguration for more context on why useCursorFetch=true is needed.
+        .put("jdbc_url", String.format("jdbc:mysql://%s:%s/%s?useCursorFetch=true",
             config.get("host").asText(),
             config.get("port").asText(),
             config.get("database").asText()));
@@ -60,8 +62,8 @@ public JsonNode toJdbcConfig(JsonNode config) {
   }
 
   @Override
-  protected List<String> getExcludedInternalSchemas() {
-    return List.of(
+  public Set<String> getExcludedInternalSchemas() {
+    return Set.of(
         "information_schema",
         "mysql",
         "performance_schema",

airbyte-integrations/connectors/source-mysql/src/test-integration/java/io/airbyte/integrations/source/mysql/MySqlIntegrationTest.java → airbyte-integrations/connectors/source-mysql/src/test-integration/java/io/airbyte/integrations/source/mysql/MySqlSourceStandardTest.java

@@ -45,7 +45,7 @@
 import org.jooq.SQLDialect;
 import org.testcontainers.containers.MySQLContainer;
 
-public class MySqlIntegrationTest extends StandardSourceTest {
+public class MySqlSourceStandardTest extends StandardSourceTest {
 
   private static final String STREAM_NAME = "id_and_name";
   private static final String STREAM_NAME2 = "public.starships";