perf(jdbc): optimize incremental sync with MapScanConcurrent by krishanu7 · Pull Request #1062 · datazip-inc/olake

krishanu7 · 2026-08-02T06:42:32Z

Description

This PR refactors the incremental sync pipeline to use MapScanConcurrent instead of the synchronous MapScan across all SQL-based relational database drivers (Postgres, MySQL, MSSQL, and Oracle).

Previously, incremental sync used synchronous scanning, which blocked database I/O while waiting for data type conversion and column sizing calculations to complete on each row. By switching to MapScanConcurrent (which uses a concurrent Producer/Consumer pipeline), data fetching from the database and CPU processing are decoupled. This increases throughput during incremental syncs (especially over networked DB connections) and drastically reduces memory allocations (~40% reduction) via pointer array reuse.

Fixes # (issue)

Type of change

Bug fix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)
Breaking change (fix or feature that would cause existing functionality to not work as expected)
This change requires a documentation update

How Has This Been Tested?

Unit & Accuracy Tests: Created benchmark and accuracy test suites in pkg/jdbc/reader_test.go verifying that MapScanConcurrent generates output identical to MapScan.
Benchmarking: Verified memory reduction (~40% allocation drop, 8MB RAM savings per batch) and pipeline concurrency behavior via local benchmark suites.
Driver Compilation: Verified full project compilation across all SQL drivers (go build ./...).

Screenshots or Recordings

N/A

Documentation

Documentation Link: [link to README, olake.io/docs, or olake-docs]
N/A (bug fix, refactor, or test changes only)

Related PR's (If Any):

PR feat: OLake full refresh sync performance improvement #657 (Introduced MapScanConcurrent for full refresh backfill)

krishanu7 · 2026-08-04T04:14:33Z

krishanu7 · 2026-08-04T04:16:38Z

krishanu7 · 2026-08-04T04:31:09Z

I updated MapScanConcurrent to use a 512-slot buffered channel instead of being unbuffered. This allows the database fetcher to run at full speed without waiting for the consumer on every row. In my tests, this improved throughput by ~36%.

I checked the edge cases, and it's completely safe:

Errors/Cancellations: If either side fails, the channel closes safely and unconsumed rows are garbage collected. No data is lost.
Memory Trade-off: If the consumer is slow, the fetcher holds up to 512 rows in memory. At 1-2kb per row, this is at most ~1MB of extra RAM, which is completely negligible for the speed boost.

The unbuffered channel took ~154 million ns (154 ms) to process the batch
while the buffered channel dropped that to ~112 million ns (112 ms). ~27% speedup

The benchmarking tool was able to run the buffered channel 9 full times within the time limit, compared to just 7 times for the unbuffered one

Both implementations used the exact same amount of memory per batch (~54.4 MB) and exact same allocations (~1 million)

Commit message of buffered channel Impl : 5e7a047

mihir-datazip · 2026-08-07T09:37:31Z

Driver Compilation: Verified full project compilation across all SQL drivers (go build ./...)

Just FYI, This does not compile driver code at all unless you ran this inside per driver folder manually... our driver code have its own go modules and ./... will be limited to the root module that builds only common code. To actually compile all the drivers, we need to use go list as we have go.work

mihir-datazip · 2026-08-07T12:16:52Z

 	}
-
-	return rows.Err()
+	


Redundant line here, install go formatter in your IDE to avoid such mistakes...

mihir-datazip · 2026-08-07T12:17:52Z

