12273047947

Committed 11 Dec 2024 09:06AM UTC coverage: 89.286% (+0.02%) from 89.266%

Build # 12273047947

Build Type

Pull #2036

gh-ci

Committed by

ostafen

Commit Message

chore(embedded/sql): Add support for core pg_catalog tables (pg_class, pg_namespace, pg_roles)

Signed-off-by: Stefano Scafiti <stefano.scafiti96@gmail.com>

Pull Request Pull Request #2036: chore(embedded/sql): Add support for core pg_catalog tables (pg_class…

Run Details

160 of 184 new or added lines in 12 files covered. (86.96%)

1 existing line in 1 file now uncovered.

37641 of 42158 relevant lines covered (89.29%)

150814.99 hits per line

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

85.56

/embedded/sql/stmt.go

/*
Copyright 2024 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 (
        "bytes"
        "context"
        "encoding/binary"
        "encoding/hex"
        "errors"
        "fmt"
        "math"
        "regexp"
        "strconv"
        "strings"
        "time"

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

const (
        catalogPrefix          = "CTL."
        catalogTablePrefix     = "CTL.TABLE."     // (key=CTL.TABLE.{1}{tableID}, value={tableNAME})
        catalogColumnPrefix    = "CTL.COLUMN."    // (key=CTL.COLUMN.{1}{tableID}{colID}{colTYPE}, value={(auto_incremental | nullable){maxLen}{colNAME}})
        catalogIndexPrefix     = "CTL.INDEX."     // (key=CTL.INDEX.{1}{tableID}{indexID}, value={unique {colID1}(ASC|DESC)...{colIDN}(ASC|DESC)})
        catalogCheckPrefix     = "CTL.CHECK."     // (key=CTL.CHECK.{1}{tableID}{checkID}, value={nameLen}{name}{expText})
        catalogPrivilegePrefix = "CTL.PRIVILEGE." // (key=CTL.COLUMN.{1}{tableID}{colID}{colTYPE}, value={(auto_incremental | nullable){maxLen}{colNAME}})

        RowPrefix    = "R." // (key=R.{1}{tableID}{0}({null}({pkVal}{padding}{pkValLen})?)+, value={count (colID valLen val)+})
        MappedPrefix = "M." // (key=M.{tableID}{indexID}({null}({val}{padding}{valLen})?)*({pkVal}{padding}{pkValLen})+, value={count (colID valLen val)+})
)

const (
        DatabaseID = uint32(1) // deprecated but left to maintain backwards compatibility
        PKIndexID  = uint32(0)
)

const (
        nullableFlag      byte = 1 << iota
        autoIncrementFlag byte = 1 << iota
)

const (
        revCol        = "_rev"
        txMetadataCol = "_tx_metadata"
)

var reservedColumns = map[string]struct{}{
        revCol:        {},
        txMetadataCol: {},
}

func isReservedCol(col string) bool {
        _, ok := reservedColumns[col]
        return ok
}

type SQLValueType = string

const (
        IntegerType   SQLValueType = "INTEGER"
        BooleanType   SQLValueType = "BOOLEAN"
        VarcharType   SQLValueType = "VARCHAR"
        UUIDType      SQLValueType = "UUID"
        BLOBType      SQLValueType = "BLOB"
        Float64Type   SQLValueType = "FLOAT"
        TimestampType SQLValueType = "TIMESTAMP"
        AnyType       SQLValueType = "ANY"
        JSONType      SQLValueType = "JSON"
)

func IsNumericType(t SQLValueType) bool {
        return t == IntegerType || t == Float64Type
}

type Permission = string

const (
        PermissionReadOnly  Permission = "READ"
        PermissionReadWrite Permission = "READWRITE"
        PermissionAdmin     Permission = "ADMIN"
        PermissionSysAdmin  Permission = "SYSADMIN"
)

func PermissionFromCode(code uint32) Permission {
        switch code {
        case 1:
                {
                        return PermissionReadOnly
                }
        case 2:
                {
                        return PermissionReadWrite
                }
        case 254:
                {
                        return PermissionAdmin
                }
        }
        return PermissionSysAdmin
}

type AggregateFn = string

const (
        COUNT AggregateFn = "COUNT"
        SUM   AggregateFn = "SUM"
        MAX   AggregateFn = "MAX"
        MIN   AggregateFn = "MIN"
        AVG   AggregateFn = "AVG"
)

type CmpOperator = int

const (
        EQ CmpOperator = iota
        NE
        LT
        LE
        GT
        GE
)

func CmpOperatorToString(op CmpOperator) string {
        switch op {
        case EQ:
                return "="
        case NE:
                return "!="
        case LT:
                return "<"
        case LE:
                return "<="
        case GT:
                return ">"
        case GE:
                return ">="
        }
        return ""
}

type LogicOperator = int

const (
        And LogicOperator = iota
        Or
)

func LogicOperatorToString(op LogicOperator) string {
        if op == And {
                return "AND"
        }
        return "OR"
}

type NumOperator = int

const (
        ADDOP NumOperator = iota
        SUBSOP
        DIVOP
        MULTOP
        MODOP
)

func NumOperatorString(op NumOperator) string {
        switch op {
        case ADDOP:
                return "+"
        case SUBSOP:
                return "-"
        case DIVOP:
                return "/"
        case MULTOP:
                return "*"
        case MODOP:
                return "%"
        }
        return ""
}

type JoinType = int

const (
        InnerJoin JoinType = iota
        LeftJoin
        RightJoin
)

type SQLStmt interface {
        readOnly() bool
        requiredPrivileges() []SQLPrivilege
        execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error)
        inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error
}

type BeginTransactionStmt struct {
}

func (stmt *BeginTransactionStmt) readOnly() bool {
        return true
}

func (stmt *BeginTransactionStmt) requiredPrivileges() []SQLPrivilege {
        return nil
}

func (stmt *BeginTransactionStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *BeginTransactionStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if tx.IsExplicitCloseRequired() {
                return nil, ErrNestedTxNotSupported
        }

        err := tx.RequireExplicitClose()
        if err == nil {
                // current tx can be reused as no changes were already made
                return tx, nil
        }

        // commit current transaction and start a fresh one

        err = tx.Commit(ctx)
        if err != nil {
                return nil, err
        }

        return tx.engine.NewTx(ctx, tx.opts.WithExplicitClose(true))
}

type CommitStmt struct {
}

func (stmt *CommitStmt) readOnly() bool {
        return true
}

func (stmt *CommitStmt) requiredPrivileges() []SQLPrivilege {
        return nil
}

func (stmt *CommitStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *CommitStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if !tx.IsExplicitCloseRequired() {
                return nil, ErrNoOngoingTx
        }

        return nil, tx.Commit(ctx)
}

type RollbackStmt struct {
}

func (stmt *RollbackStmt) readOnly() bool {
        return true
}

func (stmt *RollbackStmt) requiredPrivileges() []SQLPrivilege {
        return nil
}

func (stmt *RollbackStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *RollbackStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if !tx.IsExplicitCloseRequired() {
                return nil, ErrNoOngoingTx
        }

        return nil, tx.Cancel()
}

type CreateDatabaseStmt struct {
        DB          string
        ifNotExists bool
}

func (stmt *CreateDatabaseStmt) readOnly() bool {
        return false
}

func (stmt *CreateDatabaseStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeCreate}
}

func (stmt *CreateDatabaseStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *CreateDatabaseStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if tx.IsExplicitCloseRequired() {
                return nil, fmt.Errorf("%w: database creation can not be done within a transaction", ErrNonTransactionalStmt)
        }

        if tx.engine.multidbHandler == nil {
                return nil, ErrUnspecifiedMultiDBHandler
        }

        return nil, tx.engine.multidbHandler.CreateDatabase(ctx, stmt.DB, stmt.ifNotExists)
}

type UseDatabaseStmt struct {
        DB string
}

func (stmt *UseDatabaseStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *UseDatabaseStmt) readOnly() bool {
        return true
}

func (stmt *UseDatabaseStmt) requiredPrivileges() []SQLPrivilege {
        return nil
}

func (stmt *UseDatabaseStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if tx.IsExplicitCloseRequired() {
                return nil, fmt.Errorf("%w: database selection can NOT be executed within a transaction block", ErrNonTransactionalStmt)
        }

        if tx.engine.multidbHandler == nil {
                return nil, ErrUnspecifiedMultiDBHandler
        }

        return tx, tx.engine.multidbHandler.UseDatabase(ctx, stmt.DB)
}

type UseSnapshotStmt struct {
        period period
}

func (stmt *UseSnapshotStmt) readOnly() bool {
        return true
}

func (stmt *UseSnapshotStmt) requiredPrivileges() []SQLPrivilege {
        return nil
}

func (stmt *UseSnapshotStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *UseSnapshotStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        return nil, ErrNoSupported
}

type CreateUserStmt struct {
        username   string
        password   string
        permission Permission
}

func (stmt *CreateUserStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *CreateUserStmt) readOnly() bool {
        return false
}

func (stmt *CreateUserStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeCreate}
}

func (stmt *CreateUserStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if tx.IsExplicitCloseRequired() {
                return nil, fmt.Errorf("%w: user creation can not be done within a transaction", ErrNonTransactionalStmt)
        }

        if tx.engine.multidbHandler == nil {
                return nil, ErrUnspecifiedMultiDBHandler
        }

        return nil, tx.engine.multidbHandler.CreateUser(ctx, stmt.username, stmt.password, stmt.permission)
}

type AlterUserStmt struct {
        username   string
        password   string
        permission Permission
}

func (stmt *AlterUserStmt) readOnly() bool {
        return false
}

func (stmt *AlterUserStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeAlter}
}

func (stmt *AlterUserStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *AlterUserStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if tx.IsExplicitCloseRequired() {
                return nil, fmt.Errorf("%w: user modification can not be done within a transaction", ErrNonTransactionalStmt)
        }

        if tx.engine.multidbHandler == nil {
                return nil, ErrUnspecifiedMultiDBHandler
        }

        return nil, tx.engine.multidbHandler.AlterUser(ctx, stmt.username, stmt.password, stmt.permission)
}

type DropUserStmt struct {
        username string
}

func (stmt *DropUserStmt) readOnly() bool {
        return false
}

func (stmt *DropUserStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeDrop}
}

func (stmt *DropUserStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *DropUserStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if tx.IsExplicitCloseRequired() {
                return nil, fmt.Errorf("%w: user deletion can not be done within a transaction", ErrNonTransactionalStmt)
        }

        if tx.engine.multidbHandler == nil {
                return nil, ErrUnspecifiedMultiDBHandler
        }

        return nil, tx.engine.multidbHandler.DropUser(ctx, stmt.username)
}

type CreateTableStmt struct {
        table       string
        ifNotExists bool
        colsSpec    []*ColSpec
        checks      []CheckConstraint
        pkColNames  []string
}

func NewCreateTableStmt(table string, ifNotExists bool, colsSpec []*ColSpec, pkColNames []string) *CreateTableStmt {
        return &CreateTableStmt{table: table, ifNotExists: ifNotExists, colsSpec: colsSpec, pkColNames: pkColNames}
}

func (stmt *CreateTableStmt) readOnly() bool {
        return false
}

func (stmt *CreateTableStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeCreate}
}

func (stmt *CreateTableStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func zeroRow(tableName string, cols []*ColSpec) *Row {
        r := Row{
                ValuesByPosition: make([]TypedValue, len(cols)),
                ValuesBySelector: make(map[string]TypedValue, len(cols)),
        }

        for i, col := range cols {
                v := zeroForType(col.colType)

                r.ValuesByPosition[i] = v
                r.ValuesBySelector[EncodeSelector("", tableName, col.colName)] = v
        }
        return &r
}

func (stmt *CreateTableStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if stmt.ifNotExists && tx.catalog.ExistTable(stmt.table) {
                return tx, nil
        }

        colSpecs := make(map[uint32]*ColSpec, len(stmt.colsSpec))
        for i, cs := range stmt.colsSpec {
                colSpecs[uint32(i)+1] = cs
        }

        row := zeroRow(stmt.table, stmt.colsSpec)
        for _, check := range stmt.checks {
                value, err := check.exp.reduce(tx, row, stmt.table)
                if err != nil {
                        return nil, err
                }

                if value.Type() != BooleanType {
                        return nil, ErrInvalidCheckConstraint
                }
        }

        nextUnnamedCheck := 0
        checks := make(map[string]CheckConstraint)
        for id, check := range stmt.checks {
                name := fmt.Sprintf("%s_check%d", stmt.table, nextUnnamedCheck+1)
                if check.name != "" {
                        name = check.name
                } else {
                        nextUnnamedCheck++
                }
                check.id = uint32(id)
                check.name = name
                checks[name] = check
        }

        table, err := tx.catalog.newTable(stmt.table, colSpecs, checks, uint32(len(colSpecs)))
        if err != nil {
                return nil, err
        }

        createIndexStmt := &CreateIndexStmt{unique: true, table: table.name, cols: stmt.pkColNames}
        _, err = createIndexStmt.execAt(ctx, tx, params)
        if err != nil {
                return nil, err
        }

        for _, col := range table.cols {
                if col.autoIncrement {
                        if len(table.primaryIndex.cols) > 1 || col.id != table.primaryIndex.cols[0].id {
                                return nil, ErrLimitedAutoIncrement
                        }
                }

                err := persistColumn(tx, col)
                if err != nil {
                        return nil, err
                }
        }

        for _, check := range checks {
                if err := persistCheck(tx, table, &check); err != nil {
                        return nil, err
                }
        }

        mappedKey := MapKey(tx.sqlPrefix(), catalogTablePrefix, EncodeID(DatabaseID), EncodeID(table.id))

        err = tx.set(mappedKey, nil, []byte(table.name))
        if err != nil {
                return nil, err
        }

        tx.mutatedCatalog = true

        return tx, nil
}

func persistColumn(tx *SQLTx, col *Column) error {
        //{auto_incremental | nullable}{maxLen}{colNAME})
        v := make([]byte, 1+4+len(col.colName))

        if col.autoIncrement {
                v[0] = v[0] | autoIncrementFlag
        }

        if col.notNull {
                v[0] = v[0] | nullableFlag
        }

        binary.BigEndian.PutUint32(v[1:], uint32(col.MaxLen()))

        copy(v[5:], []byte(col.Name()))

        mappedKey := MapKey(
                tx.sqlPrefix(),
                catalogColumnPrefix,
                EncodeID(DatabaseID),
                EncodeID(col.table.id),
                EncodeID(col.id),
                []byte(col.colType),
        )

        return tx.set(mappedKey, nil, v)
}

func persistCheck(tx *SQLTx, table *Table, check *CheckConstraint) error {
        mappedKey := MapKey(
                tx.sqlPrefix(),
                catalogCheckPrefix,
                EncodeID(DatabaseID),
                EncodeID(table.id),
                EncodeID(check.id),
        )

        name := check.name
        expText := check.exp.String()

        val := make([]byte, 2+len(name)+len(expText))

        if len(name) > 256 {
                return fmt.Errorf("constraint name len: %w", ErrMaxLengthExceeded)
        }

        val[0] = byte(len(name)) - 1

        copy(val[1:], []byte(name))
        copy(val[1+len(name):], []byte(expText))

        return tx.set(mappedKey, nil, val)
}

type ColSpec struct {
        colName       string
        colType       SQLValueType
        maxLen        int
        autoIncrement bool
        notNull       bool
}

func NewColSpec(name string, colType SQLValueType, maxLen int, autoIncrement bool, notNull bool) *ColSpec {
        return &ColSpec{
                colName:       name,
                colType:       colType,
                maxLen:        maxLen,
                autoIncrement: autoIncrement,
                notNull:       notNull,
        }
}

type CreateIndexStmt struct {
        unique      bool
        ifNotExists bool
        table       string
        cols        []string
}

func NewCreateIndexStmt(table string, cols []string, isUnique bool) *CreateIndexStmt {
        return &CreateIndexStmt{unique: isUnique, table: table, cols: cols}
}

func (stmt *CreateIndexStmt) readOnly() bool {
        return false
}

func (stmt *CreateIndexStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeCreate}
}

func (stmt *CreateIndexStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *CreateIndexStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if len(stmt.cols) < 1 {
                return nil, ErrIllegalArguments
        }

        if len(stmt.cols) > MaxNumberOfColumnsInIndex {
                return nil, ErrMaxNumberOfColumnsInIndexExceeded
        }

        table, err := tx.catalog.GetTableByName(stmt.table)
        if err != nil {
                return nil, err
        }

        colIDs := make([]uint32, len(stmt.cols))

        indexKeyLen := 0

        for i, colName := range stmt.cols {
                col, err := table.GetColumnByName(colName)
                if err != nil {
                        return nil, err
                }

                if col.Type() == JSONType {
                        return nil, ErrCannotIndexJson
                }

                if variableSizedType(col.colType) && !tx.engine.lazyIndexConstraintValidation && (col.MaxLen() == 0 || col.MaxLen() > MaxKeyLen) {
                        return nil, fmt.Errorf("%w: can not create index using column '%s'. Max key length for variable columns is %d", ErrLimitedKeyType, col.colName, MaxKeyLen)
                }

                indexKeyLen += col.MaxLen()

                colIDs[i] = col.id
        }

        if !tx.engine.lazyIndexConstraintValidation && indexKeyLen > MaxKeyLen {
                return nil, fmt.Errorf("%w: can not create index using columns '%v'. Max key length is %d", ErrLimitedKeyType, stmt.cols, MaxKeyLen)
        }

        if stmt.unique && table.primaryIndex != nil {
                // check table is empty
                pkPrefix := MapKey(tx.sqlPrefix(), MappedPrefix, EncodeID(table.id), EncodeID(table.primaryIndex.id))
                _, _, err := tx.getWithPrefix(ctx, pkPrefix, nil)
                if errors.Is(err, store.ErrIndexNotFound) {
                        return nil, ErrTableDoesNotExist
                }
                if err == nil {
                        return nil, ErrLimitedIndexCreation
                } else if !errors.Is(err, store.ErrKeyNotFound) {
                        return nil, err
                }
        }

        index, err := table.newIndex(stmt.unique, colIDs)
        if errors.Is(err, ErrIndexAlreadyExists) && stmt.ifNotExists {
                return tx, nil
        }
        if err != nil {
                return nil, err
        }

        // v={unique {colID1}(ASC|DESC)...{colIDN}(ASC|DESC)}
        // TODO: currently only ASC order is supported
        colSpecLen := EncIDLen + 1

        encodedValues := make([]byte, 1+len(index.cols)*colSpecLen)

        if index.IsUnique() {
                encodedValues[0] = 1
        }

        for i, col := range index.cols {
                copy(encodedValues[1+i*colSpecLen:], EncodeID(col.id))
        }

        mappedKey := MapKey(tx.sqlPrefix(), catalogIndexPrefix, EncodeID(DatabaseID), EncodeID(table.id), EncodeID(index.id))

        err = tx.set(mappedKey, nil, encodedValues)
        if err != nil {
                return nil, err
        }

        tx.mutatedCatalog = true

        return tx, nil
}

type AddColumnStmt struct {
        table   string
        colSpec *ColSpec
}

func NewAddColumnStmt(table string, colSpec *ColSpec) *AddColumnStmt {
        return &AddColumnStmt{table: table, colSpec: colSpec}
}

func (stmt *AddColumnStmt) readOnly() bool {
        return false
}

func (stmt *AddColumnStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeAlter}
}

func (stmt *AddColumnStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *AddColumnStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        table, err := tx.catalog.GetTableByName(stmt.table)
        if err != nil {
                return nil, err
        }

        col, err := table.newColumn(stmt.colSpec)
        if err != nil {
                return nil, err
        }

        err = persistColumn(tx, col)
        if err != nil {
                return nil, err
        }

        tx.mutatedCatalog = true

        return tx, nil
}

type RenameTableStmt struct {
        oldName string
        newName string
}

func (stmt *RenameTableStmt) readOnly() bool {
        return false
}

func (stmt *RenameTableStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeAlter}
}

func (stmt *RenameTableStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *RenameTableStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        table, err := tx.catalog.renameTable(stmt.oldName, stmt.newName)
        if err != nil {
                return nil, err
        }

        // update table name
        mappedKey := MapKey(
                tx.sqlPrefix(),
                catalogTablePrefix,
                EncodeID(DatabaseID),
                EncodeID(table.id),
        )
        err = tx.set(mappedKey, nil, []byte(stmt.newName))
        if err != nil {
                return nil, err
        }

        tx.mutatedCatalog = true

        return tx, nil
}

type RenameColumnStmt struct {
        table   string
        oldName string
        newName string
}

func NewRenameColumnStmt(table, oldName, newName string) *RenameColumnStmt {
        return &RenameColumnStmt{table: table, oldName: oldName, newName: newName}
}

func (stmt *RenameColumnStmt) readOnly() bool {
        return false
}

func (stmt *RenameColumnStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeAlter}
}

func (stmt *RenameColumnStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *RenameColumnStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        table, err := tx.catalog.GetTableByName(stmt.table)
        if err != nil {
                return nil, err
        }

        col, err := table.renameColumn(stmt.oldName, stmt.newName)
        if err != nil {
                return nil, err
        }

        err = persistColumn(tx, col)
        if err != nil {
                return nil, err
        }

        tx.mutatedCatalog = true

        return tx, nil
}

type DropColumnStmt struct {
        table   string
        colName string
}

func NewDropColumnStmt(table, colName string) *DropColumnStmt {
        return &DropColumnStmt{table: table, colName: colName}
}

func (stmt *DropColumnStmt) readOnly() bool {
        return false
}

func (stmt *DropColumnStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeDrop}
}

func (stmt *DropColumnStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *DropColumnStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        table, err := tx.catalog.GetTableByName(stmt.table)
        if err != nil {
                return nil, err
        }

        col, err := table.GetColumnByName(stmt.colName)
        if err != nil {
                return nil, err
        }

        err = canDropColumn(tx, table, col)
        if err != nil {
                return nil, err
        }

        err = table.deleteColumn(col)
        if err != nil {
                return nil, err
        }

        err = persistColumnDeletion(ctx, tx, col)
        if err != nil {
                return nil, err
        }

        tx.mutatedCatalog = true

        return tx, nil
}

func canDropColumn(tx *SQLTx, table *Table, col *Column) error {
        colSpecs := make([]*ColSpec, 0, len(table.Cols())-1)
        for _, c := range table.cols {
                if c.id != col.id {
                        colSpecs = append(colSpecs, &ColSpec{colName: c.Name(), colType: c.Type()})
                }
        }

        row := zeroRow(table.Name(), colSpecs)
        for name, check := range table.checkConstraints {
                _, err := check.exp.reduce(tx, row, table.name)
                if errors.Is(err, ErrColumnDoesNotExist) {
                        return fmt.Errorf("%w %s because %s constraint requires it", ErrCannotDropColumn, col.Name(), name)
                }

                if err != nil {
                        return err
                }
        }
        return nil
}

func persistColumnDeletion(ctx context.Context, tx *SQLTx, col *Column) error {
        mappedKey := MapKey(
                tx.sqlPrefix(),
                catalogColumnPrefix,
                EncodeID(DatabaseID),
                EncodeID(col.table.id),
                EncodeID(col.id),
                []byte(col.colType),
        )

        return tx.delete(ctx, mappedKey)
}

type DropConstraintStmt struct {
        table          string
        constraintName string
}

func (stmt *DropConstraintStmt) readOnly() bool {
        return false
}

func (stmt *DropConstraintStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeDrop}
}

func (stmt *DropConstraintStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        table, err := tx.catalog.GetTableByName(stmt.table)
        if err != nil {
                return nil, err
        }

        id, err := table.deleteCheck(stmt.constraintName)
        if err != nil {
                return nil, err
        }

        err = persistCheckDeletion(ctx, tx, table.id, id)

        tx.mutatedCatalog = true

        return tx, err
}

func persistCheckDeletion(ctx context.Context, tx *SQLTx, tableID uint32, checkId uint32) error {
        mappedKey := MapKey(
                tx.sqlPrefix(),
                catalogCheckPrefix,
                EncodeID(DatabaseID),
                EncodeID(tableID),
                EncodeID(checkId),
        )
        return tx.delete(ctx, mappedKey)
}

func (stmt *DropConstraintStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

type UpsertIntoStmt struct {
        isInsert   bool
        tableRef   *tableRef
        cols       []string
        ds         DataSource
        onConflict *OnConflictDo
}

func (stmt *UpsertIntoStmt) readOnly() bool {
        return false
}

func (stmt *UpsertIntoStmt) requiredPrivileges() []SQLPrivilege {
        privileges := stmt.privileges()
        if stmt.ds != nil {
                privileges = append(privileges, stmt.ds.requiredPrivileges()...)
        }
        return privileges
}

func (stmt *UpsertIntoStmt) privileges() []SQLPrivilege {
        if stmt.isInsert {
                return []SQLPrivilege{SQLPrivilegeInsert}
        }
        return []SQLPrivilege{SQLPrivilegeInsert, SQLPrivilegeUpdate}
}

func NewUpsertIntoStmt(table string, cols []string, ds DataSource, isInsert bool, onConflict *OnConflictDo) *UpsertIntoStmt {
        return &UpsertIntoStmt{
                isInsert:   isInsert,
                tableRef:   NewTableRef(table, ""),
                cols:       cols,
                ds:         ds,
                onConflict: onConflict,
        }
}

type RowSpec struct {
        Values []ValueExp
}

func NewRowSpec(values []ValueExp) *RowSpec {
        return &RowSpec{
                Values: values,
        }
}

type OnConflictDo struct{}

func (stmt *UpsertIntoStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        ds, ok := stmt.ds.(*valuesDataSource)
        if !ok {
                return stmt.ds.inferParameters(ctx, tx, params)
        }

        emptyDescriptors := make(map[string]ColDescriptor)
        for _, row := range ds.rows {
                if len(stmt.cols) != len(row.Values) {
                        return ErrInvalidNumberOfValues
                }

                for i, val := range row.Values {
                        table, err := stmt.tableRef.referencedTable(tx)
                        if err != nil {
                                return err
                        }

                        col, err := table.GetColumnByName(stmt.cols[i])
                        if err != nil {
                                return err
                        }

                        err = val.requiresType(col.colType, emptyDescriptors, params, table.name)
                        if err != nil {
                                return err
                        }
                }
        }
        return nil
}

func (stmt *UpsertIntoStmt) validate(table *Table) (map[uint32]int, error) {
        selPosByColID := make(map[uint32]int, len(stmt.cols))

        for i, c := range stmt.cols {
                col, err := table.GetColumnByName(c)
                if err != nil {
                        return nil, err
                }

                _, duplicated := selPosByColID[col.id]
                if duplicated {
                        return nil, fmt.Errorf("%w (%s)", ErrDuplicatedColumn, col.colName)
                }

                selPosByColID[col.id] = i
        }

        return selPosByColID, nil
}

func (stmt *UpsertIntoStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        table, err := stmt.tableRef.referencedTable(tx)
        if err != nil {
                return nil, err
        }

        selPosByColID, err := stmt.validate(table)
        if err != nil {
                return nil, err
        }

        r := &Row{
                ValuesByPosition: make([]TypedValue, len(table.cols)),
                ValuesBySelector: make(map[string]TypedValue),
        }

        reader, err := stmt.ds.Resolve(ctx, tx, params, nil)
        if err != nil {
                return nil, err
        }
        defer reader.Close()

        for {
                row, err := reader.Read(ctx)
                if errors.Is(err, ErrNoMoreRows) {
                        break
                }
                if err != nil {
                        return nil, err
                }

                if len(row.ValuesByPosition) != len(stmt.cols) {
                        return nil, ErrInvalidNumberOfValues
                }

                valuesByColID := make(map[uint32]TypedValue)

                var pkMustExist bool

                for colID, col := range table.colsByID {
                        colPos, specified := selPosByColID[colID]
                        if !specified {
                                // TODO: Default values
                                if col.notNull && !col.autoIncrement {
                                        return nil, fmt.Errorf("%w (%s)", ErrNotNullableColumnCannotBeNull, col.colName)
                                }

                                // inject auto-incremental pk value
                                if stmt.isInsert && col.autoIncrement {
                                        // current implementation assumes only PK can be set as autoincremental
                                        table.maxPK++

                                        pkCol := table.primaryIndex.cols[0]
                                        valuesByColID[pkCol.id] = &Integer{val: table.maxPK}

                                        if _, ok := tx.firstInsertedPKs[table.name]; !ok {
                                                tx.firstInsertedPKs[table.name] = table.maxPK
                                        }
                                        tx.lastInsertedPKs[table.name] = table.maxPK
                                }

                                continue
                        }

                        // value was specified
                        cVal := row.ValuesByPosition[colPos]

                        val, err := cVal.substitute(params)
                        if err != nil {
                                return nil, err
                        }

                        rval, err := val.reduce(tx, nil, table.name)
                        if err != nil {
                                return nil, err
                        }

                        if rval.IsNull() {
                                if col.notNull || col.autoIncrement {
                                        return nil, fmt.Errorf("%w (%s)", ErrNotNullableColumnCannotBeNull, col.colName)
                                }

                                continue
                        }

                        if col.autoIncrement {
                                // validate specified value
                                nl, isNumber := rval.RawValue().(int64)
                                if !isNumber {
                                        return nil, fmt.Errorf("%w (expecting numeric value)", ErrInvalidValue)
                                }

                                pkMustExist = nl <= table.maxPK

                                if _, ok := tx.firstInsertedPKs[table.name]; !ok {
                                        tx.firstInsertedPKs[table.name] = nl
                                }
                                tx.lastInsertedPKs[table.name] = nl
                        }

                        valuesByColID[colID] = rval
                }

                for i, col := range table.cols {
                        v := valuesByColID[col.id]

                        if v == nil {
                                v = NewNull(AnyType)
                        } else if len(table.checkConstraints) > 0 && col.Type() == JSONType {
                                s, _ := v.RawValue().(string)
                                jsonVal, err := NewJsonFromString(s)
                                if err != nil {
                                        return nil, err
                                }
                                v = jsonVal
                        }

                        r.ValuesByPosition[i] = v
                        r.ValuesBySelector[EncodeSelector("", table.name, col.colName)] = v
                }

                if err := checkConstraints(tx, table.checkConstraints, r, table.name); err != nil {
                        return nil, err
                }

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

                // pk entry
                mappedPKey := MapKey(tx.sqlPrefix(), MappedPrefix, EncodeID(table.id), EncodeID(table.primaryIndex.id), pkEncVals, pkEncVals)
                if len(mappedPKey) > MaxKeyLen {
                        return nil, ErrMaxKeyLengthExceeded
                }

                _, err = tx.get(ctx, mappedPKey)
                if err != nil && !errors.Is(err, store.ErrKeyNotFound) {
                        return nil, err
                }

                if errors.Is(err, store.ErrKeyNotFound) && pkMustExist {
                        return nil, fmt.Errorf("%w: specified value must be greater than current one", ErrInvalidValue)
                }

                if stmt.isInsert {
                        if err == nil && stmt.onConflict == nil {
                                return nil, store.ErrKeyAlreadyExists
                        }

                        if err == nil && stmt.onConflict != nil {
                                // TODO: conflict resolution may be extended. Currently only supports "ON CONFLICT DO NOTHING"
                                continue
                        }
                }

                err = tx.doUpsert(ctx, pkEncVals, valuesByColID, table, !stmt.isInsert)
                if err != nil {
                        return nil, err
                }
        }
        return tx, nil
}

func checkConstraints(tx *SQLTx, checks map[string]CheckConstraint, row *Row, table string) error {
        for _, check := range checks {
                val, err := check.exp.reduce(tx, row, table)
                if err != nil {
                        return fmt.Errorf("%w: %s", ErrCheckConstraintViolation, err)
                }

                if val.Type() != BooleanType {
                        return ErrInvalidCheckConstraint
                }

                if !val.RawValue().(bool) {
                        return fmt.Errorf("%w: %s", ErrCheckConstraintViolation, check.exp.String())
                }
        }
        return nil
}

func (tx *SQLTx) encodeRowValue(valuesByColID map[uint32]TypedValue, table *Table) ([]byte, error) {
        valbuf := bytes.Buffer{}

        // null values are not serialized
        encodedVals := 0
        for _, v := range valuesByColID {
                if !v.IsNull() {
                        encodedVals++
                }
        }

        b := make([]byte, EncLenLen)
        binary.BigEndian.PutUint32(b, uint32(encodedVals))

        _, err := valbuf.Write(b)
        if err != nil {
                return nil, err
        }

        for _, col := range table.cols {
                rval, specified := valuesByColID[col.id]
                if !specified || rval.IsNull() {
                        continue
                }

                b := make([]byte, EncIDLen)
                binary.BigEndian.PutUint32(b, uint32(col.id))

                _, err = valbuf.Write(b)
                if err != nil {
                        return nil, fmt.Errorf("%w: table: %s, column: %s", err, table.name, col.colName)
                }

                encVal, err := EncodeValue(rval, col.colType, col.MaxLen())
                if err != nil {
                        return nil, fmt.Errorf("%w: table: %s, column: %s", err, table.name, col.colName)
                }

                _, err = valbuf.Write(encVal)
                if err != nil {
                        return nil, fmt.Errorf("%w: table: %s, column: %s", err, table.name, col.colName)
                }
        }

        return valbuf.Bytes(), nil
}

func (tx *SQLTx) doUpsert(ctx context.Context, pkEncVals []byte, valuesByColID map[uint32]TypedValue, table *Table, reuseIndex bool) error {
        var reusableIndexEntries map[uint32]struct{}

        if reuseIndex && len(table.indexes) > 1 {
                currPKRow, err := tx.fetchPKRow(ctx, table, valuesByColID)
                if err == nil {
                        currValuesByColID := make(map[uint32]TypedValue, len(currPKRow.ValuesBySelector))

                        for _, col := range table.cols {
                                encSel := EncodeSelector("", table.name, col.colName)
                                currValuesByColID[col.id] = currPKRow.ValuesBySelector[encSel]
                        }

                        reusableIndexEntries, err = tx.deprecateIndexEntries(pkEncVals, currValuesByColID, valuesByColID, table)
                        if err != nil {
                                return err
                        }
                } else if !errors.Is(err, ErrNoMoreRows) {
                        return err
                }
        }

        rowKey := MapKey(tx.sqlPrefix(), RowPrefix, EncodeID(DatabaseID), EncodeID(table.id), EncodeID(PKIndexID), pkEncVals)

        encodedRowValue, err := tx.encodeRowValue(valuesByColID, table)
        if err != nil {
                return err
        }

        err = tx.set(rowKey, nil, encodedRowValue)
        if err != nil {
                return err
        }

        // create in-memory and validate entries for secondary indexes
        for _, index := range table.indexes {
                if index.IsPrimary() {
                        continue
                }

                if reusableIndexEntries != nil {
                        _, reusable := reusableIndexEntries[index.id]
                        if reusable {
                                continue
                        }
                }

                encodedValues := make([][]byte, 2+len(index.cols))
                encodedValues[0] = EncodeID(table.id)
                encodedValues[1] = EncodeID(index.id)

                indexKeyLen := 0

                for i, col := range index.cols {
                        rval, specified := valuesByColID[col.id]
                        if !specified {
                                rval = &NullValue{t: col.colType}
                        }

                        encVal, n, err := EncodeValueAsKey(rval, col.colType, col.MaxLen())
                        if err != nil {
                                return fmt.Errorf("%w: index on '%s' and column '%s'", err, index.Name(), col.colName)
                        }

                        if n > MaxKeyLen {
                                return fmt.Errorf("%w: can not index entry for column '%s'. Max key length for variable columns is %d", ErrLimitedKeyType, col.colName, MaxKeyLen)
                        }

                        indexKeyLen += n

                        encodedValues[i+2] = encVal
                }

                if indexKeyLen > MaxKeyLen {
                        return fmt.Errorf("%w: can not index entry using columns '%v'. Max key length is %d", ErrLimitedKeyType, index.cols, MaxKeyLen)
                }

                smkey := MapKey(tx.sqlPrefix(), MappedPrefix, encodedValues...)

                // no other equivalent entry should be already indexed
                if index.IsUnique() {
                        _, valRef, err := tx.getWithPrefix(ctx, smkey, nil)
                        if err == nil && (valRef.KVMetadata() == nil || !valRef.KVMetadata().Deleted()) {
                                return store.ErrKeyAlreadyExists
                        } else if !errors.Is(err, store.ErrKeyNotFound) {
                                return err
                        }
                }

                err = tx.setTransient(smkey, nil, encodedRowValue) // only-indexable
                if err != nil {
                        return err
                }
        }

        tx.updatedRows++

        return nil
}

func encodedKey(index *Index, valuesByColID map[uint32]TypedValue) ([]byte, error) {
        valbuf := bytes.Buffer{}

        indexKeyLen := 0

        for _, col := range index.cols {
                rval, specified := valuesByColID[col.id]
                if !specified || rval.IsNull() {
                        return nil, ErrPKCanNotBeNull
                }

                encVal, n, err := EncodeValueAsKey(rval, col.colType, col.MaxLen())
                if err != nil {
                        return nil, fmt.Errorf("%w: index of table '%s' and column '%s'", err, index.table.name, col.colName)
                }

                if n > MaxKeyLen {
                        return nil, fmt.Errorf("%w: invalid key entry for column '%s'. Max key length for variable columns is %d", ErrLimitedKeyType, col.colName, MaxKeyLen)
                }

                indexKeyLen += n

                _, err = valbuf.Write(encVal)
                if err != nil {
                        return nil, err
                }
        }

        if indexKeyLen > MaxKeyLen {
                return nil, fmt.Errorf("%w: invalid key entry using columns '%v'. Max key length is %d", ErrLimitedKeyType, index.cols, MaxKeyLen)
        }

        return valbuf.Bytes(), nil
}

func (tx *SQLTx) fetchPKRow(ctx context.Context, table *Table, valuesByColID map[uint32]TypedValue) (*Row, error) {
        pkRanges := make(map[uint32]*typedValueRange, len(table.primaryIndex.cols))

        for _, pkCol := range table.primaryIndex.cols {
                pkVal := valuesByColID[pkCol.id]

                pkRanges[pkCol.id] = &typedValueRange{
                        lRange: &typedValueSemiRange{val: pkVal, inclusive: true},
                        hRange: &typedValueSemiRange{val: pkVal, inclusive: true},
                }
        }

        scanSpecs := &ScanSpecs{
                Index:         table.primaryIndex,
                rangesByColID: pkRanges,
        }

        r, err := newRawRowReader(tx, nil, table, period{}, table.name, scanSpecs)
        if err != nil {
                return nil, err
        }

        defer func() {
                r.Close()
        }()

        return r.Read(ctx)
}

// deprecateIndexEntries mark previous index entries as deleted
func (tx *SQLTx) deprecateIndexEntries(
        pkEncVals []byte,
        currValuesByColID, newValuesByColID map[uint32]TypedValue,
        table *Table) (reusableIndexEntries map[uint32]struct{}, err error) {

        encodedRowValue, err := tx.encodeRowValue(currValuesByColID, table)
        if err != nil {
                return nil, err
        }

        reusableIndexEntries = make(map[uint32]struct{})

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

                encodedValues := make([][]byte, 2+len(index.cols)+1)
                encodedValues[0] = EncodeID(table.id)
                encodedValues[1] = EncodeID(index.id)
                encodedValues[len(encodedValues)-1] = pkEncVals

                // existent index entry is deleted only if it differs from existent one
                sameIndexKey := true

                for i, col := range index.cols {
                        currVal, specified := currValuesByColID[col.id]
                        if !specified {
                                currVal = &NullValue{t: col.colType}
                        }

                        newVal, specified := newValuesByColID[col.id]
                        if !specified {
                                newVal = &NullValue{t: col.colType}
                        }

                        r, err := currVal.Compare(newVal)
                        if err != nil {
                                return nil, err
                        }

                        sameIndexKey = sameIndexKey && r == 0

                        encVal, _, _ := EncodeValueAsKey(currVal, col.colType, col.MaxLen())

                        encodedValues[i+3] = encVal
                }

                // mark existent index entry as deleted
                if sameIndexKey {
                        reusableIndexEntries[index.id] = struct{}{}
                } else {
                        md := store.NewKVMetadata()

                        md.AsDeleted(true)

                        err = tx.set(MapKey(tx.sqlPrefix(), MappedPrefix, encodedValues...), md, encodedRowValue)
                        if err != nil {
                                return nil, err
                        }
                }
        }

        return reusableIndexEntries, nil
}

type UpdateStmt struct {
        tableRef *tableRef
        where    ValueExp
        updates  []*colUpdate
        indexOn  []string
        limit    ValueExp
        offset   ValueExp
}

type colUpdate struct {
        col string
        op  CmpOperator
        val ValueExp
}

func (stmt *UpdateStmt) readOnly() bool {
        return false
}

func (stmt *UpdateStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeUpdate}
}

func (stmt *UpdateStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        selectStmt := &SelectStmt{
                ds:    stmt.tableRef,
                where: stmt.where,
        }

        err := selectStmt.inferParameters(ctx, tx, params)
        if err != nil {
                return err
        }

        table, err := stmt.tableRef.referencedTable(tx)
        if err != nil {
                return err
        }

        for _, update := range stmt.updates {
                col, err := table.GetColumnByName(update.col)
                if err != nil {
                        return err
                }

                err = update.val.requiresType(col.colType, make(map[string]ColDescriptor), params, table.name)
                if err != nil {
                        return err
                }
        }

        return nil
}

func (stmt *UpdateStmt) validate(table *Table) error {
        colIDs := make(map[uint32]struct{}, len(stmt.updates))

        for _, update := range stmt.updates {
                if update.op != EQ {
                        return ErrIllegalArguments
                }

                col, err := table.GetColumnByName(update.col)
                if err != nil {
                        return err
                }

                if table.PrimaryIndex().IncludesCol(col.id) {
                        return ErrPKCanNotBeUpdated
                }

                _, duplicated := colIDs[col.id]
                if duplicated {
                        return ErrDuplicatedColumn
                }

                colIDs[col.id] = struct{}{}
        }

        return nil
}

func (stmt *UpdateStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        selectStmt := &SelectStmt{
                ds:      stmt.tableRef,
                where:   stmt.where,
                indexOn: stmt.indexOn,
                limit:   stmt.limit,
                offset:  stmt.offset,
        }

        rowReader, err := selectStmt.Resolve(ctx, tx, params, nil)
        if err != nil {
                return nil, err
        }
        defer rowReader.Close()

        table := rowReader.ScanSpecs().Index.table

        err = stmt.validate(table)
        if err != nil {
                return nil, err
        }

        cols, err := rowReader.colsBySelector(ctx)
        if err != nil {
                return nil, err
        }

        for {
                row, err := rowReader.Read(ctx)
                if errors.Is(err, ErrNoMoreRows) {
                        break
                } else if err != nil {
                        return nil, err
                }

                valuesByColID := make(map[uint32]TypedValue, len(row.ValuesBySelector))

                for _, col := range table.cols {
                        encSel := EncodeSelector("", table.name, col.colName)
                        valuesByColID[col.id] = row.ValuesBySelector[encSel]
                }

                for _, update := range stmt.updates {
                        col, err := table.GetColumnByName(update.col)
                        if err != nil {
                                return nil, err
                        }

                        sval, err := update.val.substitute(params)
                        if err != nil {
                                return nil, err
                        }

                        rval, err := sval.reduce(tx, row, table.name)
                        if err != nil {
                                return nil, err
                        }

                        err = rval.requiresType(col.colType, cols, nil, table.name)
                        if err != nil {
                                return nil, err
                        }

                        valuesByColID[col.id] = rval
                }

                for i, col := range table.cols {
                        v := valuesByColID[col.id]

                        row.ValuesByPosition[i] = v
                        row.ValuesBySelector[EncodeSelector("", table.name, col.colName)] = v
                }

                if err := checkConstraints(tx, table.checkConstraints, row, table.name); err != nil {
                        return nil, err
                }

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

                // primary index entry
                mkey := MapKey(tx.sqlPrefix(), MappedPrefix, EncodeID(table.id), EncodeID(table.primaryIndex.id), pkEncVals, pkEncVals)

                // mkey must exist
                _, err = tx.get(ctx, mkey)
                if err != nil {
                        return nil, err
                }

                err = tx.doUpsert(ctx, pkEncVals, valuesByColID, table, true)
                if err != nil {
                        return nil, err
                }
        }

        return tx, nil
}

type DeleteFromStmt struct {
        tableRef *tableRef
        where    ValueExp
        indexOn  []string
        orderBy  []*OrdExp
        limit    ValueExp
        offset   ValueExp
}

func NewDeleteFromStmt(table string, where ValueExp, orderBy []*OrdExp, limit ValueExp) *DeleteFromStmt {
        return &DeleteFromStmt{
                tableRef: NewTableRef(table, ""),
                where:    where,
                orderBy:  orderBy,
                limit:    limit,
        }
}

func (stmt *DeleteFromStmt) readOnly() bool {
        return false
}

func (stmt *DeleteFromStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeDelete}
}

func (stmt *DeleteFromStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        selectStmt := &SelectStmt{
                ds:      stmt.tableRef,
                where:   stmt.where,
                orderBy: stmt.orderBy,
        }
        return selectStmt.inferParameters(ctx, tx, params)
}

func (stmt *DeleteFromStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        selectStmt := &SelectStmt{
                ds:      stmt.tableRef,
                where:   stmt.where,
                indexOn: stmt.indexOn,
                orderBy: stmt.orderBy,
                limit:   stmt.limit,
                offset:  stmt.offset,
        }

        rowReader, err := selectStmt.Resolve(ctx, tx, params, nil)
        if err != nil {
                return nil, err
        }
        defer rowReader.Close()

        table := rowReader.ScanSpecs().Index.table

        for {
                row, err := rowReader.Read(ctx)
                if errors.Is(err, ErrNoMoreRows) {
                        break
                }
                if err != nil {
                        return nil, err
                }

                valuesByColID := make(map[uint32]TypedValue, len(row.ValuesBySelector))

                for _, col := range table.cols {
                        encSel := EncodeSelector("", table.name, col.colName)
                        valuesByColID[col.id] = row.ValuesBySelector[encSel]
                }

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

                err = tx.deleteIndexEntries(pkEncVals, valuesByColID, table)
                if err != nil {
                        return nil, err
                }

                tx.updatedRows++
        }
        return tx, nil
}

func (tx *SQLTx) deleteIndexEntries(pkEncVals []byte, valuesByColID map[uint32]TypedValue, table *Table) error {
        encodedRowValue, err := tx.encodeRowValue(valuesByColID, table)
        if err != nil {
                return err
        }

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

                encodedValues := make([][]byte, 3+len(index.cols))
                encodedValues[0] = EncodeID(DatabaseID)
                encodedValues[1] = EncodeID(table.id)
                encodedValues[2] = EncodeID(index.id)

                for i, col := range index.cols {
                        val, specified := valuesByColID[col.id]
                        if !specified {
                                val = &NullValue{t: col.colType}
                        }

                        encVal, _, _ := EncodeValueAsKey(val, col.colType, col.MaxLen())

                        encodedValues[i+3] = encVal
                }

                md := store.NewKVMetadata()

                md.AsDeleted(true)

                err := tx.set(MapKey(tx.sqlPrefix(), RowPrefix, encodedValues...), md, encodedRowValue)
                if err != nil {
                        return err
                }
        }

        return nil
}

type ValueExp interface {
        inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error)
        requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error
        substitute(params map[string]interface{}) (ValueExp, error)
        selectors() []Selector
        reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error)
        reduceSelectors(row *Row, implicitTable string) ValueExp
        isConstant() bool
        selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error
        String() string
}

type typedValueRange struct {
        lRange *typedValueSemiRange
        hRange *typedValueSemiRange
}

type typedValueSemiRange struct {
        val       TypedValue
        inclusive bool
}

func (r *typedValueRange) unitary() bool {
        // TODO: this simplified implementation doesn't cover all unitary cases e.g. 3<=v<4
        if r.lRange == nil || r.hRange == nil {
                return false
        }

        res, _ := r.lRange.val.Compare(r.hRange.val)
        return res == 0 && r.lRange.inclusive && r.hRange.inclusive
}

func (r *typedValueRange) refineWith(refiningRange *typedValueRange) error {
        if r.lRange == nil {
                r.lRange = refiningRange.lRange
        } else if r.lRange != nil && refiningRange.lRange != nil {
                maxRange, err := maxSemiRange(r.lRange, refiningRange.lRange)
                if err != nil {
                        return err
                }
                r.lRange = maxRange
        }

        if r.hRange == nil {
                r.hRange = refiningRange.hRange
        } else if r.hRange != nil && refiningRange.hRange != nil {
                minRange, err := minSemiRange(r.hRange, refiningRange.hRange)
                if err != nil {
                        return err
                }
                r.hRange = minRange
        }

        return nil
}

func (r *typedValueRange) extendWith(extendingRange *typedValueRange) error {
        if r.lRange == nil || extendingRange.lRange == nil {
                r.lRange = nil
        } else {
                minRange, err := minSemiRange(r.lRange, extendingRange.lRange)
                if err != nil {
                        return err
                }
                r.lRange = minRange
        }

        if r.hRange == nil || extendingRange.hRange == nil {
                r.hRange = nil
        } else {
                maxRange, err := maxSemiRange(r.hRange, extendingRange.hRange)
                if err != nil {
                        return err
                }
                r.hRange = maxRange
        }

        return nil
}

func maxSemiRange(or1, or2 *typedValueSemiRange) (*typedValueSemiRange, error) {
        r, err := or1.val.Compare(or2.val)
        if err != nil {
                return nil, err
        }

        maxVal := or1.val
        if r < 0 {
                maxVal = or2.val
        }

        return &typedValueSemiRange{
                val:       maxVal,
                inclusive: or1.inclusive && or2.inclusive,
        }, nil
}

func minSemiRange(or1, or2 *typedValueSemiRange) (*typedValueSemiRange, error) {
        r, err := or1.val.Compare(or2.val)
        if err != nil {
                return nil, err
        }

        minVal := or1.val
        if r > 0 {
                minVal = or2.val
        }

        return &typedValueSemiRange{
                val:       minVal,
                inclusive: or1.inclusive || or2.inclusive,
        }, nil
}

type TypedValue interface {
        ValueExp
        Type() SQLValueType
        RawValue() interface{}
        Compare(val TypedValue) (int, error)
        IsNull() bool
}

type Tuple []TypedValue

func (t Tuple) Compare(other Tuple) (int, int, error) {
        if len(t) != len(other) {
                return -1, -1, ErrNotComparableValues
        }

        for i := range t {
                res, err := t[i].Compare(other[i])
                if err != nil || res != 0 {
                        return res, i, err
                }
        }
        return 0, -1, nil
}

func NewNull(t SQLValueType) *NullValue {
        return &NullValue{t: t}
}

type NullValue struct {
        t SQLValueType
}

func (n *NullValue) Type() SQLValueType {
        return n.t
}

func (n *NullValue) RawValue() interface{} {
        return nil
}

func (n *NullValue) IsNull() bool {
        return true
}

func (n *NullValue) String() string {
        return "NULL"
}

func (n *NullValue) Compare(val TypedValue) (int, error) {
        if n.t != AnyType && val.Type() != AnyType && n.t != val.Type() {
                return 0, ErrNotComparableValues
        }

        if val.RawValue() == nil {
                return 0, nil
        }
        return -1, nil
}

func (v *NullValue) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        return v.t, nil
}

func (v *NullValue) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if v.t == t {
                return nil
        }

        if v.t != AnyType {
                return ErrInvalidTypes
        }

        v.t = t

        return nil
}

func (v *NullValue) selectors() []Selector {
        return nil
}

func (v *NullValue) substitute(params map[string]interface{}) (ValueExp, error) {
        return v, nil
}

func (v *NullValue) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        return v, nil
}

func (v *NullValue) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return v
}

func (v *NullValue) isConstant() bool {
        return true
}

func (v *NullValue) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

type Integer struct {
        val int64
}

func NewInteger(val int64) *Integer {
        return &Integer{val: val}
}

func (v *Integer) Type() SQLValueType {
        return IntegerType
}

func (v *Integer) IsNull() bool {
        return false
}

func (v *Integer) String() string {
        return strconv.FormatInt(v.val, 10)
}

func (v *Integer) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        return IntegerType, nil
}

func (v *Integer) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if t != IntegerType && t != JSONType {
                return fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, IntegerType, t)
        }

        return nil
}

func (v *Integer) selectors() []Selector {
        return nil
}

func (v *Integer) substitute(params map[string]interface{}) (ValueExp, error) {
        return v, nil
}

func (v *Integer) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        return v, nil
}

func (v *Integer) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return v
}

func (v *Integer) isConstant() bool {
        return true
}

func (v *Integer) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (v *Integer) RawValue() interface{} {
        return v.val
}

func (v *Integer) Compare(val TypedValue) (int, error) {
        if val.IsNull() {
                return 1, nil
        }

        if val.Type() == JSONType {
                res, err := val.Compare(v)
                return -res, err
        }

        if val.Type() == Float64Type {
                r, err := val.Compare(v)
                return r * -1, err
        }

        if val.Type() != IntegerType {
                return 0, ErrNotComparableValues
        }

        rval := val.RawValue().(int64)

        if v.val == rval {
                return 0, nil
        }

        if v.val > rval {
                return 1, nil
        }

        return -1, nil
}

type Timestamp struct {
        val time.Time
}

func (v *Timestamp) Type() SQLValueType {
        return TimestampType
}

func (v *Timestamp) IsNull() bool {
        return false
}

func (v *Timestamp) String() string {
        return v.val.Format("2006-01-02 15:04:05.999999")
}

func (v *Timestamp) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        return TimestampType, nil
}

func (v *Timestamp) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if t != TimestampType {
                return fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, TimestampType, t)
        }

        return nil
}

func (v *Timestamp) selectors() []Selector {
        return nil
}

func (v *Timestamp) substitute(params map[string]interface{}) (ValueExp, error) {
        return v, nil
}

func (v *Timestamp) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        return v, nil
}

func (v *Timestamp) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return v
}

func (v *Timestamp) isConstant() bool {
        return true
}

func (v *Timestamp) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (v *Timestamp) RawValue() interface{} {
        return v.val
}

func (v *Timestamp) Compare(val TypedValue) (int, error) {
        if val.IsNull() {
                return 1, nil
        }

        if val.Type() != TimestampType {
                return 0, ErrNotComparableValues
        }

        rval := val.RawValue().(time.Time)

        if v.val.Before(rval) {
                return -1, nil
        }

        if v.val.After(rval) {
                return 1, nil
        }

        return 0, nil
}

type Varchar struct {
        val string
}

func NewVarchar(val string) *Varchar {
        return &Varchar{val: val}
}

func (v *Varchar) Type() SQLValueType {
        return VarcharType
}

func (v *Varchar) IsNull() bool {
        return false
}

func (v *Varchar) String() string {
        return fmt.Sprintf("'%s'", v.val)
}

func (v *Varchar) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        return VarcharType, nil
}

func (v *Varchar) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if t != VarcharType && t != JSONType {
                return fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, VarcharType, t)
        }
        return nil
}

func (v *Varchar) selectors() []Selector {
        return nil
}

func (v *Varchar) substitute(params map[string]interface{}) (ValueExp, error) {
        return v, nil
}

func (v *Varchar) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        return v, nil
}

func (v *Varchar) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return v
}

func (v *Varchar) isConstant() bool {
        return true
}

func (v *Varchar) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (v *Varchar) RawValue() interface{} {
        return v.val
}

func (v *Varchar) Compare(val TypedValue) (int, error) {
        if val.IsNull() {
                return 1, nil
        }

        if val.Type() == JSONType {
                res, err := val.Compare(v)
                return -res, err
        }

        if val.Type() != VarcharType {
                return 0, ErrNotComparableValues
        }

        rval := val.RawValue().(string)

        return bytes.Compare([]byte(v.val), []byte(rval)), nil
}

type UUID struct {
        val uuid.UUID
}

func NewUUID(val uuid.UUID) *UUID {
        return &UUID{val: val}
}

func (v *UUID) Type() SQLValueType {
        return UUIDType
}

func (v *UUID) IsNull() bool {
        return false
}

func (v *UUID) String() string {
        return v.val.String()
}

func (v *UUID) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        return UUIDType, nil
}

func (v *UUID) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if t != UUIDType {
                return fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, UUIDType, t)
        }

        return nil
}

func (v *UUID) selectors() []Selector {
        return nil
}

func (v *UUID) substitute(params map[string]interface{}) (ValueExp, error) {
        return v, nil
}

func (v *UUID) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        return v, nil
}

func (v *UUID) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return v
}

func (v *UUID) isConstant() bool {
        return true
}

func (v *UUID) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (v *UUID) RawValue() interface{} {
        return v.val
}

func (v *UUID) Compare(val TypedValue) (int, error) {
        if val.IsNull() {
                return 1, nil
        }

        if val.Type() != UUIDType {
                return 0, ErrNotComparableValues
        }

        rval := val.RawValue().(uuid.UUID)

        return bytes.Compare(v.val[:], rval[:]), nil
}

type Bool struct {
        val bool
}

func NewBool(val bool) *Bool {
        return &Bool{val: val}
}

func (v *Bool) Type() SQLValueType {
        return BooleanType
}

func (v *Bool) IsNull() bool {
        return false
}

func (v *Bool) String() string {
        return strconv.FormatBool(v.val)
}

func (v *Bool) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        return BooleanType, nil
}

func (v *Bool) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if t != BooleanType && t != JSONType {
                return fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, BooleanType, t)
        }
        return nil
}

func (v *Bool) selectors() []Selector {
        return nil
}

func (v *Bool) substitute(params map[string]interface{}) (ValueExp, error) {
        return v, nil
}

func (v *Bool) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        return v, nil
}

func (v *Bool) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return v
}

func (v *Bool) isConstant() bool {
        return true
}

func (v *Bool) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (v *Bool) RawValue() interface{} {
        return v.val
}

func (v *Bool) Compare(val TypedValue) (int, error) {
        if val.IsNull() {
                return 1, nil
        }

        if val.Type() == JSONType {
                res, err := val.Compare(v)
                return -res, err
        }

        if val.Type() != BooleanType {
                return 0, ErrNotComparableValues
        }

        rval := val.RawValue().(bool)

        if v.val == rval {
                return 0, nil
        }

        if v.val {
                return 1, nil
        }

        return -1, nil
}

type Blob struct {
        val []byte
}

func NewBlob(val []byte) *Blob {
        return &Blob{val: val}
}

func (v *Blob) Type() SQLValueType {
        return BLOBType
}

func (v *Blob) IsNull() bool {
        return false
}

func (v *Blob) String() string {
        return hex.EncodeToString(v.val)
}

func (v *Blob) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        return BLOBType, nil
}

func (v *Blob) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if t != BLOBType {
                return fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, BLOBType, t)
        }

        return nil
}

func (v *Blob) selectors() []Selector {
        return nil
}

func (v *Blob) substitute(params map[string]interface{}) (ValueExp, error) {
        return v, nil
}

func (v *Blob) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        return v, nil
}

func (v *Blob) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return v
}

func (v *Blob) isConstant() bool {
        return true
}

func (v *Blob) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (v *Blob) RawValue() interface{} {
        return v.val
}

func (v *Blob) Compare(val TypedValue) (int, error) {
        if val.IsNull() {
                return 1, nil
        }

        if val.Type() != BLOBType {
                return 0, ErrNotComparableValues
        }

        rval := val.RawValue().([]byte)

        return bytes.Compare(v.val, rval), nil
}

type Float64 struct {
        val float64
}

func NewFloat64(val float64) *Float64 {
        return &Float64{val: val}
}

func (v *Float64) Type() SQLValueType {
        return Float64Type
}

func (v *Float64) IsNull() bool {
        return false
}

func (v *Float64) String() string {
        return strconv.FormatFloat(float64(v.val), 'f', -1, 64)
}

func (v *Float64) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        return Float64Type, nil
}

func (v *Float64) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if t != Float64Type && t != JSONType {
                return fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, Float64Type, t)
        }
        return nil
}

func (v *Float64) selectors() []Selector {
        return nil
}

func (v *Float64) substitute(params map[string]interface{}) (ValueExp, error) {
        return v, nil
}

func (v *Float64) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        return v, nil
}

func (v *Float64) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return v
}

func (v *Float64) isConstant() bool {
        return true
}

func (v *Float64) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (v *Float64) RawValue() interface{} {
        return v.val
}

func (v *Float64) Compare(val TypedValue) (int, error) {
        if val.Type() == JSONType {
                res, err := val.Compare(v)
                return -res, err
        }

        convVal, err := mayApplyImplicitConversion(val.RawValue(), Float64Type)
        if err != nil {
                return 0, err
        }

        if convVal == nil {
                return 1, nil
        }

        rval, ok := convVal.(float64)
        if !ok {
                return 0, ErrNotComparableValues
        }

        if v.val == rval {
                return 0, nil
        }

        if v.val > rval {
                return 1, nil
        }

        return -1, nil
}

type FnCall struct {
        fn     string
        params []ValueExp
}

func (v *FnCall) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        fn, err := v.resolveFunc()
        if err != nil {
                return AnyType, nil
        }
        return fn.InferType(cols, params, implicitTable)
}

func (v *FnCall) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        fn, err := v.resolveFunc()
        if err != nil {
                return err
        }
        return fn.RequiresType(t, cols, params, implicitTable)
}

func (v *FnCall) selectors() []Selector {
        selectors := make([]Selector, 0)
        for _, param := range v.params {
                selectors = append(selectors, param.selectors()...)
        }
        return selectors
}

func (v *FnCall) substitute(params map[string]interface{}) (val ValueExp, err error) {
        ps := make([]ValueExp, len(v.params))
        for i, p := range v.params {
                ps[i], err = p.substitute(params)
                if err != nil {
                        return nil, err
                }
        }

        return &FnCall{
                fn:     v.fn,
                params: ps,
        }, nil
}

func (v *FnCall) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        fn, err := v.resolveFunc()
        if err != nil {
                return nil, err
        }

        fnInputs, err := v.reduceParams(tx, row, implicitTable)
        if err != nil {
                return nil, err
        }
        return fn.Apply(tx, fnInputs)
}

func (v *FnCall) reduceParams(tx *SQLTx, row *Row, implicitTable string) ([]TypedValue, error) {
        var values []TypedValue
        if len(v.params) > 0 {
                values = make([]TypedValue, len(v.params))
                for i, p := range v.params {
                        v, err := p.reduce(tx, row, implicitTable)
                        if err != nil {
                                return nil, err
                        }
                        values[i] = v
                }
        }
        return values, nil
}

func (v *FnCall) resolveFunc() (Function, error) {
        fn, exists := builtinFunctions[strings.ToUpper(v.fn)]
        if !exists {
                return nil, fmt.Errorf("%w: unknown function %s", ErrIllegalArguments, v.fn)
        }
        return fn, nil
}

func (v *FnCall) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return v
}

func (v *FnCall) isConstant() bool {
        return false
}

func (v *FnCall) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (v *FnCall) String() string {
        params := make([]string, len(v.params))
        for i, p := range v.params {
                params[i] = p.String()
        }
        return v.fn + "(" + strings.Join(params, ",") + ")"
}

type Cast struct {
        val ValueExp
        t   SQLValueType
}

func (c *Cast) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        _, err := c.val.inferType(cols, params, implicitTable)
        if err != nil {
                return AnyType, err
        }

        // val type may be restricted by compatible conversions, but multiple types may be compatible...

        return c.t, nil
}

func (c *Cast) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if c.t != t {
                return fmt.Errorf("%w: can not use value cast to %s as %s", ErrInvalidTypes, c.t, t)
        }

        return nil
}

func (c *Cast) substitute(params map[string]interface{}) (ValueExp, error) {
        val, err := c.val.substitute(params)
        if err != nil {
                return nil, err
        }
        c.val = val
        return c, nil
}

func (c *Cast) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        val, err := c.val.reduce(tx, row, implicitTable)
        if err != nil {
                return nil, err
        }

        conv, err := getConverter(val.Type(), c.t)
        if conv == nil {
                return nil, err
        }

        return conv(val)
}

func (v *Cast) selectors() []Selector {
        return v.val.selectors()
}

func (c *Cast) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return &Cast{
                val: c.val.reduceSelectors(row, implicitTable),
                t:   c.t,
        }
}

func (c *Cast) isConstant() bool {
        return c.val.isConstant()
}

func (c *Cast) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (c *Cast) String() string {
        return fmt.Sprintf("CAST (%s AS %s)", c.val.String(), c.t)
}

type Param struct {
        id  string
        pos int
}

func (v *Param) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        t, ok := params[v.id]
        if !ok {
                params[v.id] = AnyType
                return AnyType, nil
        }

        return t, nil
}

func (v *Param) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        currT, ok := params[v.id]
        if ok && currT != t && currT != AnyType {
                return ErrInferredMultipleTypes
        }

        params[v.id] = t

        return nil
}

func (p *Param) substitute(params map[string]interface{}) (ValueExp, error) {
        val, ok := params[p.id]
        if !ok {
                return nil, fmt.Errorf("%w(%s)", ErrMissingParameter, p.id)
        }

        if val == nil {
                return &NullValue{t: AnyType}, nil
        }

        switch v := val.(type) {
        case bool:
                {
                        return &Bool{val: v}, nil
                }
        case string:
                {
                        return &Varchar{val: v}, nil
                }
        case int:
                {
                        return &Integer{val: int64(v)}, nil
                }
        case uint:
                {
                        return &Integer{val: int64(v)}, nil
                }
        case uint64:
                {
                        return &Integer{val: int64(v)}, nil
                }
        case int64:
                {
                        return &Integer{val: v}, nil
                }
        case []byte:
                {
                        return &Blob{val: v}, nil
                }
        case time.Time:
                {
                        return &Timestamp{val: v.Truncate(time.Microsecond).UTC()}, nil
                }
        case float64:
                {
                        return &Float64{val: v}, nil
                }
        }
        return nil, ErrUnsupportedParameter
}

func (p *Param) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        return nil, ErrUnexpected
}

func (p *Param) selectors() []Selector {
        return nil
}

func (p *Param) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return p
}

func (p *Param) isConstant() bool {
        return true
}

func (v *Param) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (v *Param) String() string {
        return "@" + v.id
}

type whenThenClause struct {
        when, then ValueExp
}

type CaseWhenExp struct {
        exp      ValueExp
        whenThen []whenThenClause
        elseExp  ValueExp
}

func (ce *CaseWhenExp) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        checkType := func(e ValueExp, expectedType SQLValueType) (string, error) {
                t, err := e.inferType(cols, params, implicitTable)
                if err != nil {
                        return "", err
                }

                if expectedType == AnyType {
                        return t, nil
                }

                if t != expectedType {
                        if (t == Float64Type && expectedType == IntegerType) ||
                                (t == IntegerType && expectedType == Float64Type) {
                                return Float64Type, nil
                        }
                        return "", fmt.Errorf("%w: CASE types %s and %s cannot be matched", ErrInferredMultipleTypes, expectedType, t)
                }
                return t, nil
        }

        searchType := BooleanType
        inferredResType := AnyType
        if ce.exp != nil {
                t, err := ce.exp.inferType(cols, params, implicitTable)
                if err != nil {
                        return "", err
                }
                searchType = t
        }

        for _, e := range ce.whenThen {
                whenType, err := e.when.inferType(cols, params, implicitTable)
                if err != nil {
                        return "", err
                }

                if whenType != searchType {
                        return "", fmt.Errorf("%w: argument of CASE/WHEN must be of type %s, not type %s", ErrInvalidTypes, searchType, whenType)
                }

                t, err := checkType(e.then, inferredResType)
                if err != nil {
                        return "", err
                }
                inferredResType = t
        }

        if ce.elseExp != nil {
                return checkType(ce.elseExp, inferredResType)
        }
        return inferredResType, nil
}

func (ce *CaseWhenExp) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        inferredType, err := ce.inferType(cols, params, implicitTable)
        if err != nil {
                return err
        }

        if inferredType != t {
                return fmt.Errorf("%w: expected type %s but %s found instead", ErrInvalidTypes, t, inferredType)
        }
        return nil
}

func (ce *CaseWhenExp) substitute(params map[string]interface{}) (ValueExp, error) {
        var exp ValueExp
        if ce.exp != nil {
                e, err := ce.exp.substitute(params)
                if err != nil {
                        return nil, err
                }
                exp = e
        }

        whenThen := make([]whenThenClause, len(ce.whenThen))
        for i, wt := range ce.whenThen {
                whenValue, err := wt.when.substitute(params)
                if err != nil {
                        return nil, err
                }
                whenThen[i].when = whenValue

                thenValue, err := wt.then.substitute(params)
                if err != nil {
                        return nil, err
                }
                whenThen[i].then = thenValue
        }

        if ce.elseExp == nil {
                return &CaseWhenExp{
                        exp:      exp,
                        whenThen: whenThen,
                }, nil
        }

        elseValue, err := ce.elseExp.substitute(params)
        if err != nil {
                return nil, err
        }
        return &CaseWhenExp{
                exp:      exp,
                whenThen: whenThen,
                elseExp:  elseValue,
        }, nil
}

func (ce *CaseWhenExp) selectors() []Selector {
        selectors := make([]Selector, 0)
        if ce.exp != nil {
                selectors = append(selectors, ce.exp.selectors()...)
        }

        for _, wh := range ce.whenThen {
                selectors = append(selectors, wh.when.selectors()...)
                selectors = append(selectors, wh.then.selectors()...)
        }

        if ce.elseExp == nil {
                return selectors
        }
        return append(selectors, ce.elseExp.selectors()...)
}

func (ce *CaseWhenExp) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        var searchValue TypedValue
        if ce.exp != nil {
                v, err := ce.exp.reduce(tx, row, implicitTable)
                if err != nil {
                        return nil, err
                }
                searchValue = v
        } else {
                searchValue = &Bool{val: true}
        }

        for _, wt := range ce.whenThen {
                v, err := wt.when.reduce(tx, row, implicitTable)
                if err != nil {
                        return nil, err
                }

                if v.Type() != searchValue.Type() {
                        return nil, fmt.Errorf("%w: argument of CASE/WHEN must be type %s, not type %s", ErrInvalidTypes, v.Type(), searchValue.Type())
                }

                res, err := v.Compare(searchValue)
                if err != nil {
                        return nil, err
                }
                if res == 0 {
                        return wt.then.reduce(tx, row, implicitTable)
                }
        }

        if ce.elseExp == nil {
                return NewNull(AnyType), nil
        }
        return ce.elseExp.reduce(tx, row, implicitTable)
}

func (ce *CaseWhenExp) reduceSelectors(row *Row, implicitTable string) ValueExp {
        whenThen := make([]whenThenClause, len(ce.whenThen))
        for i, wt := range ce.whenThen {
                whenValue := wt.when.reduceSelectors(row, implicitTable)
                whenThen[i].when = whenValue

                thenValue := wt.then.reduceSelectors(row, implicitTable)
                whenThen[i].then = thenValue
        }

        if ce.elseExp == nil {
                return &CaseWhenExp{
                        whenThen: whenThen,
                }
        }

        return &CaseWhenExp{
                whenThen: whenThen,
                elseExp:  ce.elseExp.reduceSelectors(row, implicitTable),
        }
}

func (ce *CaseWhenExp) isConstant() bool {
        return false
}

func (ce *CaseWhenExp) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (ce *CaseWhenExp) String() string {
        var sb strings.Builder
        for _, wh := range ce.whenThen {
                sb.WriteString(fmt.Sprintf("WHEN %s THEN %s ", wh.when.String(), wh.then.String()))
        }

        if ce.elseExp != nil {
                sb.WriteString("ELSE " + ce.elseExp.String() + " ")
        }
        return "CASE " + sb.String() + "END"
}

type Comparison int

const (
        EqualTo Comparison = iota
        LowerThan
        LowerOrEqualTo
        GreaterThan
        GreaterOrEqualTo
)

type DataSource interface {
        SQLStmt
        Resolve(ctx context.Context, tx *SQLTx, params map[string]interface{}, scanSpecs *ScanSpecs) (RowReader, error)
        Alias() string
}

type TargetEntry struct {
        Exp ValueExp
        As  string
}

type SelectStmt struct {
        distinct  bool
        targets   []TargetEntry
        selectors []Selector
        ds        DataSource
        indexOn   []string
        joins     []*JoinSpec
        where     ValueExp
        groupBy   []*ColSelector
        having    ValueExp
        orderBy   []*OrdExp
        limit     ValueExp
        offset    ValueExp
        as        string
}

func NewSelectStmt(
        targets []TargetEntry,
        ds DataSource,
        where ValueExp,
        orderBy []*OrdExp,
        limit ValueExp,
        offset ValueExp,
) *SelectStmt {
        return &SelectStmt{
                targets: targets,
                ds:      ds,
                where:   where,
                orderBy: orderBy,
                limit:   limit,
                offset:  offset,
        }
}

func (stmt *SelectStmt) readOnly() bool {
        return true
}

func (stmt *SelectStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeSelect}
}

func (stmt *SelectStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        _, err := stmt.execAt(ctx, tx, nil)
        if err != nil {
                return err
        }

        // TODO: (jeroiraz) may be optimized so to resolve the query statement just once
        rowReader, err := stmt.Resolve(ctx, tx, nil, nil)
        if err != nil {
                return err
        }
        defer rowReader.Close()

        return rowReader.InferParameters(ctx, params)
}

func (stmt *SelectStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if stmt.groupBy == nil && stmt.having != nil {
                return nil, ErrHavingClauseRequiresGroupClause
        }

        if stmt.containsAggregations() || len(stmt.groupBy) > 0 {
                for _, sel := range stmt.targetSelectors() {
                        _, isAgg := sel.(*AggColSelector)
                        if !isAgg && !stmt.groupByContains(sel) {
                                return nil, fmt.Errorf("%s: %w", EncodeSelector(sel.resolve(stmt.Alias())), ErrColumnMustAppearInGroupByOrAggregation)
                        }
                }
        }

        if len(stmt.orderBy) > 0 {
                for _, col := range stmt.orderBy {
                        for _, sel := range col.exp.selectors() {
                                _, isAgg := sel.(*AggColSelector)
                                if (isAgg && !stmt.selectorAppearsInTargets(sel)) || (!isAgg && len(stmt.groupBy) > 0 && !stmt.groupByContains(sel)) {
                                        return nil, fmt.Errorf("%s: %w", EncodeSelector(sel.resolve(stmt.Alias())), ErrColumnMustAppearInGroupByOrAggregation)
                                }
                        }
                }
        }
        return tx, nil
}

func (stmt *SelectStmt) targetSelectors() []Selector {
        if stmt.selectors == nil {
                stmt.selectors = stmt.extractSelectors()
        }
        return stmt.selectors
}

func (stmt *SelectStmt) selectorAppearsInTargets(s Selector) bool {
        encSel := EncodeSelector(s.resolve(stmt.Alias()))

        for _, sel := range stmt.targetSelectors() {
                if EncodeSelector(sel.resolve(stmt.Alias())) == encSel {
                        return true
                }
        }
        return false
}

func (stmt *SelectStmt) groupByContains(sel Selector) bool {
        encSel := EncodeSelector(sel.resolve(stmt.Alias()))

        for _, colSel := range stmt.groupBy {
                if EncodeSelector(colSel.resolve(stmt.Alias())) == encSel {
                        return true
                }
        }
        return false
}

func (stmt *SelectStmt) extractGroupByCols() []*AggColSelector {
        cols := make([]*AggColSelector, 0, len(stmt.targets))

        for _, t := range stmt.targets {
                selectors := t.Exp.selectors()
                for _, sel := range selectors {
                        aggSel, isAgg := sel.(*AggColSelector)
                        if isAgg {
                                cols = append(cols, aggSel)
                        }
                }
        }
        return cols
}

func (stmt *SelectStmt) extractSelectors() []Selector {
        selectors := make([]Selector, 0, len(stmt.targets))
        for _, t := range stmt.targets {
                selectors = append(selectors, t.Exp.selectors()...)
        }
        return selectors
}

func (stmt *SelectStmt) Resolve(ctx context.Context, tx *SQLTx, params map[string]interface{}, _ *ScanSpecs) (ret RowReader, err error) {
        scanSpecs, err := stmt.genScanSpecs(tx, params)
        if err != nil {
                return nil, err
        }

        rowReader, err := stmt.ds.Resolve(ctx, tx, params, scanSpecs)
        if err != nil {
                return nil, err
        }
        defer func() {
                if err != nil {
                        rowReader.Close()
                }
        }()

        if stmt.joins != nil {
                var jointRowReader *jointRowReader
                jointRowReader, err = newJointRowReader(rowReader, stmt.joins)
                if err != nil {
                        return nil, err
                }
                rowReader = jointRowReader
        }

        if stmt.where != nil {
                rowReader = newConditionalRowReader(rowReader, stmt.where)
        }

        if stmt.containsAggregations() || len(stmt.groupBy) > 0 {
                if len(scanSpecs.groupBySortExps) > 0 {
                        var sortRowReader *sortRowReader
                        sortRowReader, err = newSortRowReader(rowReader, scanSpecs.groupBySortExps)
                        if err != nil {
                                return nil, err
                        }
                        rowReader = sortRowReader
                }

                var groupedRowReader *groupedRowReader
                groupedRowReader, err = newGroupedRowReader(rowReader, allAggregations(stmt.targets), stmt.extractGroupByCols(), stmt.groupBy)
                if err != nil {
                        return nil, err
                }
                rowReader = groupedRowReader

                if stmt.having != nil {
                        rowReader = newConditionalRowReader(rowReader, stmt.having)
                }
        }

        if len(scanSpecs.orderBySortExps) > 0 {
                var sortRowReader *sortRowReader
                sortRowReader, err = newSortRowReader(rowReader, stmt.orderBy)
                if err != nil {
                        return nil, err
                }
                rowReader = sortRowReader
        }

        projectedRowReader, err := newProjectedRowReader(ctx, rowReader, stmt.as, stmt.targets)
        if err != nil {
                return nil, err
        }
        rowReader = projectedRowReader

        if stmt.distinct {
                var distinctRowReader *distinctRowReader
                distinctRowReader, err = newDistinctRowReader(ctx, rowReader)
                if err != nil {
                        return nil, err
                }
                rowReader = distinctRowReader
        }

        if stmt.offset != nil {
                var offset int
                offset, err = evalExpAsInt(tx, stmt.offset, params)
                if err != nil {
                        return nil, fmt.Errorf("%w: invalid offset", err)
                }

                rowReader = newOffsetRowReader(rowReader, offset)
        }

        if stmt.limit != nil {
                var limit int
                limit, err = evalExpAsInt(tx, stmt.limit, params)
                if err != nil {
                        return nil, fmt.Errorf("%w: invalid limit", err)
                }

                if limit < 0 {
                        return nil, fmt.Errorf("%w: invalid limit", ErrIllegalArguments)
                }

                if limit > 0 {
                        rowReader = newLimitRowReader(rowReader, limit)
                }
        }
        return rowReader, nil
}

func (stmt *SelectStmt) rearrangeOrdExps(groupByCols, orderByExps []*OrdExp) ([]*OrdExp, []*OrdExp) {
        if len(groupByCols) > 0 && len(orderByExps) > 0 && !ordExpsHaveAggregations(orderByExps) {
                if ordExpsHasPrefix(orderByExps, groupByCols, stmt.Alias()) {
                        return orderByExps, nil
                }

                if ordExpsHasPrefix(groupByCols, orderByExps, stmt.Alias()) {
                        for i := range orderByExps {
                                groupByCols[i].descOrder = orderByExps[i].descOrder
                        }
                        return groupByCols, nil
                }
        }
        return groupByCols, orderByExps
}

func ordExpsHasPrefix(cols, prefix []*OrdExp, table string) bool {
        if len(prefix) > len(cols) {
                return false
        }

        for i := range prefix {
                ls := prefix[i].AsSelector()
                rs := cols[i].AsSelector()

                if ls == nil || rs == nil {
                        return false
                }

                if EncodeSelector(ls.resolve(table)) != EncodeSelector(rs.resolve(table)) {
                        return false
                }
        }
        return true
}

func (stmt *SelectStmt) groupByOrdExps() []*OrdExp {
        groupByCols := stmt.groupBy

        ordExps := make([]*OrdExp, 0, len(groupByCols))
        for _, col := range groupByCols {
                ordExps = append(ordExps, &OrdExp{exp: col})
        }
        return ordExps
}

func ordExpsHaveAggregations(exps []*OrdExp) bool {
        for _, e := range exps {
                if _, isAgg := e.exp.(*AggColSelector); isAgg {
                        return true
                }
        }
        return false
}

func (stmt *SelectStmt) containsAggregations() bool {
        for _, sel := range stmt.targetSelectors() {
                _, isAgg := sel.(*AggColSelector)
                if isAgg {
                        return true
                }
        }
        return false
}

func evalExpAsInt(tx *SQLTx, exp ValueExp, params map[string]interface{}) (int, error) {
        offset, err := exp.substitute(params)
        if err != nil {
                return 0, err
        }

        texp, err := offset.reduce(tx, nil, "")
        if err != nil {
                return 0, err
        }

        convVal, err := mayApplyImplicitConversion(texp.RawValue(), IntegerType)
        if err != nil {
                return 0, ErrInvalidValue
        }

        num, ok := convVal.(int64)
        if !ok {
                return 0, ErrInvalidValue
        }

        if num > math.MaxInt32 {
                return 0, ErrInvalidValue
        }

        return int(num), nil
}

func (stmt *SelectStmt) Alias() string {
        if stmt.as == "" {
                return stmt.ds.Alias()
        }

        return stmt.as
}

func (stmt *SelectStmt) hasTxMetadata() bool {
        for _, sel := range stmt.targetSelectors() {
                switch s := sel.(type) {
                case *ColSelector:
                        if s.col == txMetadataCol {
                                return true
                        }
                case *JSONSelector:
                        if s.ColSelector.col == txMetadataCol {
                                return true
                        }
                }
        }
        return false
}

func (stmt *SelectStmt) genScanSpecs(tx *SQLTx, params map[string]interface{}) (*ScanSpecs, error) {
        groupByCols, orderByCols := stmt.groupByOrdExps(), stmt.orderBy

        tableRef, isTableRef := stmt.ds.(*tableRef)
        if !isTableRef {
                groupByCols, orderByCols = stmt.rearrangeOrdExps(groupByCols, orderByCols)

                return &ScanSpecs{
                        groupBySortExps: groupByCols,
                        orderBySortExps: orderByCols,
                }, nil
        }

        table, err := tableRef.referencedTable(tx)
        if err != nil {
                if tx.engine.tableResolveFor(tableRef.table) != nil {
                        return &ScanSpecs{
                                groupBySortExps: groupByCols,
                                orderBySortExps: orderByCols,
                        }, nil
                }
                return nil, err
        }

        rangesByColID := make(map[uint32]*typedValueRange)
        if stmt.where != nil {
                err = stmt.where.selectorRanges(table, tableRef.Alias(), params, rangesByColID)
                if err != nil {
                        return nil, err
                }
        }

        preferredIndex, err := stmt.getPreferredIndex(table)
        if err != nil {
                return nil, err
        }

        var sortingIndex *Index
        if preferredIndex == nil {
                sortingIndex = stmt.selectSortingIndex(groupByCols, orderByCols, table, rangesByColID)
        } else {
                sortingIndex = preferredIndex
        }

        if sortingIndex == nil {
                sortingIndex = table.primaryIndex
        }

        if tableRef.history && !sortingIndex.IsPrimary() {
                return nil, fmt.Errorf("%w: historical queries are supported over primary index", ErrIllegalArguments)
        }

        var descOrder bool
        if len(groupByCols) > 0 && sortingIndex.coversOrdCols(groupByCols, rangesByColID) {
                groupByCols = nil
        }

        if len(groupByCols) == 0 && len(orderByCols) > 0 && sortingIndex.coversOrdCols(orderByCols, rangesByColID) {
                descOrder = orderByCols[0].descOrder
                orderByCols = nil
        }

        groupByCols, orderByCols = stmt.rearrangeOrdExps(groupByCols, orderByCols)

        return &ScanSpecs{
                Index:             sortingIndex,
                rangesByColID:     rangesByColID,
                IncludeHistory:    tableRef.history,
                IncludeTxMetadata: stmt.hasTxMetadata(),
                DescOrder:         descOrder,
                groupBySortExps:   groupByCols,
                orderBySortExps:   orderByCols,
        }, nil
}

func (stmt *SelectStmt) selectSortingIndex(groupByCols, orderByCols []*OrdExp, table *Table, rangesByColId map[uint32]*typedValueRange) *Index {
        sortCols := groupByCols
        if len(sortCols) == 0 {
                sortCols = orderByCols
        }

        if len(sortCols) == 0 {
                return nil
        }

        for _, idx := range table.indexes {
                if idx.coversOrdCols(sortCols, rangesByColId) {
                        return idx
                }
        }
        return nil
}

func (stmt *SelectStmt) getPreferredIndex(table *Table) (*Index, error) {
        if len(stmt.indexOn) == 0 {
                return nil, nil
        }

        cols := make([]*Column, len(stmt.indexOn))
        for i, colName := range stmt.indexOn {
                col, err := table.GetColumnByName(colName)
                if err != nil {
                        return nil, err
                }

                cols[i] = col
        }
        return table.GetIndexByName(indexName(table.name, cols))
}

type UnionStmt struct {
        distinct    bool
        left, right DataSource
}

func (stmt *UnionStmt) readOnly() bool {
        return true
}

func (stmt *UnionStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeSelect}
}

func (stmt *UnionStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        err := stmt.left.inferParameters(ctx, tx, params)
        if err != nil {
                return err
        }
        return stmt.right.inferParameters(ctx, tx, params)
}

func (stmt *UnionStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        _, err := stmt.left.execAt(ctx, tx, params)
        if err != nil {
                return tx, err
        }

        return stmt.right.execAt(ctx, tx, params)
}

func (stmt *UnionStmt) resolveUnionAll(ctx context.Context, tx *SQLTx, params map[string]interface{}) (ret RowReader, err error) {
        leftRowReader, err := stmt.left.Resolve(ctx, tx, params, nil)
        if err != nil {
                return nil, err
        }
        defer func() {
                if err != nil {
                        leftRowReader.Close()
                }
        }()

        rightRowReader, err := stmt.right.Resolve(ctx, tx, params, nil)
        if err != nil {
                return nil, err
        }
        defer func() {
                if err != nil {
                        rightRowReader.Close()
                }
        }()

        rowReader, err := newUnionRowReader(ctx, []RowReader{leftRowReader, rightRowReader})
        if err != nil {
                return nil, err
        }

        return rowReader, nil
}

func (stmt *UnionStmt) Resolve(ctx context.Context, tx *SQLTx, params map[string]interface{}, _ *ScanSpecs) (ret RowReader, err error) {
        rowReader, err := stmt.resolveUnionAll(ctx, tx, params)
        if err != nil {
                return nil, err
        }
        defer func() {
                if err != nil {
                        rowReader.Close()
                }
        }()

        if stmt.distinct {
                distinctReader, err := newDistinctRowReader(ctx, rowReader)
                if err != nil {
                        return nil, err
                }
                rowReader = distinctReader
        }

        return rowReader, nil
}

func (stmt *UnionStmt) Alias() string {
        return ""
}

func NewTableRef(table string, as string) *tableRef {
        return &tableRef{
                table: table,
                as:    as,
        }
}

type tableRef struct {
        table   string
        history bool
        period  period
        as      string
}

func (ref *tableRef) readOnly() bool {
        return true
}

func (ref *tableRef) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeSelect}
}

type period struct {
        start *openPeriod
        end   *openPeriod
}

type openPeriod struct {
        inclusive bool
        instant   periodInstant
}

type periodInstant struct {
        exp         ValueExp
        instantType instantType
}

type instantType = int

const (
        txInstant instantType = iota
        timeInstant
)

func (i periodInstant) resolve(tx *SQLTx, params map[string]interface{}, asc, inclusive bool) (uint64, error) {
        exp, err := i.exp.substitute(params)
        if err != nil {
                return 0, err
        }

        instantVal, err := exp.reduce(tx, nil, "")
        if err != nil {
                return 0, err
        }

        if i.instantType == txInstant {
                txID, ok := instantVal.RawValue().(int64)
                if !ok {
                        return 0, fmt.Errorf("%w: invalid tx range, tx ID must be a positive integer, %s given", ErrIllegalArguments, instantVal.Type())
                }

                if txID <= 0 {
                        return 0, fmt.Errorf("%w: invalid tx range, tx ID must be a positive integer, %d given", ErrIllegalArguments, txID)
                }

                if inclusive {
                        return uint64(txID), nil
                }

                if asc {
                        return uint64(txID + 1), nil
                }

                if txID <= 1 {
                        return 0, fmt.Errorf("%w: invalid tx range, tx ID must be greater than 1, %d given", ErrIllegalArguments, txID)
                }

                return uint64(txID - 1), nil
        } else {

                var ts time.Time

                if instantVal.Type() == TimestampType {
                        ts = instantVal.RawValue().(time.Time)
                } else {
                        conv, err := getConverter(instantVal.Type(), TimestampType)
                        if err != nil {
                                return 0, err
                        }

                        tval, err := conv(instantVal)
                        if err != nil {
                                return 0, err
                        }

                        ts = tval.RawValue().(time.Time)
                }

                sts := ts

                if asc {
                        if !inclusive {
                                sts = sts.Add(1 * time.Second)
                        }

                        txHdr, err := tx.engine.store.FirstTxSince(sts)
                        if err != nil {
                                return 0, err
                        }

                        return txHdr.ID, nil
                }

                if !inclusive {
                        sts = sts.Add(-1 * time.Second)
                }

                txHdr, err := tx.engine.store.LastTxUntil(sts)
                if err != nil {
                        return 0, err
                }

                return txHdr.ID, nil
        }
}

func (stmt *tableRef) referencedTable(tx *SQLTx) (*Table, error) {
        table, err := tx.catalog.GetTableByName(stmt.table)
        if err != nil {
                return nil, err
        }
        return table, nil
}

func (stmt *tableRef) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *tableRef) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        return tx, nil
}

func (stmt *tableRef) Resolve(ctx context.Context, tx *SQLTx, params map[string]interface{}, scanSpecs *ScanSpecs) (RowReader, error) {
        if tx == nil {
                return nil, ErrIllegalArguments
        }

        table, err := stmt.referencedTable(tx)
        if err == nil {
                return newRawRowReader(tx, params, table, stmt.period, stmt.as, scanSpecs)
        }

        if resolver := tx.engine.tableResolveFor(stmt.table); resolver != nil {
                return resolver.Resolve(ctx, tx, stmt.Alias())
        }
        return nil, err
}

func (stmt *tableRef) Alias() string {
        if stmt.as == "" {
                return stmt.table
        }
        return stmt.as
}

type valuesDataSource struct {
        inferTypes bool
        rows       []*RowSpec
}

func NewValuesDataSource(rows []*RowSpec) *valuesDataSource {
        return &valuesDataSource{
                rows: rows,
        }
}

func (ds *valuesDataSource) readOnly() bool {
        return true
}

func (ds *valuesDataSource) requiredPrivileges() []SQLPrivilege {
        return nil
}

func (ds *valuesDataSource) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        return tx, nil
}

func (ds *valuesDataSource) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (ds *valuesDataSource) Alias() string {
        return ""
}

func (ds *valuesDataSource) Resolve(ctx context.Context, tx *SQLTx, params map[string]interface{}, scanSpecs *ScanSpecs) (RowReader, error) {
        if tx == nil {
                return nil, ErrIllegalArguments
        }

        cols := make([]ColDescriptor, len(ds.rows[0].Values))
        for i := range cols {
                cols[i] = ColDescriptor{
                        Type:   AnyType,
                        Column: fmt.Sprintf("col%d", i),
                }
        }

        emptyColsDesc, emptyParams := map[string]ColDescriptor{}, map[string]string{}

        if ds.inferTypes {
                for i := 0; i < len(cols); i++ {
                        t := AnyType
                        for j := 0; j < len(ds.rows); j++ {
                                e, err := ds.rows[j].Values[i].substitute(params)
                                if err != nil {
                                        return nil, err
                                }

                                it, err := e.inferType(emptyColsDesc, emptyParams, "")
                                if err != nil {
                                        return nil, err
                                }

                                if t == AnyType {
                                        t = it
                                } else if t != it && it != AnyType {
                                        return nil, fmt.Errorf("cannot match types %s and %s", t, it)
                                }
                        }
                        cols[i].Type = t
                }
        }

        values := make([][]ValueExp, len(ds.rows))
        for i, rowSpec := range ds.rows {
                values[i] = rowSpec.Values
        }
        return NewValuesRowReader(tx, params, cols, ds.inferTypes, "values", values)
}

type JoinSpec struct {
        joinType JoinType
        ds       DataSource
        cond     ValueExp
        indexOn  []string
}

type OrdExp struct {
        exp       ValueExp
        descOrder bool
}

func (oc *OrdExp) AsSelector() Selector {
        sel, ok := oc.exp.(Selector)
        if ok {
                return sel
        }
        return nil
}

func NewOrdCol(table string, col string, descOrder bool) *OrdExp {
        return &OrdExp{
                exp:       NewColSelector(table, col),
                descOrder: descOrder,
        }
}

type Selector interface {
        ValueExp
        resolve(implicitTable string) (aggFn, table, col string)
}

type ColSelector struct {
        table string
        col   string
}

func NewColSelector(table, col string) *ColSelector {
        return &ColSelector{
                table: table,
                col:   col,
        }
}

func (sel *ColSelector) resolve(implicitTable string) (aggFn, table, col string) {
        table = implicitTable
        if sel.table != "" {
                table = sel.table
        }
        return "", table, sel.col
}

func (sel *ColSelector) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        _, table, col := sel.resolve(implicitTable)
        encSel := EncodeSelector("", table, col)

        desc, ok := cols[encSel]
        if !ok {
                return AnyType, fmt.Errorf("%w (%s)", ErrColumnDoesNotExist, col)
        }
        return desc.Type, nil
}

func (sel *ColSelector) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        _, table, col := sel.resolve(implicitTable)
        encSel := EncodeSelector("", table, col)

        desc, ok := cols[encSel]
        if !ok {
                return fmt.Errorf("%w (%s)", ErrColumnDoesNotExist, col)
        }

        if desc.Type != t {
                return fmt.Errorf("%w: %v(%s) can not be interpreted as type %v", ErrInvalidTypes, desc.Type, encSel, t)
        }

        return nil
}

func (sel *ColSelector) substitute(params map[string]interface{}) (ValueExp, error) {
        return sel, nil
}

func (sel *ColSelector) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        if row == nil {
                return nil, fmt.Errorf("%w: no row to evaluate in current context", ErrInvalidValue)
        }

        aggFn, table, col := sel.resolve(implicitTable)

        v, ok := row.ValuesBySelector[EncodeSelector(aggFn, table, col)]
        if !ok {
                return nil, fmt.Errorf("%w (%s)", ErrColumnDoesNotExist, col)
        }
        return v, nil
}

func (sel *ColSelector) selectors() []Selector {
        return []Selector{sel}
}

func (sel *ColSelector) reduceSelectors(row *Row, implicitTable string) ValueExp {
        aggFn, table, col := sel.resolve(implicitTable)

        v, ok := row.ValuesBySelector[EncodeSelector(aggFn, table, col)]
        if !ok {
                return sel
        }

        return v
}

func (sel *ColSelector) isConstant() bool {
        return false
}

func (sel *ColSelector) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (sel *ColSelector) String() string {
        return sel.col
}

type AggColSelector struct {
        aggFn AggregateFn
        table string
        col   string
}

func NewAggColSelector(aggFn AggregateFn, table, col string) *AggColSelector {
        return &AggColSelector{
                aggFn: aggFn,
                table: table,
                col:   col,
        }
}

func EncodeSelector(aggFn, table, col string) string {
        return aggFn + "(" + table + "." + col + ")"
}

func (sel *AggColSelector) resolve(implicitTable string) (aggFn, table, col string) {
        table = implicitTable
        if sel.table != "" {
                table = sel.table
        }
        return sel.aggFn, table, sel.col
}

func (sel *AggColSelector) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        if sel.aggFn == COUNT {
                return IntegerType, nil
        }

        colSelector := &ColSelector{table: sel.table, col: sel.col}

        if sel.aggFn == SUM || sel.aggFn == AVG {
                t, err := colSelector.inferType(cols, params, implicitTable)
                if err != nil {
                        return AnyType, err
                }

                if t != IntegerType && t != Float64Type {
                        return AnyType, fmt.Errorf("%w: %v or %v can not be interpreted as type %v", ErrInvalidTypes, IntegerType, Float64Type, t)

                }

                return t, nil
        }

        return colSelector.inferType(cols, params, implicitTable)
}

func (sel *AggColSelector) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if sel.aggFn == COUNT {
                if t != IntegerType {
                        return fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, IntegerType, t)
                }
                return nil
        }

        colSelector := &ColSelector{table: sel.table, col: sel.col}

        if sel.aggFn == SUM || sel.aggFn == AVG {
                if t != IntegerType && t != Float64Type {
                        return fmt.Errorf("%w: %v or %v can not be interpreted as type %v", ErrInvalidTypes, IntegerType, Float64Type, t)
                }
        }

        return colSelector.requiresType(t, cols, params, implicitTable)
}

func (sel *AggColSelector) substitute(params map[string]interface{}) (ValueExp, error) {
        return sel, nil
}

func (sel *AggColSelector) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        if row == nil {
                return nil, fmt.Errorf("%w: no row to evaluate aggregation (%s) in current context", ErrInvalidValue, sel.aggFn)
        }

        v, ok := row.ValuesBySelector[EncodeSelector(sel.resolve(implicitTable))]
        if !ok {
                return nil, fmt.Errorf("%w (%s)", ErrColumnDoesNotExist, sel.col)
        }
        return v, nil
}

func (sel *AggColSelector) selectors() []Selector {
        return []Selector{sel}
}

func (sel *AggColSelector) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return sel
}

func (sel *AggColSelector) isConstant() bool {
        return false
}

func (sel *AggColSelector) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (sel *AggColSelector) String() string {
        return sel.aggFn + "(" + sel.col + ")"
}

type NumExp struct {
        op          NumOperator
        left, right ValueExp
}

func (bexp *NumExp) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        // First step - check if we can infer the type of sub-expressions
        tleft, err := bexp.left.inferType(cols, params, implicitTable)
        if err != nil {
                return AnyType, err
        }
        if tleft != AnyType && tleft != IntegerType && tleft != Float64Type && tleft != JSONType {
                return AnyType, fmt.Errorf("%w: %v or %v can not be interpreted as type %v", ErrInvalidTypes, IntegerType, Float64Type, tleft)
        }

        tright, err := bexp.right.inferType(cols, params, implicitTable)
        if err != nil {
                return AnyType, err
        }
        if tright != AnyType && tright != IntegerType && tright != Float64Type && tright != JSONType {
                return AnyType, fmt.Errorf("%w: %v or %v can not be interpreted as type %v", ErrInvalidTypes, IntegerType, Float64Type, tright)
        }

        if tleft == IntegerType && tright == IntegerType {
                // Both sides are integer types - the result is also integer
                return IntegerType, nil
        }

        if tleft != AnyType && tright != AnyType {
                // Both sides have concrete types but at least one of them is float
                return Float64Type, nil
        }

        // Both sides are ambiguous
        return AnyType, nil
}

func copyParams(params map[string]SQLValueType) map[string]SQLValueType {
        ret := make(map[string]SQLValueType, len(params))
        for k, v := range params {
                ret[k] = v
        }
        return ret
}

func restoreParams(params, restore map[string]SQLValueType) {
        for k := range params {
                delete(params, k)
        }
        for k, v := range restore {
                params[k] = v
        }
}

func (bexp *NumExp) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if t != IntegerType && t != Float64Type {
                return fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, IntegerType, t)
        }

        floatArgs := 2
        paramsOrig := copyParams(params)
        err := bexp.left.requiresType(t, cols, params, implicitTable)
        if err != nil && t == Float64Type {
                restoreParams(params, paramsOrig)
                floatArgs--
                err = bexp.left.requiresType(IntegerType, cols, params, implicitTable)
        }
        if err != nil {
                return err
        }

        paramsOrig = copyParams(params)
        err = bexp.right.requiresType(t, cols, params, implicitTable)
        if err != nil && t == Float64Type {
                restoreParams(params, paramsOrig)
                floatArgs--
                err = bexp.right.requiresType(IntegerType, cols, params, implicitTable)
        }
        if err != nil {
                return err
        }

        if t == Float64Type && floatArgs == 0 {
                // Currently this case requires explicit float cast
                return fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, IntegerType, t)
        }

        return nil
}

func (bexp *NumExp) substitute(params map[string]interface{}) (ValueExp, error) {
        rlexp, err := bexp.left.substitute(params)
        if err != nil {
                return nil, err
        }

        rrexp, err := bexp.right.substitute(params)
        if err != nil {
                return nil, err
        }

        bexp.left = rlexp
        bexp.right = rrexp

        return bexp, nil
}

func (bexp *NumExp) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        vl, err := bexp.left.reduce(tx, row, implicitTable)
        if err != nil {
                return nil, err
        }

        vr, err := bexp.right.reduce(tx, row, implicitTable)
        if err != nil {
                return nil, err
        }

        vl = unwrapJSON(vl)
        vr = unwrapJSON(vr)

        return applyNumOperator(bexp.op, vl, vr)
}

func unwrapJSON(v TypedValue) TypedValue {
        if jsonVal, ok := v.(*JSON); ok {
                if sv, isSimple := jsonVal.castToTypedValue(); isSimple {
                        return sv
                }
        }
        return v
}

func (bexp *NumExp) selectors() []Selector {
        return append(bexp.left.selectors(), bexp.right.selectors()...)
}

func (bexp *NumExp) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return &NumExp{
                op:    bexp.op,
                left:  bexp.left.reduceSelectors(row, implicitTable),
                right: bexp.right.reduceSelectors(row, implicitTable),
        }
}

func (bexp *NumExp) isConstant() bool {
        return bexp.left.isConstant() && bexp.right.isConstant()
}

func (bexp *NumExp) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (bexp *NumExp) String() string {
        return fmt.Sprintf("(%s %s %s)", bexp.left.String(), NumOperatorString(bexp.op), bexp.right.String())
}

type NotBoolExp struct {
        exp ValueExp
}

func (bexp *NotBoolExp) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        err := bexp.exp.requiresType(BooleanType, cols, params, implicitTable)
        if err != nil {
                return AnyType, err
        }

        return BooleanType, nil
}

func (bexp *NotBoolExp) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if t != BooleanType {
                return fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, BooleanType, t)
        }

        return bexp.exp.requiresType(BooleanType, cols, params, implicitTable)
}

func (bexp *NotBoolExp) substitute(params map[string]interface{}) (ValueExp, error) {
        rexp, err := bexp.exp.substitute(params)
        if err != nil {
                return nil, err
        }

        bexp.exp = rexp

        return bexp, nil
}

func (bexp *NotBoolExp) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        v, err := bexp.exp.reduce(tx, row, implicitTable)
        if err != nil {
                return nil, err
        }

        r, isBool := v.RawValue().(bool)
        if !isBool {
                return nil, ErrInvalidCondition
        }

        return &Bool{val: !r}, nil
}

func (bexp *NotBoolExp) selectors() []Selector {
        return bexp.exp.selectors()
}

func (bexp *NotBoolExp) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return &NotBoolExp{
                exp: bexp.exp.reduceSelectors(row, implicitTable),
        }
}

func (bexp *NotBoolExp) isConstant() bool {
        return bexp.exp.isConstant()
}

func (bexp *NotBoolExp) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (bexp *NotBoolExp) String() string {
        return fmt.Sprintf("(NOT %s)", bexp.exp.String())
}

type LikeBoolExp struct {
        val     ValueExp
        notLike bool
        pattern ValueExp
}

func NewLikeBoolExp(val ValueExp, notLike bool, pattern ValueExp) *LikeBoolExp {
        return &LikeBoolExp{
                val:     val,
                notLike: notLike,
                pattern: pattern,
        }
}

func (bexp *LikeBoolExp) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        if bexp.val == nil || bexp.pattern == nil {
                return AnyType, fmt.Errorf("error in 'LIKE' clause: %w", ErrInvalidCondition)
        }

        err := bexp.pattern.requiresType(VarcharType, cols, params, implicitTable)
        if err != nil {
                return AnyType, fmt.Errorf("error in 'LIKE' clause: %w", err)
        }

        return BooleanType, nil
}

func (bexp *LikeBoolExp) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if bexp.val == nil || bexp.pattern == nil {
                return fmt.Errorf("error in 'LIKE' clause: %w", ErrInvalidCondition)
        }

        if t != BooleanType {
                return fmt.Errorf("error using the value of the LIKE operator as %s: %w", t, ErrInvalidTypes)
        }

        err := bexp.pattern.requiresType(VarcharType, cols, params, implicitTable)
        if err != nil {
                return fmt.Errorf("error in 'LIKE' clause: %w", err)
        }

        return nil
}

func (bexp *LikeBoolExp) substitute(params map[string]interface{}) (ValueExp, error) {
        if bexp.val == nil || bexp.pattern == nil {
                return nil, fmt.Errorf("error in 'LIKE' clause: %w", ErrInvalidCondition)
        }

        val, err := bexp.val.substitute(params)
        if err != nil {
                return nil, fmt.Errorf("error in 'LIKE' clause: %w", err)
        }

        pattern, err := bexp.pattern.substitute(params)
        if err != nil {
                return nil, fmt.Errorf("error in 'LIKE' clause: %w", err)
        }

        return &LikeBoolExp{
                val:     val,
                notLike: bexp.notLike,
                pattern: pattern,
        }, nil
}

func (bexp *LikeBoolExp) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        if bexp.val == nil || bexp.pattern == nil {
                return nil, fmt.Errorf("error in 'LIKE' clause: %w", ErrInvalidCondition)
        }

        rval, err := bexp.val.reduce(tx, row, implicitTable)
        if err != nil {
                return nil, fmt.Errorf("error in 'LIKE' clause: %w", err)
        }

        if rval.IsNull() {
                return &Bool{val: bexp.notLike}, nil
        }

        rvalStr, ok := rval.RawValue().(string)
        if !ok {
                return nil, fmt.Errorf("error in 'LIKE' clause: %w (expecting %s)", ErrInvalidTypes, VarcharType)
        }

        rpattern, err := bexp.pattern.reduce(tx, row, implicitTable)
        if err != nil {
                return nil, fmt.Errorf("error in 'LIKE' clause: %w", err)
        }

        if rpattern.Type() != VarcharType {
                return nil, fmt.Errorf("error evaluating 'LIKE' clause: %w", ErrInvalidTypes)
        }

        matched, err := regexp.MatchString(rpattern.RawValue().(string), rvalStr)
        if err != nil {
                return nil, fmt.Errorf("error in 'LIKE' clause: %w", err)
        }

        return &Bool{val: matched != bexp.notLike}, nil
}

func (bexp *LikeBoolExp) selectors() []Selector {
        return bexp.val.selectors()
}

func (bexp *LikeBoolExp) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return bexp
}

func (bexp *LikeBoolExp) isConstant() bool {
        return false
}

func (bexp *LikeBoolExp) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (bexp *LikeBoolExp) String() string {
        fmtStr := "(%s LIKE %s)"
        if bexp.notLike {
                fmtStr = "(%s NOT LIKE %s)"
        }
        return fmt.Sprintf(fmtStr, bexp.val.String(), bexp.pattern.String())
}

type CmpBoolExp struct {
        op          CmpOperator
        left, right ValueExp
}

func NewCmpBoolExp(op CmpOperator, left, right ValueExp) *CmpBoolExp {
        return &CmpBoolExp{
                op:    op,
                left:  left,
                right: right,
        }
}

func (bexp *CmpBoolExp) Left() ValueExp {
        return bexp.left
}

func (bexp *CmpBoolExp) Right() ValueExp {
        return bexp.right
}

func (bexp *CmpBoolExp) OP() CmpOperator {
        return bexp.op
}

func (bexp *CmpBoolExp) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        tleft, err := bexp.left.inferType(cols, params, implicitTable)
        if err != nil {
                return AnyType, err
        }

        tright, err := bexp.right.inferType(cols, params, implicitTable)
        if err != nil {
                return AnyType, err
        }

        // unification step

        if tleft == tright {
                return BooleanType, nil
        }

        _, ok := coerceTypes(tleft, tright)
        if !ok {
                return AnyType, fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, tleft, tright)
        }

        if tleft == AnyType {
                err = bexp.left.requiresType(tright, cols, params, implicitTable)
                if err != nil {
                        return AnyType, err
                }
        }

        if tright == AnyType {
                err = bexp.right.requiresType(tleft, cols, params, implicitTable)
                if err != nil {
                        return AnyType, err
                }
        }
        return BooleanType, nil
}

func coerceTypes(t1, t2 SQLValueType) (SQLValueType, bool) {
        switch {
        case t1 == t2:
                return t1, true
        case t1 == AnyType:
                return t2, true
        case t2 == AnyType:
                return t1, true
        case (t1 == IntegerType && t2 == Float64Type) ||
                (t1 == Float64Type && t2 == IntegerType):
                return Float64Type, true
        }
        return "", false
}

func (bexp *CmpBoolExp) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if t != BooleanType {
                return fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, BooleanType, t)
        }

        _, err := bexp.inferType(cols, params, implicitTable)
        return err
}

func (bexp *CmpBoolExp) substitute(params map[string]interface{}) (ValueExp, error) {
        rlexp, err := bexp.left.substitute(params)
        if err != nil {
                return nil, err
        }

        rrexp, err := bexp.right.substitute(params)
        if err != nil {
                return nil, err
        }

        bexp.left = rlexp
        bexp.right = rrexp

        return bexp, nil
}

func (bexp *CmpBoolExp) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        vl, err := bexp.left.reduce(tx, row, implicitTable)
        if err != nil {
                return nil, err
        }

        vr, err := bexp.right.reduce(tx, row, implicitTable)
        if err != nil {
                return nil, err
        }

        r, err := vl.Compare(vr)
        if err != nil {
                return nil, err
        }

        return &Bool{val: cmpSatisfiesOp(r, bexp.op)}, nil
}

func (bexp *CmpBoolExp) selectors() []Selector {
        return append(bexp.left.selectors(), bexp.right.selectors()...)
}

func (bexp *CmpBoolExp) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return &CmpBoolExp{
                op:    bexp.op,
                left:  bexp.left.reduceSelectors(row, implicitTable),
                right: bexp.right.reduceSelectors(row, implicitTable),
        }
}

func (bexp *CmpBoolExp) isConstant() bool {
        return bexp.left.isConstant() && bexp.right.isConstant()
}

func (bexp *CmpBoolExp) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        matchingFunc := func(_, right ValueExp) (*ColSelector, ValueExp, bool) {
                s, isSel := bexp.left.(*ColSelector)
                if isSel && s.col != revCol && bexp.right.isConstant() {
                        return s, right, true
                }
                return nil, nil, false
        }

        sel, c, ok := matchingFunc(bexp.left, bexp.right)
        if !ok {
                sel, c, ok = matchingFunc(bexp.right, bexp.left)
        }

        if !ok {
                return nil
        }

        aggFn, t, col := sel.resolve(table.name)
        if aggFn != "" || t != asTable {
                return nil
        }

        column, err := table.GetColumnByName(col)
        if err != nil {
                return err
        }

        val, err := c.substitute(params)
        if errors.Is(err, ErrMissingParameter) {
                // TODO: not supported when parameters are not provided during query resolution
                return nil
        }
        if err != nil {
                return err
        }

        rval, err := val.reduce(nil, nil, table.name)
        if err != nil {
                return err
        }

        return updateRangeFor(column.id, rval, bexp.op, rangesByColID)
}

func (bexp *CmpBoolExp) String() string {
        opStr := CmpOperatorToString(bexp.op)
        return fmt.Sprintf("(%s %s %s)", bexp.left.String(), opStr, bexp.right.String())
}

func updateRangeFor(colID uint32, val TypedValue, cmp CmpOperator, rangesByColID map[uint32]*typedValueRange) error {
        currRange, ranged := rangesByColID[colID]
        var newRange *typedValueRange

        switch cmp {
        case EQ:
                {
                        newRange = &typedValueRange{
                                lRange: &typedValueSemiRange{
                                        val:       val,
                                        inclusive: true,
                                },
                                hRange: &typedValueSemiRange{
                                        val:       val,
                                        inclusive: true,
                                },
                        }
                }
        case LT:
                {
                        newRange = &typedValueRange{
                                hRange: &typedValueSemiRange{
                                        val: val,
                                },
                        }
                }
        case LE:
                {
                        newRange = &typedValueRange{
                                hRange: &typedValueSemiRange{
                                        val:       val,
                                        inclusive: true,
                                },
                        }
                }
        case GT:
                {
                        newRange = &typedValueRange{
                                lRange: &typedValueSemiRange{
                                        val: val,
                                },
                        }
                }
        case GE:
                {
                        newRange = &typedValueRange{
                                lRange: &typedValueSemiRange{
                                        val:       val,
                                        inclusive: true,
                                },
                        }
                }
        case NE:
                {
                        return nil
                }
        }

        if !ranged {
                rangesByColID[colID] = newRange
                return nil
        }

        return currRange.refineWith(newRange)
}

func cmpSatisfiesOp(cmp int, op CmpOperator) bool {
        switch {
        case cmp == 0:
                {
                        return op == EQ || op == LE || op == GE
                }
        case cmp < 0:
                {
                        return op == NE || op == LT || op == LE
                }
        case cmp > 0:
                {
                        return op == NE || op == GT || op == GE
                }
        }
        return false
}

type BinBoolExp struct {
        op          LogicOperator
        left, right ValueExp
}

func NewBinBoolExp(op LogicOperator, lrexp, rrexp ValueExp) *BinBoolExp {
        bexp := &BinBoolExp{
                op: op,
        }

        bexp.left = lrexp
        bexp.right = rrexp

        return bexp
}

func (bexp *BinBoolExp) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        err := bexp.left.requiresType(BooleanType, cols, params, implicitTable)
        if err != nil {
                return AnyType, err
        }

        err = bexp.right.requiresType(BooleanType, cols, params, implicitTable)
        if err != nil {
                return AnyType, err
        }

        return BooleanType, nil
}

func (bexp *BinBoolExp) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        if t != BooleanType {
                return fmt.Errorf("%w: %v can not be interpreted as type %v", ErrInvalidTypes, BooleanType, t)
        }

        err := bexp.left.requiresType(BooleanType, cols, params, implicitTable)
        if err != nil {
                return err
        }

        err = bexp.right.requiresType(BooleanType, cols, params, implicitTable)
        if err != nil {
                return err
        }

        return nil
}

func (bexp *BinBoolExp) substitute(params map[string]interface{}) (ValueExp, error) {
        rlexp, err := bexp.left.substitute(params)
        if err != nil {
                return nil, err
        }

        rrexp, err := bexp.right.substitute(params)
        if err != nil {
                return nil, err
        }

        bexp.left = rlexp
        bexp.right = rrexp

        return bexp, nil
}

func (bexp *BinBoolExp) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        vl, err := bexp.left.reduce(tx, row, implicitTable)
        if err != nil {
                return nil, err
        }

        bl, isBool := vl.(*Bool)
        if !isBool {
                return nil, fmt.Errorf("%w (expecting boolean value)", ErrInvalidValue)
        }

        // short-circuit evaluation
        if (bl.val && bexp.op == Or) || (!bl.val && bexp.op == And) {
                return &Bool{val: bl.val}, nil
        }

        vr, err := bexp.right.reduce(tx, row, implicitTable)
        if err != nil {
                return nil, err
        }

        br, isBool := vr.(*Bool)
        if !isBool {
                return nil, fmt.Errorf("%w (expecting boolean value)", ErrInvalidValue)
        }

        switch bexp.op {
        case And:
                {
                        return &Bool{val: bl.val && br.val}, nil
                }
        case Or:
                {
                        return &Bool{val: bl.val || br.val}, nil
                }
        }

        return nil, ErrUnexpected
}

func (bexp *BinBoolExp) selectors() []Selector {
        return append(bexp.left.selectors(), bexp.right.selectors()...)
}

func (bexp *BinBoolExp) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return &BinBoolExp{
                op:    bexp.op,
                left:  bexp.left.reduceSelectors(row, implicitTable),
                right: bexp.right.reduceSelectors(row, implicitTable),
        }
}

func (bexp *BinBoolExp) isConstant() bool {
        return bexp.left.isConstant() && bexp.right.isConstant()
}

func (bexp *BinBoolExp) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        if bexp.op == And {
                err := bexp.left.selectorRanges(table, asTable, params, rangesByColID)
                if err != nil {
                        return err
                }

                return bexp.right.selectorRanges(table, asTable, params, rangesByColID)
        }

        lRanges := make(map[uint32]*typedValueRange)
        rRanges := make(map[uint32]*typedValueRange)

        err := bexp.left.selectorRanges(table, asTable, params, lRanges)
        if err != nil {
                return err
        }

        err = bexp.right.selectorRanges(table, asTable, params, rRanges)
        if err != nil {
                return err
        }

        for colID, lr := range lRanges {
                rr, ok := rRanges[colID]
                if !ok {
                        continue
                }

                err = lr.extendWith(rr)
                if err != nil {
                        return err
                }

                rangesByColID[colID] = lr
        }

        return nil
}

func (bexp *BinBoolExp) String() string {
        return fmt.Sprintf("(%s %s %s)", bexp.left.String(), LogicOperatorToString(bexp.op), bexp.right.String())
}

type ExistsBoolExp struct {
        q DataSource
}

func (bexp *ExistsBoolExp) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        return AnyType, fmt.Errorf("error inferring type in 'EXISTS' clause: %w", ErrNoSupported)
}

func (bexp *ExistsBoolExp) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        return fmt.Errorf("error inferring type in 'EXISTS' clause: %w", ErrNoSupported)
}

func (bexp *ExistsBoolExp) substitute(params map[string]interface{}) (ValueExp, error) {
        return bexp, nil
}

func (bexp *ExistsBoolExp) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        return nil, fmt.Errorf("'EXISTS' clause: %w", ErrNoSupported)
}

func (bexp *ExistsBoolExp) selectors() []Selector {
        return nil
}

func (bexp *ExistsBoolExp) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return bexp
}

func (bexp *ExistsBoolExp) isConstant() bool {
        return false
}

func (bexp *ExistsBoolExp) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (bexp *ExistsBoolExp) String() string {
        return ""
}

type InSubQueryExp struct {
        val   ValueExp
        notIn bool
        q     *SelectStmt
}

func (bexp *InSubQueryExp) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        return AnyType, fmt.Errorf("error inferring type in 'IN' clause: %w", ErrNoSupported)
}

func (bexp *InSubQueryExp) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        return fmt.Errorf("error inferring type in 'IN' clause: %w", ErrNoSupported)
}

func (bexp *InSubQueryExp) substitute(params map[string]interface{}) (ValueExp, error) {
        return bexp, nil
}

func (bexp *InSubQueryExp) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        return nil, fmt.Errorf("error inferring type in 'IN' clause: %w", ErrNoSupported)
}

func (bexp *InSubQueryExp) selectors() []Selector {
        return bexp.val.selectors()
}

func (bexp *InSubQueryExp) reduceSelectors(row *Row, implicitTable string) ValueExp {
        return bexp
}

func (bexp *InSubQueryExp) isConstant() bool {
        return false
}

func (bexp *InSubQueryExp) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        return nil
}

func (bexp *InSubQueryExp) String() string {
        return ""
}

// TODO: once InSubQueryExp is supported, this struct may become obsolete by creating a ListDataSource struct
type InListExp struct {
        val    ValueExp
        notIn  bool
        values []ValueExp
}

func (bexp *InListExp) inferType(cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) (SQLValueType, error) {
        t, err := bexp.val.inferType(cols, params, implicitTable)
        if err != nil {
                return AnyType, fmt.Errorf("error inferring type in 'IN' clause: %w", err)
        }

        for _, v := range bexp.values {
                err = v.requiresType(t, cols, params, implicitTable)
                if err != nil {
                        return AnyType, fmt.Errorf("error inferring type in 'IN' clause: %w", err)
                }
        }

        return BooleanType, nil
}

func (bexp *InListExp) requiresType(t SQLValueType, cols map[string]ColDescriptor, params map[string]SQLValueType, implicitTable string) error {
        _, err := bexp.inferType(cols, params, implicitTable)
        if err != nil {
                return err
        }

        if t != BooleanType {
                return fmt.Errorf("error inferring type in 'IN' clause: %w", ErrInvalidTypes)
        }

        return nil
}

func (bexp *InListExp) substitute(params map[string]interface{}) (ValueExp, error) {
        val, err := bexp.val.substitute(params)
        if err != nil {
                return nil, fmt.Errorf("error evaluating 'IN' clause: %w", err)
        }

        values := make([]ValueExp, len(bexp.values))

        for i, val := range bexp.values {
                values[i], err = val.substitute(params)
                if err != nil {
                        return nil, fmt.Errorf("error evaluating 'IN' clause: %w", err)
                }
        }

        return &InListExp{
                val:    val,
                notIn:  bexp.notIn,
                values: values,
        }, nil
}

func (bexp *InListExp) reduce(tx *SQLTx, row *Row, implicitTable string) (TypedValue, error) {
        rval, err := bexp.val.reduce(tx, row, implicitTable)
        if err != nil {
                return nil, fmt.Errorf("error evaluating 'IN' clause: %w", err)
        }

        var found bool

        for _, v := range bexp.values {
                rv, err := v.reduce(tx, row, implicitTable)
                if err != nil {
                        return nil, fmt.Errorf("error evaluating 'IN' clause: %w", err)
                }

                r, err := rval.Compare(rv)
                if err != nil {
                        return nil, fmt.Errorf("error evaluating 'IN' clause: %w", err)
                }

                if r == 0 {
                        // TODO: short-circuit evaluation may be preferred when upfront static type inference is in place
                        found = found || true
                }
        }

        return &Bool{val: found != bexp.notIn}, nil
}

func (bexp *InListExp) selectors() []Selector {
        selectors := make([]Selector, 0, len(bexp.values))
        for _, v := range bexp.values {
                selectors = append(selectors, v.selectors()...)
        }
        return append(bexp.val.selectors(), selectors...)
}

func (bexp *InListExp) reduceSelectors(row *Row, implicitTable string) ValueExp {
        values := make([]ValueExp, len(bexp.values))

        for i, val := range bexp.values {
                values[i] = val.reduceSelectors(row, implicitTable)
        }

        return &InListExp{
                val:    bexp.val.reduceSelectors(row, implicitTable),
                values: values,
        }
}

func (bexp *InListExp) isConstant() bool {
        return false
}

func (bexp *InListExp) selectorRanges(table *Table, asTable string, params map[string]interface{}, rangesByColID map[uint32]*typedValueRange) error {
        // TODO: may be determiined by smallest and bigggest value in the list
        return nil
}

func (bexp *InListExp) String() string {
        values := make([]string, len(bexp.values))
        for i, exp := range bexp.values {
                values[i] = exp.String()
        }
        return fmt.Sprintf("%s IN (%s)", bexp.val.String(), strings.Join(values, ","))
}

type FnDataSourceStmt struct {
        fnCall *FnCall
        as     string
}

func (stmt *FnDataSourceStmt) readOnly() bool {
        return true
}

func (stmt *FnDataSourceStmt) requiredPrivileges() []SQLPrivilege {
        return nil
}

func (stmt *FnDataSourceStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        return tx, nil
}

func (stmt *FnDataSourceStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

func (stmt *FnDataSourceStmt) Alias() string {
        if stmt.as != "" {
                return stmt.as
        }

        switch strings.ToUpper(stmt.fnCall.fn) {
        case DatabasesFnCall:
                {
                        return "databases"
                }
        case TablesFnCall:
                {
                        return "tables"
                }
        case TableFnCall:
                {
                        return "table"
                }
        case UsersFnCall:
                {
                        return "users"
                }
        case ColumnsFnCall:
                {
                        return "columns"
                }
        case IndexesFnCall:
                {
                        return "indexes"
                }
        case GrantsFnCall:
                return "grants"
        }

        // not reachable
        return ""
}

func (stmt *FnDataSourceStmt) Resolve(ctx context.Context, tx *SQLTx, params map[string]interface{}, scanSpecs *ScanSpecs) (rowReader RowReader, err error) {
        if stmt.fnCall == nil {
                return nil, fmt.Errorf("%w: function is unspecified", ErrIllegalArguments)
        }

        switch strings.ToUpper(stmt.fnCall.fn) {
        case DatabasesFnCall:
                {
                        return stmt.resolveListDatabases(ctx, tx, params, scanSpecs)
                }
        case TablesFnCall:
                {
                        return stmt.resolveListTables(ctx, tx, params, scanSpecs)
                }
        case TableFnCall:
                {
                        return stmt.resolveShowTable(ctx, tx, params, scanSpecs)
                }
        case UsersFnCall:
                {
                        return stmt.resolveListUsers(ctx, tx, params, scanSpecs)
                }
        case ColumnsFnCall:
                {
                        return stmt.resolveListColumns(ctx, tx, params, scanSpecs)
                }
        case IndexesFnCall:
                {
                        return stmt.resolveListIndexes(ctx, tx, params, scanSpecs)
                }
        case GrantsFnCall:
                {
                        return stmt.resolveListGrants(ctx, tx, params, scanSpecs)
                }
        }

        return nil, fmt.Errorf("%w (%s)", ErrFunctionDoesNotExist, stmt.fnCall.fn)
}

func (stmt *FnDataSourceStmt) resolveListDatabases(ctx context.Context, tx *SQLTx, params map[string]interface{}, _ *ScanSpecs) (rowReader RowReader, err error) {
        if len(stmt.fnCall.params) > 0 {
                return nil, fmt.Errorf("%w: function '%s' expect no parameters but %d were provided", ErrIllegalArguments, DatabasesFnCall, len(stmt.fnCall.params))
        }

        cols := make([]ColDescriptor, 1)
        cols[0] = ColDescriptor{
                Column: "name",
                Type:   VarcharType,
        }

        var dbs []string

        if tx.engine.multidbHandler == nil {
                return nil, ErrUnspecifiedMultiDBHandler
        } else {
                dbs, err = tx.engine.multidbHandler.ListDatabases(ctx)
                if err != nil {
                        return nil, err
                }
        }

        values := make([][]ValueExp, len(dbs))

        for i, db := range dbs {
                values[i] = []ValueExp{&Varchar{val: db}}
        }

        return NewValuesRowReader(tx, params, cols, true, stmt.Alias(), values)
}

func (stmt *FnDataSourceStmt) resolveListTables(ctx context.Context, tx *SQLTx, params map[string]interface{}, _ *ScanSpecs) (rowReader RowReader, err error) {
        if len(stmt.fnCall.params) > 0 {
                return nil, fmt.Errorf("%w: function '%s' expect no parameters but %d were provided", ErrIllegalArguments, TablesFnCall, len(stmt.fnCall.params))
        }

        cols := make([]ColDescriptor, 1)
        cols[0] = ColDescriptor{
                Column: "name",
                Type:   VarcharType,
        }

        tables := tx.catalog.GetTables()

        values := make([][]ValueExp, len(tables))

        for i, t := range tables {
                values[i] = []ValueExp{&Varchar{val: t.name}}
        }

        return NewValuesRowReader(tx, params, cols, true, stmt.Alias(), values)
}

func (stmt *FnDataSourceStmt) resolveShowTable(ctx context.Context, tx *SQLTx, params map[string]interface{}, _ *ScanSpecs) (rowReader RowReader, err error) {
        cols := []ColDescriptor{
                {
                        Column: "column_name",
                        Type:   VarcharType,
                },
                {
                        Column: "type_name",
                        Type:   VarcharType,
                },
                {
                        Column: "is_nullable",
                        Type:   BooleanType,
                },
                {
                        Column: "is_indexed",
                        Type:   VarcharType,
                },
                {
                        Column: "is_auto_increment",
                        Type:   BooleanType,
                },
                {
                        Column: "is_unique",
                        Type:   BooleanType,
                },
        }

        tableName, _ := stmt.fnCall.params[0].reduce(tx, nil, "")
        table, err := tx.catalog.GetTableByName(tableName.RawValue().(string))
        if err != nil {
                return nil, err
        }

        values := make([][]ValueExp, len(table.cols))

        for i, c := range table.cols {
                index := "NO"

                indexed, err := table.IsIndexed(c.Name())
                if err != nil {
                        return nil, err
                }
                if indexed {
                        index = "YES"
                }

                if table.PrimaryIndex().IncludesCol(c.ID()) {
                        index = "PRIMARY KEY"
                }

                var unique bool
                for _, index := range table.GetIndexesByColID(c.ID()) {
                        if index.IsUnique() && len(index.Cols()) == 1 {
                                unique = true
                                break
                        }
                }

                var maxLen string

                if c.MaxLen() > 0 && (c.Type() == VarcharType || c.Type() == BLOBType) {
                        maxLen = fmt.Sprintf("(%d)", c.MaxLen())
                }

                values[i] = []ValueExp{
                        &Varchar{val: c.colName},
                        &Varchar{val: c.Type() + maxLen},
                        &Bool{val: c.IsNullable()},
                        &Varchar{val: index},
                        &Bool{val: c.IsAutoIncremental()},
                        &Bool{val: unique},
                }
        }

        return NewValuesRowReader(tx, params, cols, true, stmt.Alias(), values)
}

func (stmt *FnDataSourceStmt) resolveListUsers(ctx context.Context, tx *SQLTx, params map[string]interface{}, _ *ScanSpecs) (rowReader RowReader, err error) {
        if len(stmt.fnCall.params) > 0 {
                return nil, fmt.Errorf("%w: function '%s' expect no parameters but %d were provided", ErrIllegalArguments, UsersFnCall, len(stmt.fnCall.params))
        }

        cols := []ColDescriptor{
                {
                        Column: "name",
                        Type:   VarcharType,
                },
                {
                        Column: "permission",
                        Type:   VarcharType,
                },
        }

        users, err := tx.ListUsers(ctx)
        if err != nil {
                return nil, err
        }

        values := make([][]ValueExp, len(users))
        for i, user := range users {
                perm := user.Permission()

                values[i] = []ValueExp{
                        &Varchar{val: user.Username()},
                        &Varchar{val: perm},
                }
        }
        return NewValuesRowReader(tx, params, cols, true, stmt.Alias(), values)
}

func (stmt *FnDataSourceStmt) resolveListColumns(ctx context.Context, tx *SQLTx, params map[string]interface{}, _ *ScanSpecs) (RowReader, error) {
        if len(stmt.fnCall.params) != 1 {
                return nil, fmt.Errorf("%w: function '%s' expect table name as parameter", ErrIllegalArguments, ColumnsFnCall)
        }

        cols := []ColDescriptor{
                {
                        Column: "table",
                        Type:   VarcharType,
                },
                {
                        Column: "name",
                        Type:   VarcharType,
                },
                {
                        Column: "type",
                        Type:   VarcharType,
                },
                {
                        Column: "max_length",
                        Type:   IntegerType,
                },
                {
                        Column: "nullable",
                        Type:   BooleanType,
                },
                {
                        Column: "auto_increment",
                        Type:   BooleanType,
                },
                {
                        Column: "indexed",
                        Type:   BooleanType,
                },
                {
                        Column: "primary",
                        Type:   BooleanType,
                },
                {
                        Column: "unique",
                        Type:   BooleanType,
                },
        }

        val, err := stmt.fnCall.params[0].substitute(params)
        if err != nil {
                return nil, err
        }

        tableName, err := val.reduce(tx, nil, "")
        if err != nil {
                return nil, err
        }

        if tableName.Type() != VarcharType {
                return nil, fmt.Errorf("%w: expected '%s' for table name but type '%s' given instead", ErrIllegalArguments, VarcharType, tableName.Type())
        }

        table, err := tx.catalog.GetTableByName(tableName.RawValue().(string))
        if err != nil {
                return nil, err
        }

        values := make([][]ValueExp, len(table.cols))

        for i, c := range table.cols {
                indexed, err := table.IsIndexed(c.Name())
                if err != nil {
                        return nil, err
                }

                var unique bool
                for _, index := range table.indexesByColID[c.id] {
                        if index.IsUnique() && len(index.Cols()) == 1 {
                                unique = true
                                break
                        }
                }

                values[i] = []ValueExp{
                        &Varchar{val: table.name},
                        &Varchar{val: c.colName},
                        &Varchar{val: c.colType},
                        &Integer{val: int64(c.MaxLen())},
                        &Bool{val: c.IsNullable()},
                        &Bool{val: c.autoIncrement},
                        &Bool{val: indexed},
                        &Bool{val: table.PrimaryIndex().IncludesCol(c.ID())},
                        &Bool{val: unique},
                }
        }

        return NewValuesRowReader(tx, params, cols, true, stmt.Alias(), values)
}

func (stmt *FnDataSourceStmt) resolveListIndexes(ctx context.Context, tx *SQLTx, params map[string]interface{}, _ *ScanSpecs) (RowReader, error) {
        if len(stmt.fnCall.params) != 1 {
                return nil, fmt.Errorf("%w: function '%s' expect table name as parameter", ErrIllegalArguments, IndexesFnCall)
        }

        cols := []ColDescriptor{
                {
                        Column: "table",
                        Type:   VarcharType,
                },
                {
                        Column: "name",
                        Type:   VarcharType,
                },
                {
                        Column: "unique",
                        Type:   BooleanType,
                },
                {
                        Column: "primary",
                        Type:   BooleanType,
                },
        }

        val, err := stmt.fnCall.params[0].substitute(params)
        if err != nil {
                return nil, err
        }

        tableName, err := val.reduce(tx, nil, "")
        if err != nil {
                return nil, err
        }

        if tableName.Type() != VarcharType {
                return nil, fmt.Errorf("%w: expected '%s' for table name but type '%s' given instead", ErrIllegalArguments, VarcharType, tableName.Type())
        }

        table, err := tx.catalog.GetTableByName(tableName.RawValue().(string))
        if err != nil {
                return nil, err
        }

        values := make([][]ValueExp, len(table.indexes))

        for i, index := range table.indexes {
                values[i] = []ValueExp{
                        &Varchar{val: table.name},
                        &Varchar{val: index.Name()},
                        &Bool{val: index.unique},
                        &Bool{val: index.IsPrimary()},
                }
        }

        return NewValuesRowReader(tx, params, cols, true, stmt.Alias(), values)
}

func (stmt *FnDataSourceStmt) resolveListGrants(ctx context.Context, tx *SQLTx, params map[string]interface{}, _ *ScanSpecs) (RowReader, error) {
        if len(stmt.fnCall.params) > 1 {
                return nil, fmt.Errorf("%w: function '%s' expect at most one parameter of type %s", ErrIllegalArguments, GrantsFnCall, VarcharType)
        }

        var username string
        if len(stmt.fnCall.params) == 1 {
                val, err := stmt.fnCall.params[0].substitute(params)
                if err != nil {
                        return nil, err
                }

                userVal, err := val.reduce(tx, nil, "")
                if err != nil {
                        return nil, err
                }

                if userVal.Type() != VarcharType {
                        return nil, fmt.Errorf("%w: expected '%s' for username but type '%s' given instead", ErrIllegalArguments, VarcharType, userVal.Type())
                }
                username, _ = userVal.RawValue().(string)
        }

        cols := []ColDescriptor{
                {
                        Column: "user",
                        Type:   VarcharType,
                },
                {
                        Column: "privilege",
                        Type:   VarcharType,
                },
        }

        var err error
        var users []User

        if tx.engine.multidbHandler == nil {
                return nil, ErrUnspecifiedMultiDBHandler
        } else {
                users, err = tx.engine.multidbHandler.ListUsers(ctx)
                if err != nil {
                        return nil, err
                }
        }

        values := make([][]ValueExp, 0, len(users))

        for _, user := range users {
                if username == "" || user.Username() == username {
                        for _, p := range user.SQLPrivileges() {
                                values = append(values, []ValueExp{
                                        &Varchar{val: user.Username()},
                                        &Varchar{val: string(p)},
                                })
                        }
                }
        }

        return NewValuesRowReader(tx, params, cols, true, stmt.Alias(), values)
}

// DropTableStmt represents a statement to delete a table.
type DropTableStmt struct {
        table string
}

func NewDropTableStmt(table string) *DropTableStmt {
        return &DropTableStmt{table: table}
}

func (stmt *DropTableStmt) readOnly() bool {
        return false
}

func (stmt *DropTableStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeDrop}
}

func (stmt *DropTableStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

/*
Exec executes the delete table statement.
It the table exists, if not it does nothing.
If the table exists, it deletes all the indexes and the table itself.
Note that this is a soft delete of the index and table key,
the data is not deleted, but the metadata is updated.
*/
func (stmt *DropTableStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if !tx.catalog.ExistTable(stmt.table) {
                return nil, ErrTableDoesNotExist
        }

        table, err := tx.catalog.GetTableByName(stmt.table)
        if err != nil {
                return nil, err
        }

        // delete table
        mappedKey := MapKey(
                tx.sqlPrefix(),
                catalogTablePrefix,
                EncodeID(DatabaseID),
                EncodeID(table.id),
        )
        err = tx.delete(ctx, mappedKey)
        if err != nil {
                return nil, err
        }

        // delete columns
        cols := table.ColumnsByID()
        for _, col := range cols {
                mappedKey := MapKey(
                        tx.sqlPrefix(),
                        catalogColumnPrefix,
                        EncodeID(DatabaseID),
                        EncodeID(col.table.id),
                        EncodeID(col.id),
                        []byte(col.colType),
                )
                err = tx.delete(ctx, mappedKey)
                if err != nil {
                        return nil, err
                }
        }

        // delete checks
        for name := range table.checkConstraints {
                key := MapKey(
                        tx.sqlPrefix(),
                        catalogCheckPrefix,
                        EncodeID(DatabaseID),
                        EncodeID(table.id),
                        []byte(name),
                )

                if err := tx.delete(ctx, key); err != nil {
                        return nil, err
                }
        }

        // delete indexes
        for _, index := range table.indexes {
                mappedKey := MapKey(
                        tx.sqlPrefix(),
                        catalogIndexPrefix,
                        EncodeID(DatabaseID),
                        EncodeID(table.id),
                        EncodeID(index.id),
                )
                err = tx.delete(ctx, mappedKey)
                if err != nil {
                        return nil, err
                }

                indexKey := MapKey(
                        tx.sqlPrefix(),
                        MappedPrefix,
                        EncodeID(table.id),
                        EncodeID(index.id),
                )
                err = tx.addOnCommittedCallback(func(sqlTx *SQLTx) error {
                        return sqlTx.engine.store.DeleteIndex(indexKey)
                })
                if err != nil {
                        return nil, err
                }
        }

        err = tx.catalog.deleteTable(table)
        if err != nil {
                return nil, err
        }

        tx.mutatedCatalog = true

        return tx, nil
}

