|
6 | 6 | findExportedJobByGcsPrefix, |
7 | 7 | findLatestExportedJobByKind, |
8 | 8 | markJobStatus, |
| 9 | + mergeJobTableRowCounts, |
9 | 10 | } from '@crowd/data-access-layer' |
10 | 11 | import { getServiceChildLogger } from '@crowd/logging' |
11 | 12 |
|
@@ -47,7 +48,7 @@ export interface BqExportToGcsOutput { |
47 | 48 | jobId: number |
48 | 49 | } |
49 | 50 |
|
50 | | -export async function bqExportToGcs(input: BqExportToGcsInput): Promise<BqExportToGcsOutput> { |
| 51 | +export async function bqExportToGcs(input: BqExportToGcsInput): Promise { |
51 | 52 | const { |
52 | 53 | jobKind, |
53 | 54 | sql, |
@@ -86,6 +87,10 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise<BqExport |
86 | 87 | { jobKind, exportName, jobId: prior.id, gcsPrefix: prior.gcsPrefix }, |
87 | 88 | 'exportName match — skipping BQ, loading from named export', |
88 | 89 | ) |
| 90 | + // A --fill-constraints run reusing an existing export must still stamp meta:fill on the |
| 91 | + // reused row, else a later --resume-job reads fill=false and reverts the merge to |
| 92 | + // ON CONFLICT DO NOTHING — silently skipping the version_constraint backfill. |
| 93 | + if (isFill) await mergeJobTableRowCounts(qx, prior.id, { 'meta:fill': 1 }) |
89 | 94 | return { |
90 | 95 | gcsPrefix: prior.gcsPrefix, |
91 | 96 | rowCount: prior.rowCountBq, |
@@ -143,6 +148,9 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise<BqExport |
143 | 148 | { jobKind, jobId: prior.id, gcsPrefix: prior.gcsPrefix }, |
144 | 149 | 'reuseExports=true — skipping BQ, loading from prior export', |
145 | 150 | ) |
| 151 | + // See named-export path above: persist meta:fill so a later --resume-job keeps the |
| 152 | + // fill-constraints merge instead of reverting to ON CONFLICT DO NOTHING. |
| 153 | + if (isFill) await mergeJobTableRowCounts(qx, prior.id, { 'meta:fill': 1 }) |
146 | 154 | return { |
147 | 155 | gcsPrefix: prior.gcsPrefix, |
148 | 156 | rowCount: prior.rowCountBq, |
@@ -171,6 +179,9 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise<BqExport |
171 | 179 | { jobKind, jobId: existing.id, gcsPrefix }, |
172 | 180 | 'GCS files already exist — reusing export', |
173 | 181 | ) |
| 182 | + // Same-runId reuse (Temporal retry): the export row already has meta:fill from its own |
| 183 | + // 'exporting' write, but re-stamp defensively so the fill flag can never be lost on reuse. |
| 184 | + if (isFill) await mergeJobTableRowCounts(qx, existing.id, { 'meta:fill': 1 }) |
174 | 185 | return { gcsPrefix, rowCount: existing.rowCountBq, bqBytesBilled: 0, jobId: existing.id } |
175 | 186 | } |
176 | 187 | } |
@@ -222,7 +233,7 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise<BqExport |
222 | 233 |
|
223 | 234 | // H7: mark exporting before we start the BQ job; store ecosystems filter + fill flag in the |
224 | 235 | // table_row_counts JSONB so --resume-job can restore the original export's settings. |
225 | | - const exportMeta: Record<string, string | number | string[]> = {} |
| 236 | + const exportMeta: Record = {} |
226 | 237 | if (ecosystems) exportMeta['meta:ecosystems'] = ecosystems |
227 | 238 | if (isFill) exportMeta['meta:fill'] = 1 |
228 | 239 | await markJobStatus(qx, jobId, 'exporting', { |
|
0 commit comments