bulk-cdk-toolkit-extract-jdbc: cosmetic changes (#44007)

Author: Marius Posta

airbyte-cdk/bulk/toolkits/extract-jdbc/src/main/kotlin/io/airbyte/cdk/command/JdbcSourceConfiguration.kt

@@ -1,7 +1,6 @@
 /* Copyright (c) 2024 Airbyte, Inc., all rights reserved. */
 package io.airbyte.cdk.command
 
-import io.airbyte.cdk.read.LimitState
 import io.micronaut.context.annotation.Factory
 import jakarta.inject.Singleton
 
@@ -19,10 +18,7 @@ interface JdbcSourceConfiguration : SourceConfiguration {
     /** Ordered set of schemas for the connector to consider. */
     val schemas: Set<String>
 
-    /** How many rows to query in the first batch. */
-    val initialLimit: LimitState
-        get() = LimitState.minimum
-
+    /** When set, each table is queried individually to check for SELECT privileges. */
     val checkPrivileges: Boolean
         get() = true
 

airbyte-cdk/bulk/toolkits/extract-jdbc/src/main/kotlin/io/airbyte/cdk/read/MemoryFetchSizeEstimator.kt

@@ -12,10 +12,10 @@ import kotlin.math.min
 class MemoryFetchSizeEstimator(
     val maxMemoryBytes: Long,
     val maxConcurrency: Int,
-) : StreamPartitionsCreatorUtils.FetchSizeEstimator {
+) {
     private val log = KotlinLogging.logger {}
 
-    override fun apply(rowByteSizeSample: Sample<Long>): Int {
+    fun apply(rowByteSizeSample: Sample<Long>): Int {
         val maxRowBytes: Long = rowByteSizeSample.sampledValues.maxOrNull() ?: 0L
         log.info {
             "maximum row size in ${rowByteSizeSample.kind.name} table is $maxRowBytes bytes"

airbyte-cdk/bulk/toolkits/extract-jdbc/src/main/kotlin/io/airbyte/cdk/read/StreamPartitionsCreatorUtils.kt

@@ -8,7 +8,6 @@ import io.airbyte.cdk.discover.Field
 import io.airbyte.cdk.util.Jsons
 import io.github.oshai.kotlinlogging.KotlinLogging
 import java.io.OutputStream
-import java.util.function.Function
 import kotlin.random.Random
 
 /** Utilities for [StreamPartitionsCreator] that don't rely directly on its input state. */
@@ -73,8 +72,6 @@ class StreamPartitionsCreatorUtils(
         return lbs.zip(ubs)
     }
 
-    fun interface FetchSizeEstimator : Function<Sample<Long>, Int>
-
     fun rowByteSizeEstimator(): (ObjectNode) -> Long {
         val countingOutputStream =
             object : OutputStream() {

airbyte-cdk/bulk/toolkits/extract-jdbc/src/main/kotlin/io/airbyte/cdk/read/StreamReadContext.kt

@@ -14,11 +14,10 @@ import java.util.concurrent.atomic.AtomicReference
 import kotlinx.coroutines.sync.Semaphore
 
 /**
- * A [StreamReadContextManager] may be injected in a
- * [io.airbyte.cdk.source.PartitionsCreatorFactory] to provide it, and the
- * [io.airbyte.cdk.source.PartitionsCreator] and [io.airbyte.cdk.source.PartitionReader] instances
- * it creates, with a set of global singletons useful for implementing stream READs for a JDBC
- * source.
+ * A [StreamReadContextManager] may be injected in a [io.airbyte.cdk.read.PartitionsCreatorFactory]
+ * to provide it, and the[io.airbyte.cdk.read.PartitionsCreator] and
+ * [io.airbyte.cdk.read.PartitionReader] instances it creates, with a set of global singletons
+ * useful for implementing stream READs for a JDBC source.
  *
  * For each stream in the configured catalog, these global singletons are packaged in a
  * [StreamReadContext] which bundles them with the corresponding [Stream] as well as a couple
@@ -61,7 +60,7 @@ class StreamReadContext(
     val outputConsumer: OutputConsumer,
     val stream: Stream,
 ) {
-    val transientLimitState: TransientState<LimitState> = TransientState(configuration.initialLimit)
+    val transientLimitState: TransientState<LimitState> = TransientState(LimitState.minimum)
 
     val transientCursorUpperBoundState: TransientState<JsonNode?> = TransientState(null)