oracledb_cdc: remove primary key ordering from unfiltered snapshot #4696
+225
−67
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -138,7 +138,8 @@ func (s *Snapshot) snapshotTable(ctx context.Context, table UserTable, maxBatchS | |
| tableName = table.FullName() | ||
| ) | ||
| l := s.log.With("src_table", tableName) | ||
| customQuery, hasFilter := s.filters[tableName] | ||
| if hasFilter { | ||
| l.Infof("Launching snapshot of table '%s' with snapshot filter", tableName) | ||
| } else { | ||
| l.Infof("Launching snapshot of table '%s'", tableName) | ||
|
|
@@ -196,34 +197,45 @@ func (s *Snapshot) snapshotTable(ctx context.Context, table UserTable, maxBatchS | |
| } | ||
| }() | ||
|
|
||
| var numRowsProcessed int | ||
| if hasFilter { | ||
| // Custom filters may join or aggregate, so a physical-order scan can't safely | ||
| // assume every row maps to a stable ROWID/heap position; keep the PK-keyset | ||
| // pagination path for these. | ||
| var tablePks []string | ||
| if tablePks, err = getTablePrimaryKeys(ctx, tx, table); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| l.Debugf("Found primary keys for table '%s': %v", table, tablePks) | ||
| lastSeenPksValues := map[string]any{} | ||
| for _, pk := range tablePks { | ||
| lastSeenPksValues[pk] = nil | ||
| } | ||
|
|
||
| var batchCount int | ||
| for { | ||
| var pksForQuery map[string]any | ||
| if numRowsProcessed > 0 { | ||
| pksForQuery = lastSeenPksValues | ||
| } | ||
| batchCount, err = s.processBatch(ctx, tx, table, tablePks, pksForQuery, lastSeenPksValues, maxBatchSize, tableName, customQuery) | ||
| if err != nil { | ||
| return fmt.Errorf("processing snapshot batch: %w", err) | ||
| } | ||
|
|
||
| numRowsProcessed += batchCount | ||
| if batchCount < maxBatchSize { | ||
| break | ||
| } | ||
| } | ||
| } else { | ||
| // No filter: read the whole table through a single unordered cursor so Oracle | ||
| // picks a full table scan (sequential multiblock reads) instead of the | ||
| // PK-index-driven, table-access-by-rowid random I/O that ORDER BY pk forces | ||
| // once the table no longer fits in buffer cache. | ||
| if numRowsProcessed, err = s.snapshotTableFullScan(ctx, tx, table, maxBatchSize, tableName); err != nil { | ||
|
josephwoodward marked this conversation as resolved.
|
||
| return fmt.Errorf("processing snapshot table scan: %w", err) | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -241,8 +253,7 @@ func (s *Snapshot) snapshotTable(ctx context.Context, table UserTable, maxBatchS | |
| // pksForQuery is passed to querySnapshotTable for cursor-based pagination (nil on first batch). | ||
| // lastSeenPksValues is mutated in place with the PK values from the last row of the batch, | ||
| // so the caller can pass it as pksForQuery on the next iteration. | ||
| func (s *Snapshot) processBatch(ctx context.Context, tx *sql.Tx, table UserTable, tablePks []string, pksForQuery map[string]any, lastSeenPksValues map[string]any, maxBatchSize int, tableName, customQuery string) (batchCount int, err error) { | ||
| batchRows, err := querySnapshotTable(ctx, tx, table, tablePks, pksForQuery, maxBatchSize, customQuery) | ||
| if err != nil { | ||
| return 0, fmt.Errorf("execute snapshot table query: %w", err) | ||
|
|
@@ -274,48 +285,116 @@ func (s *Snapshot) processBatch(ctx context.Context, tx *sql.Tx, table UserTable | |
| return 0, err | ||
| } | ||
|
|
||
| if err := s.publishRow(ctx, table, columns, types, values, mappers, colMeta, lastSeenPksValues); err != nil { | ||
| return 0, err | ||
| } | ||
| } | ||
|
|
||
| if err = batchRows.Err(); err != nil { | ||
| return 0, fmt.Errorf("iterating snapshot table row: %w", err) | ||
| } | ||
| s.snapshotRowsTotalMetric.Incr(int64(batchCount), tableName) | ||
| return batchCount, nil | ||
| } | ||
|
|
||
| // publishRow maps a single scanned row's values into a MessageEvent and publishes | ||
| // it. When lastSeenPksValues is non-nil, it's mutated in place with this row's | ||
| // primary key values so the caller can resume PK-keyset pagination from it on the | ||
| // next batch; pass nil when there's no keyset cursor to maintain (full table scan). | ||
| func (s *Snapshot) publishRow(ctx context.Context, table UserTable, columns []string, types []*sql.ColumnType, values []any, mappers []func(any) (any, error), colMeta []ColumnMeta, lastSeenPksValues map[string]any) error { | ||
| row := map[string]any{} | ||
| for idx, value := range values { | ||
| v, err := mappers[idx](value) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| if !s.lobEnabled && IsLOBTypeName(types[idx].DatabaseTypeName()) { | ||
| v = nil | ||
| } | ||
| row[columns[idx]] = v | ||
| if _, ok := lastSeenPksValues[columns[idx]]; ok { | ||
| lastSeenPksValues[columns[idx]] = value | ||
| } | ||
| } | ||
|
|
||
| m := MessageEvent{ | ||
| Table: table.Name, | ||
| Schema: table.Schema, | ||
| Data: row, | ||
| Operation: MessageOperationRead, | ||
| ColumnMeta: colMeta, | ||
| } | ||
| if s.scn != InvalidSCN { | ||
| m.SCN = s.scn | ||
| } | ||
| if !s.commitTimestamp.IsZero() { | ||
| m.CommitTimestamp = s.commitTimestamp | ||
| } | ||
|
|
||
| if err := s.publisher.Publish(ctx, &m); err != nil { | ||
| return fmt.Errorf("handling snapshot table row: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // snapshotTableFullScan reads every row of table through a single, unordered | ||
| // query so Oracle's optimizer picks a full table scan (sequential multiblock | ||
| // reads) rather than the PK-index-driven, table-access-by-rowid random I/O that | ||
| // ORDER BY pk forces once the table exceeds buffer cache. There's no query-level | ||
| // batching here — maxBatchSize only paces how often cancellation is checked. | ||
| func (s *Snapshot) snapshotTableFullScan(ctx context.Context, tx *sql.Tx, table UserTable, maxBatchSize int, tableName string) (numRowsProcessed int, err error) { | ||
|
josephwoodward marked this conversation as resolved.
|
||
| q := fmt.Sprintf(`SELECT * FROM "%s"."%s"`, table.Schema, table.Name) | ||
|
Comment on lines
+334
to
+337
There was a problem hiding this comment. This change is justified purely on throughput grounds — the doc comment here and the inline comment at CONTRIBUTING §1.3.4/§1.3.5 requires both local and real-endpoint benchmarking at various throughput levels, with results recorded under Suggested fix: re-run the snapshot benchmarks against the new scan path and update
Contributor
Author
There was a problem hiding this comment. Benchmarks included in proof of work. Will look to update these as a separate PR. |
||
| rows, err := tx.QueryContext(ctx, q) | ||
| if err != nil { | ||
| return 0, fmt.Errorf("execute snapshot table scan: %w", err) | ||
| } | ||
| defer func() { | ||
| if closeErr := rows.Close(); closeErr != nil && err == nil { | ||
| err = fmt.Errorf("closing snapshot rows: %w", closeErr) | ||
| } | ||
| }() | ||
|
|
||
| types, err := rows.ColumnTypes() | ||
| if err != nil { | ||
| return 0, fmt.Errorf("fetch column types: %w", err) | ||
| } | ||
|
|
||
| values, mappers := prepSnapshotScannerAndMappers(types) | ||
|
|
||
| columns, err := rows.Columns() | ||
| if err != nil { | ||
| return 0, fmt.Errorf("fetch columns: %w", err) | ||
| } | ||
|
|
||
| colMeta := buildColumnMeta(types) | ||
|
|
||
| var sinceCancelCheck int | ||
| for rows.Next() { | ||
| if err = rows.Scan(values...); err != nil { | ||
| return numRowsProcessed, err | ||
| } | ||
|
|
||
| if err = s.publishRow(ctx, table, columns, types, values, mappers, colMeta, nil); err != nil { | ||
| return numRowsProcessed, err | ||
| } | ||
|
|
||
| numRowsProcessed++ | ||
| s.snapshotRowsTotalMetric.Incr(1, tableName) | ||
|
|
||
| sinceCancelCheck++ | ||
| if sinceCancelCheck >= maxBatchSize { | ||
| sinceCancelCheck = 0 | ||
| if err = ctx.Err(); err != nil { | ||
| return numRowsProcessed, err | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if err = rows.Err(); err != nil { | ||
| return numRowsProcessed, fmt.Errorf("iterating snapshot table row: %w", err) | ||
| } | ||
|
|
||
| return numRowsProcessed, nil | ||
| } | ||
|
|
||
| func getTablePrimaryKeys(ctx context.Context, tx *sql.Tx, table UserTable) ([]string, error) { | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
snapshot_max_batch_sizenow carries two unrelated meanings depending on whether the table happens to have asnapshot_filtersentry: for filtered tables it is the page size of the keyset-pagination query, and for every other table (the default, sincesnapshot_filtersis optional) it degrades to "how oftenctx.Err()is polled" — seesnapshot.go#L362-L381.This is a material UX regression under CONTRIBUTING §1.1.3 ("UX should be intuitive and require minimal explanation") and §3.1.5 (consistency with the rest of the fleet, where
snapshot_max_batch_sizeuniformly means rows-per-fetch). The description text itself is the evidence — it has to explain two divergent behaviours for one knob.Concretely: a user who raised
snapshot_max_batch_sizeto tune snapshot throughput — exactly whatdocs/benchmark-results/oracledb-cdc.mddocuments withsnapshot_max_batch_size: 160000— now gets no throughput effect at all on the default path, silently.Suggested fix: keep the knob meaning "rows fetched per round trip" on both paths by wiring it into the driver's prefetch/array-fetch size for the full-scan cursor, and use a separate fixed internal constant for the cancellation-poll interval. If the two really must differ, a dedicated field for the scan path would be clearer than overloading this one.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm not sure we want
snapshot_max_batch_sizeto be theprefetch_rowsvalue, it would be better to have that as a dedicated config and deprecatesnapshot_max_batch_sizeif needed.