Skip to content

Commit

Permalink
fix: improve query performances (v1.9.x) (#446)
Browse files Browse the repository at this point in the history 
* fix: improve query performances

* ci: fix python sdk

---------

Co-authored-by: Maxence Maireaux <[email protected]>
  • Loading branch information
@paul-nicolas @flemzord
paul-nicolas and flemzord authored Jun 15, 2023
1 parent a9c6a77 commit 9a1d239
Show file tree
Hide file tree
Showing 9 changed files with 135 additions and 39 deletions.
16 changes: 8 additions & 8 deletions pkg/api/controllers/pagination_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -471,7 +471,7 @@ func TestCursor(t *testing.T) {
t.Run("GetAccounts", func(t *testing.T) {
httpResponse := internal.GetAccounts(api, url.Values{
"after": []string{"accounts:15"},
"address": []string{"acc.*"},
"address": []string{"accounts:.*"},
"metadata[foo]": []string{"bar"},
"balance": []string{"1"},
controllers.QueryKeyBalanceOperator: []string{"gte"},
Expand All @@ -483,7 +483,7 @@ func TestCursor(t *testing.T) {
res, err := base64.RawURLEncoding.DecodeString(cursor.Next)
require.NoError(t, err)
require.Equal(t,
`{"pageSize":3,"offset":3,"after":"accounts:15","address":"acc.*","metadata":{"foo":"bar"},"balance":"1","balanceOperator":"gte"}`,
`{"pageSize":3,"offset":3,"after":"accounts:15","address":"accounts:.*","metadata":{"foo":"bar"},"balance":"1","balanceOperator":"gte"}`,
string(res))

httpResponse = internal.GetAccounts(api, url.Values{
Expand All @@ -495,12 +495,12 @@ func TestCursor(t *testing.T) {
res, err = base64.RawURLEncoding.DecodeString(cursor.Previous)
require.NoError(t, err)
require.Equal(t,
`{"pageSize":3,"offset":0,"after":"accounts:15","address":"acc.*","metadata":{"foo":"bar"},"balance":"1","balanceOperator":"gte"}`,
`{"pageSize":3,"offset":0,"after":"accounts:15","address":"accounts:.*","metadata":{"foo":"bar"},"balance":"1","balanceOperator":"gte"}`,
string(res))
res, err = base64.RawURLEncoding.DecodeString(cursor.Next)
require.NoError(t, err)
require.Equal(t,
`{"pageSize":3,"offset":6,"after":"accounts:15","address":"acc.*","metadata":{"foo":"bar"},"balance":"1","balanceOperator":"gte"}`,
`{"pageSize":3,"offset":6,"after":"accounts:15","address":"accounts:.*","metadata":{"foo":"bar"},"balance":"1","balanceOperator":"gte"}`,
string(res))
})

Expand Down Expand Up @@ -545,7 +545,7 @@ func TestCursor(t *testing.T) {
t.Run("GetBalances", func(t *testing.T) {
httpResponse := internal.GetBalances(api, url.Values{
"after": []string{"accounts:15"},
"address": []string{"acc.*"},
"address": []string{"accounts:.*"},
controllers.QueryKeyPageSize: []string{"3"},
})
assert.Equal(t, http.StatusOK, httpResponse.Result().StatusCode, httpResponse.Body.String())
Expand All @@ -554,7 +554,7 @@ func TestCursor(t *testing.T) {
res, err := base64.RawURLEncoding.DecodeString(cursor.Next)
require.NoError(t, err)
require.Equal(t,
`{"pageSize":3,"offset":3,"after":"accounts:15","address":"acc.*"}`,
`{"pageSize":3,"offset":3,"after":"accounts:15","address":"accounts:.*"}`,
string(res))

httpResponse = internal.GetBalances(api, url.Values{
Expand All @@ -566,12 +566,12 @@ func TestCursor(t *testing.T) {
res, err = base64.RawURLEncoding.DecodeString(cursor.Previous)
require.NoError(t, err)
require.Equal(t,
`{"pageSize":3,"offset":0,"after":"accounts:15","address":"acc.*"}`,
`{"pageSize":3,"offset":0,"after":"accounts:15","address":"accounts:.*"}`,
string(res))
res, err = base64.RawURLEncoding.DecodeString(cursor.Next)
require.NoError(t, err)
require.Equal(t,
`{"pageSize":3,"offset":6,"after":"accounts:15","address":"acc.*"}`,
`{"pageSize":3,"offset":6,"after":"accounts:15","address":"accounts:.*"}`,
string(res))
})

Expand Down
82 changes: 64 additions & 18 deletions pkg/storage/sqlstorage/accounts.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
"encoding/base64"
"encoding/json"
"fmt"
"regexp"
"strconv"
"strings"
"time"
Expand All @@ -17,6 +18,11 @@ import (
"github.com/pkg/errors"
)

// This regexp is used to validate the account name
// If the account name is not valid, it means that the user putted a regex in
// the address filter, and we have to change the postgres operator used.
var accountNameRegex = regexp.MustCompile(`^[a-zA-Z_0-9]+$`)

func (s *Store) buildAccountsQuery(p ledger.AccountsQuery) (*sqlbuilder.SelectBuilder, AccPaginationToken) {
sb := sqlbuilder.NewSelectBuilder()
t := AccPaginationToken{}
Expand All @@ -30,11 +36,26 @@ func (s *Store) buildAccountsQuery(p ledger.AccountsQuery) (*sqlbuilder.SelectBu
)

if address != "" {
arg := sb.Args.Add("^" + address + "$")
switch s.Schema().Flavor() {
case sqlbuilder.PostgreSQL:
sb.Where("address ~* " + arg)
src := strings.Split(address, ":")
sb.Where(fmt.Sprintf("jsonb_array_length(address_json) = %d", len(src)))

for i, segment := range src {
if segment == ".*" || segment == "*" || segment == "" {
continue
}

operator := "=="
if !accountNameRegex.MatchString(segment) {
operator = "like_regex"
}

arg := sb.Args.Add(segment)
sb.Where(fmt.Sprintf("address_json @@ ('$[%d] %s \"' || %s::text || '\"')::jsonpath", i, operator, arg))
}
case sqlbuilder.SQLite:
arg := sb.Args.Add("^" + address + "$")
sb.Where("address REGEXP " + arg)
}
t.AddressRegexpFilter = address
Expand Down Expand Up @@ -206,14 +227,33 @@ func (s *Store) GetAccount(ctx context.Context, addr string) (*core.Account, err
}

func (s *Store) ensureAccountExists(ctx context.Context, account string) error {
var accountBy string
switch s.schema.Flavor() {
case sqlbuilder.PostgreSQL:
a, err := json.Marshal(strings.Split(account, ":"))
if err != nil {
return err
}
accountBy = string(a)
case sqlbuilder.SQLite:
accountBy = account
}

sb := sqlbuilder.NewInsertBuilder()
sqlq, args := sb.
InsertInto(s.schema.Table("accounts")).
Cols("address", "metadata").
Values(account, "{}").
SQL("ON CONFLICT DO NOTHING").
BuildWithFlavor(s.schema.Flavor())
sb = sb.InsertInto(s.schema.Table("accounts"))

switch s.schema.Flavor() {
case sqlbuilder.PostgreSQL:
sb = sb.Cols("address", "metadata", "address_json").
Values(account, "{}", accountBy).
SQL("ON CONFLICT DO NOTHING")
case sqlbuilder.SQLite:
sb = sb.Cols("address", "metadata").
Values(account, "{}").
SQL("ON CONFLICT DO NOTHING")
}

sqlq, args := sb.BuildWithFlavor(s.schema.Flavor())

executor, err := s.executorProvider(ctx)
if err != nil {
Expand All @@ -231,17 +271,23 @@ func (s *Store) UpdateAccountMetadata(ctx context.Context, address string, metad
if err != nil {
return err
}

placeholder := ib.Var(metadataData)
ib.
InsertInto(s.schema.Table("accounts")).
Cols("address", "metadata").
Values(address, metadataData)

switch Flavor(s.schema.Flavor()) {
case PostgreSQL:
ib.SQL(fmt.Sprintf("ON CONFLICT (address) DO UPDATE SET metadata = accounts.metadata || %s", placeholder))
case SQLite:
ib.SQL(fmt.Sprintf("ON CONFLICT (address) DO UPDATE SET metadata = json_patch(metadata, %s)", placeholder))
ib = ib.InsertInto(s.schema.Table("accounts"))

switch s.schema.Flavor() {
case sqlbuilder.PostgreSQL:
addressBy, err := json.Marshal(strings.Split(address, ":"))
if err != nil {
return err
}
ib = ib.Cols("address", "metadata", "address_json").
Values(address, metadataData, addressBy).
SQL(fmt.Sprintf("ON CONFLICT (address) DO UPDATE SET metadata = accounts.metadata || %s", placeholder))
case sqlbuilder.SQLite:
ib = ib.Cols("address", "metadata").
Values(address, metadataData).
SQL(fmt.Sprintf("ON CONFLICT (address) DO UPDATE SET metadata = json_patch(metadata, %s)", placeholder))
}

executor, err := s.executorProvider(ctx)
Expand Down
34 changes: 30 additions & 4 deletions pkg/storage/sqlstorage/balances.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"encoding/base64"
"encoding/json"
"fmt"
"strconv"
"strings"

Expand All @@ -21,11 +22,26 @@ func (s *Store) GetBalancesAggregated(ctx context.Context, q ledger.BalancesQuer
sb.GroupBy("asset")

if q.Filters.AddressRegexp != "" {
arg := sb.Args.Add("^" + q.Filters.AddressRegexp + "$")
switch s.Schema().Flavor() {
case sqlbuilder.PostgreSQL:
sb.Where("account ~* " + arg)
src := strings.Split(q.Filters.AddressRegexp, ":")
sb.Where(fmt.Sprintf("jsonb_array_length(account_json) = %d", len(src)))

for i, segment := range src {
if segment == ".*" || segment == "*" || segment == "" {
continue
}

operator := "=="
if !accountNameRegex.MatchString(segment) {
operator = "like_regex"
}

arg := sb.Args.Add(segment)
sb.Where(fmt.Sprintf("account_json @@ ('$[%d] %s \"' || %s::text || '\"')::jsonpath", i, operator, arg))
}
case sqlbuilder.SQLite:
arg := sb.Args.Add("^" + q.Filters.AddressRegexp + "$")
sb.Where("account REGEXP " + arg)
}
}
Expand Down Expand Up @@ -96,11 +112,21 @@ func (s *Store) GetBalances(ctx context.Context, q ledger.BalancesQuery) (api.Cu
}

if q.Filters.AddressRegexp != "" {
arg := sb.Args.Add("^" + q.Filters.AddressRegexp + "$")
switch s.Schema().Flavor() {
case sqlbuilder.PostgreSQL:
sb.Where("account ~* " + arg)
src := strings.Split(q.Filters.AddressRegexp, ":")
sb.Where(fmt.Sprintf("jsonb_array_length(account_json) = %d", len(src)))

for i, segment := range src {
if segment == ".*" || segment == "*" || segment == "" {
continue
}

arg := sb.Args.Add(segment)
sb.Where(fmt.Sprintf("account_json @@ ('$[%d] like_regex \"' || %s::text || '\"')::jsonpath", i, arg))
}
case sqlbuilder.SQLite:
arg := sb.Args.Add("^" + q.Filters.AddressRegexp + "$")
sb.Where("account REGEXP " + arg)
}
t.AddressRegexpFilter = q.Filters.AddressRegexp
Expand Down
2 changes: 1 addition & 1 deletion pkg/storage/sqlstorage/balances_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -114,7 +114,7 @@ func testGetBalances(t *testing.T, store *sqlstorage.Store) {
ledger.BalancesQuery{
PageSize: 10,
AfterAddress: "world",
Filters: ledger.BalancesQueryFilters{AddressRegexp: "users.+"},
Filters: ledger.BalancesQueryFilters{AddressRegexp: "users:.+"},
})
assert.NoError(t, err)
assert.Equal(t, 10, cursor.PageSize)
Expand Down
6 changes: 6 additions & 0 deletions pkg/storage/sqlstorage/migrates/18-optimized-accounts/postgres.sql
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
--statement
alter table "VAR_LEDGER_NAME".accounts add column if not exists address_json jsonb;
--statement
UPDATE "VAR_LEDGER_NAME".accounts SET address_json = to_jsonb(string_to_array(address, ':'));
--statement
create index if not exists accounts_address_json on "VAR_LEDGER_NAME".accounts using GIN(address_json);
6 changes: 6 additions & 0 deletions pkg/storage/sqlstorage/migrates/19-optimized-volumes/postgres.sql
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
--statement
alter table "VAR_LEDGER_NAME".volumes add column if not exists account_json jsonb;
--statement
UPDATE "VAR_LEDGER_NAME".volumes SET account_json = to_jsonb(string_to_array(account, ':'));
--statement
create index if not exists volumes_account_json on "VAR_LEDGER_NAME".volumes using GIN(account_json);
2 changes: 1 addition & 1 deletion pkg/storage/sqlstorage/store_ledger_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -458,7 +458,7 @@ func testGetAccounts(t *testing.T, store *sqlstorage.Store) {
accounts, err = store.GetAccounts(context.Background(), ledger.AccountsQuery{
PageSize: 10,
Filters: ledger.AccountsQueryFilters{
Address: ".*der.*",
Address: `^.*der:.*$`,
},
})
require.NoError(t, err)
Expand Down
24 changes: 19 additions & 5 deletions pkg/storage/sqlstorage/volumes.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,22 +2,36 @@ package sqlstorage

import (
"context"
"encoding/json"
"strings"

"github.com/huandu/go-sqlbuilder"
"github.com/numary/ledger/pkg/core"
)

func (s *Store) updateVolumes(ctx context.Context, volumes core.AccountsAssetsVolumes) error {
for account, accountVolumes := range volumes {
accountBy, err := json.Marshal(strings.Split(account, ":"))
if err != nil {
panic(err)
}

for asset, volumes := range accountVolumes {
ib := sqlbuilder.NewInsertBuilder()
inputArg := ib.Var(volumes.Input.String())
outputArg := ib.Var(volumes.Output.String())
ib.
InsertInto(s.schema.Table("volumes")).
Cols("account", "asset", "input", "output").
Values(account, asset, volumes.Input.String(), volumes.Output.String()).
SQL("ON CONFLICT (account, asset) DO UPDATE SET input = " + inputArg + ", output = " + outputArg)
ib = ib.InsertInto(s.schema.Table("volumes"))

switch s.schema.Flavor() {
case sqlbuilder.PostgreSQL:
ib = ib.Cols("account", "asset", "input", "output", "account_json").
Values(account, asset, volumes.Input.String(), volumes.Output.String(), accountBy).
SQL("ON CONFLICT (account, asset) DO UPDATE SET input = " + inputArg + ", output = " + outputArg)
case sqlbuilder.SQLite:
ib = ib.Cols("account", "asset", "input", "output").
Values(account, asset, volumes.Input.String(), volumes.Output.String()).
SQL("ON CONFLICT (account, asset) DO UPDATE SET input = " + inputArg + ", output = " + outputArg)
}

sqlq, args := ib.BuildWithFlavor(s.schema.Flavor())

Expand Down
2 changes: 0 additions & 2 deletions sdk/configs/python.yaml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,5 +8,3 @@ additionalProperties:
files:
Taskfile.yml.mustache:
destinationFilename: Taskfile.yml
.github/workflows/release.yml.mustache:
destinationFilename: .github/workflows/release.yml

0 comments on commit 9a1d239

Please sign in to comment.