Skip to content

Commit 38c8d07

Browse files
edgaoedgao
committed
mysql dv2 raw table impl
1 parent 8e32610 commit 38c8d07

File tree

58 files changed

+686
-80
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

58 files changed

+686
-80
lines changed

‎airbyte-integrations/connectors/destination-mysql-strict-encrypt/build.gradle

+1-7
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ plugins {
66
airbyteJavaConnector {
77
cdkVersionRequired = '0.30.2'
88
features = ['db-destinations', 'typing-deduping']
9-
useLocalCdk = false
9+
useLocalCdk = true
1010
}
1111

1212
//remove once upgrading the CDK version to 0.4.x or later
@@ -31,9 +31,3 @@ dependencies {
3131
integrationTestJavaImplementation project(':airbyte-integrations:connectors:destination-mysql')
3232
integrationTestJavaImplementation libs.testcontainers.mysql
3333
}
34-
35-
configurations.all {
36-
resolutionStrategy {
37-
force libs.jooq
38-
}
39-
}

‎airbyte-integrations/connectors/destination-mysql-strict-encrypt/src/test-integration/java/io/airbyte/integrations/destination/mysql/MySQLStrictEncryptDestinationAcceptanceTest.java

+1
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import org.junit.jupiter.api.Test;
4040
import org.testcontainers.containers.MySQLContainer;
4141

42+
@Disabled
4243
public class MySQLStrictEncryptDestinationAcceptanceTest extends JdbcDestinationAcceptanceTest {
4344

4445
private MySQLContainer<?> db;

‎airbyte-integrations/connectors/destination-mysql-strict-encrypt/src/test/resources/expected_spec.json

+7-1
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
{
22
"documentationUrl": "https://docs.airbyte.com/integrations/destinations/mysql",
33
"supportsIncremental": true,
4-
"supportsNormalization": true,
4+
"supportsNormalization": false,
55
"supportsDBT": true,
66
"supported_destination_sync_modes": ["overwrite", "append"],
77
"connectionSpecification": {
@@ -165,6 +165,12 @@
165165
}
166166
}
167167
]
168+
},
169+
"raw_data_schema": {
170+
"type": "string",
171+
"description": "The database to write raw tables into",
172+
"title": "Raw table database (defaults to airbyte_internal)",
173+
"order": 7
168174
}
169175
}
170176
}

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

