blocking local replication by avichalp · Pull Request #12 · tablelandnetwork/basin-cli

avichalp · 2023-10-26T16:55:34Z

Summary

Changes the streamer to replay the WAL records locally on a duck db instance
On replication start command, all the existing db files are uploaded before we start replicating the target database to local duck db.
The DBManager Maintains a time window. It is configurable. By default it is set to 3600 seconds. When the window is passed the current database is exported and uploaded to the provider.
This implementation is single threaded and blocking at the moment. I think, this is good enough to start with. We can parallelize this implementation when we are dealing with higher data volume. The single threaded, blocking approach greatly simplifies the current implementation.
See discussion

avichalp · 2023-11-01T13:41:20Z

internal/app/testdata/wal.input

@@ -1,2 +1,2 @@
-{"commit_lsn":957393888,"records":[{"action":"I","xid":1055,"lsn":"0/3910AB70","nextlsn":"","timestamp":"2023-08-22 10:56:08.151-03","schema":"public","table":"a","columns":[{"name":"a","type":"integer","value":4}],"pk":[]}]}
-{"commit_lsn":957398296,"records":[{"action":"I","xid":1058,"lsn":"0/3910B898","nextlsn":"","timestamp":"2023-08-22 14:44:02.043586-03","schema":"public","table":"t","columns":[{"name":"id","type":"integer","value":200232},{"name":"name","type":"text","value":"100"}],"pk":[{"name":"id","type":"integer"}]}]}
+{"commit_lsn":957398296,"records":[{"action":"I","xid":1058,"lsn":"0/3910B898","nextlsn":"","timestamp":"2023-08-22 14:44:02.043586-03","schema":"public","table":"t","columns":[{"name":"id","type":"integer","value":200232},{"name":"name","type":"text","value":"100"}],"pk":[{"name":"id","type":"integer"}]}]}


changed the first example to have the same schema for current test cases

avichalp · 2023-11-01T13:42:29Z

internal/app/streamer.go

