[RGI-1558] Let the BigQuery tap log in safely, find its tables, and copy only what changed

[soundarya-sambath-23]

In one sentence

This makes our BigQuery copy-tool able to log in two different ways, find the customer's tables reliably, and copy only the rows that changed since last time instead of all of them.

Four words you'll need

Everything below is built from these. If they're already familiar, skip ahead.

• Tap — a small program that reads from a source (here, BigQuery) and prints the rows to its output. Something else picks them up and loads them into Redshift. This repo is that program.
• Stream — one table's worth of that copying. One BigQuery table, one stream.
• Replication key — a column we use to answer "is this row new?" Usually updated_at: a timestamp the source updates whenever a row changes.
• Bookmark — the highest replication key value we saw last run, written to a state file. Next run, we ask BigQuery for rows above it. This is what makes a run take 30 seconds instead of re-copying the whole table.

Part 1 — Logging in

BigQuery needs to know who we are. Two ways to prove it:

A service account key — a JSON file with a private key in it. Like a robot's ID badge. This is the main path, because Google's RAPT security change now forces human OAuth logins to re-authenticate every few hours, which no unattended pipeline can survive.

OAuth — logging in as a human, using a refresh token. Worth explaining why this is here at all, since we just said it's fragile: the access token is what expires in an hour, and the refresh token is a long-lived voucher you trade for a fresh access token. Our old interim BigQuery connector was handed pre-fetched access tokens, so it broke every hour. This one holds the refresh token and mints new access tokens itself, so it keeps going.

If neither is configured, we fall back to whatever ambient credentials the machine has (Google calls this Application Default Credentials).

The google_application_credentials setting takes either the key JSON itself or a path to a file holding it. We try to read it as JSON; if that fails, we treat it as a filename.

Part 2 — Finding the tables

To copy a table you first have to ask BigQuery what columns it has. The library we use has a bulk "describe all these tables at once" call, and it has a bug: it puts the dataset name where the project name goes in the URL, so every request 400s.

So we ask table by table instead. Slower, works.

Part 3 — Copying only what changed

Two things had to be decided.

Which column is the replication key? If you set replication_key_column, that one. Otherwise we look for a timestamp column with a familiar name, preferring in order: updated_at, modified_at, last_modified, _sdc_batched_at, created_at. No match means we copy the whole table every time.

What do we compare it against? Here's the interesting part. The obvious query is:

WHERE updated_at >= '<bookmark>'

Greater-than-or-equal looks safer — you'd rather re-copy a row than miss one. But our source tables are loaded in batches, so fifty thousand rows can share the exact same updated_at, down to the microsecond. The bookmark lands on that shared timestamp, and >= matches all fifty thousand again. Every run. Forever. The "incremental" copy is a full copy wearing a disguise.

So we use a strict >, plus rows where the key is empty:

WHERE updated_at > '<bookmark>' OR updated_at IS NULL

The honest cost of that choice is in "What to watch for" at the bottom.

---

What the review caught

Copilot reviewed the branch and found thirteen real problems. Rather than list them, here is what was actually broken, because several were the same root cause. Each is: what you'd expect, what actually happened, why.

1. The tap could crash on a perfectly valid config

Expected: google_application_credentials: /secrets/key.json works, since the docs promise "JSON content or path."

Actual: crash. The code called json.loads() on it unconditionally, and a file path is not JSON.

Fix: try JSON, fall back to treating it as a path.

2. The tap could send rows out of order and then abort

Expected: rows come out, state gets saved.

Actual: possible mid-run crash on large tables.

Why: we'd hand-written the query as a SQL string with no ORDER BY. BigQuery makes no promise about row order without one. Meanwhile the SDK marks incremental streams as "sorted" and checks each row's timestamp against the previous one — a row that goes backwards means "we can't trust the bookmark", and it throws. So the crash only appears once BigQuery decides to parallelise your query, which is to say on exactly the big tables you care about.

Fix: build the query with SQLAlchemy instead of string-pasting, which brings back the ORDER BY the SDK expects.

3. The same query leaked columns nobody asked for

If you deselect a column, it shouldn't be copied. Our hand-written query said SELECT *. Same fix as #2 — the rebuilt query selects only selected columns.

4. And it could exhaust memory

We opened the database connection directly rather than through the SDK's helper. That helper is the thing that sets "stream the results, don't buffer them." Without it, a big extract tries to hold the whole result set in memory. Same fix again — one rewrite closed #2, #3 and MeltanoLabs#4 together.

5. Batch mode re-exported the same rows forever

This is the one worth your attention.

When google_storage_bucket is set, we don't copy row by row — we tell BigQuery to dump the table to a file in Cloud Storage, then download it. Much faster for large tables.

Expected: first run exports everything, later runs export only what changed.

Actual: every run exported the same thing, forever. The bookmark never moved off its starting value.

Why: the SDK advances the bookmark inside the row-by-row loop. Batch mode never enters that loop — it hands over a file instead. So nothing ever updated the bookmark. It writes a state message, faithfully recording the unchanged value. Nothing errors. It just quietly does the full table every night.

