[source-snowflake] Incremental sync permanently drops rows that land inside the sub-microsecond rounding gap introduced by #82705

## Summary

PR #82705 ("fix silent record loss at cursor upper bound in incremental sync", merged as `92d1a7d0e48`) changed the cursor accessor to round nanosecond timestamps **up** to the next microsecond instead of truncating them down.

The persisted incremental checkpoint is the same rounded value, so it is now **larger than any value that exists in the source table**. A row inserted afterwards whose cursor lands in that sub-microsecond gap falls below the next sync's lower bound and is never read again.

I ran the same scenario against both builds. The rounding direction decides whether a skipped row comes back:

| | pre-#82705 (`c7c248c471`) | post-#82705 (`92d1a7d0e48`) |
|---|---|---|
| checkpoint after first sync | `10:00:00.123456`, below the true max | `10:00:00.123457`, above the true max |
| gap row skipped initially | yes | yes |
| gap row after later traffic arrives | **recovered** | **never recovered** |
| final outcome | **7/7 rows present** | **6/7 rows present** |

Both builds skip the row at first. Only the old one gets it back.

## How this compares to the behaviour before #82705

Before #82705 the checkpoint was `floor(trueMax)`, which is less than or equal to every real value. Since the next sync's lower bound is inclusive (`cursor >= checkpoint`), a row that had been excluded by the upper bound still satisfied the lower bound afterwards, so it was picked up as soon as any later row extended the upper bound. In my run, sync #4 emitted 3 records: the previously skipped gap row, the new row, and one boundary row re-read as a duplicate. That last one is the price of a floored checkpoint, and it is bounded.

After #82705 the checkpoint is `roundUpToMicros(trueMax)`, which is greater than every real value. A row landing under it fails the lower bound, and because the checkpoint is monotonically non-decreasing it fails on every subsequent sync too. In my run, sync #4 emitted only the new row and the gap row stayed missing while the checkpoint moved further above it.

The compensating advantage is real and should be weighed. The old behaviour skipped the max-cursor row whenever it carried sub-microsecond digits, which is frequent, but self-healed on the next arrival and only stayed missing while the table was idle. The new behaviour triggers only when a row lands inside a window under 1 microsecond wide, which is rare, but the loss is unrecoverable short of a full refresh. The change traded a frequent, self-healing fault for a rare, permanent one.

## Affected

- Builds containing `92d1a7d0e48`.
- Reproduced on `TIMESTAMP_NTZ`. `roundUpToMicros` is applied identically in `SnowflakeOffsetDateTimeFieldType`, so `TIMESTAMP_TZ` / `TIMESTAMP_LTZ` cursors look affected by the same mechanism, but I did not reproduce those separately.

## Root cause

The cursor upper bound, the persisted checkpoint, and the emitted record value all come from one `JdbcAccessor.get()` call, so `roundUpToMicros` applies to all three:

- `SnowflakeLocalDateTimeAccessor.get()` in `airbyte-integrations/connectors/source-snowflake/src/main/kotlin/io/airbyte/integrations/source/snowflake/SnowflakeFieldTypes.kt` rounds up.
- `DefaultJdbcPartition.kt` builds the upper bound from `SelectQuerySpec(SelectColumnMaxValue(cursor), from)` and then stores `cursorCheckpoint = cursorUpperBound` in `completeState`, so the checkpoint is the rounded-up `MAX(cursor)`.
- `DefaultJdbcPartitionFactory.kt` creates the next partition with `isLowerBoundIncluded = true`, so the next sync filters `cursor >= checkpoint`.

One microsecond-precision value is therefore being asked to satisfy two contradictory constraints:

- as the **upper bound** it must be `>=` the true max, or the max row is excluded;
- as the next **lower bound** it must be `<=` the true max, or rows just above the true max are excluded.

When the true max carries sub-microsecond digits, both cannot hold. The rounding direction only chooses which side breaks, and as shown above it also chooses whether the fault is recoverable.

Note also that `LocalDateTimeCodec.PATTERN` in `airbyte-cdk/bulk/core/base/src/main/kotlin/io/airbyte/cdk/data/JsonCodec.kt` is `yyyy-MM-dd'T'HH:mm:ss.SSSSSS`, so state serialization is itself capped at 6 decimals. Keeping nanoseconds in the accessor alone would not survive the state round trip.

## Reproduction

Table with a `TIMESTAMP_NTZ(9)` cursor column, incremental sync on that column.

