[Destination-S3] Uses New Load CDK LoadStrategy Interface (temporarily disabled)

[johnny-schmidt]

What

Migrate S3 (and object storage in general) to the new ObjectLoader interface using the LoadPipeline. This has been cloud tested for performance and reliability, but I'm leaving it disabled so that other connectors can use the code w/o waiting on a full S3 release.

How

Creates a new LoadStrategy called ObjectLoader. (See the documentation on the interface for usage guidelines for the connector dev. This will also be the foundation for any GCS, Azure, or other file-like interfaces. This is also the foundation for BulkLoad, in progress here.

If the ObjectLoader bean is present, also creates

• ObjectLoaderRecordToPartAccumulator: straight port of RecordToPartAccumulator, except that it uses the per-record interface instead of the old batch iterator interface
• ObjectLoaderPartToObjectAccumulator: same as above, but a straight port of PartToObjectAccumulator
• a step for each accumulator (just the minimum necessary to wrap each accumulator in a LoadPipelineStep)
• an ObjectLoaderPartQueue for passing finished parts from the first step to the second step (ObjectLoaderPartStep and ObjectLoaderObjectStep)
• an ObjectLoaderPipeline that sequences the two steps into a LoadPipeline for injection (presence of any LoadPipeline is enough to trigger the core using the new interface)

Additionally, there are round-robin partitioners to distribute the incoming records and generated parts to the steps

Also, I made a couple of minor changes to the pipeline interface

• Pipeline doesn't passed Reserved<...> around, it just has a callback for post-processing the message (it was impossible to make a 2-step pipeline otherwise)
• I changed the accumulator interface to suspend -- this caused ~20% performance increase for s3 (and had no effect on iceberg)

[vercel]

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

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	✅ Ready (Inspect)	Visit Preview	💬 Add feedback	Mar 19, 2025 4:20pm

[edgao]

seems reasonable. Had a few minor comments, plus questions about the PartToObjectAccumulator, and about how we're injecting the queues.

Can I get a link to the PR adding bulk loader? I was a bit nervous about the micronaut wiring you described on tuesday (with all the Replaces stuff, queues, etc.) - it felt like we're moving towards the same problems that platform has, with it being difficult to understand which beans are in play at any given moment (probably even worse, since we're doing stuff across classpaths, so intellij doesn't know how to discover all the beans)

but that code isn't in this PR, so 🤷

[johnny-schmidt]

seems reasonable. Had a few minor comments, plus questions about the PartToObjectAccumulator, and about how we're injecting the queues.

Can I get a link to the PR adding bulk loader? I was a bit nervous about the micronaut wiring you described on tuesday (with all the Replaces stuff, queues, etc.) - it felt like we're moving towards the same problems that platform has, with it being difficult to understand which beans are in play at any given moment (probably even worse, since we're doing stuff across classpaths, so intellij doesn't know how to discover all the beans)

but that code isn't in this PR, so 🤷

Everything that is implicitly included by some XLoader has a @Requires(XLoader::class)

The confusing part will be when something @Replaces something, but in practice you YLoader: XLoader and some of the things that @Require(YLoader) also replace something with a @Requires(XLoader).

@edgao

#55671

[edgao]

lgtm, had a few nits

[edgao]

does streamCompletionMap need to be in the LoadPipelineStepTask constructor? seems a bit weird to require every Step implementation to pass in this hashmap

[johnny-schmidt]

it needs to be shared across tasks and it needs to be affined to the step (ie, it can't be global).

I think breaking out a factory makes sense. i was thinking something like

LoadPipelineStepTaskFactory:

  createFirstStep(... arguments except input queue/input partitioner/batch queue/num workers/stream completions/output queue)

  createMiddleStep(...same as above but also takes input queue, but injects the partitioner)

  createFinalStep(...same as middle but doesn't require output)

[johnny-schmidt]

I did that and went ahead and updated DirectLoader. (I'm punting on Object loader, since it's about to change a little with BulkLoad and I want to avoid the rebase.)

ie,

@Singleton
@Requires(bean = DirectLoaderFactory::class)
class DirectLoadPipelineStep<S : DirectLoader>(
    val directLoaderFactory: DirectLoaderFactory<S>,
    val accumulator: DirectLoadRecordAccumulator<S, StreamKey>,
    val taskFactory: LoadPipelineStepTaskFactory,
) : LoadPipelineStep {
    private val log = KotlinLogging.logger {}
    override val numWorkers: Int = directLoaderFactory.inputPartitions

    override fun taskForPartition(partition: Int): LoadPipelineStepTask<*, *, *, *, *> {
        log.info { "Creating DirectLoad pipeline step task for partition $partition" }
        return taskFactory.createOnlyStep<S, StreamKey, DirectLoadAccResult>(accumulator, partition, numWorkers)
    }
}

[johnny-schmidt]

I also reran tests for S3DataLake (temporarily unpinned)

[johnny-schmidt]

13001de (I fixed the constructor, too)

[edgao]

nit: let's have an explicit flushStrategy = null here - wasn't immediately obvious that this is how we implement the enclosing step should be configured not to flush in UploadCompleter.finish

[johnny-schmidt]

That's hidden by the factory now.

In practice, forcing only makes sense on the first step (it will finish early, and that effect will propagate downstream.)

[edgao]

seems reasonable - I'll take your word that the factory createFirstStep / etc. methods make sense, since you're only calling createOnlyStep here

(but I think you meant to replace the constructor calls with factory calls, in the ObjectFooStep classes?)

[edgao]

did you mean to replace this with a factory call?

[johnny-schmidt]

I didn't make the changes to the object steps yet, but I will go ahead and do it

[johnny-schmidt]

I was trying to avoid a rebase but the changes are actually really light

[edgao]

needs to be affined to the step (ie, it can't be global)

should this be a function so that we get a new instance on each invocation?

[johnny-schmidt]

it's worse than that 🙃

we can't do that, because each individual task would get its own map. we need the counts to be affined to the step and shared across all the workers in the step.

After some local testing, the right pattern is a map of Pair<StepIndex, Descriptor> -> Count. Not too bad since the index is deterministic on the first, final, and only steps, so the factory can set it.

[edgao]

one nonblocking comment, feel free to

[edgao]

nonblocking: maybe nicer to use taskName:String? then we don't have magic numbers like -1, 2, etc. floating around, and is maybe also useful for future debugging