diff --git a/client-dynamodb-v2/src/main/kotlin/app/cash/backfila/client/dynamodbv2/DynamoDbBackfill.kt b/client-dynamodb-v2/src/main/kotlin/app/cash/backfila/client/dynamodbv2/DynamoDbBackfill.kt
index 378aab7f7..c1061289b 100644
--- a/client-dynamodb-v2/src/main/kotlin/app/cash/backfila/client/dynamodbv2/DynamoDbBackfill.kt
+++ b/client-dynamodb-v2/src/main/kotlin/app/cash/backfila/client/dynamodbv2/DynamoDbBackfill.kt
@@ -27,7 +27,7 @@ abstract class DynamoDbBackfill<I : Any, P : Any> : Backfill {
     // Like MyBackfill.
     val thisType = TypeLiteral.get(this::class.java)
 
-    // Like Backfill<MyItem, Parameters>.
+    // Like MyBackfill<MyItem, Parameters>.
     val supertype = thisType.getSupertype(
       DynamoDbBackfill::class.java,
     ).type as ParameterizedType
diff --git a/client-dynamodb/src/main/kotlin/app/cash/backfila/client/dynamodb/DynamoDbBackfill.kt b/client-dynamodb/src/main/kotlin/app/cash/backfila/client/dynamodb/DynamoDbBackfill.kt
index 5e15e039f..856c94fc1 100644
--- a/client-dynamodb/src/main/kotlin/app/cash/backfila/client/dynamodb/DynamoDbBackfill.kt
+++ b/client-dynamodb/src/main/kotlin/app/cash/backfila/client/dynamodb/DynamoDbBackfill.kt
@@ -21,7 +21,7 @@ abstract class DynamoDbBackfill<I : Any, P : Any> : Backfill {
     // Like MyBackfill.
     val thisType = TypeLiteral.get(this::class.java)
 
-    // Like Backfill<MyItem, Parameters>.
+    // Like MyBackfill<MyItem, Parameters>.
     val supertype = thisType.getSupertype(
       DynamoDbBackfill::class.java,
     ).type as ParameterizedType
diff --git a/client-jooq/src/test/kotlin/app/cash/backfila/client/jooq/MiskJooqBackfillTests.kt b/client-jooq/src/test/kotlin/app/cash/backfila/client/jooq/MiskJooqBackfillTests.kt
index aa54a3f7a..ebdfc9c66 100644
--- a/client-jooq/src/test/kotlin/app/cash/backfila/client/jooq/MiskJooqBackfillTests.kt
+++ b/client-jooq/src/test/kotlin/app/cash/backfila/client/jooq/MiskJooqBackfillTests.kt
@@ -71,7 +71,7 @@ class MiskJooqBackfillTests {
     assertThat(run.partitionProgressSnapshot.values.single().previousEndKey).isNull()
 
     val scan1 = run.singleScan()
-    assertThat(scan1.batches).size().isEqualTo(1)
+    assertThat(scan1.batches).hasSize(1)
     assertThat(scan1.batches.single().scanned_record_count).isEqualTo(5)
     assertThat(scan1.batches.single().matching_record_count).isEqualTo(0)
     testingAssertThat(run.partitionProgressSnapshot.values.single())
@@ -204,7 +204,7 @@ class MiskJooqBackfillTests {
     run.computeCountLimit = 1L
 
     val scan1 = run.precomputeScan()
-    assertThat(scan1.batches).size().isEqualTo(1)
+    assertThat(scan1.batches).hasSize(1)
     val batch1 = scan1.batches.single()
     assertThat(batch1.batch_range.start.utf8()).isEqualTo(backfillRowKeys[0].toString())
     assertThat(batch1.matching_record_count).isEqualTo(10)
@@ -212,7 +212,7 @@ class MiskJooqBackfillTests {
 
     run.scanSize = 20L
     val scan2 = run.precomputeScan()
-    assertThat(scan2.batches).size().isEqualTo(1)
+    assertThat(scan2.batches).hasSize(1)
     val batch2 = scan2.batches.single()
     assertThat(batch2.matching_record_count).isEqualTo(10)
     // 5 extra were scanned and skipped, because they were interspersed.
@@ -247,7 +247,7 @@ class MiskJooqBackfillTests {
 
     run1.computeCountLimit = 2
     val scan = run1.singleScan()
-    assertThat(scan.batches).size().isEqualTo(2)
+    assertThat(scan.batches).hasSize(2)
     assertThat(scan.batches[0].batch_range.end).isLessThan(scan.batches[1].batch_range.start)
 
     // Requesting two batches should give the same batches as requesting one twice.
@@ -271,7 +271,7 @@ class MiskJooqBackfillTests {
     run1.scanSize = 4L
     run1.computeCountLimit = 3
     val scan = run1.singleScan()
-    assertThat(scan.batches).size().isEqualTo(3)
+    assertThat(scan.batches).hasSize(3)
     assertThat(scan.batches[0].batch_range.end).isLessThan(scan.batches[1].batch_range.start)
     assertThat(scan.batches[1].batch_range.end).isLessThan(scan.batches[2].batch_range.start)
 
diff --git a/client-misk-hibernate/src/test/kotlin/app/cash/backfila/client/misk/hibernate/SinglePartitionHibernateBackfillTest.kt b/client-misk-hibernate/src/test/kotlin/app/cash/backfila/client/misk/hibernate/SinglePartitionHibernateBackfillTest.kt
index dc56d7dd2..a5bb93714 100644
--- a/client-misk-hibernate/src/test/kotlin/app/cash/backfila/client/misk/hibernate/SinglePartitionHibernateBackfillTest.kt
+++ b/client-misk-hibernate/src/test/kotlin/app/cash/backfila/client/misk/hibernate/SinglePartitionHibernateBackfillTest.kt
@@ -44,7 +44,7 @@ abstract class SinglePartitionHibernateBackfillTest {
     assertThat(run.partitionProgressSnapshot.values.single().previousEndKey).isNull()
 
     val scan1 = run.singleScan()
-    assertThat(scan1.batches).size().isEqualTo(1)
+    assertThat(scan1.batches).hasSize(1)
     assertThat(scan1.batches.single().scanned_record_count).isEqualTo(5)
     assertThat(scan1.batches.single().matching_record_count).isEqualTo(0)
     assertThat(run.partitionProgressSnapshot.values.single()).isNotDone()
@@ -150,7 +150,7 @@ abstract class SinglePartitionHibernateBackfillTest {
     run.computeCountLimit = 1L
 
     val scan1 = run.precomputeScan()
-    assertThat(scan1.batches).size().isEqualTo(1)
+    assertThat(scan1.batches).hasSize(1)
     val batch1 = scan1.batches.single()
     assertThat(batch1.batch_range.start.utf8()).isEqualTo(expectedIds[0].toString())
     assertThat(batch1.matching_record_count).isEqualTo(10)
@@ -158,7 +158,7 @@ abstract class SinglePartitionHibernateBackfillTest {
 
     run.scanSize = 20L
     val scan2 = run.precomputeScan()
-    assertThat(scan2.batches).size().isEqualTo(1)
+    assertThat(scan2.batches).hasSize(1)
     val batch2 = scan2.batches.single()
     assertThat(batch2.matching_record_count).isEqualTo(10)
     // 5 extra were scanned and skipped, because they were interspersed.
@@ -188,7 +188,7 @@ abstract class SinglePartitionHibernateBackfillTest {
 
     run1.computeCountLimit = 2
     val scan = run1.singleScan()
-    assertThat(scan.batches).size().isEqualTo(2)
+    assertThat(scan.batches).hasSize(2)
     assertThat(scan.batches[0].batch_range.end).isLessThan(scan.batches[1].batch_range.start)
 
     // Requesting two batches should give the same batches as requesting one twice.
@@ -208,7 +208,7 @@ abstract class SinglePartitionHibernateBackfillTest {
     run1.scanSize = 4L
     run1.computeCountLimit = 3
     val scan = run1.singleScan()
-    assertThat(scan.batches).size().isEqualTo(3)
+    assertThat(scan.batches).hasSize(3)
     assertThat(scan.batches[0].batch_range.end).isLessThan(scan.batches[1].batch_range.start)
     assertThat(scan.batches[1].batch_range.end).isLessThan(scan.batches[2].batch_range.start)
 
@@ -272,7 +272,7 @@ abstract class SinglePartitionHibernateBackfillTest {
       .apply { configureForTest() }
 
     run.execute()
-    assertThat(run.backfill.idsRanDry).size().isEqualTo(5)
+    assertThat(run.backfill.idsRanDry).hasSize(5)
     assertThat(run.backfill.idsRanWet).isEmpty()
     // We got beef as a parameter
     assertThat(run.backfill.parametersLog).containsExactly(SandwichParameters("beef"))
@@ -300,7 +300,7 @@ abstract class SinglePartitionHibernateBackfillTest {
       .apply { configureForTest() }
 
     run.execute()
-    assertThat(run.backfill.idsRanDry).size().isEqualTo(20)
+    assertThat(run.backfill.idsRanDry).hasSize(20)
     assertThat(run.backfill.idsRanWet).isEmpty()
     // Null parameter used the default
     assertThat(run.backfill.parametersLog).contains(SandwichParameters("chicken"))
diff --git a/client-s3/src/test/kotlin/app/cash/backfila/client/s3/S3Utf8StringNewlineBackfillTest.kt b/client-s3/src/test/kotlin/app/cash/backfila/client/s3/S3Utf8StringNewlineBackfillTest.kt
index 302368b04..bc4d169ac 100644
--- a/client-s3/src/test/kotlin/app/cash/backfila/client/s3/S3Utf8StringNewlineBackfillTest.kt
+++ b/client-s3/src/test/kotlin/app/cash/backfila/client/s3/S3Utf8StringNewlineBackfillTest.kt
@@ -58,7 +58,7 @@ class S3Utf8StringNewlineBackfillTest {
       .containsExactlyInAnyOrder("main-blt")
 
     // See that we processed two empty lines
-    assertThat(run.backfill.backfilledIngredients.filter { it.second == "" }).size().isEqualTo(2)
+    assertThat(run.backfill.backfilledIngredients.filter { it.second == "" }).hasSize(2)
 
     // Now run efficiently aka "without empty ingredient lines"
     val optimizedRun = backfila.createWetRun<OptimizedLunchBackfill>(parameters = RecipeAttributes())
diff --git a/client-static/src/main/kotlin/app/cash/backfila/client/stat/StaticDatasourceBackfill.kt b/client-static/src/main/kotlin/app/cash/backfila/client/stat/StaticDatasourceBackfill.kt
index be5df57f8..39076bd85 100644
--- a/client-static/src/main/kotlin/app/cash/backfila/client/stat/StaticDatasourceBackfill.kt
+++ b/client-static/src/main/kotlin/app/cash/backfila/client/stat/StaticDatasourceBackfill.kt
@@ -1,57 +1,12 @@
 package app.cash.backfila.client.stat
 
-import app.cash.backfila.client.Backfill
-import app.cash.backfila.client.BackfillConfig
 import app.cash.backfila.client.PrepareBackfillConfig
-import com.google.inject.TypeLiteral
-import com.squareup.moshi.Types
-import java.lang.reflect.ParameterizedType
-import kotlin.reflect.KClass
-
-abstract class StaticDatasourceBackfill<I : Any, P : Any> : Backfill {
-  val itemType: KClass<I>
-
-  /*
-   * Extract the type parameters from the subtype's generic declaration. This uses Guice magic to
-   * read the ("I") type parameter.
-   */
-  init {
-    // Like MyBackfill.
-    val thisType = TypeLiteral.get(this::class.java)
-
-    // Like Backfill<MyItem, Parameters>.
-    val supertype = thisType.getSupertype(
-      StaticDatasourceBackfill::class.java,
-    ).type as ParameterizedType
-
-    // Like MyItem.
-    @Suppress("UNCHECKED_CAST")
-    itemType = (Types.getRawType(supertype.actualTypeArguments[0]) as Class<I>).kotlin
-  }
-
-  /**
-   * Override this and throw an exception to prevent the backfill from being created.
-   * This is also a good place to do any prep work before batches are run.
-   */
-  open fun validate(config: PrepareBackfillConfig<P>) {}
 
+abstract class StaticDatasourceBackfill<I : Any, P : Any> : StaticDatasourceBackfillBase<I, P>() {
   /**
-   * Called for each batch of matching records.
-   * Override in a backfill to process all records in a batch.
-   */
-  open fun runBatch(items: List<@JvmSuppressWildcards I>, config: BackfillConfig<P>) {
-    items.forEach { runOne(it, config) }
-  }
-
-  /**
-   * Called for each matching record.
-   * Override in a backfill to process one record at a time.
-   */
-  open fun runOne(item: I, config: BackfillConfig<P>) {
-  }
-
-  /**
-   * This invokes the static list of items that the backfill will iterate over.
+   * This provides the static list of items that the backfill will iterate over.
    */
   abstract val staticDatasource: List<I>
+
+  override fun getStaticDatasource(config: PrepareBackfillConfig<P>) = staticDatasource
 }
diff --git a/client-static/src/main/kotlin/app/cash/backfila/client/stat/StaticDatasourceBackfillBase.kt b/client-static/src/main/kotlin/app/cash/backfila/client/stat/StaticDatasourceBackfillBase.kt
new file mode 100644
index 000000000..69c322663
--- /dev/null
+++ b/client-static/src/main/kotlin/app/cash/backfila/client/stat/StaticDatasourceBackfillBase.kt
@@ -0,0 +1,54 @@
+package app.cash.backfila.client.stat
+
+import app.cash.backfila.client.Backfill
+import app.cash.backfila.client.BackfillConfig
+import app.cash.backfila.client.PrepareBackfillConfig
+import com.google.inject.TypeLiteral
+import com.squareup.moshi.Types
+import java.lang.reflect.ParameterizedType
+import kotlin.reflect.KClass
+
+abstract class StaticDatasourceBackfillBase<I : Any, P : Any> : Backfill {
+  val itemType: KClass<I>
+
+  /*
+   * Extract the type parameters from the subtype's generic declaration. This uses Guice magic to
+   * read the ("I") type parameter.
+   */
+  init {
+    // Like MyBackfill.
+    val thisType = TypeLiteral.get(this::class.java)
+
+    // Like MyBackfill<MyItem, Parameters>.
+    val supertype = thisType.getSupertype(
+      StaticDatasourceBackfillBase::class.java,
+    ).type as ParameterizedType
+
+    // Like MyItem.
+    @Suppress("UNCHECKED_CAST")
+    itemType = (Types.getRawType(supertype.actualTypeArguments[0]) as Class<I>).kotlin
+  }
+
+  /**
+   * Override this and throw an exception to prevent the backfill from being created.
+   * This is also a good place to do any prep work before batches are run.
+   */
+  open fun validate(config: PrepareBackfillConfig<P>) {}
+
+  /**
+   * Called for each batch of matching records.
+   * Override in a backfill to process all records in a batch.
+   */
+  open fun runBatch(items: List<@JvmSuppressWildcards I>, config: BackfillConfig<P>) {
+    items.forEach { runOne(it, config) }
+  }
+
+  /**
+   * Called for each matching record.
+   * Override in a backfill to process one record at a time.
+   */
+  open fun runOne(item: I, config: BackfillConfig<P>) {
+  }
+
+  abstract fun getStaticDatasource(config: PrepareBackfillConfig<P>): List<@JvmSuppressWildcards I>
+}
diff --git a/client-static/src/main/kotlin/app/cash/backfila/client/stat/StaticDatasourceBackfillModule.kt b/client-static/src/main/kotlin/app/cash/backfila/client/stat/StaticDatasourceBackfillModule.kt
index b1cfd0298..0e617ef92 100644
--- a/client-static/src/main/kotlin/app/cash/backfila/client/stat/StaticDatasourceBackfillModule.kt
+++ b/client-static/src/main/kotlin/app/cash/backfila/client/stat/StaticDatasourceBackfillModule.kt
@@ -14,7 +14,7 @@ import kotlin.reflect.jvm.jvmName
 /**
  * Installs the [BackfillBackend] for Static Datasource backfills. See the java doc for [RealBackfillModule].
  */
-class StaticDatasourceBackfillModule<T : StaticDatasourceBackfill<*, *>> private constructor(
+class StaticDatasourceBackfillModule<T : StaticDatasourceBackfillBase<*, *>> private constructor(
   private val backfillClass: KClass<T>,
 ) : AbstractModule() {
   override fun configure() {
@@ -26,15 +26,15 @@ class StaticDatasourceBackfillModule<T : StaticDatasourceBackfill<*, *>> private
   }
 
   companion object {
-    inline fun <reified T : StaticDatasourceBackfill<*, *>> create(): StaticDatasourceBackfillModule<T> = create(T::class)
+    inline fun <reified T : StaticDatasourceBackfillBase<*, *>> create(): StaticDatasourceBackfillModule<T> = create(T::class)
 
     @JvmStatic
-    fun <T : StaticDatasourceBackfill<*, *>> create(backfillClass: KClass<T>): StaticDatasourceBackfillModule<T> {
+    fun <T : StaticDatasourceBackfillBase<*, *>> create(backfillClass: KClass<T>): StaticDatasourceBackfillModule<T> {
       return StaticDatasourceBackfillModule(backfillClass)
     }
 
     @JvmStatic
-    fun <T : StaticDatasourceBackfill<*, *>> create(backfillClass: Class<T>): StaticDatasourceBackfillModule<T> {
+    fun <T : StaticDatasourceBackfillBase<*, *>> create(backfillClass: Class<T>): StaticDatasourceBackfillModule<T> {
       return StaticDatasourceBackfillModule(backfillClass.kotlin)
     }
   }
@@ -53,7 +53,7 @@ private object StaticDatasourceBackfillBackendModule : AbstractModule() {
 private fun mapBinder(binder: Binder) = MapBinder.newMapBinder(
   binder,
   object : TypeLiteral<String>() {},
-  object : TypeLiteral<KClass<out StaticDatasourceBackfill<*, *>>>() {},
+  object : TypeLiteral<KClass<out StaticDatasourceBackfillBase<*, *>>>() {},
   ForStaticBackend::class.java,
 )
 
diff --git a/client-static/src/main/kotlin/app/cash/backfila/client/stat/internal/StaticDatasourceBackend.kt b/client-static/src/main/kotlin/app/cash/backfila/client/stat/internal/StaticDatasourceBackend.kt
index 2eec154da..f2ddbee55 100644
--- a/client-static/src/main/kotlin/app/cash/backfila/client/stat/internal/StaticDatasourceBackend.kt
+++ b/client-static/src/main/kotlin/app/cash/backfila/client/stat/internal/StaticDatasourceBackend.kt
@@ -8,7 +8,7 @@ import app.cash.backfila.client.spi.BackfillBackend
 import app.cash.backfila.client.spi.BackfillOperator
 import app.cash.backfila.client.spi.BackfillRegistration
 import app.cash.backfila.client.stat.ForStaticBackend
-import app.cash.backfila.client.stat.StaticDatasourceBackfill
+import app.cash.backfila.client.stat.StaticDatasourceBackfillBase
 import com.google.inject.Injector
 import com.google.inject.TypeLiteral
 import com.squareup.moshi.Types
@@ -21,21 +21,21 @@ import kotlin.reflect.full.findAnnotation
 @Singleton
 class StaticDatasourceBackend @Inject constructor(
   private val injector: Injector,
-  @ForStaticBackend private val backfills: MutableMap<String, KClass<out StaticDatasourceBackfill<*, *>>>,
+  @ForStaticBackend private val backfills: MutableMap<String, KClass<out StaticDatasourceBackfillBase<*, *>>>,
 ) : BackfillBackend {
 
   /** Creates Backfill instances. Each backfill ID gets a new Backfill instance. */
-  private fun getBackfill(name: String): StaticDatasourceBackfill<*, *>? {
+  private fun getBackfill(name: String): StaticDatasourceBackfillBase<*, *>? {
     val backfillClass = backfills[name]
     return if (backfillClass != null) {
-      injector.getInstance(backfillClass.java) as StaticDatasourceBackfill<*, *>
+      injector.getInstance(backfillClass.java) as StaticDatasourceBackfillBase<*, *>
     } else {
       null
     }
   }
 
   private fun <E : Any, Param : Any> createStaticDatasourceOperator(
-    backfill: StaticDatasourceBackfill<E, Param>,
+    backfill: StaticDatasourceBackfillBase<E, Param>,
   ) = StaticDatasourceBackfillOperator(
     backfill,
     BackfilaParametersOperator(parametersClass(backfill::class)),
@@ -46,7 +46,7 @@ class StaticDatasourceBackend @Inject constructor(
 
     if (backfill != null) {
       @Suppress("UNCHECKED_CAST") // We don't know the types statically, so fake them.
-      return createStaticDatasourceOperator(backfill as StaticDatasourceBackfill<Any, Any>)
+      return createStaticDatasourceOperator(backfill as StaticDatasourceBackfillBase<Any, Any>)
     }
 
     return null
@@ -57,18 +57,18 @@ class StaticDatasourceBackend @Inject constructor(
       BackfillRegistration(
         name = it.key,
         description = it.value.findAnnotation<Description>()?.text,
-        parametersClass = parametersClass(it.value as KClass<StaticDatasourceBackfill<Any, Any>>),
+        parametersClass = parametersClass(it.value as KClass<StaticDatasourceBackfillBase<Any, Any>>),
         deleteBy = it.value.findAnnotation<DeleteBy>()?.parseDeleteByDate(),
       )
     }.toSet()
   }
 
-  private fun <P : Any> parametersClass(backfillClass: KClass<out StaticDatasourceBackfill<*, P>>): KClass<P> {
+  private fun <P : Any> parametersClass(backfillClass: KClass<out StaticDatasourceBackfillBase<*, P>>): KClass<P> {
     // Like MyBackfill.
     val thisType = TypeLiteral.get(backfillClass.java)
 
-    // Like StaticDatasourceBackfill<MyItemClass, MyParameterClass>.
-    val supertype = thisType.getSupertype(StaticDatasourceBackfill::class.java).type as ParameterizedType
+    // Like StaticDatasourceBackfillBase<MyItemClass, MyParameterClass>.
+    val supertype = thisType.getSupertype(StaticDatasourceBackfillBase::class.java).type as ParameterizedType
 
     // Like MyParameterClass
     return (Types.getRawType(supertype.actualTypeArguments[1]) as Class<P>).kotlin
diff --git a/client-static/src/main/kotlin/app/cash/backfila/client/stat/internal/StaticDatasourceBackfillOperator.kt b/client-static/src/main/kotlin/app/cash/backfila/client/stat/internal/StaticDatasourceBackfillOperator.kt
index 696cddef3..876299907 100644
--- a/client-static/src/main/kotlin/app/cash/backfila/client/stat/internal/StaticDatasourceBackfillOperator.kt
+++ b/client-static/src/main/kotlin/app/cash/backfila/client/stat/internal/StaticDatasourceBackfillOperator.kt
@@ -2,7 +2,7 @@ package app.cash.backfila.client.stat.internal
 
 import app.cash.backfila.client.spi.BackfilaParametersOperator
 import app.cash.backfila.client.spi.BackfillOperator
-import app.cash.backfila.client.stat.StaticDatasourceBackfill
+import app.cash.backfila.client.stat.StaticDatasourceBackfillBase
 import app.cash.backfila.protos.clientservice.GetNextBatchRangeRequest
 import app.cash.backfila.protos.clientservice.GetNextBatchRangeResponse
 import app.cash.backfila.protos.clientservice.KeyRange
@@ -14,7 +14,7 @@ import com.google.common.base.Preconditions.checkArgument
 import okio.ByteString.Companion.encodeUtf8
 
 class StaticDatasourceBackfillOperator<I : Any, P : Any>(
-  override val backfill: StaticDatasourceBackfill<I, P>,
+  override val backfill: StaticDatasourceBackfillBase<I, P>,
   val parametersOperator: BackfilaParametersOperator<P>,
 ) : BackfillOperator {
 
@@ -23,13 +23,14 @@ class StaticDatasourceBackfillOperator<I : Any, P : Any>(
   override fun prepareBackfill(request: PrepareBackfillRequest): PrepareBackfillResponse {
     val config = parametersOperator.constructBackfillConfig(request)
     backfill.validate(config)
+    val datasource = backfill.getStaticDatasource(config)
 
     val start = request.range?.start?.utf8()?.let {
       it.toIntOrNull() ?: error("Start of range must be a number")
     } ?: 0
     val end = request.range?.end?.utf8()?.let {
       it.toIntOrNull() ?: error("End of range must be a number")
-    } ?: backfill.staticDatasource.size
+    } ?: datasource.size
 
     // Sanity check that this backfill will actually process something
     require(start >= 0 && end >= 0) {
@@ -38,8 +39,8 @@ class StaticDatasourceBackfillOperator<I : Any, P : Any>(
     require(start <= end) {
       "Start must be less than or equal to end, start: $start end: $end"
     }
-    require(start <= backfill.staticDatasource.size) {
-      "Start is greater than the static datasource size, start: $start size: ${backfill.staticDatasource.size}"
+    require(start <= datasource.size) {
+      "Start is greater than the static datasource size, start: $start size: ${datasource.size}"
     }
 
     val onlyPartition = PrepareBackfillResponse.Partition.Builder()
@@ -91,7 +92,7 @@ class StaticDatasourceBackfillOperator<I : Any, P : Any>(
     val batchRange = request.batch_range.decode()
     val config = parametersOperator.constructBackfillConfig(request)
 
-    val batch = backfill.staticDatasource.subList(batchRange.start, batchRange.end)
+    val batch = backfill.getStaticDatasource(config.prepareConfig()).subList(batchRange.start, batchRange.end)
 
     backfill.runBatch(batch, config)
 
diff --git a/client-static/src/main/kotlin/app/cash/backfila/client/stat/parameters/ParametersDatasourceBackfill.kt b/client-static/src/main/kotlin/app/cash/backfila/client/stat/parameters/ParametersDatasourceBackfill.kt
new file mode 100644
index 000000000..bbe882d34
--- /dev/null
+++ b/client-static/src/main/kotlin/app/cash/backfila/client/stat/parameters/ParametersDatasourceBackfill.kt
@@ -0,0 +1,39 @@
+package app.cash.backfila.client.stat.parameters
+
+import app.cash.backfila.client.PrepareBackfillConfig
+import app.cash.backfila.client.stat.StaticDatasourceBackfillBase
+
+/**
+ * This backfill type is a sub variant of the [StaticDatasourceBackfillBase].
+ *
+ * It uses a parameter populated in the Backfila UI as the datasource for the backfill.If you have too
+ * much data to fit in a parameter consider using a different client such as the S3 client.
+ */
+abstract class ParametersDatasourceBackfill<I : Any, P : DatasourceParameters<I>> : StaticDatasourceBackfillBase<I, P>() {
+  override fun getStaticDatasource(config: PrepareBackfillConfig<P>): List<I> = config.parameters.getBackfillData()
+}
+
+interface DatasourceParameters<out I : Any> {
+  /**
+   * This produces the full list of data for the backfill. Make sure the element order is consistent.
+   */
+  fun getBackfillData(): List<I>
+}
+
+/**
+ * Simple comma parameter datasource that produces a list of strings from a comma separated parameter.
+ */
+data class CommaParameterDatasource(
+  val commaDatasource: String,
+) : DatasourceParameters<String> {
+  override fun getBackfillData() = commaDatasource.split(',')
+}
+
+/**
+ * Simple newline parameter datasource that produces a list of strings from a newline separated parameter.
+ */
+data class NewlineParameterDatasource(
+  val newlineDatasource: String,
+) : DatasourceParameters<String> {
+  override fun getBackfillData() = newlineDatasource.split('\n')
+}
diff --git a/client-static/src/test/kotlin/app/cash/backfila/client/stat/BackfillsModule.kt b/client-static/src/test/kotlin/app/cash/backfila/client/stat/BackfillsModule.kt
index 334d6a9c5..094f24715 100644
--- a/client-static/src/test/kotlin/app/cash/backfila/client/stat/BackfillsModule.kt
+++ b/client-static/src/test/kotlin/app/cash/backfila/client/stat/BackfillsModule.kt
@@ -18,5 +18,7 @@ class BackfillsModule : KAbstractModule() {
       ),
     )
     install(StaticDatasourceBackfillModule.create<StaticKotlinValBackfillTest.SaucesBackfill>())
+    install(StaticDatasourceBackfillModule.create<ParametersStaticBackfillTest.ArtisanCheeseBackfill>())
+    install(StaticDatasourceBackfillModule.create<ParametersStaticBackfillTest.ProcessedCheeseBackfill>())
   }
 }
diff --git a/client-static/src/test/kotlin/app/cash/backfila/client/stat/ParametersStaticBackfillTest.kt b/client-static/src/test/kotlin/app/cash/backfila/client/stat/ParametersStaticBackfillTest.kt
new file mode 100644
index 000000000..c2b62413f
--- /dev/null
+++ b/client-static/src/test/kotlin/app/cash/backfila/client/stat/ParametersStaticBackfillTest.kt
@@ -0,0 +1,213 @@
+package app.cash.backfila.client.stat
+
+import app.cash.backfila.client.BackfillConfig
+import app.cash.backfila.client.stat.parameters.CommaParameterDatasource
+import app.cash.backfila.client.stat.parameters.DatasourceParameters
+import app.cash.backfila.client.stat.parameters.ParametersDatasourceBackfill
+import app.cash.backfila.embedded.Backfila
+import app.cash.backfila.embedded.createDryRun
+import app.cash.backfila.embedded.createWetRun
+import com.squareup.wire.internal.newMutableList
+import javax.inject.Inject
+import kotlin.test.assertFails
+import misk.testing.MiskTest
+import misk.testing.MiskTestModule
+import okio.ByteString.Companion.encodeUtf8
+import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.SoftAssertions
+import org.junit.jupiter.api.Test
+
+@MiskTest(startService = true)
+class ParametersStaticBackfillTest {
+  @Suppress("unused")
+  @MiskTestModule
+  val module = TestingModule()
+
+  @Inject lateinit var backfila: Backfila
+
+  @Test
+  fun `backfilling artisan cheese`() {
+    val run = backfila.createWetRun<ArtisanCheeseBackfill>(
+      parameterData = mapOf("commaDatasource" to artisanCheeses.encodeUtf8()),
+    )
+    run.execute()
+
+    assertThat(run.backfill.backfilledCheese).hasSize(11)
+    assertThat(run.backfill.backfilledCheese).contains("Brie", "Blue", "Goat", "Swiss", "Havarti", "Manchego")
+  }
+
+  @Test
+  fun `backfilling processed cheese`() {
+    val run = backfila.createWetRun<ProcessedCheeseBackfill>(
+      parameterData = mapOf("cheeseCSV" to allProcessedCheese.encodeUtf8()),
+    )
+    run.execute()
+
+    assertThat(run.backfill.backfilledCheese).hasSize(6)
+    assertThat(run.backfill.backfilledCheese).contains(ProcessedCheese.LAUGHING_COW, ProcessedCheese.VELVEETA)
+  }
+
+  @Test
+  fun `backfilling lowercase processed cheese fails`() {
+    val exception = assertFails {
+      backfila.createWetRun<ProcessedCheeseBackfill>(
+        parameterData = mapOf("cheeseCSV" to allProcessedCheese.lowercase().encodeUtf8()),
+      )
+    }
+    assertThat(exception).hasMessageContaining("No enum constant")
+  }
+
+  @Test
+  fun `dry run doesn't backfill`() {
+    val run = backfila.createDryRun<ArtisanCheeseBackfill>(
+      parameterData = mapOf("commaDatasource" to artisanCheeses.encodeUtf8()),
+    )
+    run.execute()
+
+    assertThat(run.backfill.backfilledCheese).hasSize(0)
+  }
+
+  @Test
+  fun `test single batch`() {
+    val run = backfila.createWetRun<ArtisanCheeseBackfill>(
+      parameterData = mapOf("commaDatasource" to artisanCheeses.encodeUtf8()),
+    )
+    run.batchSize = 5
+    run.scanRemaining()
+    run.runBatch()
+    assertThat(run.backfill.backfilledCheese).hasSize(5)
+  }
+
+  @Test
+  fun `test rangeStart and rangeEnd`() {
+    val run = backfila.createWetRun<ArtisanCheeseBackfill>(
+      rangeStart = "2",
+      rangeEnd = "8",
+      parameterData = mapOf("commaDatasource" to artisanCheeses.encodeUtf8()),
+    )
+    run.batchSize = 3
+    run.execute()
+    assertThat(run.backfill.backfilledCheese).hasSize(6)
+  }
+
+  @Test
+  fun `test rangeStart`() {
+    val run = backfila.createWetRun<ArtisanCheeseBackfill>(
+      rangeStart = "3",
+      parameterData = mapOf("commaDatasource" to artisanCheeses.encodeUtf8()),
+    )
+    run.batchSize = 3
+    run.execute()
+    assertThat(run.backfill.backfilledCheese).hasSize(8)
+  }
+
+  @Test
+  fun `test rangeEnd`() {
+    val run = backfila.createWetRun<ArtisanCheeseBackfill>(
+      rangeEnd = "8",
+      parameterData = mapOf(
+        "commaDatasource" to artisanCheeses.encodeUtf8(),
+      ),
+    )
+    run.batchSize = 3
+    run.execute()
+    assertThat(run.backfill.backfilledCheese).hasSize(8)
+  }
+
+  @Test
+  fun `validation failures`() {
+    with(SoftAssertions()) {
+      this.assertThatCode {
+        backfila.createWetRun<ArtisanCheeseBackfill>(
+          rangeStart = "abc",
+          parameterData = mapOf("commaDatasource" to artisanCheeses.encodeUtf8()),
+        )
+      }.hasMessageContaining("must be a number")
+
+      this.assertThatCode {
+        backfila.createWetRun<ArtisanCheeseBackfill>(
+          rangeEnd = "abc",
+          parameterData = mapOf("commaDatasource" to artisanCheeses.encodeUtf8()),
+        )
+      }.hasMessageContaining("must be a number")
+
+      this.assertThatCode {
+        backfila.createWetRun<ArtisanCheeseBackfill>(
+          rangeStart = "-10",
+          parameterData = mapOf("commaDatasource" to artisanCheeses.encodeUtf8()),
+        )
+      }.hasMessageContaining("must be positive integers")
+
+      this.assertThatCode {
+        backfila.createWetRun<ArtisanCheeseBackfill>(
+          rangeEnd = "-5",
+          parameterData = mapOf("commaDatasource" to artisanCheeses.encodeUtf8()),
+        )
+      }.hasMessageContaining("must be positive integers")
+
+      this.assertThatCode {
+        backfila.createWetRun<ArtisanCheeseBackfill>(
+          rangeStart = "5",
+          rangeEnd = "2",
+          parameterData = mapOf("commaDatasource" to artisanCheeses.encodeUtf8()),
+        )
+      }.hasMessageContaining("Start must be less than or equal to end")
+
+      this.assertThatCode {
+        backfila.createWetRun<ArtisanCheeseBackfill>(
+          rangeStart = "20",
+          rangeEnd = "30",
+          parameterData = mapOf("commaDatasource" to artisanCheeses.encodeUtf8()),
+        )
+      }.hasMessageContaining("Start is greater than the static datasource size")
+
+      this.assertAll()
+    }
+  }
+
+  class ArtisanCheeseBackfill @Inject constructor() : ParametersDatasourceBackfill<String, CommaParameterDatasource>() {
+    val backfilledCheese = newMutableList<String>()
+
+    override fun runOne(item: String, config: BackfillConfig<CommaParameterDatasource>) {
+      if (!config.dryRun) {
+        backfilledCheese.add(item)
+      }
+    }
+  }
+
+  class ProcessedCheeseBackfill @Inject constructor() : ParametersDatasourceBackfill<ProcessedCheese, ProcessedCheeseBackfill.ProcessedCheeseParameters>() {
+    val backfilledCheese = newMutableList<ProcessedCheese>()
+
+    override fun runOne(item: ProcessedCheese, config: BackfillConfig<ProcessedCheeseParameters>) {
+      if (!config.dryRun) {
+        backfilledCheese.add(item)
+      }
+    }
+
+    data class ProcessedCheeseParameters(
+      val cheeseCSV: String,
+    ) : DatasourceParameters<ProcessedCheese> {
+      override fun getBackfillData(): List<ProcessedCheese> {
+        return cheeseCSV.split(',').map { ProcessedCheese.valueOf(it) }
+      }
+    }
+  }
+
+  companion object {
+    enum class ProcessedCheese {
+      YELLOW_CHEESE,
+      WHITE_CHEESE,
+      CHEESE_WHIZ,
+      VELVEETA,
+      PHILADELPHIA,
+      LAUGHING_COW,
+    }
+    val allProcessedCheese = """
+            YELLOW_CHEESE,WHITE_CHEESE,CHEESE_WHIZ,VELVEETA,PHILADELPHIA,LAUGHING_COW
+    """.trimIndent()
+
+    val artisanCheeses = """
+      Brie,Blue,Goat,Swiss,Havarti,Manchego,OKA,Cambozola,St. Andre,Monteray Jack,Smoked Gouda
+    """.trimIndent()
+  }
+}
diff --git a/client-static/src/test/kotlin/app/cash/backfila/client/stat/StaticKotlinValBackfillTest.kt b/client-static/src/test/kotlin/app/cash/backfila/client/stat/StaticKotlinValBackfillTest.kt
index fde71b333..27444facd 100644
--- a/client-static/src/test/kotlin/app/cash/backfila/client/stat/StaticKotlinValBackfillTest.kt
+++ b/client-static/src/test/kotlin/app/cash/backfila/client/stat/StaticKotlinValBackfillTest.kt
@@ -10,7 +10,7 @@ import javax.inject.Inject
 import misk.testing.MiskTest
 import misk.testing.MiskTestModule
 import okio.ByteString.Companion.encodeUtf8
-import org.assertj.core.api.Assertions
+import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.SoftAssertions
 import org.junit.jupiter.api.Test
 
@@ -27,9 +27,9 @@ class StaticKotlinValBackfillTest {
     val run = backfila.createWetRun<SaucesBackfill>()
     run.execute()
 
-    Assertions.assertThat(run.backfill.backfilledSauces.size).isEqualTo(sweetSauces.size + savourySauces.size)
-    Assertions.assertThat(run.backfill.backfilledSauces).containsAll(savourySauces)
-    Assertions.assertThat(run.backfill.backfilledSauces).containsAll(sweetSauces)
+    assertThat(run.backfill.backfilledSauces).hasSize(sweetSauces.size + savourySauces.size)
+    assertThat(run.backfill.backfilledSauces).containsAll(savourySauces)
+    assertThat(run.backfill.backfilledSauces).containsAll(sweetSauces)
   }
 
   @Test
@@ -39,9 +39,9 @@ class StaticKotlinValBackfillTest {
     )
     run.execute()
 
-    Assertions.assertThat(run.backfill.backfilledSauces.size).isEqualTo(sweetSauces.size + savourySauces.size)
-    Assertions.assertThat(run.backfill.backfilledSauces).containsAll(savourySauces)
-    Assertions.assertThat(run.backfill.backfilledSauces).containsAll(sweetSauces.map { it.markSweet() })
+    assertThat(run.backfill.backfilledSauces).hasSize(sweetSauces.size + savourySauces.size)
+    assertThat(run.backfill.backfilledSauces).containsAll(savourySauces)
+    assertThat(run.backfill.backfilledSauces).containsAll(sweetSauces.map { it.markSweet() })
   }
 
   @Test
@@ -51,7 +51,7 @@ class StaticKotlinValBackfillTest {
     )
     run.execute()
 
-    Assertions.assertThat(run.backfill.backfilledSauces.size).isEqualTo(0)
+    assertThat(run.backfill.backfilledSauces).hasSize(0)
   }
 
   @Test
@@ -60,7 +60,7 @@ class StaticKotlinValBackfillTest {
     run.batchSize = 5
     run.scanRemaining()
     run.runBatch()
-    Assertions.assertThat(run.backfill.backfilledSauces.size).isEqualTo(5)
+    assertThat(run.backfill.backfilledSauces).hasSize(5)
   }
 
   @Test
@@ -68,7 +68,7 @@ class StaticKotlinValBackfillTest {
     val run = backfila.createWetRun<SaucesBackfill>(rangeStart = "2", rangeEnd = "8")
     run.batchSize = 3
     run.execute()
-    Assertions.assertThat(run.backfill.backfilledSauces.size).isEqualTo(6)
+    assertThat(run.backfill.backfilledSauces).hasSize(6)
   }
 
   @Test
@@ -76,7 +76,7 @@ class StaticKotlinValBackfillTest {
     val run = backfila.createWetRun<SaucesBackfill>(rangeStart = "3")
     run.batchSize = 3
     run.execute()
-    Assertions.assertThat(run.backfill.backfilledSauces.size).isEqualTo(7)
+    assertThat(run.backfill.backfilledSauces).hasSize(7)
   }
 
   @Test
@@ -84,7 +84,7 @@ class StaticKotlinValBackfillTest {
     val run = backfila.createWetRun<SaucesBackfill>(rangeEnd = "8")
     run.batchSize = 3
     run.execute()
-    Assertions.assertThat(run.backfill.backfilledSauces.size).isEqualTo(8)
+    assertThat(run.backfill.backfilledSauces).hasSize(8)
   }
 
   @Test
diff --git a/service-self-backfill/src/test/kotlin/app/cash/backfila/service/selfbackfill/BackfillRegisteredParametersTest.kt b/service-self-backfill/src/test/kotlin/app/cash/backfila/service/selfbackfill/BackfillRegisteredParametersTest.kt
index 8a58f56af..10d57fca4 100644
--- a/service-self-backfill/src/test/kotlin/app/cash/backfila/service/selfbackfill/BackfillRegisteredParametersTest.kt
+++ b/service-self-backfill/src/test/kotlin/app/cash/backfila/service/selfbackfill/BackfillRegisteredParametersTest.kt
@@ -17,7 +17,7 @@ import misk.hibernate.newQuery
 import misk.scope.ActionScope
 import misk.testing.MiskTest
 import misk.testing.MiskTestModule
-import org.assertj.core.api.Assertions
+import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Test
 
 @MiskTest(startService = true)
@@ -68,10 +68,10 @@ class BackfillRegisteredParametersTest {
     transacter.transaction { session ->
       val backfill = queryFactory.newQuery<RegisteredBackfillQuery>()
         .uniqueResult(session)!!
-      Assertions.assertThat(backfill.parameters).hasSize(1)
-      Assertions.assertThat(backfill.parameters.single().name).isEqualTo("name")
-      Assertions.assertThat(backfill.parameters.single().description).isEqualTo("desc")
-      Assertions.assertThat(backfill.parameters.single().required).isFalse()
+      assertThat(backfill.parameters).hasSize(1)
+      assertThat(backfill.parameters.single().name).isEqualTo("name")
+      assertThat(backfill.parameters.single().description).isEqualTo("desc")
+      assertThat(backfill.parameters.single().required).isFalse()
       backfill.parameters.forEach { session.delete(it) }
     }
 
@@ -80,10 +80,10 @@ class BackfillRegisteredParametersTest {
     transacter.transaction { session ->
       val backfill = queryFactory.newQuery<RegisteredBackfillQuery>()
         .uniqueResult(session)!!
-      Assertions.assertThat(backfill.parameters).hasSize(1)
-      Assertions.assertThat(backfill.parameters.single().name).isEqualTo("name")
-      Assertions.assertThat(backfill.parameters.single().description).isEqualTo(null)
-      Assertions.assertThat(backfill.parameters.single().required).isFalse()
+      assertThat(backfill.parameters).hasSize(1)
+      assertThat(backfill.parameters.single().name).isEqualTo("name")
+      assertThat(backfill.parameters.single().description).isEqualTo(null)
+      assertThat(backfill.parameters.single().required).isFalse()
     }
   }
 
@@ -114,7 +114,7 @@ class BackfillRegisteredParametersTest {
     transacter.transaction { session ->
       val backfill = queryFactory.newQuery<RegisteredBackfillQuery>()
         .uniqueResult(session)!!
-      Assertions.assertThat(backfill.parameters).hasSize(2)
+      assertThat(backfill.parameters).hasSize(2)
       backfill.parameters.forEach { session.delete(it) }
     }
 
@@ -123,13 +123,13 @@ class BackfillRegisteredParametersTest {
     transacter.transaction { session ->
       val backfill = queryFactory.newQuery<RegisteredBackfillQuery>()
         .uniqueResult(session)!!
-      Assertions.assertThat(backfill.parameters).hasSize(2)
-      Assertions.assertThat(backfill.parameters.first().name).isEqualTo("abc")
-      Assertions.assertThat(backfill.parameters.first().description).isEqualTo(null)
-      Assertions.assertThat(backfill.parameters.first().required).isFalse()
-      Assertions.assertThat(backfill.parameters.last().name).isEqualTo("def")
-      Assertions.assertThat(backfill.parameters.last().description).isEqualTo(null)
-      Assertions.assertThat(backfill.parameters.last().required).isFalse()
+      assertThat(backfill.parameters).hasSize(2)
+      assertThat(backfill.parameters.first().name).isEqualTo("abc")
+      assertThat(backfill.parameters.first().description).isEqualTo(null)
+      assertThat(backfill.parameters.first().required).isFalse()
+      assertThat(backfill.parameters.last().name).isEqualTo("def")
+      assertThat(backfill.parameters.last().description).isEqualTo(null)
+      assertThat(backfill.parameters.last().required).isFalse()
     }
   }
 
@@ -158,7 +158,7 @@ class BackfillRegisteredParametersTest {
     transacter.transaction { session ->
       val backfill = queryFactory.newQuery<RegisteredBackfillQuery>()
         .uniqueResult(session)!!
-      Assertions.assertThat(backfill.parameters).hasSize(0)
+      assertThat(backfill.parameters).hasSize(0)
     }
 
     backfila.createWetRun<BackfillRegisteredParameters>().execute()
@@ -166,7 +166,7 @@ class BackfillRegisteredParametersTest {
     transacter.transaction { session ->
       val backfill = queryFactory.newQuery<RegisteredBackfillQuery>()
         .uniqueResult(session)!!
-      Assertions.assertThat(backfill.parameters).hasSize(0)
+      assertThat(backfill.parameters).hasSize(0)
     }
   }
 }