Merge pull request #60 from makasim/pg-read-preserv-order

pgdriver: preser order (do not read tx between snapshot xmin-xmax)
makasim · Nov 17, 2024 · 11c8a1f · 11c8a1f
2 parents d9f9e10 + d17296a
commit 11c8a1f
Show file tree

Hide file tree

Showing 2 changed files with 74 additions and 3 deletions.
diff --git a/pgdriver/get_states_by_labels_query.go b/pgdriver/get_states_by_labels_query.go
@@ -47,9 +47,21 @@ func (*queries) GetStatesByLabels(ctx context.Context, tx conntx, orLabels []map
 
 	q := `
 SELECT state, rev 
-FROM flowstate_states 
-WHERE ` + where + `
-ORDER BY "rev" ASC LIMIT ` + strconv.Itoa(len(ss)) + `;`
+FROM 
+    (
+		SELECT xmin::text::bigint, state, rev 
+		FROM flowstate_states 
+		WHERE ` + where + `
+		ORDER BY "rev" ASC LIMIT ` + strconv.Itoa(len(ss)) + `
+	) AS subquery
+	CROSS JOIN (
+    	SELECT 
+        split_part(pg_current_snapshot()::text, ':', 1)::bigint AS xmin, 
+        split_part(pg_current_snapshot()::text, ':', 2)::bigint AS xmax
+	) AS snapshot
+WHERE subquery.xmin < snapshot.xmin OR subquery.xmin > snapshot.xmax
+;
+`
 
 	rows, err := tx.Query(ctx, q, args...)
 	if err != nil {

diff --git a/pgdriver/get_states_by_labels_query_test.go b/pgdriver/get_states_by_labels_query_test.go
@@ -684,4 +684,63 @@ func TestQuery_GetStatesByLabels(main *testing.T) {
 			{ID: `4`, Rev: 4, Labels: map[string]string{`bar`: `barVal`}},
 		}, ss)
 	})
+
+	main.Run("PreserveOrder", func(t *testing.T) {
+		conn := openDB(t, `postgres://postgres:postgres@localhost:5432/postgres`, ``)
+
+		q := &queries{}
+
+		// past
+		require.NoError(t, q.InsertState(context.Background(), conn, &flowstate.State{ID: `1`}))
+
+		// still active
+		tx0, err := conn.Begin(context.Background())
+		require.NoError(t, err)
+		defer tx0.Rollback(context.Background())
+		require.NoError(t, q.InsertState(context.Background(), tx0, &flowstate.State{ID: `2`}))
+
+		// still active
+		tx1, err := conn.Begin(context.Background())
+		require.NoError(t, err)
+		defer tx1.Rollback(context.Background())
+		require.NoError(t, q.InsertState(context.Background(), tx1, &flowstate.State{ID: `3`}))
+
+		// commited but should not be visible
+		tx2, err := conn.Begin(context.Background())
+		require.NoError(t, err)
+		defer tx2.Rollback(context.Background())
+		require.NoError(t, q.InsertState(context.Background(), tx2, &flowstate.State{ID: `4`}))
+		require.NoError(t, tx2.Commit(context.Background()))
+
+		ss := make([]flowstate.State, 4)
+		ss, err = q.GetStatesByLabels(context.Background(), conn, nil, int64(0), ss)
+		require.NoError(t, err)
+		require.Equal(t, []flowstate.State{
+			{ID: `1`, Rev: 1},
+		}, ss)
+
+		// now, we should see 2
+		require.NoError(t, tx0.Commit(context.Background()))
+
+		ss = make([]flowstate.State, 4)
+		ss, err = q.GetStatesByLabels(context.Background(), conn, nil, int64(0), ss)
+		require.NoError(t, err)
+		require.Equal(t, []flowstate.State{
+			{ID: `1`, Rev: 1},
+			{ID: `2`, Rev: 2},
+		}, ss)
+
+		// now, we should everything
+		require.NoError(t, tx1.Commit(context.Background()))
+
+		ss = make([]flowstate.State, 4)
+		ss, err = q.GetStatesByLabels(context.Background(), conn, nil, int64(0), ss)
+		require.NoError(t, err)
+		require.Equal(t, []flowstate.State{
+			{ID: `1`, Rev: 1},
+			{ID: `2`, Rev: 2},
+			{ID: `3`, Rev: 3},
+			{ID: `4`, Rev: 4},
+		}, ss)
+	})
 }