diff options
Diffstat (limited to 'oblast/query.go')
| -rw-r--r-- | oblast/query.go | 340 |
1 files changed, 340 insertions, 0 deletions
diff --git a/oblast/query.go b/oblast/query.go new file mode 100644 index 0000000..6e375e4 --- /dev/null +++ b/oblast/query.go @@ -0,0 +1,340 @@ +// SPDX-FileCopyrightText: 2026 Stefan Majewsky <majewsky@gmx.net> +// SPDX-License-Identifier: Apache-2.0 + +package oblast + +import ( + "context" + "database/sql" + "fmt" + "reflect" + + "go.xyrillian.de/gg/errext" + "go.xyrillian.de/gg/gsql" +) + +// PrepareThreshold is a tuning parameter for the strategy used by all methods of [Store] operating on batches of records provided by the caller +// (specifically, [Store.Insert], [Store.Update] and [Store.Delete]). +// +// For large amounts of records, it is obviously advantageous to build a prepared statement for the one query that will be used repeatedly on all of them. +// However, building a prepared statement is associated with some amount of bookkeeping on the level of the database/sql library. +// When operating on individual records or small amounts of records at a time (that is, in OLTP rather than OLAP workloads), this overhead becomes a measurable performance burden. +// +// This tuning parameter defines the minimum number of records that will justify maintaining a prepared statement. +// Our benchmarking with the mattn/go-sqlite3 driver (and last checked with Go 1.26.2 on x86_64) indicates that this becomes a worthwhile investment at 8 or more records, so this is our default. +// If your benchmarking indicates a different tradeoff depending on your choice of Go version or SQL driver, you may adjust this variable accordingly. +// +// The actual effect of this setting is to control the value of the "repeated" argument in [Handle.Prepare]. +var PrepareThreshold int = 8 + +// prepare behaves like [Handle.Prepare]. +func prepare(ctx context.Context, db gsql.Handle, readOnly bool, query, operation string, inputSize int) (gsql.Statement, error) { + if query == "" { + if readOnly { + return nil, fmt.Errorf("cannot execute %s() because query planning used the ReadOnly() option", operation) + } + return nil, fmt.Errorf("cannot execute %s() because query could not be autogenerated", operation) + } + + return db.GSQLPrepare(ctx, query, inputSize >= PrepareThreshold) +} + +// Insert executes an SQL INSERT statement for each of the provided records. +// +// Fields that are declared with the "auto" tag will not be written into the DB, +// and instead their value (as auto-generated by the DB on insert) will be placed in the record. +// (This is why this method, as well as [Store.Upsert], need to take their arguments by-pointer instead of by-value). +// +// Returns an error if [NewStore] was called without the [TableNameIs] option, which is required to generate a query for this method. +// +// Returns an error if any of the `records` has a non-zero value in any column marked as `db:",auto"`. +// Records that already exist in the database should be handled with [Store.Update] instead. +// To automatically decide between INSERT and UPDATE on a per-record basis, use [Store.Upsert] instead. +func (s Store[R]) Insert(ctx context.Context, db gsql.Handle, records ...*R) error { + // NOTE: This function body should be as short as possible to reduce the binary size after monomorphization. + // Any expression that does not depend on type R should be factored out into a reusable function. + + stmt, err := prepare(ctx, db, s.plan.ReadOnly, s.plan.Insert.Query, "Insert", len(records)) + if err != nil { + return err + } + return s.insertUsing(ctx, stmt, db, records) +} + +func (s Store[R]) insertUsing(ctx context.Context, stmt gsql.Statement, db gsql.Handle, records []*R) error { + // NOTE: This function body should be as short as possible to reduce the binary size after monomorphization. + // Any expression that does not depend on type R should be factored out into a reusable function. + + var ( + argumentIndexes = s.plan.Insert.ArgumentIndexes + argumentSlots = make([]any, len(argumentIndexes)) + scanIndexes = s.plan.Insert.ScanIndexes + scanSlots = make([]any, len(scanIndexes)) + ) + + for idx, r := range records { + v := reflect.ValueOf(r).Elem() + err := checkTransparentPointerStructFieldsInitialized("INSERT", idx, v, s.plan, false) + if err != nil { + return errext.WithCleanup(err, "Stmt.Close", stmt.Close()) + } + err = insertRecord(ctx, s.plan, v, idx, stmt, argumentIndexes, argumentSlots, scanIndexes, scanSlots) + if err != nil { + return errext.WithCleanup(err, "Stmt.Close", stmt.Close()) + } + } + + return errext.WithCleanup(nil, "Stmt.Close", stmt.Close()) +} + +func insertRecord(ctx context.Context, plan plan, v reflect.Value, recordIndex int, stmt gsql.Statement, argumentIndexes [][]int, argumentSlots []any, scanIndexes [][]int, scanSlots []any) error { + for idx, index := range argumentIndexes { + argumentSlots[idx] = v.FieldByIndex(index).Interface() + } + for idx, index := range scanIndexes { + f := v.FieldByIndex(index) + if !f.IsZero() { + return fmt.Errorf(`refusing to INSERT record with idx = %d that already has non-zero values in its "auto" columns`, recordIndex) + } + scanSlots[idx] = f.Addr().Interface() + } + + var ( + result sql.Result + err error + ) + switch { + case len(scanSlots) == 0: + _, err = stmt.Exec(ctx, argumentSlots) + case plan.InsertUsesQueryRow: + err = stmt.QueryRow(ctx, argumentSlots, scanSlots) + default: + result, err = stmt.Exec(ctx, argumentSlots) + } + if err != nil { + return fmt.Errorf("while inserting record with idx = %d: %w", recordIndex, err) + } + + if result != nil { + id, err := result.LastInsertId() + if err != nil { + return fmt.Errorf("while getting LastInsertId for record with idx = %d: %w", recordIndex, err) + } + if plan.LastInsertIdIsUnsigned { + if id < 0 { + return fmt.Errorf("LastInsertId() = %d for record with idx = %d cannot be converted to uint", id, recordIndex) + } + v.FieldByIndex(scanIndexes[0]).SetUint(uint64(id)) + } else { + v.FieldByIndex(scanIndexes[0]).SetInt(id) + } + } + + return nil +} + +// This check must be performed within all query functions that access existing values using FieldByIndex(), +// to ensure that FieldByIndex() does not panic on indirection through a nil pointer. +func checkTransparentPointerStructFieldsInitialized(operation string, recordIndex int, v reflect.Value, plan plan, onlyPK bool) error { + for _, field := range plan.TransparentPointerStructFields { + f := v.FieldByIndex(field.Index) + if !f.IsZero() { + continue + } + if onlyPK { + if field.ContainsPrimaryKey { + return fmt.Errorf(`refusing to %s record with idx = %d: cannot access all primary key fields because field %q holds a nil pointer`, + operation, recordIndex, field.Name) + } + } else { + return fmt.Errorf(`refusing to %s record with idx = %d: cannot access all mapped fields because field %q holds a nil pointer`, + operation, recordIndex, field.Name) + } + } + return nil +} + +// Update executes an SQL UPDATE statement for each of the provided records, updating all non-primary-key columns with the values in the records. +// Returns [MissingRecordError] if any of the records does not exist in the database, that is, if for any of the records, the database contains no row with the same primary key values. +// +// Returns an error if [NewStore] was called without the [TableNameIs] or [PrimaryKeyIs] options, which are both required to generate a query for this method. +func (s Store[R]) Update(ctx context.Context, db gsql.Handle, records ...R) error { + // NOTE: This function body should be as short as possible to reduce the binary size after monomorphization. + // Any expression that does not depend on type R should be factored out into a reusable function. + + stmt, err := prepare(ctx, db, s.plan.ReadOnly, s.plan.Update.Query, "Update", len(records)) + if err != nil { + return err + } + + var ( + argumentIndexes = s.plan.Update.ArgumentIndexes + argumentSlots = make([]any, len(argumentIndexes)) + ) + + for idx := range records { + v := reflect.ValueOf(&records[idx]).Elem() + err := checkTransparentPointerStructFieldsInitialized("UPDATE", idx, v, s.plan, false) + if err != nil { + return errext.WithCleanup(err, "Stmt.Close", stmt.Close()) + } + rowsAffected, err := updateRecord(ctx, v, idx, stmt, argumentIndexes, argumentSlots) + if err == nil && rowsAffected == 0 { + err = MissingRecordError[R]{records[idx], s.plan} + } + if err != nil { + return errext.WithCleanup(err, "Stmt.Close", stmt.Close()) + } + } + return errext.WithCleanup(nil, "Stmt.Close", stmt.Close()) +} + +func updateRecord(ctx context.Context, v reflect.Value, recordIndex int, stmt gsql.Statement, argumentIndexes [][]int, argumentSlots []any) (int64, error) { + for idx, index := range argumentIndexes { + argumentSlots[idx] = v.FieldByIndex(index).Interface() + } + result, err := stmt.Exec(ctx, argumentSlots) + if err != nil { + return 0, fmt.Errorf("while updating record with idx = %d: %w", recordIndex, err) + } + rowsAffected, err := result.RowsAffected() + if err != nil { + return 0, fmt.Errorf("during RowsAffected() for record with idx = %d: %w", recordIndex, err) + } + return rowsAffected, nil +} + +// Delete executes an SQL DELETE statement for each of the provided records, using their primary keys to locate the respective table rows. +// +// Returns an error if [NewStore] was called without the [TableNameIs] or [PrimaryKeyIs] options, which are both required to generate a query for this method. +func (s Store[R]) Delete(ctx context.Context, db gsql.Handle, records ...R) error { + // NOTE: This function body should be as short as possible to reduce the binary size after monomorphization. + // Any expression that does not depend on type R should be factored out into a reusable function. + + stmt, err := prepare(ctx, db, s.plan.ReadOnly, s.plan.Delete.Query, "Delete", len(records)) + if err != nil { + return err + } + + var ( + argumentIndexes = s.plan.Delete.ArgumentIndexes + argumentSlots = make([]any, len(argumentIndexes)) + ) + + for idx := range records { + v := reflect.ValueOf(&records[idx]).Elem() + err := deleteRecord(ctx, s.plan, v, idx, stmt, argumentIndexes, argumentSlots) + if err != nil { + return errext.WithCleanup(err, "Stmt.Close", stmt.Close()) + } + } + + return errext.WithCleanup(nil, "Stmt.Close", stmt.Close()) +} + +func deleteRecord(ctx context.Context, plan plan, v reflect.Value, recordIndex int, stmt gsql.Statement, argumentIndexes [][]int, argumentSlots []any) error { + err := checkTransparentPointerStructFieldsInitialized("DELETE", recordIndex, v, plan, true) + if err != nil { + return errext.WithCleanup(err, "Stmt.Close", stmt.Close()) + } + for idx, index := range argumentIndexes { + argumentSlots[idx] = v.FieldByIndex(index).Interface() + } + _, err = stmt.Exec(ctx, argumentSlots) + if err != nil { + return fmt.Errorf("while deleting record with idx = %d: %w", recordIndex, err) + } + return nil +} + +// Upsert executes either an SQL INSERT or UPDATE statement for each of the provided records, +// based on whether the record already exists in the DB or not. +// +// - For record types that have fields declared with the "auto" tag, INSERT is chosen if and only if those fields hold zero values. +// Returns an error if only some of the respective fields hold zero values while others don't. +// Returns an error if [NewStore] was called without the [TableNameIs] or [PrimaryKeyIs] options, which are both required to generate the respective queries for this method. +// - For record types that do not have fields declared with the "auto" tag, an INSERT ... ON CONFLICT statement is used. +// Returns an error if [NewStore] was called without the [TableNameIs] option, which is required to generate a query for this method. +func (s Store[R]) Upsert(ctx context.Context, db gsql.Handle, records ...*R) error { + // NOTE: This function body should be as short as possible to reduce the binary size after monomorphization. + // Any expression that does not depend on type R should be factored out into a reusable function. + + if len(s.plan.AutoColumnNames) == 0 { + stmt, err := prepare(ctx, db, s.plan.ReadOnly, s.plan.Upsert.Query, "Upsert", len(records)) + if err != nil { + return err + } + return s.insertUsing(ctx, stmt, db, records) + } + + // TODO: respect PrepareThreshold (or not? may be too much bookkeeping overhead for not a whole lot of benefit) + insertStmt, err := prepare(ctx, db, s.plan.ReadOnly, s.plan.Insert.Query, "Insert", 0) + if err != nil { + return err + } + updateStmt, err := prepare(ctx, db, s.plan.ReadOnly, s.plan.Update.Query, "Update", 0) + if err != nil { + return errext.WithCleanup(err, "InsertStmt.Close", insertStmt.Close()) + } + + err = s.doUpsert(ctx, db, insertStmt, updateStmt, records) + err = errext.WithCleanup(err, "InsertStmt.Close", insertStmt.Close()) + err = errext.WithCleanup(err, "UpdateStmt.Close", updateStmt.Close()) + return err +} + +func (s Store[R]) doUpsert(ctx context.Context, db gsql.Handle, insertStmt, updateStmt gsql.Statement, records []*R) error { + // NOTE: This function body should be as short as possible to reduce the binary size after monomorphization. + // Any expression that does not depend on type R should be factored out into a reusable function. + + var ( + insertArgumentIndexes = s.plan.Insert.ArgumentIndexes + insertArgumentSlots = make([]any, len(insertArgumentIndexes)) + insertScanIndexes = s.plan.Insert.ScanIndexes + insertScanSlots = make([]any, len(insertScanIndexes)) + updateArgumentIndexes = s.plan.Update.ArgumentIndexes + updateArgumentSlots = make([]any, len(updateArgumentIndexes)) + ) + + for idx, r := range records { + v := reflect.ValueOf(r).Elem() + err := checkTransparentPointerStructFieldsInitialized("INSERT or UPDATE", idx, v, s.plan, false) + if err != nil { + return err + } + isInsert, err := upsertDecideStrategy(v, idx, insertScanIndexes) + if err != nil { + return err + } + + if isInsert { + err = insertRecord(ctx, s.plan, v, idx, insertStmt, insertArgumentIndexes, insertArgumentSlots, insertScanIndexes, insertScanSlots) + } else { + var rowsAffected int64 + rowsAffected, err = updateRecord(ctx, v, idx, updateStmt, updateArgumentIndexes, updateArgumentSlots) + if err == nil && rowsAffected == 0 { + err = MissingRecordError[R]{*r, s.plan} + } + } + if err != nil { + return err + } + } + + return nil +} + +func upsertDecideStrategy(v reflect.Value, recordIndex int, scanIndexes [][]int) (isInsert bool, err error) { + var isUpdate bool + for _, index := range scanIndexes { + if v.FieldByIndex(index).IsZero() { + isInsert = true + } else { + isUpdate = true + } + } + if isInsert && isUpdate { + return false, fmt.Errorf(`cannot decide whether to INSERT or UPDATE record with idx = %d: some "auto" columns are zero, others are not`, recordIndex) + } + return isInsert, nil +} |