Bulk Load CDK: CSV Support, S3V2Usage

[johnny-schmidt]

What

CSV. Uses the old printer with AirbyteValue/Schema directly.

Conversions aren't complete, and there are no tests. I will come back and shore all that up after I've proved out Avro and Parquet.

Bonuses

• I add a cleaner, but didn't wire it up yet because it has dependencies
• I figured out how to get java.io.Writer to work w/o resorting to runBlocking, so streamingUpload now accepts a block that works on Writer, allowing the apache CSVPrinter to wrap it; some callouts:
  • it requires async dispatch in an anonymous scope, so the s3client has to call start() and await the returned job (wonky, but limited to the client so the implementor still gets a clean interface)
  • Writer::close is a noop (the contract is that the writer is closed when the block exits)
  • flush throws, since there really isn't a need for it, but maybe later I'll wire it to uploadPart if there's a need

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Skipped Deployment

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	⬜️ Ignored (Inspect)	Visit Preview		Oct 18, 2024 11:44pm

[edgao]

wasn't sure about the multipart upload stuff, everything else makes sense

[edgao]

isn't the serialized string [1,2,3]? i.e shouldn't we also strip the square brackets (or maybe just parse it as json)

[johnny-schmidt]

Should be is ArrayType -> value.deserializeToNode().elements().asSequence().map { it.toAirbyteValue(field.items.type) }.toList().let(::ArrayValue)

[edgao]

is this supposed to be unit()? it doesn't seem like we ever actually invoke the function

[johnny-schmidt]

I'd partially converted it to a Unit channel, since all the work items were uploadPart. Fixed.

[edgao]

not sure I follow here - don't we need to call uploadPart() directly to avoid a race condition? (i.e. someone calls write() while uploadPart is still in-flight)

or should this be work.send { uploadPart(underlyingBuffer); underlyingBuffer = newBuffer() }?

[johnny-schmidt]

You're right, we need to sequence the work items. The most straightforward thing to do is just to push whatever would have been in the body of the function into the channel.

    suspend fun run(block: suspend (OutputStream) -> Unit) = coroutineScope {
        log.info {
            "Starting multipart upload to ${response.bucket}/${response.key} (${response.uploadId}"
        }
        launch {
            for (item in workQueue) {
                item()
            }
            complete()
        }
        UploadStream().use {
            block(it)
        }
    }

    inner class UploadStream : OutputStream() {
        override fun close() = runBlocking {
            workQueue.send { workQueue.close() }
        }

        override fun flush() = runBlocking {
            workQueue.send { wrappingBuffer.flush() }
        }

        override fun write(b: Int) = runBlocking {
            workQueue.send {
                wrappingBuffer.write(b)
                if (underlyingBuffer.size() >= partSize) {
                    uploadPart()
                }
            }
        }

        override fun write(b: ByteArray) = runBlocking {
            workQueue.send {
                wrappingBuffer.write(b)
                if (underlyingBuffer.size() >= partSize) {
                    uploadPart()
                }
            }
        }
    }

This also simplifies the caller to

        val upload =
            S3MultipartUpload(
                client,
                response,
                ByteArrayOutputStream(),
                streamProcessor,
                uploadConfig
            )
        upload.run(block)

Also note I converted the writer into an output stream. This will be necessary for Avro.

[edgao]

(copying from DM - let's add some comments about why we need the channel -> why we need to be an OutputStream, but otherwise lgtm)