add incremental to ~~jdbc~~ jooq source (and postgres)

airbyte-commons/src/main/java/io/airbyte/commons/stream/MoreStreams.java

@@ -0,0 +1,39 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.commons.stream;
+
+import java.util.Iterator;
+import java.util.Spliterator;
+import java.util.Spliterators;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+public class MoreStreams {
+
+  public static <T> Stream<T> toStream(Iterator<T> iterator) {
+    return StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, Spliterator.ORDERED), false);
+  }
+
+}

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/IntegrationRunner.java

@@ -29,7 +29,6 @@
 import com.google.common.base.Preconditions;
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.json.Jsons;
-import io.airbyte.config.State;
 import io.airbyte.protocol.models.AirbyteMessage;
 import io.airbyte.protocol.models.AirbyteMessage.Type;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
@@ -100,10 +99,8 @@ public void run(String[] args) throws Exception {
       case READ -> {
         final JsonNode config = parseConfig(parsed.getConfigPath());
         final ConfiguredAirbyteCatalog catalog = parseConfig(parsed.getCatalogPath(), ConfiguredAirbyteCatalog.class);
-        // todo (cgardens) - should we should only send the contents of the state field to the integration,
-        // not the whole struct. this runner obfuscates everything but the contents.
-        final Optional<State> stateOptional = parsed.getStatePath().map(path -> parseConfig(path, State.class));
-        final Stream<AirbyteMessage> messageStream = source.read(config, catalog, stateOptional.map(State::getState).orElse(null));
+        final Optional<JsonNode> stateOptional = parsed.getStatePath().map(IntegrationRunner::parseConfig);
+        final Stream<AirbyteMessage> messageStream = source.read(config, catalog, stateOptional.orElse(null));
         messageStream.map(Jsons::serialize).forEach(stdoutConsumer);
         messageStream.close();
       }

airbyte-integrations/bases/base-java/src/test/java/io/airbyte/integrations/base/IntegrationRunnerTest.java

@@ -37,7 +37,6 @@
 import com.google.common.collect.Lists;
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.json.Jsons;
-import io.airbyte.config.State;
 import io.airbyte.protocol.models.AirbyteCatalog;
 import io.airbyte.protocol.models.AirbyteConnectionStatus;
 import io.airbyte.protocol.models.AirbyteConnectionStatus.Status;
@@ -78,14 +77,13 @@ class IntegrationRunnerTest {
 
   private static final AirbyteCatalog CATALOG = new AirbyteCatalog().withStreams(Lists.newArrayList(new AirbyteStream().withName(STREAM_NAME)));
   private static final ConfiguredAirbyteCatalog CONFIGURED_CATALOG = CatalogHelpers.toDefaultConfiguredCatalog(CATALOG);
-  private static final State STATE = new State().withState(Jsons.jsonNode(ImmutableMap.of("checkpoint", "05/08/1945")));
+  private static final JsonNode STATE = Jsons.jsonNode(ImmutableMap.of("checkpoint", "05/08/1945"));
 
   private IntegrationCliParser cliParser;
   private Consumer<String> stdoutConsumer;
   private Destination destination;
   private Source source;
   private Path configPath;
-  private Path catalogPath;
   private Path configuredCatalogPath;
   private Path statePath;
 
@@ -99,7 +97,6 @@ void setup() throws IOException {
     Path configDir = Files.createTempDirectory(Files.createDirectories(TEST_ROOT), "test");
 
     configPath = IOs.writeFile(configDir, CONFIG_FILE_NAME, CONFIG_STRING);
-    catalogPath = IOs.writeFile(configDir, CATALOG_FILE_NAME, Jsons.serialize(CATALOG));
     configuredCatalogPath = IOs.writeFile(configDir, CONFIGURED_CATALOG_FILE_NAME, Jsons.serialize(CONFIGURED_CATALOG));
     statePath = IOs.writeFile(configDir, STATE_FILE_NAME, Jsons.serialize(STATE));
   }
@@ -185,11 +182,11 @@ void testRead() throws Exception {
             .withData(Jsons.jsonNode(ImmutableMap.of("names", "reginald"))));
 
     when(cliParser.parse(ARGS)).thenReturn(intConfig);
-    when(source.read(CONFIG, CONFIGURED_CATALOG, STATE.getState())).thenReturn(Stream.of(message1, message2));
+    when(source.read(CONFIG, CONFIGURED_CATALOG, STATE)).thenReturn(Stream.of(message1, message2));
 
     new IntegrationRunner(cliParser, stdoutConsumer, null, source).run(ARGS);
 
-    verify(source).read(CONFIG, CONFIGURED_CATALOG, STATE.getState());
+    verify(source).read(CONFIG, CONFIGURED_CATALOG, STATE);
     verify(stdoutConsumer).accept(Jsons.serialize(message1));
     verify(stdoutConsumer).accept(Jsons.serialize(message2));
   }

airbyte-integrations/bases/standard-source-test/src/main/java/io/airbyte/integrations/standardtest/source/TestSource.java

@@ -24,14 +24,13 @@
 
 package io.airbyte.integrations.standardtest.source;
 
-import static io.airbyte.protocol.models.SyncMode.FULL_REFRESH;
-import static io.airbyte.protocol.models.SyncMode.INCREMENTAL;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.Lists;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.config.JobGetSpecConfig;
 import io.airbyte.config.StandardCheckConnectionInput;
@@ -65,6 +64,7 @@
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.stream.Collectors;
 import org.junit.jupiter.api.AfterEach;
@@ -265,10 +265,10 @@ public void testIncrementalSyncWithState() throws Exception {
       return;
     }
 
-    ConfiguredAirbyteCatalog configuredAirbyteCatalog = withSourceDefinedCursors(getConfiguredCatalog());
-    List<AirbyteMessage> airbyteMessages = runRead(configuredAirbyteCatalog, getState());
-    List<AirbyteRecordMessage> recordMessages = filterRecords(airbyteMessages);
-    List<AirbyteStateMessage> stateMessages = airbyteMessages
+    final ConfiguredAirbyteCatalog configuredAirbyteCatalog = withSourceDefinedCursors(getConfiguredCatalog());
+    final List<AirbyteMessage> airbyteMessages = runRead(configuredAirbyteCatalog, getState());
+    final List<AirbyteRecordMessage> recordMessages = filterRecords(airbyteMessages);
+    final List<AirbyteStateMessage> stateMessages = airbyteMessages
         .stream()
         .filter(m -> m.getType() == Type.STATE)
         .map(AirbyteMessage::getState)
@@ -314,8 +314,18 @@ private List<AirbyteRecordMessage> filterRecords(Collection<AirbyteMessage> mess
   private ConfiguredAirbyteCatalog withSourceDefinedCursors(ConfiguredAirbyteCatalog catalog) {
     ConfiguredAirbyteCatalog clone = Jsons.clone(catalog);
     for (ConfiguredAirbyteStream configuredStream : clone.getStreams()) {
-      if (configuredStream.getSyncMode() == INCREMENTAL && configuredStream.getStream().getSourceDefinedCursor()) {
-        configuredStream.setCursorField(configuredStream.getStream().getDefaultCursorField());
+      if (configuredStream.getStream().getSupportedSyncModes().contains(io.airbyte.protocol.models.SyncMode.INCREMENTAL)) {
+        configuredStream.setSyncMode(io.airbyte.protocol.models.SyncMode.INCREMENTAL);
+        if (Optional.ofNullable(configuredStream.getStream().getSourceDefinedCursor()).orElse(false)) {
+          configuredStream.setCursorField(configuredStream.getStream().getDefaultCursorField());
+        } else {
+          // todo (cgardens) - this is really too terrible. there are some column types that aren't supported
+          // so you just need to order your columns such that we pick a valid one. too much guessing here. got
+          // to fix.
+          // see cursor field to an arbitrary field in the stream.
+          configuredStream
+              .setCursorField(Lists.newArrayList(new ArrayList<>(Jsons.keys(configuredStream.getStream().getJsonSchema().get("properties"))).get(0)));
+        }
       }
     }
     return clone;
@@ -324,8 +334,8 @@ private ConfiguredAirbyteCatalog withSourceDefinedCursors(ConfiguredAirbyteCatal
   private ConfiguredAirbyteCatalog withFullRefreshSyncModes(ConfiguredAirbyteCatalog catalog) {
     ConfiguredAirbyteCatalog clone = Jsons.clone(catalog);
     for (ConfiguredAirbyteStream configuredStream : clone.getStreams()) {
-      if (configuredStream.getStream().getSupportedSyncModes().contains(FULL_REFRESH)) {
-        configuredStream.setSyncMode(FULL_REFRESH);
+      if (configuredStream.getStream().getSupportedSyncModes().contains(io.airbyte.protocol.models.SyncMode.FULL_REFRESH)) {
+        configuredStream.setSyncMode(io.airbyte.protocol.models.SyncMode.FULL_REFRESH);
       }
     }
     return clone;
@@ -334,7 +344,7 @@ private ConfiguredAirbyteCatalog withFullRefreshSyncModes(ConfiguredAirbyteCatal
   private boolean sourceSupportsIncremental() throws Exception {
     ConfiguredAirbyteCatalog catalog = getConfiguredCatalog();
     for (ConfiguredAirbyteStream stream : catalog.getStreams()) {
-      if (stream.getStream().getSupportedSyncModes().contains(INCREMENTAL)) {
+      if (stream.getStream().getSupportedSyncModes().contains(io.airbyte.protocol.models.SyncMode.INCREMENTAL)) {
         return true;
       }
     }

airbyte-integrations/connectors/source-jdbc/build.gradle

@@ -1,15 +1,18 @@
+import org.jsonschema2pojo.SourceType
+
 plugins {
     id 'application'
     id 'airbyte-docker'
     id 'airbyte-integration-test-java'
-    // todo: needs standard source test
+    id 'com.github.eirnym.js2p' version '1.0'
 }
 
 application {
     mainClass = 'io.airbyte.integrations.source.jdbc.JdbcSource'
 }
 
 dependencies {
+    implementation project(':airbyte-commons')
     implementation project(':airbyte-db')
     implementation project(':airbyte-integrations:bases:base-java')
     implementation project(':airbyte-protocol:models')
@@ -22,3 +25,17 @@ dependencies {
 
     implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
 }
+
+jsonSchema2Pojo {
+    sourceType = SourceType.YAMLSCHEMA
+    source = files("${sourceSets.main.output.resourcesDir}/jdbc_models")
+    targetDirectory = new File(project.buildDir, 'generated/src/gen/java/')
+    removeOldOutput = true
+
+    targetPackage = 'io.airbyte.integrations.source.jdbc.models'
+
+    useLongIntegers = true
+    generateBuilders = true
+    includeConstructors = false
+    includeSetters = true
+}