-		signature, err := b.sign(capnpTx)
-		if err != nil {
-			return fmt.Errorf("sign: %s", err)
+		if err := b.dbMngr.Replay(ctx, tx); err != nil {


local WAL replay before committing LSN

avichalp · 2023-11-01T13:43:23Z

internal/app/streamer.go

 	return nil
 }
-
-func (b *BasinStreamer) sign(tx basincapnp.Tx) ([]byte, error) {


we wouldn't need sign and push because we can re use the file uploader's functions

avichalp · 2023-11-01T13:44:25Z

internal/app/streamer.go

+
+	// Create sink table in local DB
+	if err := b.dbMngr.Setup(ctx); err != nil {
+		return fmt.Errorf("cannot setup db: %s", err)


creates and sets up a new local db before starting replication

avichalp · 2023-11-01T13:45:27Z

internal/app/db.go

+}
+
+// queryFromWAL creates a query for a WAL TX records.
+func (dbm *DBManager) queryFromWAL(tx *pgrepl.Tx) string {


pg -> ddb type remappings is a todo for a separate PR

avichalp · 2023-11-01T14:02:36Z

cmd/basin/publication.go

-		WHERE c.table_name = $1; 
-		`, rel,
-	)
+	columns, err := inspectTable(ctx, tx, rel)


extracted this logic to get table schema out into a new function

avichalp · 2023-11-01T14:03:26Z

cmd/basin/publication.go

-	}
-	defer rows.Close()
-
-	type column struct {


made the column public and moved to the app package since it needs to be shared with the local db

avichalp · 2023-11-01T14:06:38Z

internal/app/db.go

+func (dbm *DBManager) Replay(ctx context.Context, tx *pgrepl.Tx) error {
+	if dbm.windowPassed() {
+		slog.Info("replacing current db before replaying further txs")
+		if err := dbm.replace(ctx); err != nil {


If export or upload fails, the replicator will exit. the next time it starts it will first upload all existing db files and then start replicating on a fresh db

avichalp · 2023-11-01T14:08:42Z

internal/app/db.go

+	createdAT time.Time
+	cols      []Column
+	winSize   time.Duration
+	uploader  *BasinUploader


reuses file uploader to upload the exported files

avichalp · 2023-11-01T14:09:33Z

internal/app/db.go

+	return strings.Join(queries, "\n")
+}
+
+func (dbm *DBManager) replace(ctx context.Context) error {


this is how the db is being replaced when the winSize exceeds

avichalp · 2023-11-01T14:11:39Z

internal/app/streamer_test.go

 	return nil
 }
+
+func (bp *basinProviderMock) Upload(


the mock impl, makes a copy of the file that was uploaded. to test, we will import this file into a db and make assertions

avichalp · 2023-11-02T11:44:34Z

internal/app/db.go

+}
+
+// UploadAll uploads all db dumps in the db dir.
+func (dbm *DBManager) UploadAll(ctx context.Context) error {


UploadAll is called when the replication starts to send any remaining data left.

avichalp · 2023-11-02T12:15:39Z

cmd/basin/publication.go

+				return fmt.Errorf("upload all: %s", err)
+			}
+
+			basinStreamer := app.NewBasinStreamer(ns, r, dbm)


Streamer doesn't bind to BasinProvider anymore because going forward, BasinProvider is only used for fetching and updating metadata. Uploader is used to upload files. DBManager is instantiated with an Uploader.

Just a comment about an alternative more decoupled architecture. No need to change it.

One option would be to not have the Uploader as a dependency of DBManager. You could use a channel to synchronize both components and make them independent. Every time a file is ready to be uploaded, DBManager notifies the channel, Uploader picks up the notification and uploads the file. Notice that I'm not saying to make it async, it would still be blocking, but more decoupled and easier to test the parts independently.

avichalp · 2023-11-02T12:24:19Z

internal/app/basin_provider.go

 // BasinProvider ...
 type BasinProvider interface {
 	Create(context.Context, string, string, basincapnp.Schema, common.Address) (bool, error)
-	Push(context.Context, string, string, basincapnp.Tx, []byte) error


removing Push from here as it will not be used now. I left it intact in the capnp definition maybe we can clean that up when we remove capnp otherwise we need to regenerate the files in this PR

avichalp · 2023-11-02T12:30:54Z

pkg/basinprovider/provider.go

 }

-// Push pushes Postgres tx to the server.
-func (bp *BasinProvider) Push(ctx context.Context, ns string, rel string, tx basincapnp.Tx, sig []byte) error {


Push will not be used anymore

brunocalza

Nice work! In general, it looks good and correct to me. Left some comments for your considerations. Nothing critical, just minor improvements from my perspective.

brunocalza · 2023-11-02T14:05:35Z

cmd/basin/publication.go


 func newPublicationCreateCommand() *cli.Command {
-	var owner, dburi, provider string
+	var owner, dburi, provider, winSize string


I think you could switch the winSize type from string to int64, in L82 use the &cli.Int64Flag struct, avoin the parseInt in L127

brunocalza · 2023-11-02T14:11:30Z

cmd/basin/publication.go

+}
+
+func mkDBDir(dir, pub string) error {
+	if err := os.Mkdir(path.Join(dir, pub), 0o755); err != nil {


Maybe better using MkdirAll so it ignores if the dir already exists?

MkdirAll creates a directory named path, along with any necessary parents, and returns nil, or else returns an error. The permission bits perm (before umask) are used for all directories that MkdirAll creates. If path is already a directory, MkdirAll does nothing and returns nil.

brunocalza · 2023-11-02T14:26:10Z

internal/app/db.go

+
+		cols := strings.Join(columnNames, ", ")
+		valsStr := strings.Join(vals, ", ")
+		query := fmt.Sprintf(


you don't have to create multiple insert stmts (one for each record), you can do all into one statement, e.g. INSERT INTO %s (%s) values (%s), (%s), (%s), ...

nice! i didn't know that you could write an insert statement like that

brunocalza · 2023-11-02T14:41:31Z

internal/app/db.go

+	}
+
+	// Create a new db
+	db, err := dbm.NewDB()


this pattern is weird

db, err := dbm.NewDB() ... dbm.db = db

you're returning a db and setting into in the dbm later. Why not simply do the dbm.db = db inside NewDB?

also, if every time you call NewDB you have to call Setup, you can make Setup private and be called inside NewDb.

makes sense 👍

brunocalza · 2023-11-02T14:52:06Z

cmd/basin/publication.go

+				return fmt.Errorf("upload all: %s", err)
+			}
+
+			basinStreamer := app.NewBasinStreamer(ns, r, dbm)


Just a comment about an alternative more decoupled architecture. No need to change it.

One option would be to not have the Uploader as a dependency of DBManager. You could use a channel to synchronize both components and make them independent. Every time a file is ready to be uploaded, DBManager notifies the channel, Uploader picks up the notification and uploads the file. Notice that I'm not saying to make it async, it would still be blocking, but more decoupled and easier to test the parts independently.

brunocalza · 2023-11-02T14:54:43Z

internal/app/db.go

+		return err
+	}
+
+	oldDBFname := dbm.dbFname


nit: simply do oldDBPath := path.Join(dbm.dbDir, dbm.dbFname) in L99

brunocalza · 2023-11-02T14:56:47Z

internal/app/db.go

+		return fmt.Errorf("cannot stat file: %s", err)
+	}
+
+	progress := progressbar.DefaultBytes(


not sure if the progress bar makes sense in a background routine, this is more useful for a user when executing a command

can pass in nil to remove it but it looks okay to me since you can tell that the "replication is paused while this file uploads".

may we can change it later based on the user feedback

brunocalza · 2023-11-02T15:01:23Z

internal/app/streamer.go

 func (b *BasinStreamer) Run(ctx context.Context) error {
-	txs, table, err := b.replicator.StartReplication(ctx)
+	// Open a local DB for replaying txs
+	db, err := b.dbMngr.NewDB()


see: #12 (comment)

sanderpick

awesome work.

Signed-off-by: avichalp <hi@avichalp.me>

avichalp force-pushed the avichalp/local-replication-blocking branch from 01f18f0 to ff3dae6 Compare October 31, 2023 12:29

avichalp commented Nov 1, 2023

View reviewed changes

avichalp force-pushed the avichalp/local-replication-blocking branch from d01db14 to d7d0f6f Compare November 1, 2023 22:30

avichalp commented Nov 2, 2023

View reviewed changes

avichalp marked this pull request as ready for review November 2, 2023 12:43

avichalp requested review from brunocalza and sanderpick November 2, 2023 12:44

brunocalza approved these changes Nov 2, 2023

View reviewed changes

brunocalza reviewed Nov 2, 2023

View reviewed changes

brunocalza approved these changes Nov 6, 2023

View reviewed changes

sanderpick approved these changes Nov 6, 2023

View reviewed changes

avichalp force-pushed the avichalp/local-replication-blocking branch from 1dd5599 to 9ea6305 Compare November 9, 2023 07:22

avichalp changed the base branch from main to next November 9, 2023 07:44

avichalp added 4 commits November 9, 2023 22:45

Add duckdb go module

3cfe50d

Signed-off-by: avichalp <hi@avichalp.me>

init local replication

5e84e7c

Signed-off-by: avichalp <hi@avichalp.me>

refactor and add tests

4b9d505

Signed-off-by: avichalp <hi@avichalp.me>

Add unit tests for local replication

6f53834

Signed-off-by: avichalp <hi@avichalp.me>

avichalp added 17 commits November 9, 2023 22:45

impl blocking db uploads

09a5c50

Signed-off-by: avichalp <hi@avichalp.me>

fix lint

dfd62b3

Signed-off-by: avichalp <hi@avichalp.me>

cleanup gitignore

c6a6eb8

Signed-off-by: avichalp <hi@avichalp.me>

resolve merge conflicts

5ab217d

Signed-off-by: avichalp <hi@avichalp.me>

fix window size and progress bar

2cbc1bc

Signed-off-by: avichalp <hi@avichalp.me>

refactor

3bd3218

Signed-off-by: avichalp <hi@avichalp.me>

refactor blocking impl

c35f6bf

Signed-off-by: avichalp <hi@avichalp.me>

cleanup tests

32b6bba

Signed-off-by: avichalp <hi@avichalp.me>

refactor tests

06d99b0

Signed-off-by: avichalp <hi@avichalp.me>

cleanup tests, improve comments

d033d46

Signed-off-by: avichalp <hi@avichalp.me>

rename threshold fn

f6b36dc

Signed-off-by: avichalp <hi@avichalp.me>

fix lint

0b81d1d

Signed-off-by: avichalp <hi@avichalp.me>

fix cleanup bug

40b2195

Signed-off-by: avichalp <hi@avichalp.me>

refactor created at timestamps

ea2311a

Signed-off-by: avichalp <hi@avichalp.me>

fix error msg

10fc8cc

Signed-off-by: avichalp <hi@avichalp.me>

removes unused Push functions

4ff185d

Signed-off-by: avichalp <hi@avichalp.me>

address comments

e1358d9

Signed-off-by: avichalp <hi@avichalp.me>

avichalp force-pushed the avichalp/local-replication-blocking branch from 9ea6305 to e1358d9 Compare November 9, 2023 17:17

avichalp changed the base branch from next to main November 10, 2023 09:03

avichalp changed the base branch from main to next November 10, 2023 09:26

avichalp merged commit cf7afc6 into next Nov 10, 2023

avichalp deleted the avichalp/local-replication-blocking branch November 10, 2023 09:27

brunocalza mentioned this pull request Nov 15, 2023

Removes exporter and streaming mode tablelandnetwork/basin-provider#14

Merged

avichalp mentioned this pull request Nov 27, 2023

Local replication changes #22

Merged

Conversation

avichalp commented Oct 26, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

avichalp Nov 1, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

avichalp Nov 1, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

avichalp Nov 1, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

avichalp Nov 2, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

brunocalza left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sanderpick left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

avichalp commented Oct 26, 2023 •

edited

Loading

avichalp Nov 1, 2023 •

edited

Loading

avichalp Nov 1, 2023 •

edited

Loading

avichalp Nov 1, 2023 •

edited

Loading

avichalp Nov 2, 2023 •

edited

Loading