Skip to content

Commit 54b0a7b

Browse files
jbfbelljbfbell
authored
MSSQL remove normalization (#36050)
1 parent 8f6036e commit 54b0a7b

File tree

15 files changed

+243
-58
lines changed

15 files changed

+243
-58
lines changed

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

+3-1
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,12 @@ plugins {
44
}
55

66
airbyteJavaConnector {
7-
cdkVersionRequired = '0.2.0'
7+
cdkVersionRequired = '0.30.2'
88
features = [
99
'db-sources', // required for tests
1010
'db-destinations',
11+
's3-destinations',
12+
'typing-deduping'
1113
]
1214
useLocalCdk = false
1315
}

‎airbyte-integrations/connectors/destination-mssql-strict-encrypt/metadata.yaml

+11-5
Original file line numberDiff line numberDiff line change
@@ -7,17 +7,23 @@ data:
77
connectorSubtype: database
88
connectorType: destination
99
definitionId: d4353156-9217-4cad-8dd7-c108fd4f74cf
10-
dockerImageTag: 0.2.0
10+
dockerImageTag: 1.0.0
1111
dockerRepository: airbyte/destination-mssql-strict-encrypt
1212
githubIssueLabel: destination-mssql
1313
icon: mssql.svg
1414
license: ELv2
1515
name: MS SQL Server
16-
normalizationConfig:
17-
normalizationIntegrationType: mssql
18-
normalizationRepository: airbyte/normalization-mssql
19-
normalizationTag: 0.4.1
2016
releaseStage: alpha
17+
releases:
18+
breakingChanges:
19+
1.0.0:
20+
upgradeDeadline: "2024-05-25"
21+
message: >
22+
This version removes the option to use "normalization" with MSSQL. It also changes
23+
the schema and database of Airbyte's "raw" tables to be compatible with the new
24+
[Destinations V2](https://docs.airbyte.com/release_notes/upgrading_to_destinations_v2/#what-is-destinations-v2)
25+
format. These changes will likely require updates to downstream dbt / SQL models.
26+
Selecting `Upgrade` will upgrade **all** connections using this destination at their next sync.
2127
documentationUrl: https://docs.airbyte.com/integrations/destinations/mssql
2228
supportsDbt: true
2329
tags:

‎airbyte-integrations/connectors/destination-mssql-strict-encrypt/src/test-integration/java/io/airbyte/integrations/destination/mssql_strict_encrypt/MssqlStrictEncryptDestinationAcceptanceTest.java

+3-1
Original file line numberDiff line numberDiff line change
@@ -30,9 +30,11 @@
3030
import org.jooq.DSLContext;
3131
import org.junit.jupiter.api.AfterAll;
3232
import org.junit.jupiter.api.BeforeAll;
33+
import org.junit.jupiter.api.Disabled;
3334
import org.junit.jupiter.api.Test;
3435
import org.testcontainers.containers.MSSQLServerContainer;
3536

37+
@Disabled("Disabled after DV2 migration. Re-enable with fixtures updated to DV2.")
3638
public class MssqlStrictEncryptDestinationAcceptanceTest extends DestinationAcceptanceTest {
3739

3840
private static MSSQLServerContainer<?> db;
@@ -167,7 +169,7 @@ protected void setup(final TestDestinationEnv testEnv, final HashSet<String> TES
167169

168170
@Override
169171
protected void tearDown(final TestDestinationEnv testEnv) {
170-
dslContext.close();
172+
// do nothing
171173
}
172174

173175
@AfterAll

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

+3-1
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,12 @@ plugins {
44
}
55

66
airbyteJavaConnector {
7-
cdkVersionRequired = '0.2.0'
7+
cdkVersionRequired = '0.30.2'
88
features = [
99
'db-sources', // required for tests
1010
'db-destinations',
11+
's3-destinations',
12+
'typing-deduping'
1113
]
1214
useLocalCdk = false
1315
}

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

+11-5
Original file line numberDiff line numberDiff line change
@@ -2,23 +2,29 @@ data:
22
connectorSubtype: database
33
connectorType: destination
44
definitionId: d4353156-9217-4cad-8dd7-c108fd4f74cf
5-
dockerImageTag: 0.2.0
5+
dockerImageTag: 1.0.0
66
dockerRepository: airbyte/destination-mssql
77
githubIssueLabel: destination-mssql
88
icon: mssql.svg
99
license: ELv2
1010
name: MS SQL Server
11-
normalizationConfig:
12-
normalizationIntegrationType: mssql
13-
normalizationRepository: airbyte/normalization-mssql
14-
normalizationTag: 0.4.3
1511
registries:
1612
cloud:
1713
dockerRepository: airbyte/destination-mssql-strict-encrypt
1814
enabled: true
1915
oss:
2016
enabled: true
2117
releaseStage: alpha
18+
releases:
19+
breakingChanges:
20+
1.0.0:
21+
upgradeDeadline: "2024-05-25"
22+
message: >
23+
This version removes the option to use "normalization" with MSSQL. It also changes
24+
the schema and database of Airbyte's "raw" tables to be compatible with the new
25+
[Destinations V2](https://docs.airbyte.com/release_notes/upgrading_to_destinations_v2/#what-is-destinations-v2)
26+
format. These changes will likely require updates to downstream dbt / SQL models.
27+
Selecting `Upgrade` will upgrade **all** connections using this destination at their next sync.
2228
documentationUrl: https://docs.airbyte.com/integrations/destinations/mssql
2329
supportsDbt: true
2430
tags:

‎airbyte-integrations/connectors/destination-mssql/src/main/java/io/airbyte/integrations/destination/mssql/MSSQLDestination.java

+46
Original file line numberDiff line numberDiff line change
@@ -7,16 +7,28 @@
77
import com.fasterxml.jackson.databind.JsonNode;
88
import com.google.common.collect.ImmutableMap;
99
import io.airbyte.cdk.db.factory.DatabaseDriver;
10+
import io.airbyte.cdk.db.jdbc.JdbcDatabase;
1011
import io.airbyte.cdk.db.jdbc.JdbcUtils;
1112
import io.airbyte.cdk.integrations.base.Destination;
1213
import io.airbyte.cdk.integrations.base.IntegrationRunner;
1314
import io.airbyte.cdk.integrations.base.ssh.SshWrappedDestination;
1415
import io.airbyte.cdk.integrations.destination.jdbc.AbstractJdbcDestination;
16+
import io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.JdbcDestinationHandler;
17+
import io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.JdbcSqlGenerator;
18+
import io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.NoOpJdbcDestinationHandler;
19+
import io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.RawOnlySqlGenerator;
1520
import io.airbyte.commons.json.Jsons;
21+
import io.airbyte.integrations.base.destination.typing_deduping.DestinationHandler;
22+
import io.airbyte.integrations.base.destination.typing_deduping.SqlGenerator;
23+
import io.airbyte.integrations.base.destination.typing_deduping.migrators.Migration;
24+
import io.airbyte.integrations.base.destination.typing_deduping.migrators.MinimumDestinationState;
1625
import java.io.File;
1726
import java.util.HashMap;
27+
import java.util.List;
1828
import java.util.Map;
1929
import java.util.Optional;
30+
import org.jetbrains.annotations.NotNull;
31+
import org.jooq.SQLDialect;
2032
import org.slf4j.Logger;
2133
import org.slf4j.LoggerFactory;
2234

@@ -30,6 +42,7 @@ public MSSQLDestination() {
3042
super(DRIVER_CLASS, new MSSQLNameTransformer(), new SqlServerOperations());
3143
}
3244

45+
@NotNull
3346
@Override
3447
protected Map<String, String> getDefaultConnectionProperties(final JsonNode config) {
3548
final HashMap<String, String> properties = new HashMap<>();
@@ -57,6 +70,7 @@ protected Map<String, String> getDefaultConnectionProperties(final JsonNode conf
5770
return properties;
5871
}
5972

73+
@NotNull
6074
@Override
6175
public JsonNode toJdbcConfig(final JsonNode config) {
6276
final String schema = Optional.ofNullable(config.get("schema")).map(JsonNode::asText).orElse("public");
@@ -81,6 +95,22 @@ public JsonNode toJdbcConfig(final JsonNode config) {
8195
return Jsons.jsonNode(configBuilder.build());
8296
}
8397

98+
@Override
99+
protected JdbcDestinationHandler<? extends MinimumDestinationState> getDestinationHandler(final String databaseName,
100+
final JdbcDatabase database,
101+
final String rawTableSchema) {
102+
return new NoOpJdbcDestinationHandler<>(databaseName, database, rawTableSchema, SQLDialect.DEFAULT);
103+
}
104+
105+
@NotNull
106+
@Override
107+
protected List<Migration> getMigrations(final JdbcDatabase database,
108+
final String databaseName,
109+
final SqlGenerator sqlGenerator,
110+
final DestinationHandler destinationHandler) {
111+
return List.of();
112+
}
113+
84114
private String getTrustStoreLocation() {
85115
// trust store location code found at https://stackoverflow.com/a/56570588
86116
final String trustStoreLocation = Optional.ofNullable(System.getProperty("javax.net.ssl.trustStore"))
@@ -104,4 +134,20 @@ public static void main(final String[] args) throws Exception {
104134
LOGGER.info("completed destination: {}", MSSQLDestination.class);
105135
}
106136

137+
@Override
138+
public boolean isV2Destination() {
139+
return true;
140+
}
141+
142+
@Override
143+
protected boolean shouldAlwaysDisableTypeDedupe() {
144+
return true;
145+
}
146+
147+
@NotNull
148+
@Override
149+
protected JdbcSqlGenerator getSqlGenerator(@NotNull final JsonNode config) {
150+
return new RawOnlySqlGenerator(new MSSQLNameTransformer());
151+
}
152+
107153
}

‎airbyte-integrations/connectors/destination-mssql/src/main/java/io/airbyte/integrations/destination/mssql/SqlServerOperations.java

+59-18
Original file line numberDiff line numberDiff line change
@@ -5,17 +5,26 @@
55
package io.airbyte.integrations.destination.mssql;
66

77
import com.fasterxml.jackson.databind.JsonNode;
8+
import com.fasterxml.jackson.databind.ObjectMapper;
89
import com.google.common.collect.Lists;
910
import io.airbyte.cdk.db.jdbc.JdbcDatabase;
1011
import io.airbyte.cdk.integrations.base.JavaBaseConstants;
12+
import io.airbyte.cdk.integrations.destination.async.model.PartialAirbyteMessage;
1113
import io.airbyte.cdk.integrations.destination.jdbc.SqlOperations;
12-
import io.airbyte.cdk.integrations.destination.jdbc.SqlOperationsUtils;
13-
import io.airbyte.protocol.models.v0.AirbyteRecordMessage;
1414
import java.sql.SQLException;
15+
import java.sql.Timestamp;
16+
import java.time.Instant;
1517
import java.util.List;
18+
import java.util.Objects;
19+
import java.util.UUID;
20+
import org.slf4j.Logger;
21+
import org.slf4j.LoggerFactory;
1622

1723
public class SqlServerOperations implements SqlOperations {
1824

25+
private static final Logger LOGGER = LoggerFactory.getLogger(SqlServerOperations.class);
26+
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
27+
1928
@Override
2029
public void createSchemaIfNotExists(final JdbcDatabase database, final String schemaName) throws Exception {
2130
final String query = String.format("IF NOT EXISTS ( SELECT * FROM sys.schemas WHERE name = '%s') EXEC('CREATE SCHEMA [%s]')",
@@ -37,10 +46,12 @@ public String createTableQuery(final JdbcDatabase database, final String schemaN
3746
+ "CREATE TABLE %s.%s ( \n"
3847
+ "%s VARCHAR(64) PRIMARY KEY,\n"
3948
+ "%s NVARCHAR(MAX),\n" // Microsoft SQL Server specific: NVARCHAR can store Unicode meanwhile VARCHAR - not
40-
+ "%s DATETIMEOFFSET(7) DEFAULT SYSDATETIMEOFFSET()\n"
49+
+ "%s DATETIMEOFFSET(7) DEFAULT SYSDATETIMEOFFSET(),\n"
50+
+ "%s DATETIMEOFFSET(7),\n"
51+
+ "%s NVARCHAR(MAX),\n"
4152
+ ");\n",
42-
schemaName, tableName, schemaName, tableName, JavaBaseConstants.COLUMN_NAME_AB_ID, JavaBaseConstants.COLUMN_NAME_DATA,
43-
JavaBaseConstants.COLUMN_NAME_EMITTED_AT);
53+
schemaName, tableName, schemaName, tableName, JavaBaseConstants.COLUMN_NAME_AB_RAW_ID, JavaBaseConstants.COLUMN_NAME_DATA,
54+
JavaBaseConstants.COLUMN_NAME_AB_EXTRACTED_AT, JavaBaseConstants.COLUMN_NAME_AB_LOADED_AT, JavaBaseConstants.COLUMN_NAME_AB_META);
4455
}
4556

4657
@Override
@@ -60,30 +71,60 @@ public String truncateTableQuery(final JdbcDatabase database, final String schem
6071

6172
@Override
6273
public void insertRecords(final JdbcDatabase database,
63-
final List<AirbyteRecordMessage> records,
74+
final List<PartialAirbyteMessage> records,
6475
final String schemaName,
6576
final String tempTableName)
6677
throws SQLException {
6778
// MSSQL has a limitation of 2100 parameters used in a query
6879
// Airbyte inserts data with 3 columns (raw table) this limits to 700 records.
6980
// Limited the variable to 500 records to
70-
final int MAX_BATCH_SIZE = 500;
81+
final int MAX_BATCH_SIZE = 400;
7182
final String insertQueryComponent = String.format(
72-
"INSERT INTO %s.%s (%s, %s, %s) VALUES\n",
83+
"INSERT INTO %s.%s (%s, %s, %s, %s, %s) VALUES\n",
7384
schemaName,
7485
tempTableName,
75-
JavaBaseConstants.COLUMN_NAME_AB_ID,
86+
JavaBaseConstants.COLUMN_NAME_AB_RAW_ID,
7687
JavaBaseConstants.COLUMN_NAME_DATA,
77-
JavaBaseConstants.COLUMN_NAME_EMITTED_AT);
78-
final String recordQueryComponent = "(?, ?, ?),\n";
79-
final List<List<AirbyteRecordMessage>> batches = Lists.partition(records, MAX_BATCH_SIZE);
80-
batches.forEach(record -> {
81-
try {
82-
SqlOperationsUtils.insertRawRecordsInSingleQuery(insertQueryComponent, recordQueryComponent, database, record);
83-
} catch (final SQLException e) {
84-
e.printStackTrace();
88+
JavaBaseConstants.COLUMN_NAME_AB_EXTRACTED_AT,
89+
JavaBaseConstants.COLUMN_NAME_AB_LOADED_AT,
90+
JavaBaseConstants.COLUMN_NAME_AB_META);
91+
final String recordQueryComponent = "(?, ?, ?, ?, ?),\n";
92+
final List<List<PartialAirbyteMessage>> batches = Lists.partition(records, MAX_BATCH_SIZE);
93+
for (List<PartialAirbyteMessage> batch : batches) {
94+
if (batch.isEmpty()) {
95+
continue;
8596
}
86-
});
97+
database.execute(connection -> {
98+
final StringBuilder sqlStatement = new StringBuilder(insertQueryComponent);
99+
for (PartialAirbyteMessage ignored : batch) {
100+
sqlStatement.append(recordQueryComponent);
101+
}
102+
final var sql = sqlStatement.substring(0, sqlStatement.length() - 2) + ";";
103+
try (final var statement = connection.prepareStatement(sql)) {
104+
int i = 1;
105+
for (PartialAirbyteMessage record : batch) {
106+
final var id = UUID.randomUUID().toString();
107+
statement.setString(i++, id);
108+
statement.setString(i++, record.getSerialized());
109+
statement.setTimestamp(i++, Timestamp.from(Instant.ofEpochMilli(Objects.requireNonNull(record.getRecord()).getEmittedAt())));
110+
statement.setTimestamp(i++, null);
111+
String metadata;
112+
if (record.getRecord().getMeta() != null) {
113+
try {
114+
metadata = OBJECT_MAPPER.writeValueAsString(record.getRecord().getMeta());
115+
} catch (Exception e) {
116+
LOGGER.error("Failed to serialize record metadata for record {}", id, e);
117+
metadata = null;
118+
}
119+
} else {
120+
metadata = null;
121+
}
122+
statement.setString(i++, metadata);
123+
}
124+
statement.execute();
125+
}
126+
});
127+
}
87128
}
88129

89130
@Override

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

+6
Original file line numberDiff line numberDiff line change
@@ -114,6 +114,12 @@
114114
}
115115
}
116116
]
117+
},
118+
"raw_data_schema": {
119+
"type": "string",
120+
"description": "The schema to write raw tables into (default: airbyte_internal)",
121+
"title": "Raw Table Schema Name",
122+
"order": 7
117123
}
118124
}
119125
}

‎airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/MSSQLDestinationAcceptanceTest.java

+16-15
Original file line numberDiff line numberDiff line change
@@ -26,14 +26,15 @@
2626
import org.jooq.DSLContext;
2727
import org.junit.jupiter.api.AfterAll;
2828
import org.junit.jupiter.api.BeforeAll;
29+
import org.junit.jupiter.api.Disabled;
2930
import org.testcontainers.containers.MSSQLServerContainer;
3031

32+
@Disabled("Disabled after DV2 migration. Re-enable with fixtures updated to DV2.")
3133
public class MSSQLDestinationAcceptanceTest extends JdbcDestinationAcceptanceTest {
3234

3335
private static MSSQLServerContainer<?> db;
3436
private final StandardNameTransformer namingResolver = new StandardNameTransformer();
3537
private JsonNode config;
36-
private DSLContext dslContext;
3738

3839
@Override
3940
protected String getImageName() {
@@ -93,17 +94,16 @@ protected List<JsonNode> retrieveNormalizedRecords(final TestDestinationEnv env,
9394
}
9495

9596
private List<JsonNode> retrieveRecordsFromTable(final String tableName, final String schemaName) throws SQLException {
96-
try (final DSLContext dslContext = DatabaseConnectionHelper.createDslContext(db, null)) {
97-
return getDatabase(dslContext).query(
98-
ctx -> {
99-
ctx.fetch(String.format("USE %s;", config.get(JdbcUtils.DATABASE_KEY)));
100-
return ctx
101-
.fetch(String.format("SELECT * FROM %s.%s ORDER BY %s ASC;", schemaName, tableName, JavaBaseConstants.COLUMN_NAME_EMITTED_AT))
102-
.stream()
103-
.map(this::getJsonFromRecord)
104-
.collect(Collectors.toList());
105-
});
106-
}
97+
final DSLContext dslContext = DatabaseConnectionHelper.createDslContext(db, null);
98+
return getDatabase(dslContext).query(
99+
ctx -> {
100+
ctx.fetch(String.format("USE %s;", config.get(JdbcUtils.DATABASE_KEY)));
101+
return ctx
102+
.fetch(String.format("SELECT * FROM %s.%s ORDER BY %s ASC;", schemaName, tableName, JavaBaseConstants.COLUMN_NAME_AB_EXTRACTED_AT))
103+
.stream()
104+
.map(this::getJsonFromRecord)
105+
.collect(Collectors.toList());
106+
});
107107
}
108108

109109
@BeforeAll
@@ -134,7 +134,7 @@ private static Database getDatabase(final DSLContext dslContext) {
134134
protected void setup(final TestDestinationEnv testEnv, final HashSet<String> TEST_SCHEMAS) throws SQLException {
135135
final JsonNode configWithoutDbName = getConfig(db);
136136
final String dbName = Strings.addRandomSuffix("db", "_", 10);
137-
dslContext = getDslContext(configWithoutDbName);
137+
DSLContext dslContext = getDslContext(configWithoutDbName);
138138
final Database database = getDatabase(dslContext);
139139
database.query(ctx -> {
140140
ctx.fetch(String.format("CREATE DATABASE %s;", dbName));
@@ -150,8 +150,9 @@ protected void setup(final TestDestinationEnv testEnv, final HashSet<String> TES
150150
}
151151

152152
@Override
153-
protected void tearDown(final TestDestinationEnv testEnv) {
154-
dslContext.close();
153+
protected void tearDown(final TestDestinationEnv testEnv) throws Exception {
154+
db.stop();
155+
db.close();
155156
}
156157

157158
@Override

0 commit comments

Comments
 (0)