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

Class BigQuerySourceBuilder<T>

java.lang.Object
io.github.flink.gcp.connector.bigquery.source.BigQuerySourceBuilder<T>
Type Parameters:
T - type of the records produced by the source
@Public public class BigQuerySourceBuilder<T> extends Object
Builds a BigQuerySource.

  • Field Details

    • DEFAULT_MAX_RECORDS_PER_FETCH

      public static final int DEFAULT_MAX_RECORDS_PER_FETCH
      The most rows one fetch hands to the task thread.

      A BigQuery response block carries up to about 128 MiB of rows, so a cap is what lets a checkpoint be taken part-way through one. The value follows the reference connector's own default and complements DEFAULT_MAX_BYTES_PER_FETCH: this cap bounds small decoded records by count while the byte target bounds large variable-width records.

      See Also:

    • DEFAULT_MAX_BYTES_PER_FETCH

      public static final long DEFAULT_MAX_BYTES_PER_FETCH
      Default for maxBytesPerFetch(long): 8 MiB of serialized Avro rows per fetch.

      The byte cap complements the row cap for variable-width rows. A local benchmark over 4-256 KiB rows retained 92% of the count-only throughput while reducing the first retained batch from 64 MiB to 8.1 MiB. It is a target rather than a hard heap limit: one row is always allowed to make progress, and the cursor also retains the current Storage Read response block.

      See Also:

    • DEFAULT_RETRY_MAX_ATTEMPTS

      public static final int DEFAULT_RETRY_MAX_ATTEMPTS
      Default for retryMaxAttempts(int): consecutive ReadRows attempts without progress.

      Read off the sequence it bounds rather than chosen for its own sake, and how long it is depends on the failure. An UNAVAILABLE backs off exponentially — nominally 100 ms growing by 1.3 — with gax picking each wait uniformly between zero and that value, which puts twenty-five consecutive failures at about three minutes at worst and half that on average: long enough to ride out the restarts a long read meets, short enough that a stream which is never coming back is reported instead of retried for the SDK's own twenty-four hours. The INTERNAL transport faults the client also resumes are retried a fixed millisecond apart with no growth, so for those the bound is reached almost at once.

      See Also:

  • Method Details

    • table

      public BigQuerySourceBuilder<T> table(TableDestination table)
      Sets the table to read.

      Either this or query(String), never both and never neither. A view is not a table here: the Storage Read API reads storage, and a view has none — pass its query, or SELECT * FROM the_view, to query(String) instead.

      Parameters:
      table - the table
      Returns:
      this builder

    • query

      public BigQuerySourceBuilder<T> query(String query)
      Sets a query whose result is read, instead of reading a table directly.

      The query is run once, at job start, as an ordinary BigQuery query job, and the source reads the table its result landed in. That is what makes a view readable — the Storage Read API cannot read a logical or materialized view at all, because it reads storage and a view has none.

      It is billed twice: once for the bytes the query scans, and again for the bytes the read session scans out of its result. Prune inside the query itself rather than relying on selectedFields(String...) and rowRestriction(String), which BigQuery applies to the result and so cannot make the query cheaper.

      Where the result lands is queryResultDataset(String)'s choice, and by default it is BigQuery's own anonymous dataset — nothing this connector has to create, expire or delete.

      Requires parentProject(String): with no table named, nothing else says which project the query job is submitted to and billed to.

      Parameters:
      query - the query, in GoogleSQL
      Returns:
      this builder

    • materializeViews

      public BigQuerySourceBuilder<T> materializeViews()
      Reads a table(TableDestination) that turns out to be a view by materializing it.

      Optional, and off by default. With it, the source asks BigQuery once, at job start, what the configured name is; a view — logical or materialized — is then read the way query(String) reads one, by running SELECT … FROM the_view and reading its result. An ordinary table is read directly, exactly as without this. Spark's and the Dataproc connector's equivalent is spelled viewsEnabled.

      Off by default because it costs a metadata call, and a source pointed at a table should not pay a round trip to be told it is a table — the read path otherwise makes no REST call at all. Asking for this is also asking to be billed for a query nobody typed, which is the other reason it is not the default.

      selectedFields(String...) is folded into the generated SELECT, so a view is not scanned column by column for data that is then discarded. rowRestriction(String) is not: BigQuery's restriction syntax is not a SQL WHERE, so it stays where a table source applies it, on the read session.

      Where the materialized result lands, and what it costs, is queryResultDataset(String)'s choice, exactly as for query(String).

      Returns:
      this builder

    • queryLocation

      public BigQuerySourceBuilder<T> queryLocation(String queryLocation)
      Sets the BigQuery location the query job runs in.

      Optional; defaults to letting BigQuery infer it from the tables the query names, which is what it does for a query submitted without one. Set it where the inference has nothing to go on, or where the job must be pinned to a region.

      Parameters:
      queryLocation - the location, for example "US" or "asia-northeast1"
      Returns:
      this builder

    • queryResultDataset

      public BigQuerySourceBuilder<T> queryResultDataset(String queryResultDataset)
      Sets a dataset the query's result is written to, instead of BigQuery's anonymous dataset.

      Optional, and the two choices differ in who owns the result:

      • Unset — BigQuery's anonymous dataset. The query is submitted with no destination table, so BigQuery writes the result into a hidden dataset of its own, expires it after about a day and charges no storage for it. Nothing is created here, so nothing is left to clean up, and an identical query re-run within that window is answered from cache — free, and landing on the same table. Its constraints are BigQuery's: access is restricted to the identity that ran the query, Google advises against depending on a cached result table, and a result above the maximum response size is not kept.
      • Set — a table in this dataset. The result is written to a table this connector creates there, with an expiration of a day set on it. Storage is charged for it until then, and nothing deletes it earlier: teardown also runs on a JobManager failover, where the restored job is still reading the read session that table backs.

      The dataset must already exist, must be in the query's own location, and must live in parentProject(String).

      Parameters:
      queryResultDataset - the dataset id
      Returns:
      this builder

    • reuseQueryResultWithin

      public BigQuerySourceBuilder<T> reuseQueryResultWithin(Duration reuseQueryResultWithin)
      Lets a re-planned job reuse a previous attempt's query job instead of running the query again, for attempts within the given window.

      Optional, and off by default — the job id is then random and every plan runs the query. With it, the job id is derived from the Flink job name, a digest of the query configuration, and the window, so a JobManager failover before the first checkpoint — the one failure that re-plans a source — finds the first attempt's job and adopts it: a job still running is waited for instead of racing it with a second scan, and a finished one has its result table checked and read — a table that vanished meanwhile makes the attempt run the query again instead. queryJobsReattached reports each reuse.

      What it treats as "the same job" is the Flink job name, because the name is the identifier the user controls: rename the job and nothing is reused. The rest of the id is derived — a digest over the query, project, location, result dataset and this window — so two pipelines can only ever share a job when they would run the identical query to the identical place, in which case sharing it is correct. The flip side is deliberate: attempts of the same name and query inside one window reuse each other's result even across an intentional redeploy, so the result can be up to a window old. Size the window to how stale a result the pipeline can read, or rename the job to force a fresh one.

      At most 24 hours, because both places a result can land expire at about a day: past that there is nothing left to reuse — an adoption whose table has expired falls back to running the query — so a longer window could only ever pay for the query again while appearing to deduplicate it.

      Requires queryLocation(String): BigQuery scopes a job to (project, location, id), and a look-up that names no location sees only the US multi-region — outside it the previous attempt's job would never be found, so the reuse this knob asks for could never happen (measured 2026-08-10 against a us-central1 dataset).

      Parameters:
      reuseQueryResultWithin - the window, positive and at most 24 hours
      Returns:
      this builder

    • parentProject

      public BigQuerySourceBuilder<T> parentProject(String parentProject)
      Sets the project the read session belongs to and is billed to.

      Optional beside table(TableDestination), where it defaults to the table's own project; set it to read a table in another project — a public dataset, say — where the read cannot be billed to the project that owns the table. Required beside query(String), which names no table to take a default from, and where it is also the project the query job runs in and is billed to.

      Parameters:
      parentProject - the Google Cloud project id
      Returns:
      this builder

    • deserializer

      public BigQuerySourceBuilder<T> deserializer(BigQueryRowDeserializationSchema<T> deserializer)
      Sets the deserializer converting each row into zero or more records.
      Parameters:
      deserializer - the deserializer
      Returns:
      this builder

    • selectedFields

      public BigQuerySourceBuilder<T> selectedFields(String... fields)
      Sets the columns to read.

      Optional; defaults to every column. The projection is applied by BigQuery when the read session is created, so unread columns are neither scanned nor billed.

      Parameters:
      fields - the column names
      Returns:
      this builder

    • selectedFields

      public BigQuerySourceBuilder<T> selectedFields(Collection<String> fields)
      Sets the columns to read, rejecting a blank or repeated column name.
      Parameters:
      fields - the column names
      Returns:
      this builder
      See Also:

    • rowRestriction

      public BigQuerySourceBuilder<T> rowRestriction(String rowRestriction)
      Sets a filter BigQuery applies before any row is sent.

      Optional; defaults to no filter. The expression is BigQuery's own restriction syntax, a WHERE clause without the keyword, for example "state = 'CA' AND year > 2020". Rows it excludes are neither transferred nor billed.

      Parameters:
      rowRestriction - the restriction
      Returns:
      this builder

    • snapshotTime

      public BigQuerySourceBuilder<T> snapshotTime(Instant snapshotTime)
      Sets the instant the table is read as of.

      Optional; defaults to the table's contents when the read session is created. BigQuery serves this from its time-travel window and rejects an instant outside it.

      Parameters:
      snapshotTime - the instant to read the table as of
      Returns:
      this builder

    • maxStreamCount

      public BigQuerySourceBuilder<T> maxStreamCount(int maxStreamCount)
      Sets an upper bound on the number of read streams BigQuery creates.

      Optional; defaults to 0, which lets BigQuery choose. It is a cap and not a target: BigQuery returns at most this many streams and may return far fewer — a small table is read by a single stream however many are asked for (measured 2026-08-09). Since one stream is read by one subtask at a time, capping it below the job's parallelism leaves subtasks idle.

      Parameters:
      maxStreamCount - the upper bound, or 0 to let BigQuery choose
      Returns:
      this builder

    • preferredMinStreamCount

      public BigQuerySourceBuilder<T> preferredMinStreamCount(int preferredMinStreamCount)
      Sets the number of read streams to ask BigQuery for.

      Optional; defaults to 0, which asks for no particular number. BigQuery makes a best effort to provide at least this many and may provide fewer. Asking for more streams than there are subtasks is how this source gets its elasticity: readers take the next stream as they finish one, so over-provisioning is what keeps a slow stream from holding a subtask idle.

      Parameters:
      preferredMinStreamCount - the preferred lower bound, or 0 for none
      Returns:
      this builder

    • maxRecordsPerFetch

      public BigQuerySourceBuilder<T> maxRecordsPerFetch(int maxRecordsPerFetch)
      Sets the most rows one fetch hands to the task thread.

      Optional; defaults to DEFAULT_MAX_RECORDS_PER_FETCH. This remains an independent count bound for small rows and checkpoint cadence; maxBytesPerFetch(long) bounds variable-width rows by their serialized Avro size.

      Parameters:
      maxRecordsPerFetch - the cap
      Returns:
      this builder

    • maxBytesPerFetch

      public BigQuerySourceBuilder<T> maxBytesPerFetch(long maxBytesPerFetch)
      Sets the target serialized Avro bytes one fetch hands to the task thread.

      Optional; defaults to DEFAULT_MAX_BYTES_PER_FETCH. The reader stops before adding a row that would take a non-empty batch over the target. A row larger than the target is emitted by itself so the source always makes progress.

      Parameters:
      maxBytesPerFetch - the positive byte target
      Returns:
      this builder

    • retryMaxAttempts

      public BigQuerySourceBuilder<T> retryMaxAttempts(int retryMaxAttempts)
      Sets how many consecutive ReadRows attempts without progress the client makes before the read fails.

      Optional; defaults to DEFAULT_RETRY_MAX_ATTEMPTS. The retry itself is the client library's, not this connector's: it resumes a broken stream at the row it had reached, and decides for itself which failures deserve a resume. This knob only stops it. Without one the client retries for twenty-four hours, so a stream that will never come back holds a reader for a day while reporting nothing.

      An attempt that produced rows resets the count, so this bounds a stream that is stuck rather than one that is slow. Raise it for a read that must survive a long outage; the job fails and restarts from its last checkpoint either way, resuming each stream at the offset that checkpoint holds rather than reading it from the top.

      Parameters:
      retryMaxAttempts - the maximum number of consecutive attempts without progress
      Returns:
      this builder

    • serviceAccountKeyFile

      public BigQuerySourceBuilder<T> serviceAccountKeyFile(String serviceAccountKeyFile)
      Uses the service account in the given JSON key file for every BigQuery client this source opens.

      Optional; absent uses application-default credentials. Only the path enters the job graph. The source loads the file when its runtime clients are first opened: the JobManager creates read sessions and runs query or view-materialization jobs, and TaskManagers open the assigned read streams. The same key file must therefore exist at that path on both process types, including after failover or rescaling.

      Only service-account JSON is accepted. This setting cannot be combined with either emulator endpoint because emulator connections are credential-free.

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

    • emulatorEndpoint

      public BigQuerySourceBuilder<T> emulatorEndpoint(String emulatorEndpoint)
      Sends the source's traffic to a BigQuery emulator at host:port, over plaintext and without credentials.

      For testing against a local emulator and nothing else. This is the whole of it for a source reading a table(TableDestination) that does not ask for materializeViews(): the read session carries the schema, so nothing on that path makes a REST call. A source reading a query(String), or one that asked for materializeViews(), does make one — the query job, and the view lookup — and needs emulatorRestEndpoint(String) as well.

      The value is parsed here, so a malformed host:port is rejected on the client instead of surfacing as a connection failure once the job has been deployed.

      Parameters:
      emulatorEndpoint - the emulator's gRPC endpoint as host:port
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the endpoint is not host:port with a port in 1..65535

    • emulatorRestEndpoint

      public BigQuerySourceBuilder<T> emulatorRestEndpoint(String emulatorRestEndpoint)
      Sends the source's REST traffic — its query job, and the view lookup materializeViews() makes — to a BigQuery emulator at host:port, over plain HTTP and without credentials.

      This is the REST half of emulatorEndpoint(String). Two sources reach it: query(String), whose query job is a REST call, and table(TableDestination) with materializeViews(), whose view lookup is one. The lookup is made whether or not the name turns out to be a view, so a materializeViews() source over an ordinary table needs this endpoint and runs no query job at all. The two are separate because they are separate transports on separate ports, as they are on the sink side.

      Parameters:
      emulatorRestEndpoint - the emulator's REST endpoint as host:port
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the endpoint is not host:port with a port in 1..65535

    • build

      public org.apache.flink.api.connector.source.Source<T,ReadStreamSplit,BigQueryReadEnumeratorState> build()
      Builds the source.
      Returns:
      the source