adds support for streaming multi tables #40

brunocalza · 2024-03-06T20:04:23Z

Summary

This PR adds support for streaming changes of multiple Postgres tables at the same time

Context

When you created a vault and started streaming database changes to that vault, only the changes from one table could be streamed. The reason for that was that the table was picked from the vault's name. For example, if the vault was called demo.test, the CLI assumed that the test would be the table. That meant that if you wanted to stream multiple tables, you would have to start multiple CLI processes, which is not ideal

This PR changes this behavior, to allow multiple tables to be picked and its changes streamed to a particular vault, all in a simple vaults stream call.

Changes

There's no relation between the vault's name and the table anymore. When you create a vault it simply creates the vault in the provider and smart contract. That means vaults create command doesn't accept dburi flag anymore, because there's no need to connect to the database at this point.
In order to provide the tables you want to capture changes you have to provide the dburi and tables flag to the vaults stream command. The tables value is a list of tables separated by comma (e.g. t1,t2). You can also provide the window-size flag.
Each table can have its own schema and one parquet file will be generated for each table and uploaded to Basin Provider

Usage

Creating a vault

vaults create --account [ETH_ADDRESS] demo.test

Streaming changes

Suppose you want to stream tables t1 and t2 to vault demo.test, this is the command:

vaults stream --dburi [DB_URI] --tables t1,t2 --private-key [PRIVATE_KEY] demo.test

Note: There's one caveat with the current implementation. Once a stream command is called for a particular vault with a particular dburi and tables there's no way of changing that. For example, suppose that now you want to add another table, that wouldn't work. I wonder what would be the ideal flow here.

Signed-off-by: Bruno Calza <brunoangelicalza@gmail.com>

brunocalza · 2024-03-13T17:00:08Z

cmd/vaults/commands.go

