Skip to content

Commit 781be94

Browse files
DoNotPanicUAsuhomud
DoNotPanicUA
authored and
suhomud
committed
Destination Postgres : Enable DAT and fix the data fetch. (#12543)
* Enable DAT for Postgres and fix the data fetch. Move JDBC abstract part for tests to the JdbcDestinationAcceptanceTest.java * Remove unnecessary deserialization + add jsonb to json transformation. * Remove unnecessary deserialization from ssh
1 parent 38390e5 commit 781be94

File tree

6 files changed

+164
-89
lines changed

6 files changed

+164
-89
lines changed

‎airbyte-integrations/bases/standard-destination-test/build.gradle

+1
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ plugins {
22
id 'java-library'
33
}
44
dependencies {
5+
implementation project(':airbyte-db:lib')
56
implementation project(':airbyte-config:models')
67
implementation project(':airbyte-integrations:bases:base-java')
78
implementation project(':airbyte-protocol:models')

‎airbyte-integrations/bases/standard-destination-test/src/main/java/io/airbyte/integrations/standardtest/destination/JdbcDestinationAcceptanceTest.java

+41
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
/*
2+
* Copyright (c) 2021 Airbyte, Inc., all rights reserved.
3+
*/
4+
5+
package io.airbyte.integrations.standardtest.destination;
6+
7+
import com.fasterxml.jackson.databind.JsonNode;
8+
import com.fasterxml.jackson.databind.ObjectMapper;
9+
import com.fasterxml.jackson.databind.node.ObjectNode;
10+
import io.airbyte.commons.json.Jsons;
11+
import java.util.Arrays;
12+
import org.jooq.Record;
13+
14+
public abstract class JdbcDestinationAcceptanceTest extends DestinationAcceptanceTest {
15+
16+
protected final ObjectMapper mapper = new ObjectMapper();
17+
18+
protected JsonNode getJsonFromRecord(Record record) {
19+
ObjectNode node = mapper.createObjectNode();
20+
21+
Arrays.stream(record.fields()).forEach(field -> {
22+
var value = record.get(field);
23+
24+
switch (field.getDataType().getTypeName()) {
25+
case "varchar", "jsonb", "other":
26+
var stringValue = (value != null ? value.toString() : null);
27+
if (stringValue != null && (stringValue.replaceAll("[^\\x00-\\x7F]", "").matches("^\\[.*\\]$")
28+
|| stringValue.replaceAll("[^\\x00-\\x7F]", "").matches("^\\{.*\\}$"))) {
29+
node.set(field.getName(), Jsons.deserialize(stringValue));
30+
} else {
31+
node.put(field.getName(), stringValue);
32+
}
33+
break;
34+
default:
35+
node.put(field.getName(), (value != null ? value.toString() : null));
36+
}
37+
});
38+
return node;
39+
}
40+
41+
}

‎airbyte-integrations/connectors/destination-postgres/src/test-integration/java/io/airbyte/integrations/destination/postgres/PostgresDestinationAcceptanceTest.java

+31-31
Original file line numberDiff line numberDiff line change
@@ -8,17 +8,16 @@
88
import com.google.common.collect.ImmutableMap;
99
import io.airbyte.commons.json.Jsons;
1010
import io.airbyte.db.Databases;
11-
import io.airbyte.db.jdbc.JdbcUtils;
1211
import io.airbyte.integrations.base.JavaBaseConstants;
1312
import io.airbyte.integrations.destination.ExtendedNameTransformer;
14-
import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
13+
import io.airbyte.integrations.standardtest.destination.JdbcDestinationAcceptanceTest;
14+
import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
1515
import java.sql.SQLException;
16-
import java.util.ArrayList;
1716
import java.util.List;
1817
import java.util.stream.Collectors;
1918
import org.testcontainers.containers.PostgreSQLContainer;
2019

21-
public class PostgresDestinationAcceptanceTest extends DestinationAcceptanceTest {
20+
public class PostgresDestinationAcceptanceTest extends JdbcDestinationAcceptanceTest {
2221

2322
private PostgreSQLContainer<?> db;
2423
private final ExtendedNameTransformer namingResolver = new ExtendedNameTransformer();
@@ -62,7 +61,7 @@ protected List<JsonNode> retrieveRecords(final TestDestinationEnv env,
6261
throws Exception {
6362
return retrieveRecordsFromTable(namingResolver.getRawTableName(streamName), namespace)
6463
.stream()
65-
.map(r -> Jsons.deserialize(r.get(JavaBaseConstants.COLUMN_NAME_DATA).asText()))
64+
.map(r -> r.get(JavaBaseConstants.COLUMN_NAME_DATA))
6665
.collect(Collectors.toList());
6766
}
6867

@@ -81,41 +80,42 @@ protected boolean implementsNamespaces() {
8180
return true;
8281
}
8382

83+
@Override
84+
protected TestDataComparator getTestDataComparator() {
85+
return new PostgresTestDataComparator();
86+
}
87+
88+
@Override
89+
protected boolean supportBasicDataTypeTest() {
90+
return true;
91+
}
92+
93+
@Override
94+
protected boolean supportArrayDataTypeTest() {
95+
return true;
96+
}
97+
98+
@Override
99+
protected boolean supportObjectDataTypeTest() {
100+
return true;
101+
}
102+
84103
@Override
85104
protected List<JsonNode> retrieveNormalizedRecords(final TestDestinationEnv env, final String streamName, final String namespace)
86105
throws Exception {
87106
final String tableName = namingResolver.getIdentifier(streamName);
88-
// Temporarily disabling the behavior of the ExtendedNameTransformer, see (issue #1785) so we don't
89-
// use quoted names
90-
// if (!tableName.startsWith("\"")) {
91-
// // Currently, Normalization always quote tables identifiers
92-
// //tableName = "\"" + tableName + "\"";
93-
// }
94107
return retrieveRecordsFromTable(tableName, namespace);
95108
}
96109

97-
@Override
98-
protected List<String> resolveIdentifier(final String identifier) {
99-
final List<String> result = new ArrayList<>();
100-
final String resolved = namingResolver.getIdentifier(identifier);
101-
result.add(identifier);
102-
result.add(resolved);
103-
if (!resolved.startsWith("\"")) {
104-
result.add(resolved.toLowerCase());
105-
result.add(resolved.toUpperCase());
106-
}
107-
return result;
108-
}
109-
110110
private List<JsonNode> retrieveRecordsFromTable(final String tableName, final String schemaName) throws SQLException {
111111
return Databases.createPostgresDatabase(db.getUsername(), db.getPassword(),
112-
db.getJdbcUrl()).query(
113-
ctx -> ctx
114-
.fetch(String.format("SELECT * FROM %s.%s ORDER BY %s ASC;", schemaName, tableName, JavaBaseConstants.COLUMN_NAME_EMITTED_AT))
115-
.stream()
116-
.map(r -> r.formatJSON(JdbcUtils.getDefaultJSONFormat()))
117-
.map(Jsons::deserialize)
118-
.collect(Collectors.toList()));
112+
db.getJdbcUrl()).query(ctx -> {
113+
ctx.execute("set time zone 'UTC';");
114+
return ctx.fetch(String.format("SELECT * FROM %s.%s ORDER BY %s ASC;", schemaName, tableName, JavaBaseConstants.COLUMN_NAME_EMITTED_AT))
115+
.stream()
116+
.map(this::getJsonFromRecord)
117+
.collect(Collectors.toList());
118+
});
119119
}
120120

121121
@Override

‎airbyte-integrations/connectors/destination-postgres/src/test-integration/java/io/airbyte/integrations/destination/postgres/PostgresTestDataComparator.java

+58
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
/*
2+
* Copyright (c) 2021 Airbyte, Inc., all rights reserved.
3+
*/
4+
5+
package io.airbyte.integrations.destination.postgres;
6+
7+
import io.airbyte.integrations.destination.ExtendedNameTransformer;
8+
import io.airbyte.integrations.standardtest.destination.comparator.AdvancedTestDataComparator;
9+
import java.time.LocalDate;
10+
import java.time.LocalDateTime;
11+
import java.time.ZoneOffset;
12+
import java.time.ZonedDateTime;
13+
import java.time.format.DateTimeFormatter;
14+
import java.util.ArrayList;
15+
import java.util.List;
16+
17+
public class PostgresTestDataComparator extends AdvancedTestDataComparator {
18+
19+
private final ExtendedNameTransformer namingResolver = new ExtendedNameTransformer();
20+
21+
private static final String POSTGRES_DATETIME_FORMAT = "yyyy-MM-dd'T'HH:mm:ss'Z'";
22+
private static final String POSTGRES_DATETIME_WITH_TZ_FORMAT = "yyyy-MM-dd'T'HH:mm:ss'Z'";
23+
24+
@Override
25+
protected List<String> resolveIdentifier(final String identifier) {
26+
final List<String> result = new ArrayList<>();
27+
final String resolved = namingResolver.getIdentifier(identifier);
28+
result.add(identifier);
29+
result.add(resolved);
30+
if (!resolved.startsWith("\"")) {
31+
result.add(resolved.toLowerCase());
32+
result.add(resolved.toUpperCase());
33+
}
34+
return result;
35+
}
36+
37+
private LocalDate parseLocalDate(String dateTimeValue) {
38+
if (dateTimeValue != null) {
39+
var format = (dateTimeValue.matches(".+Z") ? POSTGRES_DATETIME_FORMAT : AIRBYTE_DATETIME_FORMAT);
40+
return LocalDate.parse(dateTimeValue, DateTimeFormatter.ofPattern(format));
41+
} else {
42+
return null;
43+
}
44+
}
45+
46+
@Override
47+
protected boolean compareDateTimeValues(String expectedValue, String actualValue) {
48+
var destinationDate = parseLocalDate(actualValue);
49+
var expectedDate = LocalDate.parse(expectedValue, DateTimeFormatter.ofPattern(AIRBYTE_DATETIME_FORMAT));
50+
return expectedDate.equals(destinationDate);
51+
}
52+
53+
@Override
54+
protected ZonedDateTime parseDestinationDateWithTz(String destinationValue) {
55+
return ZonedDateTime.of(LocalDateTime.parse(destinationValue, DateTimeFormatter.ofPattern(POSTGRES_DATETIME_WITH_TZ_FORMAT)), ZoneOffset.UTC);
56+
}
57+
58+
}

‎airbyte-integrations/connectors/destination-postgres/src/test-integration/java/io/airbyte/integrations/destination/postgres/SshPostgresDestinationAcceptanceTest.java

+31-31
Original file line numberDiff line numberDiff line change
@@ -10,13 +10,12 @@
1010
import io.airbyte.commons.json.Jsons;
1111
import io.airbyte.db.Database;
1212
import io.airbyte.db.Databases;
13-
import io.airbyte.db.jdbc.JdbcUtils;
1413
import io.airbyte.integrations.base.JavaBaseConstants;
1514
import io.airbyte.integrations.base.ssh.SshBastionContainer;
1615
import io.airbyte.integrations.base.ssh.SshTunnel;
1716
import io.airbyte.integrations.destination.ExtendedNameTransformer;
18-
import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
19-
import java.util.ArrayList;
17+
import io.airbyte.integrations.standardtest.destination.JdbcDestinationAcceptanceTest;
18+
import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
2019
import java.util.List;
2120
import java.util.stream.Collectors;
2221
import org.apache.commons.lang3.RandomStringUtils;
@@ -29,7 +28,7 @@
2928
* Abstract class that allows us to avoid duplicating testing logic for testing SSH with a key file
3029
* or with a password.
3130
*/
32-
public abstract class SshPostgresDestinationAcceptanceTest extends DestinationAcceptanceTest {
31+
public abstract class SshPostgresDestinationAcceptanceTest extends JdbcDestinationAcceptanceTest {
3332

3433
private final ExtendedNameTransformer namingResolver = new ExtendedNameTransformer();
3534
private static final String schemaName = RandomStringUtils.randomAlphabetic(8).toLowerCase();
@@ -63,7 +62,7 @@ protected List<JsonNode> retrieveRecords(final TestDestinationEnv env,
6362
throws Exception {
6463
return retrieveRecordsFromTable(namingResolver.getRawTableName(streamName), namespace)
6564
.stream()
66-
.map(r -> Jsons.deserialize(r.get(JavaBaseConstants.COLUMN_NAME_DATA).asText()))
65+
.map(r -> r.get(JavaBaseConstants.COLUMN_NAME_DATA))
6766
.collect(Collectors.toList());
6867
}
6968

@@ -82,32 +81,33 @@ protected boolean implementsNamespaces() {
8281
return true;
8382
}
8483

84+
@Override
85+
protected TestDataComparator getTestDataComparator() {
86+
return new PostgresTestDataComparator();
87+
}
88+
89+
@Override
90+
protected boolean supportBasicDataTypeTest() {
91+
return true;
92+
}
93+
94+
@Override
95+
protected boolean supportArrayDataTypeTest() {
96+
return true;
97+
}
98+
99+
@Override
100+
protected boolean supportObjectDataTypeTest() {
101+
return true;
102+
}
103+
85104
@Override
86105
protected List<JsonNode> retrieveNormalizedRecords(final TestDestinationEnv env, final String streamName, final String namespace)
87106
throws Exception {
88107
final String tableName = namingResolver.getIdentifier(streamName);
89-
// Temporarily disabling the behavior of the ExtendedNameTransformer, see (issue #1785) so we don't
90-
// use quoted names
91-
// if (!tableName.startsWith("\"")) {
92-
// // Currently, Normalization always quote tables identifiers
93-
// //tableName = "\"" + tableName + "\"";
94-
// }
95108
return retrieveRecordsFromTable(tableName, namespace);
96109
}
97110

98-
@Override
99-
protected List<String> resolveIdentifier(final String identifier) {
100-
final List<String> result = new ArrayList<>();
101-
final String resolved = namingResolver.getIdentifier(identifier);
102-
result.add(identifier);
103-
result.add(resolved);
104-
if (!resolved.startsWith("\"")) {
105-
result.add(resolved.toLowerCase());
106-
result.add(resolved.toUpperCase());
107-
}
108-
return result;
109-
}
110-
111111
private static Database getDatabaseFromConfig(final JsonNode config) {
112112
return Databases.createPostgresDatabase(
113113
config.get("username").asText(),
@@ -123,13 +123,13 @@ private List<JsonNode> retrieveRecordsFromTable(final String tableName, final St
123123
PostgresDestination.HOST_KEY,
124124
PostgresDestination.PORT_KEY,
125125
(CheckedFunction<JsonNode, List<JsonNode>, Exception>) mangledConfig -> getDatabaseFromConfig(mangledConfig)
126-
.query(
127-
ctx -> ctx
128-
.fetch(String.format("SELECT * FROM %s.%s ORDER BY %s ASC;", schemaName, tableName, JavaBaseConstants.COLUMN_NAME_EMITTED_AT))
129-
.stream()
130-
.map(r -> r.formatJSON(JdbcUtils.getDefaultJSONFormat()))
131-
.map(Jsons::deserialize)
132-
.collect(Collectors.toList())));
126+
.query(ctx -> {
127+
ctx.execute("set time zone 'UTC';");
128+
return ctx.fetch(String.format("SELECT * FROM %s.%s ORDER BY %s ASC;", schemaName, tableName, JavaBaseConstants.COLUMN_NAME_EMITTED_AT))
129+
.stream()
130+
.map(this::getJsonFromRecord)
131+
.collect(Collectors.toList());
132+
}));
133133
}
134134

135135
@Override

‎airbyte-integrations/connectors/destination-redshift/src/test-integration/java/io/airbyte/integrations/destination/redshift/RedshiftCopyDestinationAcceptanceTest.java

+2-27
Original file line numberDiff line numberDiff line change
@@ -13,22 +13,20 @@
1313
import io.airbyte.db.Database;
1414
import io.airbyte.db.Databases;
1515
import io.airbyte.integrations.base.JavaBaseConstants;
16-
import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
16+
import io.airbyte.integrations.standardtest.destination.JdbcDestinationAcceptanceTest;
1717
import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
1818
import java.nio.file.Path;
1919
import java.sql.SQLException;
20-
import java.util.Arrays;
2120
import java.util.List;
2221
import java.util.stream.Collectors;
23-
import org.jooq.Record;
2422
import org.slf4j.Logger;
2523
import org.slf4j.LoggerFactory;
2624

2725
/**
2826
* Integration test testing {@link RedshiftCopyS3Destination}. The default Redshift integration test
2927
* credentials contain S3 credentials - this automatically causes COPY to be selected.
3028
*/
31-
public class RedshiftCopyDestinationAcceptanceTest extends DestinationAcceptanceTest {
29+
public class RedshiftCopyDestinationAcceptanceTest extends JdbcDestinationAcceptanceTest {
3230

3331
private static final Logger LOGGER = LoggerFactory.getLogger(RedshiftCopyDestinationAcceptanceTest.class);
3432

@@ -121,29 +119,6 @@ protected List<JsonNode> retrieveNormalizedRecords(final TestDestinationEnv test
121119
return retrieveRecordsFromTable(tableName, namespace);
122120
}
123121

124-
private JsonNode getJsonFromRecord(Record record) {
125-
ObjectNode node = mapper.createObjectNode();
126-
127-
Arrays.stream(record.fields()).forEach(field -> {
128-
var value = record.get(field);
129-
130-
switch (field.getDataType().getTypeName()) {
131-
case "varchar", "other":
132-
var stringValue = (value != null ? value.toString() : null);
133-
if (stringValue != null && (stringValue.replaceAll("[^\\x00-\\x7F]", "").matches("^\\[.*\\]$")
134-
|| stringValue.replaceAll("[^\\x00-\\x7F]", "").matches("^\\{.*\\}$"))) {
135-
node.set(field.getName(), Jsons.deserialize(stringValue));
136-
} else {
137-
node.put(field.getName(), stringValue);
138-
}
139-
break;
140-
default:
141-
node.put(field.getName(), (value != null ? value.toString() : null));
142-
}
143-
});
144-
return node;
145-
}
146-
147122
private List<JsonNode> retrieveRecordsFromTable(final String tableName, final String schemaName) throws SQLException {
148123
return getDatabase().query(
149124
ctx -> ctx

0 commit comments

Comments
 (0)