Incremental CSV destination

airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/8be1cf83-fde1-477f-a4ad-318d23c9f3c6.json

@@ -2,6 +2,6 @@
   "destinationDefinitionId": "8be1cf83-fde1-477f-a4ad-318d23c9f3c6",
   "name": "Local CSV",
   "dockerRepository": "airbyte/destination-csv",
-  "dockerImageTag": "0.1.4",
+  "dockerImageTag": "0.1.5",
   "documentationUrl": "https://hub.docker.com/r/airbyte/integration-singer-csv-destination"
 }

airbyte-config/init/src/main/resources/seed/destination_definitions.yaml

@@ -1,7 +1,7 @@
 - destinationDefinitionId: 8be1cf83-fde1-477f-a4ad-318d23c9f3c6
   name: Local CSV
   dockerRepository: airbyte/destination-csv
-  dockerImageTag: 0.1.4
+  dockerImageTag: 0.1.5
   documentationUrl: https://hub.docker.com/r/airbyte/integration-singer-csv-destination
 - destinationDefinitionId: 25c5221d-dce2-4163-ade9-739ef790f503
   name: Postgres

airbyte-integrations/connectors/destination-csv/Dockerfile

@@ -7,5 +7,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
 
 RUN tar xf ${APPLICATION}.tar --strip-components=1
 
-LABEL io.airbyte.version=0.1.4
+LABEL io.airbyte.version=0.1.5
 LABEL io.airbyte.name=airbyte/destination-csv

airbyte-integrations/connectors/destination-csv/src/main/java/io/airbyte/integrations/destination/csv/CsvDestination.java

@@ -40,11 +40,16 @@
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
 import io.airbyte.protocol.models.ConfiguredAirbyteStream;
 import io.airbyte.protocol.models.ConnectorSpecification;
+import io.airbyte.protocol.models.SyncMode;
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
 import java.io.FileWriter;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
 import java.time.Instant;
 import java.util.HashMap;
 import java.util.Map;
@@ -109,11 +114,12 @@ public DestinationConsumer<AirbyteMessage> write(JsonNode config, ConfiguredAirb
       final String streamName = stream.getStream().getName();
       final String tableName = getNamingResolver().getRawTableName(streamName);
       final String tmpTableName = getNamingResolver().getTmpTableName(streamName);
+      final SyncMode syncMode = stream.getSyncMode();
       final Path tmpPath = destinationDir.resolve(tmpTableName + ".csv");
       final Path finalPath = destinationDir.resolve(tableName + ".csv");
       final FileWriter fileWriter = new FileWriter(tmpPath.toFile());
       final CSVPrinter printer = new CSVPrinter(fileWriter, CSVFormat.DEFAULT.withHeader(COLUMN_AB_ID, COLUMN_EMITTED_AT, COLUMN_DATA));
-      writeConfigs.put(stream.getStream().getName(), new WriteConfig(printer, tmpPath, finalPath));
+      writeConfigs.put(stream.getStream().getName(), new WriteConfig(printer, tmpPath, finalPath, syncMode));
     }
 
     return new CsvConsumer(writeConfigs, catalog);
@@ -183,14 +189,41 @@ protected void close(boolean hasFailed) throws IOException {
       // do not persist the data, if there are any failures.
       if (!hasFailed) {
         for (final WriteConfig writeConfig : writeConfigs.values()) {
-          Files.move(writeConfig.getTmpPath(), writeConfig.getFinalPath(), StandardCopyOption.REPLACE_EXISTING);
+          final boolean fileAlreadyExists = writeConfig.getFinalPath().toFile().exists();
+          if (writeConfig.getSyncMode() == SyncMode.FULL_REFRESH || !fileAlreadyExists) {
+            Files.move(writeConfig.getTmpPath(), writeConfig.getFinalPath(), StandardCopyOption.REPLACE_EXISTING);
+          } else if (writeConfig.getSyncMode() == SyncMode.INCREMENTAL) {
+            insertCsvFile(writeConfig.getTmpPath(), writeConfig.getFinalPath());
+          }
         }
       }
       // clean up tmp files.
       for (final WriteConfig writeConfig : writeConfigs.values()) {
         Files.deleteIfExists(writeConfig.getTmpPath());
       }
+    }
 
+    /**
+     * Copy and append Csv file to another
+     *
+     * @param srcFilePath CSV file to append data from
+     * @param dstFilePath CSV file to append data to
+     * @throws IOException
+     */
+    private static void insertCsvFile(Path srcFilePath, Path dstFilePath) throws IOException {
+      try (
+          final BufferedReader reader = Files.newBufferedReader(srcFilePath);
+          final BufferedWriter writer = Files.newBufferedWriter(dstFilePath, StandardCharsets.UTF_8, StandardOpenOption.CREATE,
+              StandardOpenOption.APPEND)) {
+        // Skip header line
+        reader.readLine();
+        String line;
+        while ((line = reader.readLine()) != null) {
+          writer.write(line);
+          writer.newLine();
+        }
+        writer.flush();
+      }
     }
 
   }
@@ -200,11 +233,13 @@ private static class WriteConfig {
     private final CSVPrinter writer;
     private final Path tmpPath;
     private final Path finalPath;
+    private final SyncMode syncMode;
 
-    public WriteConfig(CSVPrinter writer, Path tmpPath, Path finalPath) {
+    public WriteConfig(CSVPrinter writer, Path tmpPath, Path finalPath, SyncMode syncMode) {
       this.writer = writer;
       this.tmpPath = tmpPath;
       this.finalPath = finalPath;
+      this.syncMode = syncMode;
     }
 
     public CSVPrinter getWriter() {
@@ -219,6 +254,10 @@ public Path getFinalPath() {
       return finalPath;
     }
 
+    public SyncMode getSyncMode() {
+      return syncMode;
+    }
+
   }
 
   public static void main(String[] args) throws Exception {

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

@@ -1,5 +1,6 @@
 {
   "documentationUrl": "https://docs.airbyte.io/integrations/destinations/local-csv",
+  "supportsIncremental": true,
   "connectionSpecification": {
     "$schema": "http://json-schema.org/draft-07/schema#",
     "title": "CSV Destination Spec",