Import axon as an additional internal project, support JSONB fields in Changeset and expand demo #9

13rac1 · 2020-04-16T03:40:18Z

Demo run:

$ make demo
./scripts/setup_demo.sh
Creating warp-pipe_target_db_1 ... done
Creating warp-pipe_source_db_1 ... done
Creating warp-pipe_demo_service_1 ... done
db available
Setup Schema on Source
CREATE TABLE
Setup Schema on Target
CREATE TABLE
INFO[0000] registering trigger for table public.users   
INFO[0000] registering trigger for table public.pets    
Successfully created `warp_pipe` schema
Load the websocket demo app: http://localhost:8080/
Then press <Enter> to import SQL fixtures to see Warp Pipe Changesets get created...
Add data to Source
INSERT 0 5
Run axon to sync all changes. CTRL-C to exit when done.
INFO[0000] Starting notify listener for `warp_pipe_new_changeset`  component=listener
2020/04/15 20:38:33 row inserted: {timestamp: 2020-04-15 20:38:33.157167 -0700 PDT, kind: insert, schema: public, table: users}
2020/04/15 20:38:33 row inserted: {timestamp: 2020-04-15 20:38:33.157167 -0700 PDT, kind: insert, schema: public, table: users}
2020/04/15 20:38:33 row inserted: {timestamp: 2020-04-15 20:38:33.157167 -0700 PDT, kind: insert, schema: public, table: users}
2020/04/15 20:38:33 row inserted: {timestamp: 2020-04-15 20:38:33.157167 -0700 PDT, kind: insert, schema: public, table: users}
2020/04/15 20:38:33 row inserted: {timestamp: 2020-04-15 20:38:33.157167 -0700 PDT, kind: insert, schema: public, table: users}
2020/04/15 20:38:33 row inserted: {timestamp: 2020-04-15 20:38:33.157167 -0700 PDT, kind: insert, schema: public, table: pets}
2020/04/15 20:38:33 row inserted: {timestamp: 2020-04-15 20:38:33.157167 -0700 PDT, kind: insert, schema: public, table: pets}
2020/04/15 20:38:33 row inserted: {timestamp: 2020-04-15 20:38:33.157167 -0700 PDT, kind: insert, schema: public, table: pets}
2020/04/15 20:38:34 row inserted: {timestamp: 2020-04-15 20:38:33.157167 -0700 PDT, kind: insert, schema: public, table: pets}
2020/04/15 20:38:34 row inserted: {timestamp: 2020-04-15 20:38:33.157167 -0700 PDT, kind: insert, schema: public, table: pets}

Axon is a separate project which should be integrated into Warp Pipe

JSON/B fields are stored in the NewValue ChangesetColumn as a map[string]interface {} which the "sql" package does not support. Solution: Attempt to convert the map to a string or Fatal.

13rac1 · 2020-04-16T03:51:27Z

cmd/axon/sql.go

 	}
+	log.Printf("row inserted: %s", change)


These may be too noisy. Perhaps should be a roll up reported every 10-30 seconds? See #1

Resolves error from sql package: unsupported type []interface {}, a slice of interface.

Resolves: could not find name id in map[string]interface {}

bernielomax

This looks pretty good to me, plus it has already been proven. Just some minor comments.

bernielomax · 2020-04-24T00:20:16Z

cmd/axon/schema.go

