destination: add implementation for mysql as destination #3242

subodh1810 · 2021-05-05T15:05:32Z

What

Issue : #1483

How

The PR introduces module for mysql-destination. This mainly tackles loading data into the mysql tables without making too many changes to the existing architecture/code. We are using LOAD DATA INFILE mechanism to load data into mysql. This is the recommended way of doing it. Ref : https://dev.mysql.com/doc/refman/8.0/en/load-data.html

This implementation is not compatible with normalization. The MySQL integration tests pass locally.

Pre-merge Checklist

Run integration tests
Publish Docker images

Recommended reading order

test.java
component.ts
the rest

ChristopheDuong · 2021-05-05T15:07:41Z

/test connector=destination-mysql

🕑 destination-mysql https://github.com/airbytehq/airbyte/actions/runs/813853351
❌ destination-mysql https://github.com/airbytehq/airbyte/actions/runs/813853351

subodh1810 · 2021-05-05T15:14:22Z

...ava/io/airbyte/integrations/destination/buffered_stream_consumer/BufferedStreamConsumer.java

@@ -79,7 +79,7 @@

  private static final long GRACEFUL_SHUTDOWN_MINUTES = 5L;
  private static final int MIN_RECORDS = 500;
-  private static final int BATCH_SIZE = 10000;
+  private static final int BATCH_SIZE = 500_000;


Since both MySQL and Postgres are now using bulk load mechanism, the batch size should be increased. I dont have any solid reasoning behind 500K but based on few stack overflows and other articles like https://www.citusdata.com/blog/2017/11/08/faster-bulk-loading-in-postgresql-with-copy/ it seems like bulk loads are super efficient for around 1M records and perhaps we can increase 500K to 1M as well after observing how this performs.

There are two problems with this change:

We usually aren't hitting the 10k batch limit consistently. For this to have the intended effect we probably need to increase MIN_RECORDS as well. Otherwise we would likely continue writing 1-5k records anyways.

We need this record size to be more byte-size based than row based. Some use cases may have large rows. Even in a case where you have 500k 2mb rows we'd run into serious disk space issues on some installations.

I do agree that we need to increase this batch size and for most common cases 500k-1M makes perfect sense; it's just more work than changing this value unfortunately.

Also, just fyi, after making a change to BufferedStreamConsumer we would need to bump and deploy a version for all downstream deestinations (meilisearch and all jdbc-based destinations).

Subodh, let's leave this at 10000 for now.

@jrhizor any idea if this work is already planned/noted or should we create an issue to tackle making this byte based?

Reverted it to 10K

jrhizor

Generally looks good.

We'll also need to add a page to the docs for this destination.

jrhizor · 2021-05-05T16:22:24Z

...jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/JdbcBufferedConsumerFactory.java