-		if err != nil {
-			return fmt.Errorf("failed to scan record: %s", err)
-		}
+	setter := jdbc.NewReader(ctx, incrementalQuery, func(ctx context.Context, query string, args ...any) (*sql.Rows, error) {


We need to handle closing the opened rows explicitly. JDBC backfills use defer tx.Rollback() which handles this but I don't think this is the case here...

Find out if there is a way to do this from here, but if we need to change the jdbc.MapScanConcurrent then we will need to test it thoroughly

mihir-datazip · 2026-08-07T12:56:37Z

@krishanu7 I see we still use MapScan in drivers/mssql/internal/cdc.go as well, this PR should handle its removal there as well I think.

		// Use MapScan to properly convert data types including binary types
		// TODO: check if we can use MapScanConcurrent for mssql
		// rowBytes is the after-image data-column byte sum (excludes __$* metadata columns), attached to the emitted change below.
		record := make(map[string]interface{})
		rowBytes, err := jdbc.MapScan(rows, record, m.dataTypeConverter, mssqlCDCColumnSizer)
		if err != nil {
			return fmt.Errorf("failed to scan MSSQL CDC row: %s", err)
		}

Also we can then remove the function and the TODO completely then

…n, add tests

krishanu7 · 2026-08-11T04:26:08Z

Hey @mihir-datazip — addressed your review feedback. Quick summary:

Fixed the rows leak: Reader.Capture() now calls defer rows.Close()
Moved MSSQL CDC off MapScan → MapScanConcurrent (same pattern as incremental/backfill)
Removed dead MapScan + the TODO
Added pkg/jdbc/reader_test.go (9 tests for close lifecycle + MapScanConcurrent behavior)
Minor gofmt cleanup in postgres incremental

Tests Ran

go test -v ./pkg/jdbc/... -count=1 # 9/9 pass
make golangci # pass

built postgres, mysql, mssql, oracle modules — all good

Manual CDC check Mssql

make db.mssql.start

make dev.mssql.build

./drivers/mssql/olake discover --config /tmp/olake-mssql-cdc-test/source.json

Sync 1 — backfill (CDC initial load)

./drivers/mssql/olake sync
--config /tmp/olake-mssql-cdc-test/source.json
--catalog /tmp/olake-mssql-cdc-test/streams.json
--destination /tmp/olake-mssql-cdc-test/destination.json

Insert a row, wait for CDC capture agent

INSERT INTO dbo.users (email) VALUES ('cdc-test@example.com');
sleep 5

Sync 2 — CDC insert

./drivers/mssql/olake sync ... --state /tmp/olake-mssql-cdc-test/state.json

Result:

Update + delete, wait for CDC agent

UPDATE dbo.users SET email='updated@example.com' WHERE id=1;
DELETE FROM dbo.users WHERE email='cdc-test@example.com';
sleep 6

Sync 3 — CDC update + delete

./drivers/mssql/olake sync ... --state /tmp/olake-mssql-cdc-test/state.json

mihir-datazip · 2026-08-13T06:46:54Z

@krishanu7 lint is failing on your PR, please enable precommit hooks and also add go formatter in your IDE to avoid lint failures

krishanu7 · 2026-08-13T07:24:46Z

@krishanu7 lint is failing on your PR, please enable precommit hooks and also add go formatter in your IDE to avoid lint failures

Okay thanks

mihir-datazip · 2026-08-13T07:16:49Z

I don't think these UTs are helping to assert the changes we are adding...

database/sql auto-closes Rows whenever Next() returns false — both on exhaustion (EOF) and on an iteration error (nextLocked sets doClose whenever lasterr != nil). That means the driver-level closeCount reaches 1 with or without the defer Close() you added to Capture:

TestReaderCapture_closesRowsOnSuccess — passes even if you delete the defer (auto-close on EOF).

TestReaderCapture_closesRowsOnRowsErr — passes without the defer (auto-close on iteration error).

TestReaderCapture_doesNotLeakConnection — iterates to exhaustion every loop, so the conn is returned via auto-close; passes without the defer.

TestReaderCapture_closesRowsOnCaptureError — the only test that actually fails without your fix, because an early return from onCapture is the one path where rows are still open.

So the suite mostly re-verifies database/sql's own behavior. You can confirm with a quick mutation test: comment out the defer block in reader.go:46-48 and run go test ./pkg/jdbc/ — only closesRowsOnCaptureError fails.

mihir-datazip · 2026-08-13T07:19:48Z

I would recommend to add testing for Reader with a fake iterable, that way we can assert the correct behaviour of Capture easily without need for sql at all.

mihir-datazip · 2026-08-13T07:25:28Z

 	if err != nil {
 		return err
 	}
+	if closer, ok := any(rows).(interface{ Close() error }); ok {


Instead of doing this, we should just add Close() in iterable as we already use sql.Rows which needs closing so its better to have it in the interface by design rather than guessing

krishanu7 · 2026-08-26T12:43:25Z

@mihir-datazip I found a potential loophole in MapConncurent around producer consumer cancellation. When the consumer fails and ctx.Done() fires, the producer currently returns nill "case <-ctx.Done(): return nil " , which I think may cause Capture to continue scanning the remaining rows instead of stopping early. let me know whether my understanding is correct.

Add Close() to types.Iterable so Capture always releases cursors by contract instead of a runtime type assertion. Replace sql-mock close tests with a fake iterable that does not auto-close, and stop the MapScanConcurrent producer when context is canceled so rows close promptly on downstream errors.

perf(jdbc): optimize incremental sync with MapScanConcurrent

89488dc

krishanu7 mentioned this pull request Aug 4, 2026

TODO: Use MapScanConcurrent instead of MapScan for incremental as well #1055

Open

krishanu7 temporarily deployed to integration_tests August 7, 2026 10:34 — with GitHub Actions Inactive

mihir-datazip reviewed Aug 7, 2026

View reviewed changes

fix(jdbc): close rows in Capture, complete MapScanConcurrent migratio…

18e2227

…n, add tests

krishanu7 requested a review from mihir-datazip August 11, 2026 04:26

Merge branch 'staging' into perf/incremental-mapscan-concurrent

06020f6

mihir-datazip temporarily deployed to integration_tests August 13, 2026 06:36 — with GitHub Actions Inactive

mihir-datazip requested changes Aug 13, 2026

View reviewed changes

Merge branch 'staging' into perf/incremental-mapscan-concurrent

94408cf

vaibhav-datazip requested a deployment to integration_tests August 18, 2026 09:22 — with GitHub Actions Waiting

krishanu7 requested a review from mihir-datazip August 26, 2026 19:06

Uh oh!

Conversation

krishanu7 commented Aug 2, 2026

Description

Type of change

How Has This Been Tested?

Screenshots or Recordings

Documentation

Related PR's (If Any):

Uh oh!

krishanu7 commented Aug 4, 2026

Uh oh!

krishanu7 commented Aug 4, 2026

Uh oh!

krishanu7 commented Aug 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mihir-datazip commented Aug 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mihir-datazip Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

mihir-datazip Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

mihir-datazip commented Aug 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

krishanu7 commented Aug 11, 2026

Tests Ran

built postgres, mysql, mssql, oracle modules — all good

Manual CDC check Mssql

Sync 1 — backfill (CDC initial load)

Insert a row, wait for CDC capture agent

Sync 2 — CDC insert

Update + delete, wait for CDC agent

Sync 3 — CDC update + delete

Uh oh!

mihir-datazip commented Aug 13, 2026

Uh oh!

krishanu7 commented Aug 13, 2026

Uh oh!

mihir-datazip Aug 13, 2026

Choose a reason for hiding this comment

Uh oh!

mihir-datazip Aug 13, 2026

Choose a reason for hiding this comment

Uh oh!

mihir-datazip Aug 13, 2026

Choose a reason for hiding this comment

Uh oh!

krishanu7 commented Aug 26, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

krishanu7 commented Aug 4, 2026 •

edited

Loading

mihir-datazip commented Aug 7, 2026 •

edited

Loading

mihir-datazip commented Aug 7, 2026 •

edited

Loading