+	// TODO: More thorough version checks
+	major, err := strconv.Atoi(version[0])
+	if err != nil {
+		return err


I think these could be panics, or have some context added to the messages.

Absolutely. There's a lot of UX to add. I'll add this as a todo.

cmd/axon/main.go

bernielomax · 2020-04-24T00:29:04Z

cmd/axon/schema.go

+
+	for _, r := range rows {
+		colDefault := strings.Split(r.ColumnDefault, "'")
+		sequenceName := colDefault[1]


Im guessing it is safe not to check the length?

There's a lot of additional error checking needed in this entire codebase, but, yes, unless there is a database is broken this should never fail.

cmd/axon/sql.go

Sequence values are not updated when a row is inserted including the sequenced value, often the ID. Checks for sequenced columns on startup and then runs setval() after each insert to manually update the value for that column.

InsertRow needs to load data from the sourceDB in the next commit

There is no way to detect sequence updates directly, so we'll assume they are changed on INSERT and update the orphaned sequences.

Handles: unsupported type []interface {}, a slice of interface when the array is not empty

bernielomax

Lgtm

cmd/axon/main.go

perangel · 2020-05-27T15:07:10Z

cmd/axon/schema.go

+var primaryKeys = make(map[string][]string)
+
+// maps serial key columns by table
+var columnSequences = make(map[string]string)


nit: perhaps rename this since it read as if its the order of the columns in a table, and not the colums that are of a sequence type. maybe something like sequenceColumns instead

Swapped variable name

cmd/axon/schema.go

perangel · 2020-05-27T15:26:27Z

cmd/axon/schema.go

+	err := conn.Select(&rows, `
+		SELECT
+			t.table_name,
+			string_to_array(string_agg(c.column_name, ','), ',') AS primary_key


It's been a while since I wrote this, but iirc this query is slightly different for other versions of PG (older, I believe). Just making note because i can't remember if its the meta-data tables or the method for going making the array that I had to modify at some point.

perangel · 2020-05-27T15:28:51Z

cmd/axon/schema.go

+	return "", false
+}
+
+func updateColumnSequence(conn *sqlx.DB, table string, columns []*warppipe.ChangesetColumn) error {


Please pluralize the name since it takes multiple columns s/updateColumnSequence/updateColumnSequences/g

perangel · 2020-05-27T15:29:47Z

cmd/axon/schema.go

+	return nil
+}
+
+// loadUnconnectedSequences loads all sequences in the source database not


Docstring and function name do not match. Should be loadOrphanSequences.

perangel · 2020-05-27T15:42:45Z

cmd/axon/schema.go

+	for _, sequenceName := range orphanSequences {
+		var lastVal int64 // PG bigint is 8bit
+
+		err := sourceDB.QueryRow(`


nit: you could also do

err := sourceDB.Get(&lastVal, "SELECT last_value FROM " + sequenceName)

unless the target pointer needs to be a struct, in which case nvm what I said :P

Yeah your version is cleaner, switching

perangel · 2020-05-27T15:45:46Z

cmd/axon/schema.go

+		}
+
+		var setVal string
+		err = targetDB.QueryRow(`


ditto here for all the other QueryRow calls

perangel · 2020-05-27T15:47:44Z

cmd/axon/sql.go

+	warppipe "github.com/perangel/warp-pipe"
+)
+
+var regexOneSpace = regexp.MustCompile(`\s+`)


nit: s/regexOneSpace/regexSpace/g since this regex is technically 1 or more spaces

perangel · 2020-05-27T16:26:34Z

db/schema.go

+				log.Printf("%v+", pgErr)
+			}
+			// Failure case resolution:
+			// GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA public TO master;


This is RDS specific, please remove this.

…s on err

Import axon as an additional internal project, support JSONB fields in Changeset and expand demo

13rac1 · 2020-05-29T16:53:41Z

fixups have been rebased out. Merge commit changed to 02f4352

perangel and others added 3 commits April 14, 2020 13:56

chore: Import axon from commit 553527c

efdc763

Axon is a separate project which should be integrated into Warp Pipe

chore: Add axon dependencies

458d5a7

feat: Support JSON/B fields in Changesets

f06083a

JSON/B fields are stored in the NewValue ChangesetColumn as a map[string]interface {} which the "sql" package does not support. Solution: Attempt to convert the map to a string or Fatal.

13rac1 force-pushed the import-axon branch 2 times, most recently from dc71281 to ff9ed0f Compare April 16, 2020 03:45

13rac1 commented Apr 16, 2020

View reviewed changes

feat: Expand Demo to sync data to target DB with axon

ee11232

13rac1 force-pushed the import-axon branch from ff9ed0f to ee11232 Compare April 16, 2020 16:48

feat: Additional detail to SQL error messages

9f23ed3

13rac1 force-pushed the import-axon branch from d58f781 to 5302bb1 Compare April 16, 2020 18:58

feat: Block Target DB versions less than 9.5

6198f5e

13rac1 force-pushed the import-axon branch from 5302bb1 to 6198f5e Compare April 16, 2020 19:10

13rac1 added 2 commits April 17, 2020 23:57

feat: Support empty array/slice fields in Changeset

9e989ac

Resolves error from sql package: unsupported type []interface {}, a slice of interface.

fix: prepareDeleteQuery will find query args in changeset.OldValues

f570d62

Resolves: could not find name id in map[string]interface {}

13rac1 force-pushed the import-axon branch 2 times, most recently from 01f9c73 to 781fc4c Compare April 18, 2020 20:23

bernielomax approved these changes Apr 24, 2020

View reviewed changes

13rac1 added 5 commits May 8, 2020 16:08

feat: Update SERIAL sequence after INSERT queries

b375b66

Sequence values are not updated when a row is inserted including the sequenced value, often the ID. Checks for sequenced columns on startup and then runs setval() after each insert to manually update the value for that column.

feat: Add sourceDB connection and differentiate targetDB

885aeb5

InsertRow needs to load data from the sourceDB in the next commit

feat: Update orphaned sequences on every INSERT

8f94419

There is no way to detect sequence updates directly, so we'll assume they are changed on INSERT and update the orphaned sequences.

feat: Changeset count on start, move Signal handler, print args on err

0a73e83

fix: Handle changesets including arrays of strings

c3ad1af

Handles: unsupported type []interface {}, a slice of interface when the array is not empty

13rac1 force-pushed the import-axon branch from 959163a to c3ad1af Compare May 8, 2020 23:08

bernielomax approved these changes May 8, 2020

View reviewed changes

perangel reviewed May 27, 2020

View reviewed changes

13rac1 added 5 commits May 27, 2020 13:46

fixup! feat: Changeset count on start, move Signal handler, print arg…

dc2d02a

…s on err

fixup! feat: Update SERIAL sequence after INSERT queries

fce1fde

fixup! feat: Block Target DB versions less than 9.5

0f5da22

fixup! feat: Block Target DB versions less than 9.5

3b5cf68

fixup! feat: Changeset count on start, move Signal handler, print arg…

1aa6b53

…s on err

13rac1 added 15 commits May 27, 2020 13:46

fixup! feat: Update SERIAL sequence after INSERT queries

c7d7a3c

fixup! feat: Update SERIAL sequence after INSERT queries

201c16f

fixup! feat: Update SERIAL sequence after INSERT queries

aa3e175

fixup! feat: Update SERIAL sequence after INSERT queries

7d186ba

fixup! feat: Update orphaned sequences on every INSERT

8c6f48e

fixup! feat: Update orphaned sequences on every INSERT

734ea16

fixup! feat: Update orphaned sequences on every INSERT

a7e5fd2

fixup! feat: Update orphaned sequences on every INSERT

438f6ad

fixup! feat: Update orphaned sequences on every INSERT

3f7044e

fixup! feat: Update orphaned sequences on every INSERT

343361c

fixup! feat: Update orphaned sequences on every INSERT

145d83a

fixup! feat: Update orphaned sequences on every INSERT

e51e722

fixup! feat: Additional detail to SQL error messages

808fc72

fixup! feat: Additional detail to SQL error messages

0f22315

fixup! feat: Update orphaned sequences on every INSERT

7dc5d18

perangel approved these changes May 29, 2020

View reviewed changes

perangel merged commit 49f5fe5 into perangel:master May 29, 2020

13rac1 pushed a commit to 13rac1/warp-pipe that referenced this pull request May 29, 2020

Merge pull request perangel#9 from 13rac1/import-axon

02f4352

Import axon as an additional internal project, support JSONB fields in Changeset and expand demo

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Import axon as an additional internal project, support JSONB fields in Changeset and expand demo #9

Import axon as an additional internal project, support JSONB fields in Changeset and expand demo #9

13rac1 commented Apr 16, 2020

13rac1 Apr 16, 2020

bernielomax left a comment

bernielomax Apr 24, 2020

13rac1 May 8, 2020

bernielomax Apr 24, 2020

13rac1 May 8, 2020

bernielomax left a comment

perangel May 27, 2020

13rac1 May 27, 2020

perangel May 27, 2020

perangel May 27, 2020

13rac1 May 27, 2020

perangel May 27, 2020

perangel May 27, 2020

perangel May 27, 2020

13rac1 May 27, 2020

perangel May 27, 2020

13rac1 May 27, 2020

perangel May 27, 2020

13rac1 May 27, 2020

perangel May 27, 2020

13rac1 May 27, 2020

13rac1 commented May 29, 2020

Import axon as an additional internal project, support JSONB fields in Changeset and expand demo #9

Import axon as an additional internal project, support JSONB fields in Changeset and expand demo #9

Conversation

13rac1 commented Apr 16, 2020

Choose a reason for hiding this comment

bernielomax 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

bernielomax 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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

13rac1 commented May 29, 2020