fluidB
/
fluidb-old
1
0
Fork 0
Code Pull Requests Activity

Minor optimizations to Requests on Store

Browse Source
master
Philip O'Toole 8 months ago
parent 6a7ab4df83
commit e64985ff82
3 changed files with 79 additions and 47 deletions
Show Stats Download Patch File Download Diff File

1
CHANGELOG.md
Unescape Escape View File

@ -1,6 +1,7 @@
## 8.19.1 (unreleased)
### Implementation changes and bug fixes
- [PR #1670](https://github.com/rqlite/rqlite/pull/1670): Improve error message when query on remote node fails.
- [PR #1671](https://github.com/rqlite/rqlite/pull/1670): Minor optimizations to Unified Request processing.
## 8.19.0 (February 3rd 2024)
This release allows you to set a maximum amount of a time a query will run. If the query does not complete within the set time, an error will be returned.

42
store/store.go
Unescape Escape View File

@ -1158,15 +1158,17 @@ func (s *Store) Request(eqr *proto.ExecuteQueryRequest) ([]*proto.ExecuteQueryRe
if !s.open {
return nil, ErrNotOpen
}
nRW, nRO := s.RORWCount(eqr)
isLeader := s.raft.State() == raft.Leader
if s.QueriesOnly(eqr) {
if nRW == 0 {
// It's a little faster just to do a Query of the DB if we know there are no writes.
if eqr.Request.Transaction {
// Transaction requested during query, but not going through consensus. This means
// we need to block any database serialization during the query.
s.queryTxMu.RLock()
defer s.queryTxMu.RUnlock()
}
convertFn := func(qr []*proto.QueryRows) []*proto.ExecuteQueryResponse {
resp := make([]*proto.ExecuteQueryResponse, len(qr))
for i := range qr {
@ -1176,9 +1178,8 @@ func (s *Store) Request(eqr *proto.ExecuteQueryRequest) ([]*proto.ExecuteQueryRe
}
return resp
}
if eqr.Level == proto.QueryRequest_QUERY_REQUEST_LEVEL_NONE {
if s.raft.State() != raft.Leader && eqr.Freshness > 0 &&
if !isLeader && eqr.Freshness > 0 &&
time.Since(s.raft.LastContact()).Nanoseconds() > eqr.Freshness {
return nil, ErrStaleRead
}
@ -1193,13 +1194,30 @@ func (s *Store) Request(eqr *proto.ExecuteQueryRequest) ([]*proto.ExecuteQueryRe
}
}
if s.raft.State() != raft.Leader {
// At least one write in the request, or STRONG consistency requested, so
// we need to go through consensus. Check that we can do that.
if !isLeader {
return nil, ErrNotLeader
}
if !s.Ready() {
return nil, ErrNotReady
}
if nRO == 0 {
// No read-only requests, so we can go through the faster path.
convertFn := func(er []*proto.ExecuteResult) []*proto.ExecuteQueryResponse {
resp := make([]*proto.ExecuteQueryResponse, len(er))
for i := range er {
resp[i] = &proto.ExecuteQueryResponse{
Result: &proto.ExecuteQueryResponse_E{E: er[i]},
}
}
return resp
}
er, err := s.db.Execute(eqr.Request, eqr.Timings)
return convertFn(er), err
}
b, compressed, err := s.tryCompress(eqr)
if err != nil {
return nil, err
@ -1659,20 +1677,22 @@ func (s *Store) Noop(id string) (raft.ApplyFuture, error) {
return s.raft.Apply(bc, s.ApplyTimeout), nil
}
// QueriesOnly returns whether the given ExecuteQueryRequest contains only
// queries.
func (s *Store) QueriesOnly(eqr *proto.ExecuteQueryRequest) bool {
// RORWCount returns the number of read-only and read-write statements in the
// given ExecuteQueryRequest.
func (s *Store) RORWCount(eqr *proto.ExecuteQueryRequest) (nRW, nRO int) {
for _, stmt := range eqr.Request.Statements {
sql := stmt.Sql
if sql == "" {
continue
}
ro, err := s.db.StmtReadOnly(sql)
if err != nil || !ro {
return false
if err == nil && ro {
nRO++
} else {
nRW++
}
}
return true
return
}
// setLogInfo records some key indexes about the log.

83
store/store_test.go
Unescape Escape View File

@ -789,7 +789,14 @@ func Test_SingleNodeRequest(t *testing.T) {
},
{
stmts: []string{
`SELECT * FROM foo`,
`INSERT INTO foo(id, name) VALUES(1234, "dana")`,
`INSERT INTO foo(id, name) VALUES(5678, "bob")`,
},
expected: `[{"last_insert_id":1234,"rows_affected":1},{"last_insert_id":5678,"rows_affected":1}]`,
},
{
stmts: []string{
`SELECT * FROM foo WHERE name='fiona'`,
`INSERT INTO foo(id, name) VALUES(66, "declan")`,
},
expected: `[{"columns":["id","name"],"types":["integer","text"],"values":[[1,"fiona"]]},{"last_insert_id":66,"rows_affected":1}]`,
@ -799,7 +806,7 @@ func Test_SingleNodeRequest(t *testing.T) {
`INSERT INTO foo(id, name) VALUES(77, "fiona")`,
`SELECT COUNT(*) FROM foo`,
},
expected: `[{"last_insert_id":77,"rows_affected":1},{"columns":["COUNT(*)"],"types":["integer"],"values":[[3]]}]`,
expected: `[{"last_insert_id":77,"rows_affected":1},{"columns":["COUNT(*)"],"types":["integer"],"values":[[5]]}]`,
},
{
stmts: []string{
@ -2448,7 +2455,7 @@ func Test_IsVoter(t *testing.T) {
}
}
func Test_QueriesOnly(t *testing.T) {
func Test_RWROCount(t *testing.T) {
s, ln := mustNewStore(t)
defer ln.Close()
@ -2471,62 +2478,66 @@ func Test_QueriesOnly(t *testing.T) {
}
tests := []struct {
name string
stmts []string
queriesOnly bool
name string
stmts []string
expRW int
expRO int
}{
{
name: "Empty SQL",
stmts: []string{""},
queriesOnly: true,
name: "Empty SQL",
stmts: []string{""},
},
{
name: "Junk SQL",
stmts: []string{"asdkflj asgkdj"},
queriesOnly: false,
name: "Junk SQL",
stmts: []string{"asdkflj asgkdj"},
expRW: 1,
},
{
name: "CREATE TABLE statement, already exists",
stmts: []string{"CREATE TABLE foo (id INTEGER NOT NULL PRIMARY KEY, name TEXT)"},
queriesOnly: false,
name: "CREATE TABLE statement, already exists",
stmts: []string{"CREATE TABLE foo (id INTEGER NOT NULL PRIMARY KEY, name TEXT)"},
expRW: 1,
},
{
name: "Single INSERT",
stmts: []string{"INSERT INTO foo(id, name) VALUES(1, 'fiona')"},
queriesOnly: false,
name: "Single INSERT",
stmts: []string{"INSERT INTO foo(id, name) VALUES(1, 'fiona')"},
expRW: 1,
},
{
name: "Single INSERT, non-existent table",
stmts: []string{"INSERT INTO qux(id, name) VALUES(1, 'fiona')"},
queriesOnly: false,
name: "Single INSERT, non-existent table",
stmts: []string{"INSERT INTO qux(id, name) VALUES(1, 'fiona')"},
expRW: 1,
},
{
name: "Single SELECT",
stmts: []string{"SELECT * FROM foo"},
queriesOnly: true,
name: "Single SELECT",
stmts: []string{"SELECT * FROM foo"},
expRO: 1,
},
{
name: "Single SELECT from non-existent table",
stmts: []string{"SELECT * FROM qux"},
queriesOnly: false,
name: "Single SELECT from non-existent table",
stmts: []string{"SELECT * FROM qux"},
expRW: 1, // Yeah, this is unfortunate, but it's how SQLite works.
},
{
name: "Double SELECT",
stmts: []string{"SELECT * FROM foo", "SELECT * FROM foo WHERE id = 1"},
queriesOnly: true,
name: "Double SELECT",
stmts: []string{"SELECT * FROM foo", "SELECT * FROM foo WHERE id = 1"},
expRO: 2,
},
{
name: "Mix queries and executes",
stmts: []string{"SELECT * FROM foo", "INSERT INTO foo(id, name) VALUES(1, 'fiona')"},
queriesOnly: false,
name: "Mix queries and executes",
stmts: []string{"SELECT * FROM foo", "INSERT INTO foo(id, name) VALUES(1, 'fiona')"},
expRW: 1,
expRO: 1,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
requires := s.QueriesOnly(executeQueryRequestFromStrings(tt.stmts, proto.QueryRequest_QUERY_REQUEST_LEVEL_NONE, false, false))
if requires != tt.queriesOnly {
t.Fatalf(" test %s failed, unexpected requires: expected %v, got %v", tt.name, tt.queriesOnly, requires)
rwN, roN := s.RORWCount(executeQueryRequestFromStrings(tt.stmts, proto.QueryRequest_QUERY_REQUEST_LEVEL_NONE, false, false))
if rwN != tt.expRW {
t.Fatalf("wrong number of RW statements, exp %d, got %d", tt.expRW, rwN)
}
if roN != tt.expRO {
t.Fatalf("wrong number of RO statements, exp %d, got %d", tt.expRO, roN)
}
})
}

Write Preview
Loading…
Cancel
Save