Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

BigQuery: Skip merge if no primary keys present #1290

Merged
merged 6 commits into from
Feb 14, 2024
Merged

BigQuery: Skip merge if no primary keys present #1290

Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
2590c6f
skip merge if no pkey
Amogh-Bharadwaj Feb 14, 2024
872072b
refactor the skipping
Amogh-Bharadwaj Feb 14, 2024
a6ac593
another refactor
Amogh-Bharadwaj Feb 14, 2024
7b95c8a
minor change
Amogh-Bharadwaj Feb 14, 2024
a2e3813
change to error log
Amogh-Bharadwaj Feb 14, 2024
34ee18c
change to warn
Amogh-Bharadwaj Feb 14, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions flow/connectors/bigquery/bigquery.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -629,12 +629,16 @@ func (c *BigQueryConnector) NormalizeRecords(req *model.NormalizeRecordsRequest)
if err != nil {
return nil, fmt.Errorf("couldn't get tablename to unchanged cols mapping: %w", err)
}

// append all the statements to one list
c.logger.Info(fmt.Sprintf("merge raw records to corresponding tables: %s %s %v",
c.datasetID, rawTableName, distinctTableNames))

for _, tableName := range distinctTableNames {
normalizeTableSchema := req.TableNameSchemaMapping[tableName]
if len(normalizeTableSchema.PrimaryKeyColumns) == 0 {
c.logger.Warn(fmt.Sprintf("skipping merge for table %s as it has no primary key", tableName))
continue
}
unchangedToastColumns := tableNametoUnchangedToastCols[tableName]
dstDatasetTable, _ := c.convertToDatasetTable(tableName)
mergeGen := &mergeStmtGenerator{
Expand All @@ -645,7 +649,7 @@ func (c *BigQueryConnector) NormalizeRecords(req *model.NormalizeRecordsRequest)
},
dstTableName: tableName,
dstDatasetTable: dstDatasetTable,
normalizedTableSchema: req.TableNameSchemaMapping[tableName],
normalizedTableSchema: normalizeTableSchema,
syncBatchID: req.SyncBatchID,
normalizeBatchID: normBatchID,
peerdbCols: &protos.PeerDBColumns{
Expand Down
Loading