Fix: before exporting, ask SELECT MAX(updated_at). After the export and download succeed, save that as the new bookmark.

Note the ordering: we read the ceiling before the export, not after. Rows that land in between get exported now and exported again next run. That's deliberate — duplicates are absorbed by de-duplication downstream, whereas reading the ceiling afterwards would silently skip those rows. Given a choice, copy twice rather than lose one.

6. A DATE column would have failed at runtime

We accept DATE, DATETIME and TIMESTAMP columns as replication keys, but the query always compared against TIMESTAMP('...'). BigQuery refuses to compare a DATE to a TIMESTAMP — it's a type error, not a silent coercion. Any customer whose date column got picked would have hit a hard failure.

Fix: the bookmark now carries the same type as the column it's compared to.

7. The bookmark was pasted straight into SQL

The bookmark comes from a state file. We were formatting it directly into the query text, so a quote character in it could change what the query means. Same for column names.

Fix: the value is passed as a query parameter — handed to BigQuery separately from the SQL, so it can only ever be read as a value. Column names are backtick-quoted.

8. A typo in auth_type silently used the wrong identity

Writing oauth2 instead of oauth didn't fail. It fell through to the service-account branch, and with no key configured, to whatever ambient credentials the machine had. You'd get a run under an unintended identity rather than an error.

Fix: unknown values are rejected outright.

9. OAuth couldn't read its own export files

In batch mode we write files to Cloud Storage, download them, then delete them. But the OAuth login only ever asked Google for BigQuery permission, not Storage. Correct IAM roles, still denied — the token wasn't scoped for it.