-		return exists, nil
-	}
+// DatabaseStreamSetup setups the database for streaming.
+type DatabaseStreamSetup struct {


created this object to make it easier to setup the database for replication. it can:

create the publication in Postgres

fetch the schemas of the tables in Postgres so that we can create a similar table in Duckdb

brunocalza · 2024-03-13T17:01:36Z

cmd/vaults/commands.go

@@ -32,8 +32,8 @@ import (
 var vaultNameRx = regexp.MustCompile(`^([a-zA-Z_][a-zA-Z0-9_]*)[.]([a-zA-Z_][a-zA-Z0-9_]*$)`)



lots of changes in this file, but in essence:

moved the flags from create to stream

we're now passing multiple tables and schemas to the database manager

brunocalza · 2024-03-13T17:03:11Z

internal/app/db.go

 // NewDBManager creates a new DBManager.
 func NewDBManager(
-	dbDir, table string, cols []Column, windowInterval time.Duration, uploader *VaultsUploader,
+	dbDir string, schemas []TableSchema, windowInterval time.Duration, uploader *VaultsUploader,


passing multiple tables and schemas now

brunocalza · 2024-03-13T17:03:48Z

internal/app/db.go

-		return true, nil
-	}
+	exportedFiles := []string{}
+	for _, schema := range dbm.schemas {


for each schema we check if the table has rows, if not we will not export it

brunocalza · 2024-03-13T17:04:18Z

internal/app/db.go

@@ -214,13 +225,13 @@ func (dbm *DBManager) UploadAll(ctx context.Context) error {
 		if re.MatchString(fname) {
 			dbPath := path.Join(dbm.dbDir, fname)
 			exportAt := dbPath + ".parquet"
-			isEmpty, err := dbm.Export(ctx, exportAt)
+			files, err := dbm.Export(ctx, exportAt)


only exported files will be returned

brunocalza · 2024-03-13T17:05:29Z

internal/app/db.go

-
-	recordVals := []string{}
+	// build an insert stmt for each record inside tx
+	stmts := []string{}
 	for _, r := range tx.Records {


a tx can have records of different tables now, we gotta consider this

brunocalza · 2024-03-13T17:06:03Z

internal/app/db.go

-			if column.IsPrimary {
-				pks = column.Name
+	stmts := []string{}
+	for _, schema := range dbm.schemas {


we gotta create multiple tables now, one for each schema

dtbuchholz

lgtm

avichalp · 2024-03-15T07:53:42Z

internal/app/db.go

 ) *DBManager {
 	return &DBManager{
 		dbDir:          dbDir,
-		table:          table,
-		cols:           cols,
+		schemas:        schemas,


avichalp · 2024-03-15T08:00:01Z

cmd/vaults/commands.go

@@ -156,7 +133,8 @@ func newVaultCreateCommand() *cli.Command {
 }

 func newStreamCommand() *cli.Command {
-	var privateKey string
+	var privateKey, dburi, tables string
+	var winSize int64


should winSize be configurable per table?

do you think there's a good reason for it? we could, but we would need an export gouroutine per table which sounds more complicated.

yes, def sounds complicated for now. maybe @dtbuchholz can tell if people have asked for it or likely to ask for?

agreed—i think a single window size makes sense as a first feature. when i stream multiple tables, I'm essentially just trying to replicate the full db, so one window size is sufficient.

multiple windows could be useful, i suppose, but that's something we can wait on to hear directly from a dev. personally, i don't think i'd need it unless i was trying to really optimize for something.

avichalp

Looks good!

brunocalza added 2 commits March 6, 2024 17:03

adds support for streaming multi tables

d3c4931

Signed-off-by: Bruno Calza <brunoangelicalza@gmail.com>

fixes flaky test

feb7e3b

Signed-off-by: Bruno Calza <brunoangelicalza@gmail.com>

brunocalza self-assigned this Mar 7, 2024

moves window-size to stream command

8a7aa9c

Signed-off-by: Bruno Calza <brunoangelicalza@gmail.com>

brunocalza marked this pull request as ready for review March 13, 2024 16:57

brunocalza requested review from dtbuchholz and avichalp March 13, 2024 16:57

remove unnecessary config

fdb2c1c

Signed-off-by: Bruno Calza <brunoangelicalza@gmail.com>

brunocalza commented Mar 13, 2024

View reviewed changes

dtbuchholz approved these changes Mar 14, 2024

View reviewed changes

avichalp reviewed Mar 15, 2024

View reviewed changes

internal/app/db.go

) *DBManager {

return &DBManager{

dbDir: dbDir,

table: table,

cols: cols,

schemas: schemas,

Copy link

Contributor

avichalp Mar 15, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍

avichalp reviewed Mar 15, 2024

View reviewed changes

avichalp approved these changes Mar 15, 2024

View reviewed changes

brunocalza merged commit ad95dcb into main Mar 15, 2024
2 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

adds support for streaming multi tables #40

adds support for streaming multi tables #40

brunocalza commented Mar 6, 2024 •

edited

Loading

brunocalza Mar 13, 2024

brunocalza Mar 13, 2024

brunocalza Mar 13, 2024

brunocalza Mar 13, 2024

brunocalza Mar 13, 2024

brunocalza Mar 13, 2024

brunocalza Mar 13, 2024

dtbuchholz left a comment

avichalp Mar 15, 2024

avichalp Mar 15, 2024

brunocalza Mar 15, 2024

avichalp Mar 15, 2024

dtbuchholz Mar 15, 2024

avichalp left a comment

		@@ -32,8 +32,8 @@ import (
		var vaultNameRx = regexp.MustCompile(`^([a-zA-Z_][a-zA-Z0-9_])[.]([a-zA-Z_][a-zA-Z0-9_]$)`)

adds support for streaming multi tables #40

adds support for streaming multi tables #40

Conversation

brunocalza commented Mar 6, 2024 • edited Loading

Summary

Context

Changes

Usage

Creating a vault

Streaming changes

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dtbuchholz left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

avichalp left a comment

Choose a reason for hiding this comment

brunocalza commented Mar 6, 2024 •

edited

Loading