Note that the first sync after a reset is a snapshot and reads every row regardless of cursor bounds, so the divergence only appears on the incremental syncs that follow.

**Step 1.** Load 5 rows. IDs 1 to 4 at `2026-03-01 10:00:00.000000000`, ID 5 at `2026-03-01 10:00:00.123456789` (the max, with non-zero sub-microsecond digits). Sync.

Both builds deliver all 5 rows. The checkpoints differ:

```
pre-#82705 : {"cursors": {"MODIFICATION_DATE": "2026-03-01T10:00:00.123456"}}
post-#82705: {"cursors": {"MODIFICATION_DATE": "2026-03-01T10:00:00.123457"}}
```

The post-fix checkpoint is 211ns above every value that exists in the table.

**Step 2.** Insert ID 6 at `2026-03-01 10:00:00.123456999`, strictly greater than the previous max. Sync.

Both builds: sync **succeeds**, `recordsEmitted = 0`, ID 6 absent, state unchanged. No query is issued, because the recomputed upper bound equals the stored checkpoint and `DefaultJdbcPartitionFactory` takes its `// Incremental complete.` branch.

**Step 3.** Sync again with no changes. Both builds: `recordsEmitted = 0`, ID 6 still absent.

**Step 4.** Insert ID 7 at `2026-03-01 10:00:01.000000000` and sync. This is where the builds diverge.

- pre-#82705: `recordsEmitted = 3`. ID 6 is recovered, ID 7 arrives, and ID 5 is re-read as a duplicate because the inclusive lower bound `>= .123456` still covers it. Final: **7/7 present**.
- post-#82705: `recordsEmitted = 1`. Only ID 7 arrives. ID 6 remains absent and the checkpoint advances to `2026-03-01T10:00:01.000000`, further above ID 6 than before. Final: **6/7 present**.

Post-fix final state:

| ID | source value | destination | outcome |
|---|---|---|---|
| 1-4 | `10:00:00.000000000` | `10:00:00.000000` | ok |
| 5 | `10:00:00.123456789` | `10:00:00.123457` | ok |
| 6 | `10:00:00.123456999` | absent | **silently lost** |
| 7 | `10:00:01.000000000` | `10:00:01.000000` | ok |

The only way ID 6 is ever delivered post-fix is if the row is later updated to a value above the checkpoint, which arrives as an update rather than as the original insert.

## When this triggers

The gap is `(trueMax, roundUpToMicros(trueMax))`, open at both ends since the lower bound is inclusive, so at most 999ns wide, and non-empty only when the current max carries non-zero sub-microsecond digits.

Worth noting for triage: `CURRENT_TIMESTAMP()` returned millisecond precision in my test account (samples all ended `.xxx000000`), so a plain `DEFAULT CURRENT_TIMESTAMP` cursor column will not hit this. The realistic exposure is tables holding genuinely sub-microsecond timestamps, for example values written explicitly or loaded from nanosecond-capable producers. Unqualified `TIMESTAMP_NTZ` columns do have `DATETIME_PRECISION = 9` by default, so the type itself is not a safeguard.

## Possible directions

Both need a CDK change, since `JdbcAccessor.get()` currently feeds the record value, the upper bound, and the checkpoint from a single call.

1. **Asymmetric rounding, no codec change.** Keep the rounded-up value for the query's upper bound, but store a rounded-**down** checkpoint. This is the combination that produced 7/7 above, at the cost of re-reading the boundary microsecond on each sync. Duplicates rather than data loss, which fits incremental's at-least-once contract.
2. **Full-precision cursor bound and state.** Widen the state encoding beyond 6 decimals and decouple the bound/checkpoint value from the truncated record payload. Correct with no duplicates, but it touches `LocalDateTimeCodec` and therefore every JDBC source.

Happy to put up a PR for either once maintainers indicate which direction is preferred.

## How this was verified

Both connector builds were compiled locally from this repo and checked before use by disassembling the shipped jar:

- pre-#82705 (`c7c248c471`): truncation only, no `roundUpToMicros` and no `plusNanos` in the bytecode.
- post-#82705 (`92d1a7d0e48`): `access$roundUpToMicros` and `LocalDateTime.plusNanos` present.

Each run used a Snowflake source and a Postgres destination. Sync pods were checked against the built image IDs to confirm the intended build actually ran, and every checkpoint quoted above was read from the connection state API rather than inferred.

---
**Internal Tracking:** https://github.com/airbytehq/oncall/issues/13308