24841644892

Committed 23 Apr 2026 02:44PM UTC coverage: 85.279% (-4.0%) from 89.306%

Build # 24841644892

Build Type

push

gh-ci

Committed by

web-flow

Commit Message

feat: v1.11.0 PostgreSQL compatibility and SQL feature expansion (#2090)

* Add structured audit logging with immutable audit trail

Introduces a new --audit-log flag that records all gRPC operations as
structured JSON events in immudb's tamper-proof KV store. Events are
stored under the audit: key prefix in systemdb, queryable via Scan and
verifiable via VerifiableGet. An async buffered writer ensures minimal
latency impact. Configurable event filtering (all/write/admin) via
--audit-log-events flag.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Add PostgreSQL ORM compatibility layer and verification functions

Extend the pgsql wire protocol with immudb verification functions
(immudb_state, immudb_verify_row, immudb_verify_tx, immudb_history,
immudb_tx) accessible via standard SQL SELECT statements.

Add pg_catalog resolvers (pg_attribute, pg_index, pg_constraint,
pg_type, pg_settings, pg_description) and information_schema
resolvers (tables, columns, schemata, key_column_usage) to support
ORM introspection from Django, SQLAlchemy, GORM, and ActiveRecord.

Add PostgreSQL compatibility functions: current_database,
current_schema, current_user, format_type, pg_encoding_to_char,
pg_get_expr, pg_get_constraintdef, obj_description, col_description,
has_table_privilege, has_schema_privilege, and others.

Add SHOW statement emulation for common ORM config queries and
schema-qualified name stripping for information_schema and public
schema references.

* Implement EXISTS and IN subquery support in SQL engine

Replace the previously stubbed ExistsBoolExp and InSubQueryExp
implementations with working non-correlated subquery execution.

EXISTS subqueries resolve the inner SELECT and check if any rows
are returned. IN subqueries resolve the inner SELECT, iterate the
result set, and compare each value against the outer expression.
Both support NOT variants (NOT EXISTS, NOT IN).

Correlated subqueries (referencing outer query columns) ar... (continued)

Coverage Stats

7254 of 10471 new or added lines in 124 files covered. (69.28%)

115 existing lines in 18 files now uncovered.

44599 of 52298 relevant lines covered (85.28%)

127676.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.27

/embedded/sql/engine.go

/*
Copyright 2026 Codenotary Inc. All rights reserved.

SPDX-License-Identifier: BUSL-1.1
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    https://mariadb.com/bsl11/

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package sql

import (
        "context"
        "encoding/binary"
        "errors"
        "fmt"
        "strings"
        "sync"
        "sync/atomic"

        "github.com/codenotary/immudb/embedded/store"
)

var (
        ErrNoSupported                            = errors.New("not supported")
        ErrIllegalArguments                       = store.ErrIllegalArguments
        ErrMultiIndexingNotEnabled                = fmt.Errorf("%w: multi-indexing must be enabled", store.ErrIllegalState)
        ErrParsingError                           = errors.New("parsing error")
        ErrDDLorDMLTxOnly                         = errors.New("transactions can NOT combine DDL and DML statements")
        ErrUnspecifiedMultiDBHandler              = fmt.Errorf("%w: unspecified multidbHanlder", store.ErrIllegalState)
        ErrDatabaseDoesNotExist                   = errors.New("database does not exist")
        ErrDatabaseAlreadyExists                  = errors.New("database already exists")
        ErrTableAlreadyExists                     = errors.New("table already exists")
        ErrTableDoesNotExist                      = errors.New("table does not exist")
        ErrColumnDoesNotExist                     = errors.New("column does not exist")
        ErrColumnAlreadyExists                    = errors.New("column already exists")
        ErrCannotDropColumn                       = errors.New("cannot drop column")
        ErrSameOldAndNewNames                     = errors.New("same old and new names")
        ErrColumnNotIndexed                       = errors.New("column is not indexed")
        ErrFunctionDoesNotExist                   = errors.New("function does not exist")
        ErrLimitedKeyType                         = errors.New("indexed key of unsupported type or exceeded length")
        ErrLimitedAutoIncrement                   = errors.New("only INTEGER single-column primary keys can be set as auto incremental")
        ErrLimitedMaxLen                          = errors.New("only VARCHAR and BLOB types support max length")
        ErrDuplicatedColumn                       = errors.New("duplicated column")
        ErrInvalidColumn                          = errors.New("invalid column")
        ErrInvalidCheckConstraint                 = errors.New("invalid check constraint")
        ErrCheckConstraintViolation               = errors.New("check constraint violation")
        ErrReservedWord                           = errors.New("reserved word")
        ErrNoPrimaryKey                           = errors.New("no primary key specified")
        ErrPKCanNotBeNull                         = errors.New("primary key can not be null")
        ErrPKCanNotBeUpdated                      = errors.New("primary key can not be updated")
        ErrMultiplePrimaryKeys                    = errors.New("multiple primary keys are not allowed")
        ErrNotNullableColumnCannotBeNull          = errors.New("not nullable column can not be null")
        ErrNewColumnMustBeNullable                = errors.New("new column must be nullable")
        ErrIndexAlreadyExists                     = errors.New("index already exists")
        ErrMaxNumberOfColumnsInIndexExceeded      = errors.New("number of columns in multi-column index exceeded")
        ErrIndexNotFound                          = errors.New("index not found")
        ErrConstraintNotFound                     = errors.New("constraint not found")
        ErrInvalidNumberOfValues                  = errors.New("invalid number of values provided")
        ErrInvalidValue                           = errors.New("invalid value provided")
        ErrInferredMultipleTypes                  = errors.New("inferred multiple types")
        ErrExpectingDQLStmt                       = errors.New("illegal statement. DQL statement expected")
        ErrColumnMustAppearInGroupByOrAggregation = errors.New("must appear in the group by clause or be used in an aggregated function")
        ErrIllegalMappedKey                       = errors.New("error illegal mapped key")
        ErrCorruptedData                          = store.ErrCorruptedData
        ErrBrokenCatalogColSpecExpirable          = fmt.Errorf("%w: catalog column entry set as expirable", ErrCorruptedData)
        ErrBrokenCatalogCheckConstraintExpirable  = fmt.Errorf("%w: catalog check constraint set as expirable", ErrCorruptedData)
        ErrNoMoreRows                             = store.ErrNoMoreEntries
        ErrInvalidTypes                           = errors.New("invalid types")
        ErrUnsupportedJoinType                    = errors.New("unsupported join type")
        ErrInvalidCondition                       = errors.New("invalid condition")
        ErrHavingClauseRequiresGroupClause        = errors.New("having clause requires group clause")
        ErrNotComparableValues                    = errors.New("values are not comparable")
        ErrNumericTypeExpected                    = errors.New("numeric type expected")
        ErrUnexpected                             = errors.New("unexpected error")
        ErrMaxKeyLengthExceeded                   = errors.New("max key length exceeded")
        ErrMaxLengthExceeded                      = errors.New("max length exceeded")
        ErrColumnIsNotAnAggregation               = errors.New("column is not an aggregation")
        ErrLimitedCount                           = errors.New("only unbounded counting is supported i.e. COUNT(*)")
        ErrWindowRowsLimitExceeded                = errors.New("window function result exceeds max_window_rows limit")
        ErrTxDoesNotExist                         = errors.New("tx does not exist")
        ErrNestedTxNotSupported                   = errors.New("nested tx are not supported")
        ErrNoOngoingTx                            = errors.New("no ongoing transaction")
        ErrNonTransactionalStmt                   = errors.New("non transactional statement")
        ErrDivisionByZero                         = errors.New("division by zero")
        ErrMissingParameter                       = errors.New("missing parameter")
        ErrUnsupportedParameter                   = errors.New("unsupported parameter")
        ErrDuplicatedParameters                   = errors.New("duplicated parameters")
        ErrLimitedIndexCreation                   = errors.New("unique index creation is only supported on empty tables")
        ErrTooManyRows                            = errors.New("too many rows")
        ErrAlreadyClosed                          = store.ErrAlreadyClosed
        ErrAmbiguousSelector                      = errors.New("ambiguous selector")
        ErrUnsupportedCast                        = fmt.Errorf("%w: unsupported cast", ErrInvalidValue)
        ErrColumnMismatchInUnionStmt              = errors.New("column mismatch in union statement")
        ErrCannotIndexJson                        = errors.New("cannot index column of type JSON")
        ErrInvalidTxMetadata                      = errors.New("invalid transaction metadata")
        ErrAccessDenied                           = errors.New("access denied")
        ErrDiffRequiresPeriod                     = errors.New("DIFF requires both SINCE/AFTER and UNTIL/BEFORE clauses")
)

// MaxKeyLen caps the length of variable-width indexed columns (the
// real on-disk constraint is in the store/btree layer, currently 1024
// bytes — see embedded/store/immustore.go MaxKeyLen and the btree
// DefaultMaxKeySize). Engine code historically used 512 bytes, which
// is well under the storage cap; raising it to 1024 unblocks dumps
// that declare longer text PKs (e.g. netflix's `show_id text`) without
// any on-disk format change. The variable stays mutable so that
// embedded/sql.Options can override it at engine init time.
var MaxKeyLen = 1024

// CatalogCacheHitObserver and CatalogCacheMissObserver are package-level
// hooks the upper layers (pkg/server/metrics.go) populate to translate
// catalog cache events into Prometheus counters. Kept as func() vars to
// avoid pulling a Prometheus dependency into embedded/sql, which is a
// pure storage primitive shared across all higher layers.
//
// Default no-op implementations make these safe to call from NewTx
// regardless of whether the server has wired observers in yet.
var (
        CatalogCacheHitObserver  = func() {}
        CatalogCacheMissObserver = func() {}
)

const (
        EncIDLen  = 4
        EncLenLen = 4
)

const MaxNumberOfColumnsInIndex = 8

type Engine struct {
        store *store.ImmuStore

        prefix                        []byte
        distinctLimit                 int
        distinctSpillThreshold        int
        sortBufferSize                int
        maxWindowRows                 int // max rows materialized for window functions (0 = unlimited)
        autocommit                    bool
        lazyIndexConstraintValidation bool
        parseTxMetadata               func([]byte) (map[string]interface{}, error)
        multidbHandler                MultiDBHandler
        tableResolvers                map[string]TableResolver
        sequences                     map[string]*Sequence

        // catalogMu guards cachedCatalog. The cached catalog is built once from the
        // store and reused across read-only transactions to avoid redundant B-tree
        // scans on every autocommit SELECT. It is invalidated whenever a DDL
        // statement commits (mutatedCatalog == true).
        catalogMu     sync.RWMutex
        cachedCatalog *Catalog

        // cachedCatalogVersion is bumped on every invalidateCatalogCache call.
        // Each SQLTx records this counter in openCatalogVersion at NewTx time;
        // on commit, tryPopulateCatalogCache only accepts a populate when the
        // version still matches — i.e. no DDL has invalidated the cache since
        // this tx opened. Without this guard, a non-DDL RW tx whose snapshot
        // pre-dates a concurrent DDL commit could overwrite the cache with a
        // stale schema view (the doc's "stale-view race").
        cachedCatalogVersion atomic.Uint64
}

// Sequence represents a named auto-incrementing counter
type Sequence struct {
        name      string
        currValue int64
        increment int64
        started   bool
}

func (e *Engine) CreateSequence(name string, startValue, increment int64) {
        if e.sequences == nil {
                e.sequences = make(map[string]*Sequence)
        }
        e.sequences[name] = &Sequence{
                name:      name,
                currValue: startValue - increment, // first NEXTVAL will return startValue
                increment: increment,
        }
}

func (e *Engine) DropSequence(name string) bool {
        if e.sequences == nil {
                return false
        }
        _, exists := e.sequences[name]
        if exists {
                delete(e.sequences, name)
        }
        return exists
}

func (e *Engine) NextVal(name string) (int64, error) {
        if e.sequences == nil {
                return 0, fmt.Errorf("sequence does not exist (%s)", name)
        }
        seq, exists := e.sequences[name]
        if !exists {
                return 0, fmt.Errorf("sequence does not exist (%s)", name)
        }
        seq.currValue += seq.increment
        seq.started = true
        return seq.currValue, nil
}

func (e *Engine) CurrVal(name string) (int64, error) {
        if e.sequences == nil {
                return 0, fmt.Errorf("sequence does not exist (%s)", name)
        }
        seq, exists := e.sequences[name]
        if !exists {
                return 0, fmt.Errorf("sequence does not exist (%s)", name)
        }
        if !seq.started {
                return 0, fmt.Errorf("sequence '%s' has not been used yet", name)
        }
        return seq.currValue, nil
}

func (e *Engine) loadViews(ctx context.Context, tx *store.OngoingTx) {
        prefix := MapKey(e.prefix, catalogViewPrefix, EncodeID(DatabaseID))

        _ = iteratePrefix(ctx, tx, prefix, func(key, value []byte, deleted bool) error {
                if deleted || len(value) == 0 {
                        return nil
                }

                // Extract view name from key
                nameStart := len(prefix)
                if nameStart >= len(key) {
                        return nil
                }
                viewName := string(key[nameStart:])
                sqlText := string(value)

                // Parse the view query and register it
                stmts, err := ParseSQL(strings.NewReader(sqlText))
                if err != nil {
                        return nil // skip unparseable views
                }

                for _, stmt := range stmts {
                        if ds, ok := stmt.(DataSource); ok {
                                e.registerTableResolver(viewName, &viewResolver{
                                        name:  viewName,
                                        query: ds,
                                })
                                break
                        }
                }

                return nil
        })
}

func (e *Engine) loadSequences(ctx context.Context, tx *store.OngoingTx) error {
        prefix := MapKey(e.prefix, catalogSequencePrefix, EncodeID(DatabaseID))

        return iteratePrefix(ctx, tx, prefix, func(key, value []byte, deleted bool) error {
                if deleted || len(value) < 16 {
                        return nil
                }

                // Extract name from key: prefix + DatabaseID + name
                nameStart := len(prefix)
                if nameStart >= len(key) {
                        return nil
                }
                name := string(key[nameStart:])

                currValue := int64(binary.BigEndian.Uint64(value[0:8]))
                increment := int64(binary.BigEndian.Uint64(value[8:16]))

                if e.sequences == nil {
                        e.sequences = make(map[string]*Sequence)
                }

                e.sequences[name] = &Sequence{
                        name:      name,
                        currValue: currValue,
                        increment: increment,
                        started:   currValue != 0,
                }

                return nil
        })
}

type MultiDBHandler interface {
        ListDatabases(ctx context.Context) ([]string, error)
        CreateDatabase(ctx context.Context, db string, ifNotExists bool) error
        UseDatabase(ctx context.Context, db string) error
        GetLoggedUser(ctx context.Context) (User, error)
        ListUsers(ctx context.Context) ([]User, error)
        CreateUser(ctx context.Context, username, password string, permission Permission) error
        AlterUser(ctx context.Context, username, password string, permission Permission) error
        GrantSQLPrivileges(ctx context.Context, database, username string, privileges []SQLPrivilege) error
        RevokeSQLPrivileges(ctx context.Context, database, username string, privileges []SQLPrivilege) error
        DropUser(ctx context.Context, username string) error
        ExecPreparedStmts(ctx context.Context, opts *TxOptions, stmts []SQLStmt, params map[string]interface{}) (ntx *SQLTx, committedTxs []*SQLTx, err error)
}

type TableResolver interface {
        Table() string
        Resolve(ctx context.Context, tx *SQLTx, alias string) (RowReader, error)
}

type User interface {
        Username() string
        Permission() Permission
        SQLPrivileges() []SQLPrivilege
}

func NewEngine(st *store.ImmuStore, opts *Options) (*Engine, error) {
        if st == nil {
                return nil, ErrIllegalArguments
        }

        if !st.MultiIndexingEnabled() {
                return nil, ErrMultiIndexingNotEnabled
        }

        err := opts.Validate()
        if err != nil {
                return nil, err
        }

        // Apply the engine-side variable-key cap override, if requested.
        // This mutates a package-level variable that other call sites
        // (EncodeRawValueAsKey, the document layer, etc.) consult — the
        // global is intentional, matching the existing MaxKeyLen design.
        // Zero leaves the package default in place.
        if opts.maxKeyLen != 0 {
                MaxKeyLen = opts.maxKeyLen
        }

        e := &Engine{
                store:                         st,
                prefix:                        make([]byte, len(opts.prefix)),
                distinctLimit:                 opts.distinctLimit,
                distinctSpillThreshold:        opts.distinctSpillThreshold,
                sortBufferSize:                opts.sortBufferSize,
                maxWindowRows:                 opts.maxWindowRows,
                autocommit:                    opts.autocommit,
                lazyIndexConstraintValidation: opts.lazyIndexConstraintValidation,
                parseTxMetadata:               opts.parseTxMetadata,
                multidbHandler:                opts.multidbHandler,
        }

        copy(e.prefix, opts.prefix)

        err = st.InitIndexing(&store.IndexSpec{
                SourcePrefix:     append(e.prefix, []byte(catalogPrefix)...),
                TargetPrefix:     append(e.prefix, []byte(catalogPrefix)...),
                InjectiveMapping: true,
        })
        if err != nil && !errors.Is(err, store.ErrIndexAlreadyInitialized) {
                return nil, err
        }

        for _, r := range opts.tableResolvers {
                e.registerTableResolver(r.Table(), r)
        }

        // TODO: find a better way to handle parsing errors
        yyErrorVerbose = true

        return e, nil
}

func (e *Engine) NewTx(ctx context.Context, opts *TxOptions) (*SQLTx, error) {
        err := opts.Validate()
        if err != nil {
                return nil, err
        }

        var mode store.TxMode
        if opts.ReadOnly {
                mode = store.ReadOnlyTx
        } else {
                mode = store.ReadWriteTx
        }

        txOpts := &store.TxOptions{
                Mode:                    mode,
                SnapshotMustIncludeTxID: opts.SnapshotMustIncludeTxID,
                SnapshotRenewalPeriod:   opts.SnapshotRenewalPeriod,
                UnsafeMVCC:              opts.UnsafeMVCC,
        }

        tx, err := e.store.NewTx(ctx, txOpts)
        if err != nil {
                return nil, err
        }

        if len(opts.Extra) > 0 {
                txmd := store.NewTxMetadata()
                err := txmd.WithExtra(opts.Extra)
                if err != nil {
                        return nil, err
                }

                tx.WithMetadata(txmd)
        }

        // Try to reuse the engine-level catalog cache.
        // catalog.load() scans all table/column/index metadata from the B-tree on
        // every transaction open — skipping it eliminates the dominant per-query
        // overhead for autocommit SELECT workloads, and (via Clone) for DML-heavy
        // read-write workloads as well.
        //
        // openCatalogVersion is captured once here under the same RLock as
        // cachedCatalog so the (catalog, version) pair seen by this tx is
        // coherent. tryPopulateCatalogCache rejects populates whose version
        // disagrees with what subsequently lives on the engine.
        e.catalogMu.RLock()
        cached := e.cachedCatalog
        openVersion := e.cachedCatalogVersion.Load()
        e.catalogMu.RUnlock()

        if cached != nil && opts.ReadOnly {
                // Read-only transactions cannot mutate the schema, so they share the
                // cached catalog directly. Sequences and views are already loaded in
                // engine fields.
                CatalogCacheHitObserver()
                return &SQLTx{
                        engine:             e,
                        opts:               opts,
                        tx:                 tx,
                        catalog:            cached,
                        openCatalogVersion: openVersion,
                        lastInsertedPKs:    make(map[string]int64),
                        firstInsertedPKs:   make(map[string]int64),
                }, nil
        }

        var catalog *Catalog
        if cached != nil {
                // Read-write tx with a warm cache: clone so any DDL this tx performs
                // lands on the local copy, not the shared cache. Reset maxPK on the
                // clones — the authoritative value comes from the snapshot-dependent
                // loadMaxPK loop below and must not leak the cache's stale value into
                // an empty-table case (where loadMaxPK returns ErrNoMoreEntries and
                // the existing code falls through without resetting).
                CatalogCacheHitObserver()
                catalog = cached.Clone()
                for _, table := range catalog.GetTables() {
                        table.maxPK = 0
                }
                // D2 Phase 2 correctness: catalog.load was skipped, so the MVCC
                // read-set has no record of the catalog rows this tx implicitly
                // observed. Without this seed, two concurrent txs doing
                // conflicting DDL would both commit (no read-conflict tripwire).
                // See immudb-improvements.md "Open question #2".
                if err := e.seedCatalogReadSet(ctx, tx, catalog.GetTables()); err != nil {
                        return nil, err
                }
        } else {
                CatalogCacheMissObserver()
                catalog = newCatalog(e.prefix)
                if err := catalog.load(ctx, tx); err != nil {
                        return nil, err
                }
        }

        // Load persisted sequences (only once, on first tx)
        if e.sequences == nil {
                if err := e.loadSequences(ctx, tx); err != nil {
                        // Non-fatal: sequences are optional
                        _ = err
                }
        }

        // Load persisted views (only once, on first tx open)
        if len(e.tableResolvers) == 0 {
                e.loadViews(ctx, tx)
        }

        for _, table := range catalog.GetTables() {
                primaryIndex := table.primaryIndex

                rowEntryPrefix := MapKey(
                        e.prefix,
                        RowPrefix,
                        EncodeID(DatabaseID),
                        EncodeID(table.id),
                        EncodeID(primaryIndex.id),
                )

                mappedPKEntryPrefix := MapKey(
                        e.prefix,
                        MappedPrefix,
                        EncodeID(table.id),
                        EncodeID(primaryIndex.id),
                )

                err = e.store.InitIndexing(&store.IndexSpec{
                        SourcePrefix: rowEntryPrefix,

                        TargetEntryMapper: indexEntryMapperFor(primaryIndex, primaryIndex),
                        TargetPrefix:      mappedPKEntryPrefix,

                        InjectiveMapping: true,
                })
                if err != nil && !errors.Is(err, store.ErrIndexAlreadyInitialized) {
                        return nil, err
                }

                for _, index := range table.indexes {
                        if index.IsPrimary() {
                                continue
                        }

                        mappedEntryPrefix := MapKey(
                                e.prefix,
                                MappedPrefix,
                                EncodeID(table.id),
                                EncodeID(index.id),
                        )

                        err = e.store.InitIndexing(&store.IndexSpec{
                                SourcePrefix:      rowEntryPrefix,
                                SourceEntryMapper: indexEntryMapperFor(primaryIndex, primaryIndex),
                                TargetEntryMapper: indexEntryMapperFor(index, primaryIndex),
                                TargetPrefix:      mappedEntryPrefix,

                                InjectiveMapping: true,
                        })
                        if errors.Is(err, store.ErrIndexAlreadyInitialized) {
                                continue
                        }
                        if err != nil {
                                return nil, err
                        }
                }

                if table.autoIncrementPK {
                        encMaxPK, err := loadMaxPK(ctx, e.prefix, tx, table)
                        if errors.Is(err, store.ErrNoMoreEntries) {
                                continue
                        }
                        if err != nil {
                                return nil, err
                        }

                        if len(encMaxPK) != 9 {
                                return nil, ErrCorruptedData
                        }

                        if encMaxPK[0] != KeyValPrefixNotNull {
                                return nil, ErrCorruptedData
                        }

                        // map to signed integer space
                        encMaxPK[1] ^= 0x80

                        table.maxPK = int64(binary.BigEndian.Uint64(encMaxPK[1:]))
                }
        }

        // Populate the read-only catalog cache so subsequent autocommit SELECTs
        // bypass catalog.load() entirely. Only cache when the caller did not
        // mutate the schema (write transactions are never cached).
        if opts.ReadOnly {
                e.catalogMu.Lock()
                if e.cachedCatalog == nil {
                        e.cachedCatalog = catalog
                }
                e.catalogMu.Unlock()
        }

        return &SQLTx{
                engine:             e,
                opts:               opts,
                tx:                 tx,
                catalog:            catalog,
                openCatalogVersion: openVersion,
                lastInsertedPKs:    make(map[string]int64),
                firstInsertedPKs:   make(map[string]int64),
        }, nil
}

// seedCatalogReadSet registers prefix-range fingerprints on the
// OngoingTx read-set for every catalog prefix that catalog.load
// would have walked. Used by NewTx's Clone fast path (D2 Phase 2)
// so DDL-vs-DDL conflict detection still fires even when the cached
// catalog skips the parse step.
//
// Uses OngoingTx.MarkPrefixScanned (R1 from the agent review) which
// records ONE fingerprint per prefix instead of per-entry
// expectedReads + expectedNoMoreEntries — same conflict guarantee,
// without the O(N) bookkeeping tax on every warm-cache NewTx.
func (e *Engine) seedCatalogReadSet(ctx context.Context, tx *store.OngoingTx, tables []*Table) error {
        prefixes := [][]byte{
                MapKey(e.prefix, catalogTablePrefix, EncodeID(DatabaseID)),
        }
        for _, t := range tables {
                prefixes = append(prefixes,
                        MapKey(e.prefix, catalogColumnPrefix, EncodeID(DatabaseID), EncodeID(t.id)),
                        MapKey(e.prefix, catalogIndexPrefix, EncodeID(DatabaseID), EncodeID(t.id)),
                        MapKey(e.prefix, catalogCheckPrefix, EncodeID(DatabaseID), EncodeID(t.id)),
                )
        }
        for _, p := range prefixes {
                if err := tx.MarkPrefixScanned(ctx, store.KeyReaderSpec{Prefix: p}); err != nil {
                        return err
                }
        }
        return nil
}

func drainKeyReader(ctx context.Context, tx *store.OngoingTx, prefix []byte) error {
        rdr, err := tx.NewKeyReader(store.KeyReaderSpec{Prefix: prefix})
        if err != nil {
                return err
        }
        defer rdr.Close()

        for {
                _, _, err := rdr.Read(ctx)
                if errors.Is(err, store.ErrNoMoreEntries) {
                        return nil
                }
                if err != nil {
                        return err
                }
        }
}

// invalidateCatalogCache clears the engine-level catalog cache. Called after
// any DDL transaction commits so the next read-only tx reloads from the store.
//
// The version counter is bumped under the same lock so a concurrent
// tryPopulateCatalogCache call sees a coherent (catalog, version) pair —
// either the old cached catalog with version V, or nil with version V+1.
// Never the inconsistent intermediate state where the catalog is nil but
// the version is still V.
func (e *Engine) invalidateCatalogCache() {
        e.catalogMu.Lock()
        e.cachedCatalog = nil
        e.cachedCatalogVersion.Add(1)
        e.catalogMu.Unlock()
}

// tryPopulateCatalogCache opportunistically promotes a freshly-committed
// (non-DDL) tx's catalog into the engine cache so subsequent NewTx calls
// can take the Clone fast path. No-op when:
//   - The cache is already warm (first-fill-wins, preserves coherence
//     even if the existing entry came from a different tx).
//   - cachedCatalogVersion has changed since this tx opened — meaning a
//     concurrent DDL invalidated the cache, and our snapshot is by now
//     stale relative to the next valid view.
//
// The catalog reference handed in is shared post-publication: callers
// MUST NOT mutate the catalog after handing it over.
func (e *Engine) tryPopulateCatalogCache(catalog *Catalog, openVersion uint64) {
        if catalog == nil {
                return
        }
        e.catalogMu.Lock()
        defer e.catalogMu.Unlock()

        if e.cachedCatalog != nil {
                return
        }
        if e.cachedCatalogVersion.Load() != openVersion {
                return
        }
        e.cachedCatalog = catalog
}

func indexEntryMapperFor(index, primaryIndex *Index) store.EntryMapper {
        // value={count (colID valLen val)+})
        // key=M.{tableID}{indexID}({null}({val}{padding}{valLen})?)+({pkVal}{padding}{pkValLen})+

        valueExtractor := func(value []byte, valuesByColID map[uint32]TypedValue) error {
                voff := 0

                cols := int(binary.BigEndian.Uint32(value[voff:]))
                voff += EncLenLen

                for i := 0; i < cols; i++ {
                        if len(value) < EncIDLen {
                                return fmt.Errorf("key is lower than required")
                        }

                        colID := binary.BigEndian.Uint32(value[voff:])
                        voff += EncIDLen

                        col, err := index.table.GetColumnByID(colID)
                        if errors.Is(err, ErrColumnDoesNotExist) {
                                vlen := int(binary.BigEndian.Uint32(value[voff:]))
                                voff += EncLenLen + vlen
                                continue
                        } else if err != nil {
                                return err
                        }

                        val, n, err := DecodeValue(value[voff:], col.colType)
                        if err != nil {
                                return err
                        }

                        voff += n

                        valuesByColID[colID] = val
                }

                return nil
        }

        return func(key, value []byte) ([]byte, error) {
                encodedValues := make([][]byte, 2+len(index.cols)+1)
                encodedValues[0] = EncodeID(index.table.id)
                encodedValues[1] = EncodeID(index.id)

                valuesByColID := make(map[uint32]TypedValue, len(index.cols))

                for _, col := range index.table.cols {
                        valuesByColID[col.id] = &NullValue{t: col.colType}
                }

                err := valueExtractor(value, valuesByColID)
                if err != nil {
                        return nil, err
                }

                for i, col := range index.cols {
                        encKey, _, err := EncodeValueAsKey(valuesByColID[col.id], col.Type(), col.MaxLen())
                        if err != nil {
                                return nil, err
                        }

                        encodedValues[2+i] = encKey
                }

                pkEncVals, err := encodedKey(primaryIndex, valuesByColID)
                if err != nil {
                        return nil, err
                }

                encodedValues[len(encodedValues)-1] = pkEncVals

                return MapKey(index.enginePrefix(), MappedPrefix, encodedValues...), nil
        }
}

func (e *Engine) Exec(ctx context.Context, tx *SQLTx, sql string, params map[string]interface{}) (ntx *SQLTx, committedTxs []*SQLTx, err error) {
        stmts, err := ParseSQL(strings.NewReader(sql))
        if err != nil {
                return nil, nil, fmt.Errorf("%w: %v", ErrParsingError, err)
        }

        return e.ExecPreparedStmts(ctx, tx, stmts, params)
}

func (e *Engine) ExecPreparedStmts(ctx context.Context, tx *SQLTx, stmts []SQLStmt, params map[string]interface{}) (ntx *SQLTx, committedTxs []*SQLTx, err error) {
        ntx, ctxs, pendingStmts, err := e.execPreparedStmts(ctx, tx, stmts, params)
        if err != nil {
                return ntx, ctxs, err
        }

        if len(pendingStmts) > 0 {
                // a different database was selected

                if e.multidbHandler == nil || ntx != nil {
                        return ntx, ctxs, fmt.Errorf("%w: all statements should have been executed when not using a multidbHandler", ErrUnexpected)
                }

                var opts *TxOptions

                if tx != nil {
                        opts = tx.opts
                } else {
                        opts = DefaultTxOptions()
                }

                ntx, hctxs, err := e.multidbHandler.ExecPreparedStmts(ctx, opts, pendingStmts, params)

                return ntx, append(ctxs, hctxs...), err
        }

        return ntx, ctxs, nil
}

func (e *Engine) execPreparedStmts(ctx context.Context, tx *SQLTx, stmts []SQLStmt, params map[string]interface{}) (ntx *SQLTx, committedTxs []*SQLTx, pendingStmts []SQLStmt, err error) {
        if len(stmts) == 0 {
                return nil, nil, stmts, ErrIllegalArguments
        }

        nparams, err := normalizeParams(params)
        if err != nil {
                return nil, nil, stmts, err
        }

        currTx := tx

        execStmts := 0

        for _, stmt := range stmts {
                if stmt == nil {
                        return nil, nil, stmts[execStmts:], ErrIllegalArguments
                }

                _, isDBSelectionStmt := stmt.(*UseDatabaseStmt)

                // handle the case when working in non-autocommit mode outside a transaction block
                if isDBSelectionStmt && (currTx != nil && !currTx.Closed()) && !currTx.IsExplicitCloseRequired() {
                        err = currTx.Commit(ctx)
                        if err == nil {
                                committedTxs = append(committedTxs, currTx)
                        }
                        if err != nil {
                                return nil, committedTxs, stmts[execStmts:], err
                        }
                }

                if currTx == nil || currTx.Closed() {
                        var opts *TxOptions

                        if currTx != nil {
                                opts = currTx.opts
                        } else if tx != nil {
                                opts = tx.opts
                        } else {
                                opts = DefaultTxOptions()
                        }

                        // begin tx with implicit commit
                        currTx, err = e.NewTx(ctx, opts)
                        if err != nil {
                                return nil, committedTxs, stmts[execStmts:], err
                        }
                }

                if e.multidbHandler != nil {
                        if err := e.checkUserPermissions(ctx, stmt); err != nil {
                                currTx.Cancel()
                                return nil, committedTxs, stmts[execStmts:], err
                        }
                }

                ntx, err := stmt.execAt(ctx, currTx, nparams)
                if err != nil {
                        currTx.Cancel()
                        return nil, committedTxs, stmts[execStmts:], err
                }

                if !currTx.Closed() && !currTx.IsExplicitCloseRequired() && e.autocommit {
                        err = currTx.Commit(ctx)
                        if err != nil {
                                return nil, committedTxs, stmts[execStmts:], err
                        }
                }

                if currTx.Closed() {
                        committedTxs = append(committedTxs, currTx)
                }

                currTx = ntx

                execStmts++

                if isDBSelectionStmt && e.multidbHandler != nil {
                        break
                }
        }

        if currTx != nil && !currTx.Closed() && !currTx.IsExplicitCloseRequired() {
                err = currTx.Commit(ctx)
                if err != nil {
                        return nil, committedTxs, stmts[execStmts:], err
                }

                committedTxs = append(committedTxs, currTx)
        }

        if currTx != nil && currTx.Closed() {
                currTx = nil
        }

        return currTx, committedTxs, stmts[execStmts:], nil
}

func (e *Engine) checkUserPermissions(ctx context.Context, stmt SQLStmt) error {
        user, err := e.multidbHandler.GetLoggedUser(ctx)
        if err != nil {
                return err
        }

        if !stmt.readOnly() && user.Permission() == PermissionReadOnly {
                return fmt.Errorf("%w: statement requires %s permission", ErrAccessDenied, PermissionReadWrite)
        }

        requiredPrivileges := stmt.requiredPrivileges()
        if !hasAllPrivileges(user.SQLPrivileges(), requiredPrivileges) {
                return fmt.Errorf("%w: statement requires %v privileges", ErrAccessDenied, requiredPrivileges)
        }
        return nil
}

func hasAllPrivileges(userPrivileges, privileges []SQLPrivilege) bool {
        for _, p := range privileges {
                has := false
                for _, up := range userPrivileges {
                        if up == p {
                                has = true
                                break
                        }
                }

                if !has {
                        return false
                }
        }
        return true
}

func (e *Engine) queryAll(ctx context.Context, tx *SQLTx, sql string, params map[string]interface{}) ([]*Row, error) {
        reader, err := e.Query(ctx, tx, sql, params)
        if err != nil {
                return nil, err
        }
        defer reader.Close()

        return ReadAllRows(ctx, reader)
}

func (e *Engine) Query(ctx context.Context, tx *SQLTx, sql string, params map[string]interface{}) (RowReader, error) {
        stmts, err := ParseSQL(strings.NewReader(sql))
        if err != nil {
                return nil, fmt.Errorf("%w: %v", ErrParsingError, err)
        }
        if len(stmts) != 1 {
                return nil, ErrExpectingDQLStmt
        }

        stmt, ok := stmts[0].(DataSource)
        if !ok {
                return nil, ErrExpectingDQLStmt
        }

        return e.QueryPreparedStmt(ctx, tx, stmt, params)
}

func (e *Engine) QueryPreparedStmt(ctx context.Context, tx *SQLTx, stmt DataSource, params map[string]interface{}) (rowReader RowReader, err error) {
        if stmt == nil {
                return nil, ErrIllegalArguments
        }

        qtx := tx

        if qtx == nil {
                readOnly := stmt.readOnly()
                qtx, err = e.NewTx(ctx, DefaultTxOptions().WithReadOnly(readOnly))
                if err != nil {
                        return nil, err
                }
                defer func() {
                        if err != nil {
                                qtx.Cancel()
                        }
                }()
        }

        nparams, err := normalizeParams(params)
        if err != nil {
                return nil, err
        }

        if e.multidbHandler != nil {
                if err := e.checkUserPermissions(ctx, stmt); err != nil {
                        return nil, err
                }
        }

        _, err = stmt.execAt(ctx, qtx, nparams)
        if err != nil {
                return nil, err
        }

        r, err := stmt.Resolve(ctx, qtx, nparams, nil)
        if err != nil {
                return nil, err
        }

        if tx == nil {
                if stmt.readOnly() {
                        r.onClose(func() {
                                qtx.Cancel()
                        })
                } else {
                        // DML with RETURNING: commit the transaction on close
                        r.onClose(func() {
                                qtx.Commit(ctx)
                        })
                }
        }

        return r, nil
}

func (e *Engine) Catalog(ctx context.Context, tx *SQLTx) (catalog *Catalog, err error) {
        qtx := tx

        if qtx == nil {
                qtx, err = e.NewTx(ctx, DefaultTxOptions().WithReadOnly(true))
                if err != nil {
                        return nil, err
                }
                defer qtx.Cancel()
        }

        return qtx.Catalog(), nil
}

func (e *Engine) InferParameters(ctx context.Context, tx *SQLTx, sql string) (params map[string]SQLValueType, err error) {
        stmts, err := ParseSQL(strings.NewReader(sql))
        if err != nil {
                return nil, fmt.Errorf("%w: %v", ErrParsingError, err)
        }
        return e.InferParametersPreparedStmts(ctx, tx, stmts)
}

func (e *Engine) InferParametersPreparedStmts(ctx context.Context, tx *SQLTx, stmts []SQLStmt) (params map[string]SQLValueType, err error) {
        if len(stmts) == 0 {
                return nil, ErrIllegalArguments
        }

        qtx := tx

        if qtx == nil {
                qtx, err = e.NewTx(ctx, DefaultTxOptions().WithReadOnly(true))
                if err != nil {
                        return nil, err
                }
                defer qtx.Cancel()
        }

        params = make(map[string]SQLValueType)

        for _, stmt := range stmts {
                err = stmt.inferParameters(ctx, qtx, params)
                if err != nil {
                        return nil, err
                }
        }

        return params, nil
}

func normalizeParams(params map[string]interface{}) (map[string]interface{}, error) {
        nparams := make(map[string]interface{}, len(params))

        for name, value := range params {
                nname := strings.ToLower(name)

                _, exists := nparams[nname]
                if exists {
                        return nil, ErrDuplicatedParameters
                }

                nparams[nname] = value
        }

        return nparams, nil
}

// CopyCatalogToTx copies the current sql catalog to the ongoing transaction.
func (e *Engine) CopyCatalogToTx(ctx context.Context, tx *store.OngoingTx) error {
        if tx == nil {
                return ErrIllegalArguments
        }

        catalog := newCatalog(e.prefix)

        err := catalog.addSchemaToTx(ctx, tx)
        if err != nil {
                return err
        }

        return nil
}

func (e *Engine) GetStore() *store.ImmuStore {
        return e.store
}

func (e *Engine) GetPrefix() []byte {
        return e.prefix
}

func (e *Engine) tableResolveFor(tableName string) TableResolver {
        if e.tableResolvers == nil {
                return nil
        }
        return e.tableResolvers[tableName]
}

func (e *Engine) registerTableResolver(tableName string, r TableResolver) {
        if e.tableResolvers == nil {
                e.tableResolvers = make(map[string]TableResolver)
        }
        e.tableResolvers[tableName] = r
}

codenotary / immudb / 24841644892

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous