(java) postgres destination #539

cgardens · 2020-10-10T07:15:11Z

What

Move GracefulShutdownHandler for ExecutorServices to commons.
Remove tests that rely on ordering from the sync test suite.
Add transaction method to DatabaseHelper.
Oh yeah, and add the java impl of the postgres-destination. 😉

closes #540

First Impressions

It really didn't feel too hard to write this integration. The simplifying assumptions that we are doing full refresh and writing a singe object column makes life much easier. Feeling optimistic that we can bang out a couple other warehouses quickly. Having the test suite to run again made developing go faster too. The most time-consumnig part was writing the buffering logic (presented in separate PR)

Checklist

Add / update docs
Manual testing with the app

Recommended reading order

I recommend focusing on the postgres-destination code and then the little updates to commons and utility methods will make sense from there.

PostgresDestination.java, PostgresDestinationTest.java, PostgresDestinationIntegrationTest.java
the rest

cgardens · 2020-10-10T07:20:55Z

airbyte-integrations/postgres-destination/src/main/resources/spec.json

@@ -0,0 +1,44 @@
+{
+  "destinationId": "",


need to adapt this to the new format.

michel-tricot · 2020-10-12T01:20:11Z

airbyte-commons/src/main/java/io/airbyte/commons/concurrency/GracefulShutdownHandler.java

  private final ExecutorService[] threadPools;

-  public SchedulerShutdownHandler(final ExecutorService... threadPools) {
+  public GracefulShutdownHandler(


NIT: weird line break

michel-tricot · 2020-10-12T01:24:18Z

...-integration/java/io/airbyte/integrations/destination/csv/CsvDestinationIntegrationTest.java

@@ -60,7 +61,7 @@ protected JsonNode getInvalidConfig() {
  }

  @Override
-  protected List<JsonNode> recordRetriever(TestDestinationEnv testEnv) throws Exception {
+  protected Set<JsonNode> recordRetriever(TestDestinationEnv testEnv, String streamName) throws Exception {


You should keep List and sort or create an assert that is not sensitive to order. Set will make it hard to test for duplicated values.

good point.

michel-tricot · 2020-10-12T01:27:38Z

...tination/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestination.java

+      final String tableName = stream.getName();
+      final String tmpTableName = stream.getName() + "_" + Instant.now().toEpochMilli();
+      DatabaseHelper.query(connectionPool, ctx -> ctx.execute(String.format(
+          "CREATE EXTENSION IF NOT EXISTS \"uuid-ossp\";\n"


Why do you need this?

I am injecting our own id in the output as a uuid. in order to do that we need to make sure that the uuid lib is installed in the database. the alternative here is to just save the uuid as a string so that we don't need to install the lib.

We should prefer inserting as a string so we aren't modifying the user's db in unexpected ways. I wouldn't expect any ETL tool to install a plugin in my database.

Onboard with @jrhizor

michel-tricot · 2020-10-12T01:29:18Z

...tination/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestination.java

+     * @param writeBuffers - map of stream name to its respective buffer.
+     * @param connectionPool - connection to the db.
+     */
+    private static void writeStreamsWithNRecords(


NIT: weird line break.

michel-tricot · 2020-10-12T01:32:26Z

...tination/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestination.java

+      final StringBuilder query = new StringBuilder(String.format("INSERT INTO %s(%s)\n", tmpTableName, COLUMN_NAME))
+          .append("VALUES \n");
+      boolean firstRecordInQuery = true;
+      for (int i = 0; i <= batchSize; i++) {


Do you have an index error here? If it is expected, you should add a comment to explain it since it not a normal code structure

made it standard.

michel-tricot · 2020-10-12T01:33:52Z

...tination/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestination.java

+        writerPool.shutdown();
+        writerPool.awaitTermination(1, TimeUnit.SECONDS);
+      } else {
+        LOGGER.error("executing on success close procedure.");


This shouldn't be an error log

michel-tricot · 2020-10-12T01:34:07Z

...tination/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestination.java

+      writeConfigs.values().forEach(writeConfig -> writeConfig.getWriteBuffer().closeInput());
+
+      if (hasFailed) {
+        LOGGER.error("executing on failed close procedure.");


This shoudln't be an error log

airbyte-integrations/postgres-destination/Dockerfile

jrhizor · 2020-10-12T16:33:46Z

airbyte-integrations/postgres-destination/build.gradle

+    integrationTestImplementation project(':airbyte-integrations:integration-test-lib')
+    integrationTestImplementation project(':airbyte-integrations:postgres-destination')
+
+    testImplementation "org.testcontainers:postgresql:1.15.0-rc2"


nit: group integrationTestImplementation together

jrhizor · 2020-10-12T16:36:44Z

...tion/src/test/java/io/airbyte/integrations/destination/postgres/PostgresDestinationTest.java

+    return DatabaseHelper.getConnectionPool(db.getUsername(), db.getPassword(), db.getJdbcUrl());
+  }
+
+  private Set<JsonNode> recordRetriever(String streamName) throws Exception {


this is duplicated from the integration test?

jrhizor · 2020-10-12T16:38:45Z

airbyte-commons/src/main/java/io/airbyte/commons/concurrency/GracefulShutdownHandler.java

  private final ExecutorService[] threadPools;

-  public SchedulerShutdownHandler(final ExecutorService... threadPools) {
+  public GracefulShutdownHandler(
+                                 long terminationWaitTime,


I slightly prefer Duration here so the time and unit are tied together by whatever is providing it.

jrhizor · 2020-10-12T16:41:39Z

...tination/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestination.java

+      return new StandardCheckConnectionOutput().withStatus(Status.FAILURE).withMessage(e.getMessage());
+    }
+
+    return new StandardCheckConnectionOutput().withStatus(Status.SUCCESS);


Can we move this into the end of the try so the method reads linearly except in the exceptional case?

jrhizor · 2020-10-12T16:42:26Z

...tination/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestination.java

+    try {
+      final BasicDataSource connectionPool = getConnectionPool(config);
+      DatabaseHelper.query(connectionPool, ctx -> ctx.execute(
+          "SELECT *\n"


Is there a reason to do this query instead of SELECT 1 or something for a faster check?

jrhizor · 2020-10-12T16:43:32Z

...tination/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestination.java

+  }
+
+  @Override
+  public StandardDiscoverSchemaOutput discover(JsonNode config) {


Does Destination need this as part of the interface?

eh. i dunno. we got to decide what we want out of discover with destination.

jrhizor · 2020-10-12T16:45:22Z

...tination/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestination.java

+      final String tableName = stream.getName();
+      final String tmpTableName = stream.getName() + "_" + Instant.now().toEpochMilli();
+      DatabaseHelper.query(connectionPool, ctx -> ctx.execute(String.format(
+          "CREATE EXTENSION IF NOT EXISTS \"uuid-ossp\";\n"


We should prefer inserting as a string so we aren't modifying the user's db in unexpected ways. I wouldn't expect any ETL tool to install a plugin in my database.

jrhizor · 2020-10-12T16:50:03Z

...tination/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestination.java

+          tmpTableName, COLUMN_NAME)));
+
+      final Path queueRoot = Files.createTempDirectory("queues");
+      final BigQueue writeBuffer = new BigQueue(queueRoot.resolve(stream.getName()), stream.getName());


How do we handle two sources with the same named streams syncing at the same time?

I assumed these would be running in separate workers, but I guess the source shouldn't assume it?

different workers. why shouldn't the source assume this? we add noise at the end of the paths to fix this, but i think the contract is that the source runs on its own docker image (with its own filesystem).

you're right -- each source should run in its own container. i was being silly in my head .

sherifnada

LGTM % comment about batch size/missed batches

sherifnada · 2020-10-12T16:50:56Z

...tination/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestination.java

+    private final Map<String, WriteConfig> writeConfigs;
+    private final Schema schema;
+
+    public RecordConsumer(BasicDataSource connectionPool, Map<String, WriteConfig> writeConfigs, Schema schema) {


Suggested change

public RecordConsumer(BasicDataSource connectionPool, Map<String, WriteConfig> writeConfigs, Schema schema) {

public RecordConsumer(BasicDataSource connectionPool, Map<String, WriteConfig> streamNameToWriteConfigs, Schema schema) {

sherifnada · 2020-10-12T16:56:42Z

...tination/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestination.java

+     * @param writeBuffers - map of stream name to its respective buffer.
+     * @param connectionPool - connection to the db.
+     */
+    private static void writeStreamsWithNRecords(int minRecords,


wouldn't this miss the last batch if it is smaller than the minimum batch size?

check out the close function on consumer.

sherifnada · 2020-10-12T17:03:25Z

...t-integration/java/io/airbyte/integrations/destination/postgres/PostgresIntegrationTest.java

+            .stream()
+            .map(Record::intoMap)
+            .map(r -> r.entrySet().stream().map(e -> {
+              if (e.getValue().getClass().equals(org.jooq.JSONB.class)) {


why would this not be the case for all records?

it's a jooq issue. it couldn't create a record with a value that had a jsonb class:
adding this comment:

// jooq needs more configuration to handle jsonb natively. coerce it to a string for now and handle deserializing later.

awesome, thanks for retaining the context in a comment

sherifnada · 2020-10-12T17:04:31Z

...t-integration/java/io/airbyte/integrations/destination/postgres/PostgresIntegrationTest.java

+  }
+
+  @Override
+  protected List<JsonNode> recordRetriever(TestDestinationEnv env, String streamName) throws Exception {


Suggested change

protected List<JsonNode> recordRetriever(TestDestinationEnv env, String streamName) throws Exception {

protected List<JsonNode> retrieveRecords(TestDestinationEnv env, String streamName) throws Exception {

cgardens force-pushed the cgardens/java_postgres2 branch from 4ba6055 to 45cbc82 Compare October 10, 2020 07:19

cgardens commented Oct 10, 2020

View reviewed changes

cgardens force-pushed the cgardens/java_postgres2 branch 2 times, most recently from bee0350 to e95f4cd Compare October 10, 2020 07:27

cgardens marked this pull request as ready for review October 10, 2020 07:28

cgardens requested review from michel-tricot, jrhizor and sherifnada October 10, 2020 07:28

cgardens changed the title ~~wip~~ postgres destination Oct 10, 2020

cgardens changed the title ~~postgres destination~~ (java) postgres destination Oct 10, 2020

cgardens mentioned this pull request Oct 10, 2020

java postgres destination #527

Closed

3 tasks

michel-tricot approved these changes Oct 12, 2020

View reviewed changes

cgardens force-pushed the cgardens/java_postgres2 branch from e95f4cd to 3be63f8 Compare October 12, 2020 16:48

jrhizor approved these changes Oct 12, 2020

View reviewed changes

sherifnada approved these changes Oct 12, 2020

View reviewed changes

cgardens force-pushed the cgardens/persistent_queue branch 2 times, most recently from a0c96ec to 2f525f9 Compare October 13, 2020 16:21

Base automatically changed from cgardens/persistent_queue to cgardens/integration_test_runner October 13, 2020 16:43

cgardens force-pushed the cgardens/integration_test_runner branch from c9e65b6 to 936a3e9 Compare October 13, 2020 17:23

Base automatically changed from cgardens/integration_test_runner to master October 13, 2020 17:40

cgardens force-pushed the cgardens/java_postgres2 branch from 3be63f8 to ff18d12 Compare October 13, 2020 18:00

This was referenced Oct 13, 2020

move graceful thread shutdown to a helper #555

Merged

add transaction support to DatabaseHelper #556

Merged

cgardens added 5 commits October 13, 2020 11:52

postgres

95ed51e

clean up

738f1d0

clean up

9e68b6f

clean up

cad49c2

rename retrieve records

718528f

cgardens force-pushed the cgardens/java_postgres2 branch from 8b763e3 to 718528f Compare October 13, 2020 18:53

cgardens merged commit adc41a7 into master Oct 13, 2020

cgardens deleted the cgardens/java_postgres2 branch October 13, 2020 18:53

	public RecordConsumer(BasicDataSource connectionPool, Map<String, WriteConfig> writeConfigs, Schema schema) {
	public RecordConsumer(BasicDataSource connectionPool, Map<String, WriteConfig> streamNameToWriteConfigs, Schema schema) {

	protected List<JsonNode> recordRetriever(TestDestinationEnv env, String streamName) throws Exception {
	protected List<JsonNode> retrieveRecords(TestDestinationEnv env, String streamName) throws Exception {

(java) postgres destination #539

(java) postgres destination #539

Conversation

cgardens commented Oct 10, 2020 • edited Loading

What

First Impressions

Checklist

Recommended reading order

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

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

sherifnada 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

cgardens commented Oct 10, 2020 •

edited

Loading