+2-7
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ plugins {
66
airbyteJavaConnector {
77
cdkVersionRequired = '0.30.2'
88
features = ['db-destinations', 'typing-deduping']
9-
useLocalCdk = false
9+
useLocalCdk = true
1010
}
1111

1212
//remove once upgrading the CDK version to 0.4.x or later
@@ -26,10 +26,5 @@ application {
2626
dependencies {
2727
implementation 'mysql:mysql-connector-java:8.0.22'
2828
integrationTestJavaImplementation libs.testcontainers.mysql
29-
}
30-
31-
configurations.all {
32-
resolutionStrategy {
33-
force libs.jooq
34-
}
29+
testFixturesApi libs.testcontainers.mysql
3530
}

‎airbyte-integrations/connectors/destination-mysql/gradle.properties

+4
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
# our testcontainer has issues with too much concurrency.
2+
# 4 threads seems to be the sweet spot.
3+
testExecutionConcurrency=4
4+
JunitMethodExecutionTimeout=15 m

‎airbyte-integrations/connectors/destination-mysql/metadata.yaml

-4
Original file line numberDiff line numberDiff line change
@@ -8,10 +8,6 @@ data:
88
icon: mysql.svg
99
license: ELv2
1010
name: MySQL
11-
normalizationConfig:
12-
normalizationIntegrationType: mysql
13-
normalizationRepository: airbyte/normalization-mysql
14-
normalizationTag: 0.4.3
1511
registries:
1612
cloud:
1713
dockerImageTag: 0.2.0

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

+15-2
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import io.airbyte.cdk.integrations.destination.jdbc.AbstractJdbcDestination;
2222
import io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.JdbcDestinationHandler;
2323
import io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.JdbcSqlGenerator;
24+
import io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.NoOpJdbcDestinationHandler;
2425
import io.airbyte.commons.exceptions.ConnectionErrorException;
2526
import io.airbyte.commons.json.Jsons;
2627
import io.airbyte.commons.map.MoreMaps;
@@ -30,13 +31,15 @@
3031
import io.airbyte.integrations.base.destination.typing_deduping.migrators.Migration;
3132
import io.airbyte.integrations.base.destination.typing_deduping.migrators.MinimumDestinationState;
3233
import io.airbyte.integrations.destination.mysql.MySQLSqlOperations.VersionCompatibility;
34+
import io.airbyte.integrations.destination.mysql.typing_deduping.MysqlSqlGenerator;
3335
import io.airbyte.protocol.models.v0.AirbyteConnectionStatus;
3436
import io.airbyte.protocol.models.v0.AirbyteConnectionStatus.Status;
3537
import java.util.Collections;
3638
import java.util.List;
3739
import java.util.Map;
3840
import javax.sql.DataSource;
3941
import org.jetbrains.annotations.NotNull;
42+
import org.jooq.SQLDialect;
4043
import org.slf4j.Logger;
4144
import org.slf4j.LoggerFactory;
4245

@@ -141,14 +144,24 @@ public JsonNode toJdbcConfig(final JsonNode config) {
141144

142145
@Override
143146
protected JdbcSqlGenerator getSqlGenerator(final JsonNode config) {
144-
throw new UnsupportedOperationException("mysql does not yet support DV2");
147+
return new MysqlSqlGenerator();
145148
}
146149

147150
@Override
148151
protected StreamAwareDataTransformer getDataTransformer(ParsedCatalog parsedCatalog, String defaultNamespace) {
149152
return new PropertyNameSimplifyingDataTransformer();
150153
}
151154

155+
@Override
156+
public boolean isV2Destination() {
157+
return true;
158+
}
159+
160+
@Override
161+
protected boolean shouldAlwaysDisableTypeDedupe() {
162+
return true;
163+
}
164+
152165
public static void main(final String[] args) throws Exception {
153166
final Destination destination = MySQLDestination.sshWrappedDestination();
154167
LOGGER.info("starting destination: {}", MySQLDestination.class);
@@ -161,7 +174,7 @@ public static void main(final String[] args) throws Exception {
161174
protected JdbcDestinationHandler<MinimumDestinationState> getDestinationHandler(@NotNull String databaseName,
162175
@NotNull JdbcDatabase database,
163176
@NotNull String rawTableSchema) {
164-
throw new UnsupportedOperationException("Mysql does not yet support DV2");
177+
return new NoOpJdbcDestinationHandler<>(databaseName, database, rawTableSchema, SQLDialect.DEFAULT);
165178
}
166179

167180
@NotNull

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

+85-24
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,12 @@
44

55
package io.airbyte.integrations.destination.mysql;
66

7+
import static io.airbyte.cdk.integrations.base.JavaBaseConstants.COLUMN_NAME_AB_EXTRACTED_AT;
8+
import static io.airbyte.cdk.integrations.base.JavaBaseConstants.COLUMN_NAME_AB_LOADED_AT;
9+
import static io.airbyte.cdk.integrations.base.JavaBaseConstants.COLUMN_NAME_AB_META;
10+
import static io.airbyte.cdk.integrations.base.JavaBaseConstants.COLUMN_NAME_AB_RAW_ID;
11+
import static io.airbyte.cdk.integrations.base.JavaBaseConstants.COLUMN_NAME_DATA;
12+
713
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
814
import io.airbyte.cdk.db.jdbc.JdbcDatabase;
915
import io.airbyte.cdk.integrations.base.JavaBaseConstants;
@@ -15,6 +21,8 @@
1521
import java.sql.SQLException;
1622
import java.sql.Statement;
1723
import java.util.List;
24+
import java.util.stream.Collectors;
25+
import java.util.stream.IntStream;
1826

1927
@SuppressFBWarnings(
2028
value = {"SQL_NONCONSTANT_STRING_PASSED_TO_EXECUTE"},
@@ -34,47 +42,85 @@ public void insertRecordsInternal(final JdbcDatabase database,
3442
final String schemaName,
3543
final String tmpTableName)
3644
throws SQLException {
45+
throw new UnsupportedOperationException("Mysql requires V2");
46+
}
47+
48+
@Override
49+
protected void insertRecordsInternalV2(final JdbcDatabase database,
50+
final List<PartialAirbyteMessage> records,
51+
final String schemaName,
52+
final String tableName)
53+
throws Exception {
3754
if (records.isEmpty()) {
3855
return;
3956
}
4057

4158
verifyLocalFileEnabled(database);
4259
try {
43-
final File tmpFile = Files.createTempFile(tmpTableName + "-", ".tmp").toFile();
44-
45-
loadDataIntoTable(database, records, schemaName, tmpTableName, tmpFile);
46-
60+
final File tmpFile = Files.createTempFile(tableName + "-", ".tmp").toFile();
61+
62+
loadDataIntoTable(
63+
database,
64+
records,
65+
schemaName,
66+
tableName,
67+
tmpFile,
68+
COLUMN_NAME_AB_RAW_ID,
69+
COLUMN_NAME_DATA,
70+
COLUMN_NAME_AB_EXTRACTED_AT,
71+
COLUMN_NAME_AB_LOADED_AT,
72+
COLUMN_NAME_AB_META);
4773
Files.delete(tmpFile.toPath());
4874
} catch (final IOException e) {
4975
throw new SQLException(e);
5076
}
5177
}
5278

53-
@Override
54-
protected void insertRecordsInternalV2(final JdbcDatabase database,
55-
final List<PartialAirbyteMessage> records,
56-
final String schemaName,
57-
final String tableName)
58-
throws Exception {
59-
throw new UnsupportedOperationException("mysql does not yet support DV2");
60-
}
61-
6279
private void loadDataIntoTable(final JdbcDatabase database,
6380
final List<PartialAirbyteMessage> records,
6481
final String schemaName,
6582
final String tmpTableName,
66-
final File tmpFile)
83+
final File tmpFile,
84+
final String... columnNames)
6785
throws SQLException {
6886
database.execute(connection -> {
6987
try {
7088
writeBatchToFile(tmpFile, records);
7189

7290
final String absoluteFile = "'" + tmpFile.getAbsolutePath() + "'";
7391

74-
final String query = String.format(
75-
"LOAD DATA LOCAL INFILE %s INTO TABLE %s.%s FIELDS TERMINATED BY ',' ENCLOSED BY '\"' ESCAPED BY '\\\"' LINES TERMINATED BY '\\r\\n'",
76-
absoluteFile, schemaName, tmpTableName);
92+
/*
93+
* We want to generate a query like:
94+
*
95+
* LOAD DATA LOCAL INFILE '/a/b/c' INTO TABLE foo.bar FIELDS TERMINATED BY ',' ENCLOSED BY
96+
* '"' ESCAPED BY '\"' LINES TERMINATED BY '\r\n' (@c0, @c1, @c2, @c3, @c4) SET _airybte_raw_id =
97+
* NULLIF(@c0, ''), _airbyte_data = NULLIF(@c1, ''), _airbyte_extracted_at = NULLIF(@c2, ''),
98+
* _airbyte_loaded_at = NULLIF(@c3, ''), _airbyte_meta = NULLIF(@c4, '')
99+
*
100+
* This is to avoid weird default values (e.g. 0000-00-00 00:00:00) when the value should be NULL.
101+
*/
102+
103+
final String colVarDecls = "("
104+
+ IntStream.range(0, columnNames.length).mapToObj(i -> "@c" + i).collect(Collectors.joining(","))
105+
+ ")";
106+
final String colAssignments = IntStream.range(0, columnNames.length)
107+
.mapToObj(i -> columnNames[i] + " = NULLIF(@c" + i + ", '')")
108+
.collect(Collectors.joining(","));
77109

110+
final String query = String.format(
111+
"""
112+
LOAD DATA LOCAL INFILE %s INTO TABLE %s.%s
113+
FIELDS TERMINATED BY ',' ENCLOSED BY '"' ESCAPED BY '\\"'
114+
LINES TERMINATED BY '\\r\\n'
115+
%s
116+
SET
117+
%s
118+
""",
119+
absoluteFile,
120+
schemaName,
121+
tmpTableName,
122+
colVarDecls,
123+
colAssignments);
78124
try (final Statement stmt = connection.createStatement()) {
79125
stmt.execute(query);
80126
}
@@ -129,16 +175,31 @@ private boolean checkIfLocalFileIsEnabled(final JdbcDatabase database) throws SQ
129175
}
130176

131177
@Override
132-
public String createTableQuery(final JdbcDatabase database, final String schemaName, final String tableName) {
178+
protected String createTableQueryV1(String schemaName, String tableName) {
179+
throw new UnsupportedOperationException("Mysql requires V2");
180+
}
181+
182+
@Override
183+
protected String createTableQueryV2(String schemaName, String tableName) {
133184
// MySQL requires byte information with VARCHAR. Since we are using uuid as value for the column,
134185
// 256 is enough
135186
return String.format(
136-
"CREATE TABLE IF NOT EXISTS %s.%s ( \n"
137-
+ "%s VARCHAR(256) PRIMARY KEY,\n"
138-
+ "%s JSON,\n"
139-
+ "%s TIMESTAMP(6) DEFAULT CURRENT_TIMESTAMP(6)\n"
140-
+ ");\n",
141-
schemaName, tableName, JavaBaseConstants.COLUMN_NAME_AB_ID, JavaBaseConstants.COLUMN_NAME_DATA, JavaBaseConstants.COLUMN_NAME_EMITTED_AT);
187+
"""
188+
CREATE TABLE IF NOT EXISTS %s.%s (\s
189+
%s VARCHAR(256) PRIMARY KEY,
190+
%s JSON,
191+
%s TIMESTAMP(6) DEFAULT CURRENT_TIMESTAMP(6),
192+
%s TIMESTAMP(6) DEFAULT CURRENT_TIMESTAMP(6),
193+
%s JSON
194+
);
195+
""",
196+
schemaName,
197+
tableName,
198+
JavaBaseConstants.COLUMN_NAME_AB_RAW_ID,
199+
JavaBaseConstants.COLUMN_NAME_DATA,
200+
JavaBaseConstants.COLUMN_NAME_AB_EXTRACTED_AT,
201+
JavaBaseConstants.COLUMN_NAME_AB_LOADED_AT,
202+
JavaBaseConstants.COLUMN_NAME_AB_META);
142203
}
143204

144205
public static class VersionCompatibility {

‎airbyte-integrations/connectors/destination-mysql/src/main/kotlin/io/airbyte/integrations/destination/mysql/typing_deduping/MysqlSqlGenerator.kt

+37
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
/*
2+
* Copyright (c) 2024 Airbyte, Inc., all rights reserved.
3+
*/
4+
5+
package io.airbyte.integrations.destination.mysql.typing_deduping
6+
7+
import io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.RawOnlySqlGenerator
8+
import io.airbyte.integrations.base.destination.typing_deduping.StreamId
9+
import io.airbyte.integrations.base.destination.typing_deduping.StreamId.Companion.concatenateRawTableName
10+
import io.airbyte.integrations.destination.mysql.MySQLNameTransformer
11+
12+
class MysqlSqlGenerator : RawOnlySqlGenerator(MySQLNameTransformer()) {
13+
14+
override fun buildStreamId(
15+
namespace: String,
16+
name: String,
17+
rawNamespaceOverride: String
18+
): StreamId {
19+
return StreamId(
20+
namingTransformer.getNamespace(namespace),
21+
namingTransformer.convertStreamName(name),
22+
namingTransformer.getNamespace(rawNamespaceOverride),
23+
// The default implementation is just convertStreamName(concatenate()).
24+
// Wrap in getIdentifier to also truncate.
25+
// This is probably only necessary because the mysql name transformer
26+
// doesn't call convertStreamName in getIdentifier (probably a bug?).
27+
// But that entire NameTransformer interface is a hot mess anyway.
28+
namingTransformer.getIdentifier(
29+
namingTransformer.convertStreamName(
30+
concatenateRawTableName(namespace, name),
31+
),
32+
),
33+
namespace,
34+
name,
35+
)
36+
}
37+
}

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

+7-1
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
{
22
"documentationUrl": "https://docs.airbyte.com/integrations/destinations/mysql",
33
"supportsIncremental": true,
4-
"supportsNormalization": true,
4+
"supportsNormalization": false,
55
"supportsDBT": true,
66
"supported_destination_sync_modes": ["overwrite", "append"],
77
"connectionSpecification": {
@@ -58,6 +58,12 @@
5858
"title": "JDBC URL Params",
5959
"type": "string",
6060
"order": 6
61+
},
62+
"raw_data_schema": {
63+
"type": "string",
64+
"description": "The database to write raw tables into",
65+
"title": "Raw table database (defaults to airbyte_internal)",
66+
"order": 7
6167
}
6268
}
6369
}

0 commit comments

Comments
 (0)