Package io.github.flink.gcp.connector.spanner.source

Class SpannerSourceBuilder<T>

java.lang.Object
io.github.flink.gcp.connector.spanner.source.SpannerSourceBuilder<T>
Type Parameters:
T - the record type produced
@Public public class SpannerSourceBuilder<T> extends Object
Builds a SpannerSource.

Everything is validated at build() or at the setter that took it, so a configuration mistake fails where the job is assembled rather than on a JobManager once the read is planned.

  • Field Details

    • DEFAULT_MAX_ROWS_PER_FETCH

      public static final int DEFAULT_MAX_ROWS_PER_FETCH
      The default maximum rows one fetch hands to the task thread.
      See Also:

    • DEFAULT_MAX_BYTES_PER_FETCH

      public static final long DEFAULT_MAX_BYTES_PER_FETCH
      The default target maximum decoded input bytes one fetch hands to the task thread.
      See Also:

  • Method Details

    • database

      public SpannerSourceBuilder<T> database(DatabaseDestination database)
      Sets the database to read. Required.
      Parameters:
      database - the database
      Returns:
      this builder

    • readOperation

      public SpannerSourceBuilder<T> readOperation(SpannerReadOperation readOperation)
      Sets what to read: a query, or a table with its columns and key set. Required.
      Parameters:
      readOperation - the read operation
      Returns:
      this builder

    • deserializer

      public SpannerSourceBuilder<T> deserializer(SpannerStructDeserializationSchema<T> deserializer)
      Sets the deserializer turning rows into records. Required.
      Parameters:
      deserializer - the deserializer
      Returns:
      this builder

    • timestampBound

      public SpannerSourceBuilder<T> timestampBound(com.google.cloud.spanner.TimestampBound timestampBound)
      Sets the snapshot to read at. Optional; the default is TimestampBound.strong(), the latest committed data.

      Only strong(), ofReadTimestamp and ofExactStaleness can bound a batch read. The other two modes are refused here rather than on a JobManager: Spanner restricts them to single-use transactions, and a batch read is by construction a multi-use one — every partition rejoins the same transaction.

      A stale read is cheaper for the service to serve and can be answered by any replica; a strong one may have to reach the leader. Nothing about the source changes either way — every partition is read at the one timestamp the plan fixed.

      Parameters:
      timestampBound - the bound
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the bound is one a batch read cannot take

    • maxPartitions

      public SpannerSourceBuilder<T> maxPartitions(long maxPartitions)
      Hints how many partitions the read should be divided into. Optional; unset leaves the choice entirely to the service.

      A hint, and Spanner documents it as one: it may return more or fewer. Setting it to the job's parallelism is the reasonable thing to ask for, not something to rely on having got.

      Parameters:
      maxPartitions - the desired maximum number of partitions
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the value is not positive

    • partitionSizeBytes

      public SpannerSourceBuilder<T> partitionSizeBytes(long partitionSizeBytes)
      Hints how much data each partition should cover. Optional; unset leaves the choice entirely to the service.

      A hint, like maxPartitions(long), and the two are asked for together at the discretion rather than combined by this connector.

      Parameters:
      partitionSizeBytes - the desired size of one partition, in bytes
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the value is not positive

    • dataBoostEnabled

      public SpannerSourceBuilder<T> dataBoostEnabled(boolean dataBoostEnabled)
      Runs the read on Data Boost's independent compute. Optional; off by default.

      Data Boost serves a partitioned read from compute that is not the instance's, so a large scan does not contend with the workload the instance is serving. Three things come with it: the caller needs spanner.databases.useDataBoost on the database, the read is billed separately, and its concurrency has a quota of its own — so RESOURCE_EXHAUSTED is a shape a boosted read can meet that an ordinary one does not.

      Parameters:
      dataBoostEnabled - whether to enable Data Boost
      Returns:
      this builder

    • rpcPriority

      public SpannerSourceBuilder<T> rpcPriority(SpannerRpcPriority rpcPriority)
      Sets the priority Spanner schedules the reads at. Optional; unset leaves the service's own handling in place, which is the same as HIGH.

      LOW is what a backfill of a large table wants: Spanner sheds low-priority work first when an instance is at capacity, so the read yields to the traffic the instance is serving rather than competing with it. MEDIUM is a step down from the default rather than a restatement of it, since Spanner treats an unspecified priority as HIGH.

      The priority applies to the reads that move the rows, which is where a job's load on the instance actually is. It does not apply to the one call that plans the partitions.

      Not a substitute for dataBoostEnabled(boolean): a low-priority read still runs on the instance's own compute and still competes for it, while Data Boost does not use it at all. Lowering the priority costs nothing extra; Data Boost is billed separately.

      Parameters:
      rpcPriority - the priority
      Returns:
      this builder

    • serviceAccountKeyFile

      public SpannerSourceBuilder<T> serviceAccountKeyFile(String serviceAccountKeyFile)
      Authenticates the source with the service-account JSON key at the given path instead of application-default credentials. The JobManager reads the file when it creates or restores the enumerator, and each TaskManager reads it when it creates a reader. Every eligible process must therefore see the same path. Optional; when unset the real-service path uses application-default credentials.

      Service-account keys are long-lived secrets. Prefer an attached service account or Workload Identity where the deployment supports one. This setting cannot be combined with emulatorEndpoint(String), whose plaintext channel carries no credentials.

      Parameters:
      serviceAccountKeyFile - the service-account JSON key-file path
      Returns:
      this builder

    • emulatorEndpoint

      public SpannerSourceBuilder<T> emulatorEndpoint(String emulatorEndpoint)
      Points the source at an emulator, over a plaintext channel with no credentials. Never production.
      Parameters:
      emulatorEndpoint - the emulator endpoint as host:port
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the endpoint is not host:port with a port in 1..65535

    • maxRowsPerFetch

      public SpannerSourceBuilder<T> maxRowsPerFetch(int maxRowsPerFetch)
      Sets the maximum input rows one fetch hands to Flink's element queue. Optional; defaults to 1000.

      The fetch returns when either this row limit or maxBytesPerFetch(long) is reached. Lower values can reduce the source reader's queued memory and the delay before a checkpoint or cancellation can be observed, at the cost of more fetch hand-offs. This is a TaskManager hand-off bound, not a hint for how Spanner plans read partitions.

      Parameters:
      maxRowsPerFetch - the positive maximum number of rows per fetch
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the value is not positive

    • maxBytesPerFetch

      public SpannerSourceBuilder<T> maxBytesPerFetch(long maxBytesPerFetch)
      Sets the target maximum decoded input bytes one fetch hands to Flink's element queue. Optional; defaults to 12582912L bytes.

      The estimate counts logical field content rather than JVM object overhead. A fetch stops before a row would take a non-empty batch over the target, while one oversized row is handed over alone so the source always makes progress. Lower values can reduce the source reader's queued memory for multi-row batches, while higher values can reduce fetch hand-offs for wide rows. This is a TaskManager hand-off bound, not a Spanner transport or partition-planning setting.

      Parameters:
      maxBytesPerFetch - the positive target maximum estimated bytes per fetch
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the value is not positive

    • build

      public org.apache.flink.api.connector.source.Source<T,BatchReadSplit,SpannerBatchReadEnumeratorState> build()
      Builds the source.
      Returns:
      the source
      Throws:
      IllegalStateException - if a required option was not set