🎉 Destination Redshift (copy): accept bucket path for staging data

airbyte-integrations/connectors/destination-databricks/src/main/java/io/airbyte/integrations/destination/databricks/DatabricksStreamCopier.java

@@ -10,6 +10,7 @@
 import io.airbyte.integrations.destination.ExtendedNameTransformer;
 import io.airbyte.integrations.destination.jdbc.SqlOperations;
 import io.airbyte.integrations.destination.jdbc.copy.StreamCopier;
+import io.airbyte.integrations.destination.jdbc.copy.s3.LegacyS3StreamCopier;
 import io.airbyte.integrations.destination.s3.S3DestinationConfig;
 import io.airbyte.integrations.destination.s3.parquet.S3ParquetFormatConfig;
 import io.airbyte.integrations.destination.s3.parquet.S3ParquetWriter;
@@ -24,7 +25,7 @@
 
 /**
  * This implementation is similar to
- * {@link io.airbyte.integrations.destination.jdbc.copy.s3.S3StreamCopier}. The difference is that
+ * {@link LegacyS3StreamCopier}. The difference is that
  * this implementation creates Parquet staging files, instead of CSV ones.
  * <p>
  * </p>

airbyte-integrations/connectors/destination-gcs/src/main/java/io/airbyte/integrations/destination/gcs/avro/GcsAvroWriter.java

@@ -38,11 +38,11 @@ public class GcsAvroWriter extends BaseGcsWriter implements S3Writer {
   private final DataFileWriter<GenericData.Record> dataFileWriter;
 
   public GcsAvroWriter(final GcsDestinationConfig config,
-                       final AmazonS3 s3Client,
-                       final ConfiguredAirbyteStream configuredStream,
-                       final Timestamp uploadTimestamp,
-                       final Schema schema,
-                       final JsonAvroConverter converter)
+      final AmazonS3 s3Client,
+      final ConfiguredAirbyteStream configuredStream,
+      final Timestamp uploadTimestamp,
+      final Schema schema,
+      final JsonAvroConverter converter)
       throws IOException {
     super(config, s3Client, configuredStream);
 
@@ -85,4 +85,9 @@ protected void closeWhenFail() throws IOException {
     uploadManager.abort();
   }
 
+  @Override
+  public String getObjectKey() {
+    // TODO
+    throw new UnsupportedOperationException("not yet implemented");
+  }
 }

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

@@ -38,9 +38,9 @@ public class GcsCsvWriter extends BaseGcsWriter implements S3Writer {
   private final String gcsCsvFileLocation; // this used in destination-bigquery (GCS upload type)
 
   public GcsCsvWriter(final GcsDestinationConfig config,
-                      final AmazonS3 s3Client,
-                      final ConfiguredAirbyteStream configuredStream,
-                      final Timestamp uploadTimestamp)
+      final AmazonS3 s3Client,
+      final ConfiguredAirbyteStream configuredStream,
+      final Timestamp uploadTimestamp)
       throws IOException {
     super(config, s3Client, configuredStream);
 
@@ -90,4 +90,9 @@ public CSVPrinter getCsvPrinter() {
     return csvPrinter;
   }
 
+  @Override
+  public String getObjectKey() {
+    // TODO
+    throw new UnsupportedOperationException("not yet implemented");
+  }
 }

airbyte-integrations/connectors/destination-gcs/src/main/java/io/airbyte/integrations/destination/gcs/jsonl/GcsJsonlWriter.java

@@ -37,9 +37,9 @@ public class GcsJsonlWriter extends BaseGcsWriter implements S3Writer {
   private final PrintWriter printWriter;
 
   public GcsJsonlWriter(final GcsDestinationConfig config,
-                        final AmazonS3 s3Client,
-                        final ConfiguredAirbyteStream configuredStream,
-                        final Timestamp uploadTimestamp) {
+      final AmazonS3 s3Client,
+      final ConfiguredAirbyteStream configuredStream,
+      final Timestamp uploadTimestamp) {
     super(config, s3Client, configuredStream);
 
     final String outputFilename = BaseGcsWriter.getOutputFilename(uploadTimestamp, S3Format.JSONL);
@@ -78,4 +78,9 @@ protected void closeWhenFail() {
     uploadManager.abort();
   }
 
+  @Override
+  public String getObjectKey() {
+    // TODO
+    throw new UnsupportedOperationException("not yet implemented");
+  }
 }

airbyte-integrations/connectors/destination-gcs/src/main/java/io/airbyte/integrations/destination/gcs/parquet/GcsParquetWriter.java

@@ -43,11 +43,11 @@ public class GcsParquetWriter extends BaseGcsWriter implements S3Writer {
   private final AvroRecordFactory avroRecordFactory;
 
   public GcsParquetWriter(final GcsDestinationConfig config,
-                          final AmazonS3 s3Client,
-                          final ConfiguredAirbyteStream configuredStream,
-                          final Timestamp uploadTimestamp,
-                          final Schema schema,
-                          final JsonAvroConverter converter)
+      final AmazonS3 s3Client,
+      final ConfiguredAirbyteStream configuredStream,
+      final Timestamp uploadTimestamp,
+      final Schema schema,
+      final JsonAvroConverter converter)
       throws URISyntaxException, IOException {
     super(config, s3Client, configuredStream);
 
@@ -109,4 +109,9 @@ public void close(final boolean hasFailed) throws IOException {
     }
   }
 
+  @Override
+  public String getObjectKey() {
+    // TODO
+    throw new UnsupportedOperationException("not yet implemented");
+  }
 }

airbyte-integrations/connectors/destination-gcs/src/main/java/io/airbyte/integrations/destination/gcs/writer/BaseGcsWriter.java

@@ -44,8 +44,8 @@ public abstract class BaseGcsWriter implements S3Writer {
   protected final String outputPrefix;
 
   protected BaseGcsWriter(final GcsDestinationConfig config,
-                          final AmazonS3 s3Client,
-                          final ConfiguredAirbyteStream configuredStream) {
+      final AmazonS3 s3Client,
+      final ConfiguredAirbyteStream configuredStream) {
     this.config = config;
     this.s3Client = s3Client;
     this.stream = configuredStream.getStream();
@@ -90,9 +90,8 @@ public void initialize() {
   }
 
   /**
-   * {@link AmazonS3#doesBucketExistV2} should be used to check the bucket existence. However, this
-   * method does not work for GCS. So we use {@link AmazonS3#headBucket} instead, which will throw an
-   * exception if the bucket does not exist, or there is no permission to access it.
+   * {@link AmazonS3#doesBucketExistV2} should be used to check the bucket existence. However, this method does not work for GCS. So we use {@link
+   * AmazonS3#headBucket} instead, which will throw an exception if the bucket does not exist, or there is no permission to access it.
    */
   public boolean gcsBucketExist(final AmazonS3 s3Client, final String bucket) {
     try {
@@ -141,4 +140,9 @@ public static String getOutputFilename(final Timestamp timestamp, final S3Format
         format.getFileExtension());
   }
 
+  @Override
+  public String getObjectKey() {
+    // TODO
+    throw new UnsupportedOperationException("not yet implemented");
+  }
 }

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

