use SqlStruct when query includes pagination

Signed-off-by: Hayim Shaul <[email protected]>

Author: Hayim Shaul

platform/common/utils/collections/iterators.go

@@ -57,19 +57,23 @@ func ReadFirst[T any](it Iterator[*T], limit int) ([]T, error) {
 }
 
 func ReadLast[T any](it Iterator[*T]) (*T, *sliceIterator[*T], error) {
-	var items []*T
-	// create a copy of it so not to ruin it
-	for item, err := it.Next(); item != nil || err != nil; item, err = it.Next() {
-		if err != nil {
-			return nil, nil, err
-		}
-		items = append(items, item)
+	var items []T
+	var ptrItems []*T
+	var err error
+	items, err = ReadAll(it)
+
+	if err != nil {
+		return nil, nil, err
+	}
+	// convert from []T to []*T
+	for i := range items {
+		ptrItems = append(ptrItems, &items[i])
 	}
-	si := NewSliceIterator(items)
+	si := NewSliceIterator(ptrItems)
 	if len(items) == 0 {
 		return nil, si, nil
 	}
-	return items[len(items)-1], si, nil
+	return ptrItems[len(ptrItems)-1], si, nil
 }
 
 func ReadAll[T any](it Iterator[*T]) ([]T, error) {

platform/view/services/db/driver/sql/common/pagination.go

@@ -154,27 +154,35 @@ func NewPaginationInterpreter() *paginationInterpreter {
 }
 
 type PaginationInterpreter interface {
-	Interpret(p driver.Pagination) (string, error)
+	Interpret(p driver.Pagination, sql SqlQuery) (SqlQuery, error)
 }
 
 type paginationInterpreter struct{}
 
-func (i *paginationInterpreter) Interpret(p driver.Pagination) (string, error) {
+func (i *paginationInterpreter) Interpret(p driver.Pagination, sql SqlQuery) (SqlQuery, error) {
 	switch pagination := p.(type) {
 	case *NoPagination:
-		return "", nil
+		return sql, nil
 	case *OffsetPagination:
-		return fmt.Sprintf("LIMIT %d OFFSET %d", pagination.pageSize, pagination.offset), nil
+		sql.limit = fmt.Sprintf("%d", pagination.pageSize)
+		sql.offset = fmt.Sprintf("%d", pagination.offset)
+		return sql, nil
 	case *KeysetPagination:
-		// TODO: add OrderBy?
+		sql.order = fmt.Sprintf("%s ASC", pagination.sqlIdName)
+		sql.limit = fmt.Sprintf("%d", pagination.pageSize)
 		if (pagination.lastOffset != -1) && (pagination.offset == pagination.lastOffset+pagination.pageSize) {
-			return fmt.Sprintf("WHERE %s>'%s' ORDER BY %s ASC LIMIT %d", pagination.sqlIdName, pagination.lastId, pagination.sqlIdName, pagination.pageSize), nil
+			lastId := sql.AddParam(fmt.Sprintf("%d", pagination.lastId))
+			sql.where = append(sql.where, fmt.Sprintf("%s>'$%d'", pagination.sqlIdName, lastId))
+		} else {
+			sql.offset = fmt.Sprintf("%d", pagination.offset)
 		}
-		return fmt.Sprintf("ORDER BY %s ASC LIMIT %d OFFSET %d", pagination.sqlIdName, pagination.pageSize, pagination.offset), nil
+		return sql, nil
 	case *EmptyPagination:
-		return "LIMIT 0 OFFSET 0", nil
+		sql.limit = "0"
+		sql.offset = "0"
+		return sql, nil
 	default:
-		return "", errors.Errorf("invalid pagination option %+v", pagination)
+		return sql, errors.Errorf("invalid pagination option %+v", pagination)
 	}
 }
 

platform/view/services/db/driver/sql/common/vault.go

@@ -13,6 +13,7 @@ import (
 	"encoding/gob"
 	errors2 "errors"
 	"fmt"
+	"strings"
 	"sync"
 
 	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
@@ -62,6 +63,60 @@ type VaultPersistence struct {
 	il           IsolationLevelMapper
 }
 
+type SqlQuery struct {
+	table  string
+	fields []string
+	where  []string
+	limit  string
+	offset string
+	order  string
+	params []any
+}
+
+func (q *SqlQuery) AddWhere(p string) {
+	q.where = append(q.where, p)
+}
+
+func (q *SqlQuery) AddParam(p any) int {
+	q.params = append(q.params, p)
+	return len(q.params) - 1
+}
+
+func (q *SqlQuery) FormatQuery() string {
+	if q.table == "" {
+		return ""
+	}
+
+	// SELECT fields
+	fields := "*"
+	if len(q.fields) > 0 {
+		fields = strings.Join(q.fields, ", ")
+	}
+	query := fmt.Sprintf("SELECT %s FROM %s", fields, q.table)
+
+	// WHERE clause
+	if len(q.where) > 0 {
+		query += " WHERE " + strings.Join(q.where, " AND ")
+	}
+
+	// ORDER BY
+	if q.order != "" {
+		query += " ORDER BY " + q.order
+	}
+
+	// LIMIT
+	if q.limit != "" {
+		query += " LIMIT " + q.limit
+	}
+
+	// OFFSET
+	if q.offset != "" {
+		query += " OFFSET " + q.offset
+	}
+
+	return query
+}
+
 func (db *VaultPersistence) NewTxLockVaultReader(ctx context.Context, txID driver.TxID, isolationLevel driver.IsolationLevel) (driver.LockedVaultReader, error) {
 	logger.Debugf("Acquire tx id lock for [%s]", txID)
 	span := trace.SpanFromContext(ctx)
@@ -439,11 +494,16 @@ func (db *vaultReader) GetLast(ctx context.Context) (*driver.TxStatus, error) {
 	span.AddEvent("start_get_last")
 	defer span.AddEvent("end_get_last")
 	it, err := db.queryStatus(fmt.Sprintf("WHERE pos=(SELECT max(pos) FROM %s WHERE code!=$1)", db.tables.StatusTable), []any{driver.Busy}, "")
+	// sql := SqlQuery{}
+	// driverBusy := sql.AddParam(driver.Busy)
+	// sql.AddWhere(fmt.Sprintf("WHERE pos=(SELECT max(pos) FROM %s WHERE code!=$%d)", db.tables.StatusTable, driverBusy))
+	// it, err := db.queryStatus(sql)
 	if err != nil {
 		return nil, err
 	}
 	return collections.GetUnique(it)
 }
+
 func (db *vaultReader) GetTxStatus(ctx context.Context, txID driver.TxID) (*driver.TxStatus, error) {
 	span := trace.SpanFromContext(ctx)
 	span.AddEvent("start_get_tx_status")
@@ -469,11 +529,12 @@ func (db *vaultReader) GetAllTxStatuses(ctx context.Context, pagination driver.P
 	if pagination == nil {
 		return nil, fmt.Errorf("invalid input pagination: %+v", pagination)
 	}
-	limit, err := db.pi.Interpret(pagination)
+	sql := SqlQuery{}
+	sql, err := db.pi.Interpret(pagination, sql)
 	if err != nil {
 		return nil, err
 	}
-	txStatusIterator, err := db.queryStatus("", []any{}, limit)
+	txStatusIterator, err := db.queryStatusWithPagination(sql)
 	if err != nil {
 		return nil, err
 	}
@@ -494,6 +555,18 @@ func (db *vaultReader) queryStatus(where string, params []any, limit string) (dr
 	return &TxCodeIterator{rows: rows}, nil
 }
 
+func (db *vaultReader) queryStatusWithPagination(sql SqlQuery) (driver.TxStatusIterator, error) {
+	sql.table = db.tables.StatusTable
+	query := sql.FormatQuery()
+	logger.Debug(query, sql.params)
+
+	rows, err := db.readDB.Query(query, sql.params...)
+	if err != nil {
+		return nil, err
+	}
+	return &TxCodeIterator{rows: rows}, nil
+}
+
 func (db *VaultPersistence) Close() error {
 	return errors2.Join(db.writeDB.Close(), db.readDB.Close())
 }

platform/view/services/storage/vault/vaultstore_test.go

@@ -8,6 +8,7 @@ package vault
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/hyperledger-labs/fabric-smart-client/platform/common/driver"
@@ -167,9 +168,11 @@ func testPagination(t *testing.T, store driver.VaultStore) {
 		pagination := item.pagination
 		page := 0
 		for ; true; page++ {
-			sql, err := interpreter.Interpret((pagination))
+			sql := common.SqlQuery{}
+			sql, err := interpreter.Interpret(pagination, sql)
 			Expect(err).ToNot(HaveOccurred())
-			Expect(sql).To(Equal(item.sqlForward[page]))
+			fmt.Printf("sql (forward) = %s\n", sql.FormatQuery())
+			Expect(sql.FormatQuery()).To(Equal(item.sqlForward[page]))
 			statuses, err := getAllTxStatuses(store, pagination)
 			Expect(err).ToNot(HaveOccurred())
 			Expect(err).ToNot(HaveOccurred())
@@ -192,9 +195,11 @@ func testPagination(t *testing.T, store driver.VaultStore) {
 			Expect(err).ToNot(HaveOccurred())
 		}
 		for page := len(item.matcher) - 1; page >= 0; page-- {
-			sql, err := interpreter.Interpret((pagination))
+			sql := common.SqlQuery{}
+			sql, err := interpreter.Interpret(pagination, sql)
 			Expect(err).ToNot(HaveOccurred())
-			Expect(sql).To(Equal(item.sqlBackward[page]))
+			fmt.Printf("sql (backward) = %s\n", sql.FormatQuery())
+			Expect(sql.FormatQuery()).To(Equal(item.sqlBackward[page]))
 			statuses, err := getAllTxStatuses(store, pagination)
 			Expect(err).ToNot(HaveOccurred())
 			Expect(statuses).To(item.matcher[page])