+      String tmpTableName = Names.concatQuotedNames("_airbyte_" + now.toEpochMilli() + "_", tableName);
+
+      //This is for MySQL destination, the table names cant have more than 64 characters.
+      if (tmpTableName.length() > 64) {


This needs collision handling in case the 0-31 and 32-63 are the same for different tables.

Actually @ChristopheDuong mentioned that this is a more general problem. Beyond the scope of this PR.

Yeap. Discussed this with Subodh. Will create a follow up ticket to track this.

yes, a related issue on Postgres is here:
https://github.com/airbytehq/airbyte/issues/2948

We'd probably need to handle these table name truncations and collisions in a common class, maybe in StandardNameTransformer as each destination defines some character limits anyway

Cool, Subodh, can you leave a TODO here and reference the issue?

e.g.

TODO (#2948): Refactor into StandardNameTransformed

or something of the sort so we remember when we get to this.

jrhizor · 2021-05-05T16:23:36Z

airbyte-db/src/main/java/io/airbyte/db/jdbc/JdbcDatabase.java

@@ -50,6 +50,18 @@ default void execute(String sql) throws SQLException {
    execute(connection -> connection.createStatement().execute(sql));
  }

+  default void executeWithinTransaction(List<String> queries) throws SQLException {


this cleans things up a lot 👍

jrhizor · 2021-05-05T16:24:53Z

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

+    implementation project(':airbyte-integrations:connectors:destination-jdbc')
+
+    // https://mvnrepository.com/artifact/mysql/mysql-connector-java
+    implementation group: 'mysql', name: 'mysql-connector-java', version: '8.0.22'


Suggested change

implementation group: 'mysql', name: 'mysql-connector-java', version: '8.0.22'

implementation 'mysql:mysql-connector-java:8.0.22'

We generally try to use this format

jrhizor · 2021-05-05T16:25:30Z

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

+    implementation project(':airbyte-protocol:models')
+    implementation project(':airbyte-integrations:connectors:destination-jdbc')
+
+    // https://mvnrepository.com/artifact/mysql/mysql-connector-java


nit: not really necessary to include this since all of our public mvn dependencies can be looked up the same way

jrhizor · 2021-05-05T16:25:53Z

airbyte-integrations/connectors/destination-mysql/Dockerfile

+
+RUN tar xf ${APPLICATION}.tar --strip-components=1
+
+LABEL io.airbyte.version=0.3.0


Suggested change

LABEL io.airbyte.version=0.3.0

LABEL io.airbyte.version=0.1.0

jrhizor · 2021-05-05T16:36:14Z

...tination-mysql/src/main/java/io/airbyte/integrations/destination/mysql/MySQLDestination.java

+    try (final JdbcDatabase database = getDatabase(config)) {
+      String outputSchema = namingResolver.getIdentifier(config.get("database").asText());
+      attemptSQLCreateAndDropTableOperations(outputSchema, database, namingResolver, sqlOperations);
+      boolean localFileEnabled = ((MySQLSqlOperations) sqlOperations)


Can this check go inside tryEnableLocalFile? That would DRY up this and insertRecords

jrhizor · 2021-05-05T16:37:00Z

...tination-mysql/src/main/java/io/airbyte/integrations/destination/mysql/MySQLDestination.java

+      boolean localFileEnabled = ((MySQLSqlOperations) sqlOperations)
+          .checkIfLocalFileIsEnabled(database);
+      if (!localFileEnabled) {
+        ((MySQLSqlOperations) sqlOperations).tryEnableLocalFile(database);


nit: could re-use a single casted version instead of casting each usage

jrhizor · 2021-05-05T16:38:04Z

airbyte-integrations/connectors/destination-mysql/src/main/resources/spec.json

+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "title": "MySQL Destination Spec",
+    "type": "object",
+    "required": ["host", "port", "username", "database", "password"],


Suggested change

"required": ["host", "port", "username", "database", "password"],

"required": ["host", "port", "username", "database"],

I don't think password is actually required? You already have checks to see if it's set before using it elsewhere.

Jared, when is this true? When a DB doesn't set a password?

Yeah. We do this for other dbs as well. Mostly allows users to test if they have a local db they didn't add a password for.

jrhizor · 2021-05-05T16:43:05Z

...nation-mysql/src/main/java/io/airbyte/integrations/destination/mysql/MySQLSqlOperations.java

+      try (Statement statement = connection.createStatement()) {
+        statement.execute("set global local_infile=true");
+      } catch (Exception e) {
+        throw new RuntimeException("local_infile attribute could not be enabled", e);


It seems like if people are using an Airbyte-specific user it's unlikely it will have permissions to do this.

We should provide as much context as possible in the RuntimeException as possible so it's clear what they need to do.

Something like:

The DB user X was unable to set the local_infile attribute on the MySQL server. As an admin user, you will need to run "SET GLOBAL local_infile = true" before syncing data with Airbyte.

We'll also need to detail this requirement in the docs.

jrhizor · 2021-05-05T16:48:27Z

...ava/io/airbyte/integrations/destination/buffered_stream_consumer/BufferedStreamConsumer.java

@@ -79,7 +79,7 @@

  private static final long GRACEFUL_SHUTDOWN_MINUTES = 5L;
  private static final int MIN_RECORDS = 500;
-  private static final int BATCH_SIZE = 10000;
+  private static final int BATCH_SIZE = 500_000;


There are two problems with this change:

We usually aren't hitting the 10k batch limit consistently. For this to have the intended effect we probably need to increase MIN_RECORDS as well. Otherwise we would likely continue writing 1-5k records anyways.

We need this record size to be more byte-size based than row based. Some use cases may have large rows. Even in a case where you have 500k 2mb rows we'd run into serious disk space issues on some installations.

I do agree that we need to increase this batch size and for most common cases 500k-1M makes perfect sense; it's just more work than changing this value unfortunately.

Also, just fyi, after making a change to BufferedStreamConsumer we would need to bump and deploy a version for all downstream deestinations (meilisearch and all jdbc-based destinations).

michel-tricot · 2021-05-06T06:05:04Z

...rc/test-integration/java/io.airbyte.integrations.destination.mysql/MySQLIntegrationTest.java

@@ -0,0 +1,128 @@
+/*


the path of this file is incorrect

davinchia · 2021-05-06T06:48:47Z

/test connector=destination-mysql

🕑 destination-mysql https://github.com/airbytehq/airbyte/actions/runs/815923454
❌ destination-mysql https://github.com/airbytehq/airbyte/actions/runs/815923454

subodh1810 · 2021-05-06T08:30:58Z

/test connector=destination-mysql

🕑 destination-mysql https://github.com/airbytehq/airbyte/actions/runs/816185443
✅ destination-mysql https://github.com/airbytehq/airbyte/actions/runs/816185443

davinchia

Quick work! Generally makes sense!

Jared's comments aside, some minor comments around readability and small code tweaks.

One thing came to mind: since we are modifying the JdbcBufferedConsumerFactory, let's also run the integration test for Postgres, Redshift and Snowflake. All these destinations use this class under the hood so it's good to double check.

davinchia · 2021-05-06T07:58:26Z

...ion-jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/AbstractJdbcDestination.java

-  private final String driverClass;
-  private final NamingConventionTransformer namingResolver;
-  private final SqlOperations sqlOperations;
+  protected final String driverClass;


nit: either use getter methods or the @VisibleForTesting. I prefer the getter methods, but either one works.

davinchia · 2021-05-06T08:00:23Z

...jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/JdbcBufferedConsumerFactory.java

+      String tmpTableName = Names.concatQuotedNames("_airbyte_" + now.toEpochMilli() + "_", tableName);
+
+      //This is for MySQL destination, the table names cant have more than 64 characters.
+      if (tmpTableName.length() > 64) {


Yeap. Discussed this with Subodh. Will create a follow up ticket to track this.

...s/destination-jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/SqlOperations.java

davinchia · 2021-05-06T08:04:45Z

airbyte-protocol/models/src/main/java/io/airbyte/protocol/models/CatalogHelpers.java

@@ -157,12 +157,4 @@ public static JsonNode fieldsToJsonSchema(List<Field> fields) {
    return allFieldNames;
  }

-  public static Set<String> getStreamNames(ConfiguredAirbyteCatalog catalog) {


oops sorry this is me when I formatted the files. I was going to get rid of it in a follow up PR so up to you to leave this here or remove it.

it's not used any where in our code today.

Its fine. Lets keep it

davinchia · 2021-05-06T08:06:30Z

...jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/JdbcBufferedConsumerFactory.java

-      final String tmpTableName = Names.concatQuotedNames("_airbyte_" + now.toEpochMilli() + "_", tableName);
+      String tmpTableName = Names.concatQuotedNames("_airbyte_" + now.toEpochMilli() + "_", tableName);
+
+      // This is for MySQL destination, the table names cant have more than 64 characters.


Suggested change

// This is for MySQL destination, the table names cant have more than 64 characters.

// This is for MySQL destination, the table names can't have more than 64 characters.

davinchia · 2021-05-06T08:08:44Z

...tination-mysql/src/main/java/io/airbyte/integrations/destination/mysql/MySQLDestination.java

+  @Override
+  public AirbyteConnectionStatus check(JsonNode config) {
+    try (final JdbcDatabase database = getDatabase(config)) {
+      String outputSchema = namingResolver.getIdentifier(config.get("database").asText());


MySQL doesn't have the schema concept. There are only databases.

davinchia · 2021-05-06T08:10:52Z

airbyte-integrations/connectors/destination-mysql/src/main/resources/spec.json

+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "title": "MySQL Destination Spec",
+    "type": "object",
+    "required": ["host", "port", "username", "database", "password"],


Jared, when is this true? When a DB doesn't set a password?

...nation-mysql/src/main/java/io/airbyte/integrations/destination/mysql/MySQLSqlOperations.java

davinchia · 2021-05-06T08:41:31Z

...ion-jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/AbstractJdbcDestination.java

-  private final SqlOperations sqlOperations;
+  protected final String driverClass;
+  protected final NamingConventionTransformer namingResolver;
+  protected final SqlOperations sqlOperations;


I don't think sqlOperations needs to be protected?

subodh1810 · 2021-05-06T10:31:40Z

/test connector=destination-mysql

🕑 destination-mysql https://github.com/airbytehq/airbyte/actions/runs/816551452
✅ destination-mysql https://github.com/airbytehq/airbyte/actions/runs/816551452

subodh1810 · 2021-05-06T10:31:48Z

/test connector=destination-postgres

🕑 destination-postgres https://github.com/airbytehq/airbyte/actions/runs/816552168
✅ destination-postgres https://github.com/airbytehq/airbyte/actions/runs/816552168

davinchia

LGTM!

Let's make sure the additional tests pass first before merging. I'd also want to let @jrhizor sign off since he started the original review.

subodh1810 · 2021-05-06T10:54:45Z

/test connector=destination-redshift

🕑 destination-redshift https://github.com/airbytehq/airbyte/actions/runs/816578825
❌ destination-redshift https://github.com/airbytehq/airbyte/actions/runs/816578825

subodh1810 · 2021-05-06T10:54:55Z

/test connector=destination-snowflake

🕑 destination-snowflake https://github.com/airbytehq/airbyte/actions/runs/816579071
✅ destination-snowflake https://github.com/airbytehq/airbyte/actions/runs/816579071

subodh1810 · 2021-05-06T11:32:59Z

/test connector=destination-redshift

🕑 destination-redshift https://github.com/airbytehq/airbyte/actions/runs/816673930
✅ destination-redshift https://github.com/airbytehq/airbyte/actions/runs/816673930

…s required

subodh1810 · 2021-05-06T18:02:11Z

/test connector=destination-mysql

🕑 destination-mysql https://github.com/airbytehq/airbyte/actions/runs/817781018
✅ destination-mysql https://github.com/airbytehq/airbyte/actions/runs/817781018

jrhizor

Looks great!

Either in this PR before merging or in a separate PR, you'll still need to:

add docs
add mysql to destination_definitions.yaml and run ./gradlew generateSeed to make this connector available in the UI by default.

subodh1810 · 2021-05-07T06:14:42Z

@jrhizor we are going to create a new PR for documentation and other changes

destination: add implementation for mysql as destination

f1b070f

subodh1810 requested review from ChristopheDuong, cgardens and davinchia May 5, 2021 15:05

subodh1810 self-assigned this May 5, 2021

auto-assign bot requested a review from jrhizor May 5, 2021 15:05

subodh1810 mentioned this pull request May 5, 2021

destination: mysql implementation #3238

Closed

2 tasks

subodh1810 commented May 5, 2021

View reviewed changes

jrhizor suggested changes May 5, 2021

View reviewed changes

Fix formatting errors.

1aa3314

michel-tricot reviewed May 6, 2021

View reviewed changes

subodh1810 added 2 commits May 6, 2021 13:45

address review comments + fix flaky test

d0ff264

fix formatting

d77b1eb

subodh1810 requested review from jrhizor and michel-tricot May 6, 2021 08:30

davinchia requested changes May 6, 2021

View reviewed changes

address Davin's review comments

a4cb890

subodh1810 requested a review from davinchia May 6, 2021 10:24

davinchia approved these changes May 6, 2021

View reviewed changes

add missing todo

775210c

Merge branch 'master' into subodh/destination-mysql

ea4ca83

enable namespace test + only provide test user the minimum permission…

ecaad3a

…s required

jrhizor approved these changes May 6, 2021

View reviewed changes

subodh1810 merged commit 601ea5e into master May 7, 2021

subodh1810 deleted the subodh/destination-mysql branch May 7, 2021 06:14

subodh1810 mentioned this pull request May 7, 2021

mysql destination docs #3275

Merged

2 tasks

karinakuz added connectors/destinations-warehouse connectors/destination/snowflake labels Jan 18, 2022

	implementation group: 'mysql', name: 'mysql-connector-java', version: '8.0.22'
	implementation 'mysql:mysql-connector-java:8.0.22'


		RUN tar xf ${APPLICATION}.tar --strip-components=1

		LABEL io.airbyte.version=0.3.0

	LABEL io.airbyte.version=0.3.0
	LABEL io.airbyte.version=0.1.0

	"required": ["host", "port", "username", "database", "password"],
	"required": ["host", "port", "username", "database"],

	// This is for MySQL destination, the table names cant have more than 64 characters.
	// This is for MySQL destination, the table names can't have more than 64 characters.

destination: add implementation for mysql as destination #3242

destination: add implementation for mysql as destination #3242

Conversation

subodh1810 commented May 5, 2021 • edited Loading

What

How

Pre-merge Checklist

Recommended reading order

ChristopheDuong commented May 5, 2021 • edited by github-actions bot Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davinchia May 6, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jrhizor left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davinchia commented May 6, 2021 • edited by github-actions bot Loading

subodh1810 commented May 6, 2021 • edited by github-actions bot Loading

davinchia left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

subodh1810 commented May 6, 2021 • edited by github-actions bot Loading

subodh1810 commented May 6, 2021 • edited by github-actions bot Loading

davinchia left a comment

Choose a reason for hiding this comment

subodh1810 commented May 6, 2021 • edited by github-actions bot Loading

subodh1810 commented May 6, 2021 • edited by github-actions bot Loading

subodh1810 commented May 6, 2021 • edited by github-actions bot Loading

subodh1810 commented May 6, 2021 • edited by github-actions bot Loading

jrhizor left a comment

Choose a reason for hiding this comment

subodh1810 commented May 7, 2021

subodh1810 commented May 5, 2021 •

edited

Loading

ChristopheDuong commented May 5, 2021 •

edited by github-actions bot

Loading

davinchia May 6, 2021 •

edited

Loading

davinchia commented May 6, 2021 •

edited by github-actions bot

Loading

subodh1810 commented May 6, 2021 •

edited by github-actions bot

Loading

davinchia left a comment •

edited

Loading

subodh1810 commented May 6, 2021 •

edited by github-actions bot

Loading

subodh1810 commented May 6, 2021 •

edited by github-actions bot

Loading

subodh1810 commented May 6, 2021 •

edited by github-actions bot

Loading

subodh1810 commented May 6, 2021 •

edited by github-actions bot

Loading

subodh1810 commented May 6, 2021 •

edited by github-actions bot

Loading

subodh1810 commented May 6, 2021 •

edited by github-actions bot

Loading