@@ -19,6 +19,7 @@ dependencies {
     implementation 'com.fasterxml.jackson.core:jackson-databind'
 
     testImplementation "org.testcontainers:postgresql:1.15.3"
+    testImplementation "org.mockito:mockito-inline:4.1.0"
 
     integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
     integrationTestJavaImplementation "org.testcontainers:postgresql:1.15.3"

airbyte-integrations/connectors/destination-jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/copy/CopyConsumerFactory.java

@@ -29,7 +29,7 @@ public class CopyConsumerFactory {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(CopyConsumerFactory.class);
 
-  private static final int MAX_BATCH_SIZE_BYTES = 1024 * 1024 * 1024 / 4; // 256 mib
+  private static final int MAX_BATCH_SIZE_BYTES = 1024 * 1024 * 1024 / 4; // 256 MiB
 
   public static <T> AirbyteMessageConsumer create(final Consumer<AirbyteMessage> outputRecordCollector,
                                                   final JdbcDatabase database,

airbyte-integrations/connectors/destination-jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/copy/StreamCopier.java

@@ -8,8 +8,7 @@
 import java.util.UUID;
 
 /**
- * StreamCopier is responsible for writing to a staging persistence and providing methods to remove
- * the staged data.
+ * StreamCopier is responsible for writing to a staging persistence and providing methods to remove the staged data.
  */
 public interface StreamCopier {
 
@@ -19,8 +18,8 @@ public interface StreamCopier {
   void write(UUID id, AirbyteRecordMessage recordMessage, String fileName) throws Exception;
 
   /**
-   * Closes the writer for the stream to the staging persistence. This method should block until all
-   * buffered data has been written to the persistence.
+   * Closes the writer for the stream to the staging persistence. This method should block until all buffered data has been written to the
+   * persistence.
    */
   void closeStagingUploader(boolean hasFailed) throws Exception;
 
@@ -30,8 +29,7 @@ public interface StreamCopier {
   void createTemporaryTable() throws Exception;
 
   /**
-   * Copies the staging file to the temporary table. This method should block until the copy/upload
-   * has completed.
+   * Copies the staging file to the temporary table. This method should block until the copy/upload has completed.
    */
   void copyStagingFileToTemporaryTable() throws Exception;
 
@@ -53,15 +51,14 @@ public interface StreamCopier {
   String generateMergeStatement(String destTableName) throws Exception;
 
   /**
-   * Cleans up the copier by removing the staging file and dropping the temporary table after
-   * completion or failure.
+   * Cleans up the copier by removing the staging file and dropping the temporary table after completion or failure.
    */
   void removeFileAndDropTmpTable() throws Exception;
 
   /**
    * Creates the staging file and all the necessary items to write data to this file.
    *
-   * @return the name of the staging file
+   * @return A string that unqiuely identifies the file. E.g. the filename, or a unique suffix that is appended to a shared filename prefix
    */
   String prepareStagingFile();