// DropIndexStmt represents a statement to delete a table.
type DropIndexStmt struct {
        table string
        cols  []string
}

func NewDropIndexStmt(table string, cols []string) *DropIndexStmt {
        return &DropIndexStmt{table: table, cols: cols}
}

func (stmt *DropIndexStmt) readOnly() bool {
        return false
}

func (stmt *DropIndexStmt) requiredPrivileges() []SQLPrivilege {
        return []SQLPrivilege{SQLPrivilegeDrop}
}

func (stmt *DropIndexStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

/*
Exec executes the delete index statement.
If the index exists, it deletes it. Note that this is a soft delete of the index
the data is not deleted, but the metadata is updated.
*/
func (stmt *DropIndexStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if !tx.catalog.ExistTable(stmt.table) {
                return nil, ErrTableDoesNotExist
        }

        table, err := tx.catalog.GetTableByName(stmt.table)
        if err != nil {
                return nil, err
        }

        cols := make([]*Column, len(stmt.cols))

        for i, colName := range stmt.cols {
                col, err := table.GetColumnByName(colName)
                if err != nil {
                        return nil, err
                }

                cols[i] = col
        }

        index, err := table.GetIndexByName(indexName(table.name, cols))
        if err != nil {
                return nil, err
        }

        // delete index
        mappedKey := MapKey(
                tx.sqlPrefix(),
                catalogIndexPrefix,
                EncodeID(DatabaseID),
                EncodeID(table.id),
                EncodeID(index.id),
        )
        err = tx.delete(ctx, mappedKey)
        if err != nil {
                return nil, err
        }

        indexKey := MapKey(
                tx.sqlPrefix(),
                MappedPrefix,
                EncodeID(table.id),
                EncodeID(index.id),
        )

        err = tx.addOnCommittedCallback(func(sqlTx *SQLTx) error {
                return sqlTx.engine.store.DeleteIndex(indexKey)
        })
        if err != nil {
                return nil, err
        }

        err = table.deleteIndex(index)
        if err != nil {
                return nil, err
        }

        tx.mutatedCatalog = true

        return tx, nil
}

type SQLPrivilege string

const (
        SQLPrivilegeSelect SQLPrivilege = "SELECT"
        SQLPrivilegeCreate SQLPrivilege = "CREATE"
        SQLPrivilegeInsert SQLPrivilege = "INSERT"
        SQLPrivilegeUpdate SQLPrivilege = "UPDATE"
        SQLPrivilegeDelete SQLPrivilege = "DELETE"
        SQLPrivilegeDrop   SQLPrivilege = "DROP"
        SQLPrivilegeAlter  SQLPrivilege = "ALTER"
)

var allPrivileges = []SQLPrivilege{
        SQLPrivilegeSelect,
        SQLPrivilegeCreate,
        SQLPrivilegeInsert,
        SQLPrivilegeUpdate,
        SQLPrivilegeDelete,
        SQLPrivilegeDrop,
        SQLPrivilegeAlter,
}

func DefaultSQLPrivilegesForPermission(p Permission) []SQLPrivilege {
        switch p {
        case PermissionSysAdmin, PermissionAdmin, PermissionReadWrite:
                return allPrivileges
        case PermissionReadOnly:
                return []SQLPrivilege{SQLPrivilegeSelect}
        }
        return nil
}

type AlterPrivilegesStmt struct {
        database   string
        user       string
        privileges []SQLPrivilege
        isGrant    bool
}

func (stmt *AlterPrivilegesStmt) readOnly() bool {
        return false
}

func (stmt *AlterPrivilegesStmt) requiredPrivileges() []SQLPrivilege {
        return nil
}

func (stmt *AlterPrivilegesStmt) execAt(ctx context.Context, tx *SQLTx, params map[string]interface{}) (*SQLTx, error) {
        if tx.IsExplicitCloseRequired() {
                return nil, fmt.Errorf("%w: user privileges modification can not be done within a transaction", ErrNonTransactionalStmt)
        }

        if tx.engine.multidbHandler == nil {
                return nil, ErrUnspecifiedMultiDBHandler
        }

        var err error
        if stmt.isGrant {
                err = tx.engine.multidbHandler.GrantSQLPrivileges(ctx, stmt.database, stmt.user, stmt.privileges)
        } else {
                err = tx.engine.multidbHandler.RevokeSQLPrivileges(ctx, stmt.database, stmt.user, stmt.privileges)
        }
        return nil, err
}

func (stmt *AlterPrivilegesStmt) inferParameters(ctx context.Context, tx *SQLTx, params map[string]SQLValueType) error {
        return nil
}

codenotary / immudb / 12273047947

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