Fix: request the Storage scope too when a bucket is configured. (Existing refresh tokens must already carry that scope; scopes can't be widened at refresh time. Noted in the README.)

10–13. Smaller things

• reflect_indices=False asked us to skip index reflection; we reflected anyway. Now honoured.
• Every normal service-account startup logged a scary "not valid json" warning. It's the documented path form — now a debug line.
• A docstring described consuming an access token from Argo, which is not what the code does.
• Two README claims didn't match the code: the OAuth section omitted permission requirements, and the replication-key rules understated when we fall back to full-table copies.

---

How we know it works

Tests went from 12 to 35, and CI passes on Python 3.9 through 3.12.

The new ones cover the things above that used to be untested: which column gets picked as the replication key and every way that can fall back; that both query paths use strict >, keep the ORDER BY, fetch only selected columns, and refuse to let a quote in the bookmark change the query; that a DATE key produces a DATE parameter; and that batch mode actually moves the bookmark.

One honest limitation: these are unit tests against a mocked BigQuery. They prove we generate the right SQL and make the right calls. They cannot prove BigQuery accepts it. A staging run is still needed — see below.

What to watch for

• Strict > has a real edge. A row committed after a sync but stamped with an updated_at exactly equal to the bookmark is never picked up. There's no lookback window. We chose this over re-copying entire batches every run, but it's a trade, not a free win. If that edge matters for a given source, the fix is a small trailing lookback.
• Rows with an empty replication key are re-sent every run, by design — we can't tell if they changed. De-duplication downstream has to hold.
• This has never run end to end. No full tap → S3 → COPY → dbt → stage → merge run has been completed against staging. The batch bookmark fix in particular changes behaviour that no test can fully verify.
• Another repo pins this branch. mk-data-ingestion-core #278 points its pip_url at @feat/mk-auth-layer. Merging here and deleting the branch breaks every MDI image build until that's re-pinned to a merged SHA or tag. Please coordinate.
• @claude review doesn't work on this repo. Both attempts failed with empty credentials — the Claude Code action has no CLAUDE_CODE_OAUTH_TOKEN secret configured. Needs a repo admin.

This branch is 2 commits behind main and has not been rebased.

🤖 Generated with Claude Code

[soundarya-sambath-23]

@claude

[claude]

Claude encountered an error —— View job

---

I'll analyze this and get back to you.

[amit-HGIN]

@claude review

[claude]

Claude encountered an error —— View job

---

I'll analyze this and get back to you.

[Copilot]

Pull request overview

Adds dual BigQuery authentication, reliable catalog discovery, and timestamp-based incremental replication.

Changes:

• Supports service-account, OAuth, and ADC authentication.
• Uses per-table discovery and automatically selects replication keys.
• Applies strict incremental filters to streaming and batch extraction paths.

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
tap_bigquery/connector.py	Implements authentication, discovery, and replication-key selection.
tap_bigquery/client.py	Adds incremental streaming and batch queries.
tap_bigquery/tap.py	Defines new configuration settings.
tests/test_connector_auth.py	Tests credential selection.
tests/test_client.py	Stubs credential-dependent clients.
tests/test_core.py	Stubs credential-dependent clients.
README.md	Documents authentication, settings, and replication.

Suppressed comments (6)

tap_bigquery/connector.py:299

• DATE and DATETIME columns are marked as valid incremental keys, but both extraction paths compare them with TIMESTAMP('<bookmark>'). BigQuery rejects DATE > TIMESTAMP and DATETIME > TIMESTAMP with a type-signature error, so such discovered streams fail at runtime. Either restrict discovery to TIMESTAMP or retain each key's type and generate a matching DATE/DATETIME expression in both query paths.

                type_name = type(col_type).__name__.upper()
                if type_name in ("TIMESTAMP", "DATETIME", "DATE"):
                    timestamp_columns.append(column_name)

tap_bigquery/connector.py:94

• OAuth credentials are scoped only for BigQuery, but the batch path later uses the same credentials with GCSFileSystem to download and delete exported objects (client.py:159-169). With google_storage_bucket configured, OAuth runs will therefore receive Cloud Storage authorization failures even when IAM permissions are present. Request the Storage read/write scope when batch extraction is enabled, and ensure the refresh token was originally granted that scope.

                scopes=["https://www.googleapis.com/auth/bigquery"],

tap_bigquery/connector.py:78

• Any value other than the exact string oauth silently falls through to service-account/ADC authentication. A typo such as auth_type: oauth2 can therefore run under an unintended ambient identity instead of failing startup, despite the public setting advertising only two modes. Reject unsupported values before selecting credentials.

        if auth_type == "oauth":

tap_bigquery/connector.py:116

• A documented and valid credentials path always reaches this branch, so logging that it is “not valid json” at warning level produces a misleading warning on every normal service-account startup. Treat the JSON parse miss as an expected path-selection branch and log it at debug level.

            if creds_dict is None:
                self.logger.warning(
                    "'google_application_credentials' not valid json trying path",
                )

README.md:111

• This step is incomplete: the implementation also falls back to FULL_TABLE when timestamp columns exist but none has a preferred name and replication_key_column is unset or invalid. Document the actual fallback condition so users do not expect an arbitrary timestamp column to be selected.

3. If the table has no timestamp column at all, the stream is `FULL_TABLE`.

tap_bigquery/client.py:116

• This bypasses SQLConnector._connect(), which is where Singer SDK 0.46 enables stream_results=True. On large row-by-row extracts, using the engine directly can allow result buffering and substantially increase memory use. Preserve the connector's streaming execution options.

        with self.connector._engine.connect() as conn:

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[richard-merrick]

@claude

[claude]

Claude encountered an error —— View job

---

I'll analyze this and get back to you.

[richard-merrick]

this seems like a good way of handling missing key vals, but perhaps that should be handled upstream on the provider side ? it would suggest corrupted or incorrect data (eg no created/updated at/id) on their side which they should fix, rather than us pulling it repeatedly

[amit-HGIN]

Agreed that a NULL key signals a data quality problem on the provider side, and I've made that visible rather than silent — but I'd argue against dropping the rows, because excluding them doesn't just stop the re-pulling, it means they're never extracted at all.

The case that worries me is a source that leaves updated_at NULL until a row is first updated, setting only created_at on insert. That's a common and entirely non-corrupt pattern. Since updated_at is first in our preference order, such a table would silently lose every never-updated row. Duplication is recoverable; that isn't.

So behaviour is unchanged, but as of 6687c19 we now count them and log one warning per run naming the count and the column:

1234 row(s) in 'dataset.table' have a NULL updated_at and are re-extracted on every run; the source should populate this column

The batch path gets the count from the watermark query via COUNTIF, so it costs no extra job. That gives us a number to take to the provider rather than an assertion, and if it stays high for a tenant we can escalate with evidence.

Happy to add a config flag to exclude them for a tenant where NULL genuinely means garbage — say the word if you'd rather have that now. COALESCE(updated_at, created_at) is a third option, but that's a semantics change I'd want decided separately.

[richard-merrick]

I think BQ streams a snapshotted result of the query so in theory this should be a static dataset and handled corectly.
That said it might be worth including an upper boundas a safety measure, thereby avoiding the risk of endless jobs if I'm mistaken and there is situation where it keeps returning newly updated data

[amit-HGIN]

You're right on the snapshot, and it turns out the suggestion still earns its place — just on the other path.

A BigQuery SELECT reads a consistent snapshot taken at job start, so this row-by-row query can't grow while it's being read. One job, one snapshot, finite result set — no endless-job risk. I've left it unbounded and added a comment saying why, so it doesn't read as an oversight later.

The batch path is different, and this is a genuine gap you've caught. It issues two jobs — SELECT MAX(key) for the watermark, then EXPORT DATA — so it sees two snapshots. I was committing the watermark as the new bookmark but not bounding the export with it, so rows landing between the two jobs got exported and re-exported next run. I'd documented that as deliberate at-least-once; bounding it is strictly better.

As of 6687c19:

WHERE (updated_at > @bookmark AND updated_at <= @watermark) OR updated_at IS NULL

The exported window now matches the committed bookmark exactly — no duplicates, no gap. It applies on the first run too, where there's no bookmark yet but still a ceiling.

One thing I considered and rejected for the row path: the SDK already computes a signpost (utc_now() for timestamp keys) that could serve as an upper bound. But it's our clock, not the source's — if the source stamps updated_at slightly ahead of us, rows get deferred a run for no benefit — and a DATE key compared against a datetime signpost reintroduces the type error fixed earlier in this PR. Not worth it on a path that's already snapshot-safe, but happy to revisit if you disagree.