24841644892

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

Build # 24841644892

Build Type

push

gh-ci

Committed by

web-flow

Commit Message

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

* Add structured audit logging with immutable audit trail

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

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

* Add PostgreSQL ORM compatibility layer and verification functions

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

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

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

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

* Implement EXISTS and IN subquery support in SQL engine

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

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

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

Coverage Stats

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

115 existing lines in 18 files now uncovered.

44599 of 52298 relevant lines covered (85.28%)

127676.6 hits per line

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

91.67

/embedded/sql/system_tables.go

/*
Copyright 2026 Codenotary Inc. All rights reserved.

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

    https://mariadb.com/bsl11/

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

package sql

import (
        "context"
        "fmt"
        "sort"
        "sync"
)

// SystemTableDef describes a catalog-registered virtual table whose
// schema is visible to the SQL engine but whose rows come from Go code
// rather than persistent storage. Used to expose PG-compatibility
// catalogs (pg_type today; pg_class, pg_attribute, pg_index, etc. in
// future phases) without requiring DDL at server start.
//
// Registered definitions are installed into every Catalog on creation
// (see newCatalog) and survive Catalog.Clone(). They are effectively
// immutable: the registry is a package-level map populated via
// RegisterSystemTable at init() time.
type SystemTableDef struct {
        // Name is the SQL name used to reference the table (e.g. "pg_type").
        // Must be unique across all registrations.
        Name string

        // Columns is the table's column schema. Must be non-empty. The
        // ValuesByPosition slices of Rows returned by Scan must match this
        // order 1:1.
        Columns []SystemTableColumn

        // PKColumn is the name of the primary-key column. Must appear in
        // Columns. The PK index is created as a unique single-column index.
        // System tables that don't have a natural single-column PK should
        // synthesise one (typically a stable oid).
        PKColumn string

        // Scan returns the rows for a query against this table. When nil,
        // the table appears empty at the SQL engine layer — the historic
        // behaviour for pg_type, where rows are synthesised by the PG wire
        // layer instead.
        //
        // The returned Rows' ValuesByPosition must match Columns order;
        // ValuesBySelector is populated by newSystemTableRowReader from
        // ValuesByPosition, so Scan implementations can leave it nil.
        Scan func(ctx context.Context, tx *SQLTx) ([]*Row, error)
}

// SystemTableColumn describes a column in a SystemTableDef.
type SystemTableColumn struct {
        Name   string
        Type   SQLValueType
        MaxLen int
}

var (
        sysTablesMu sync.RWMutex
        sysTables   = map[string]*SystemTableDef{}
)

// RegisterSystemTable registers a system-table definition. Panics on
// duplicate Name or malformed definition — errors here are programmer
// bugs, not runtime conditions. Call at package init().
func RegisterSystemTable(def *SystemTableDef) {
        if def == nil {
                panic("sql: RegisterSystemTable: nil definition")
        }
        if def.Name == "" {
                panic("sql: RegisterSystemTable: empty Name")
        }
        if len(def.Columns) == 0 {
                panic(fmt.Sprintf("sql: RegisterSystemTable(%s): no columns", def.Name))
        }
        if def.PKColumn == "" {
                panic(fmt.Sprintf("sql: RegisterSystemTable(%s): empty PKColumn", def.Name))
        }
        pkFound := false
        for _, c := range def.Columns {
                if c.Name == def.PKColumn {
                        pkFound = true
                        break
                }
        }
        if !pkFound {
                panic(fmt.Sprintf("sql: RegisterSystemTable(%s): PKColumn %q not in Columns",
                        def.Name, def.PKColumn))
        }

        sysTablesMu.Lock()
        defer sysTablesMu.Unlock()
        if _, dup := sysTables[def.Name]; dup {
                panic(fmt.Sprintf("sql: RegisterSystemTable(%s): already registered", def.Name))
        }
        sysTables[def.Name] = def
}

// registeredSystemTables returns the registry snapshot in
// name-sorted order so catalog construction is deterministic across
// restarts (affects test stability and any future oid-hash assignment
// that depends on iteration order).
func registeredSystemTables() []*SystemTableDef {
        sysTablesMu.RLock()
        defer sysTablesMu.RUnlock()

        out := make([]*SystemTableDef, 0, len(sysTables))
        for _, d := range sysTables {
                out = append(out, d)
        }
        sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
        return out
}

// installSystemTable materialises def as a *Table on catlg. The table
// is installed only into tablesByName (not tables/tablesByID) so user
// DDL enumeration remains unaffected — matching the historic pg_type
// treatment. The PK index is single-column, unique, with id
// PKIndexID; this mirrors how user tables build their primary index
// via newTable, and keeps SELECT planning (which expects every table
// to have a primaryIndex) working uniformly.
func installSystemTable(catlg *Catalog, def *SystemTableDef) {
        // A nil Scan from the def means "zero rows at the SQL engine
        // layer" (the historic pg_type contract — rows are fabricated by
        // the PG wire layer). Install a sentinel no-op Scan so the
        // systemScan field is always non-nil on system tables; call sites
        // use that pointer as a "is this a system table?" marker.
        scan := def.Scan
        if scan == nil {
                scan = func(ctx context.Context, tx *SQLTx) ([]*Row, error) {
                        return nil, nil
                }
        }

        t := &Table{
                catalog:    catlg,
                name:       def.Name,
                cols:       make([]*Column, 0, len(def.Columns)),
                colsByID:   make(map[uint32]*Column, len(def.Columns)),
                colsByName: make(map[string]*Column, len(def.Columns)),
                systemScan: scan,
        }

        for i, cSpec := range def.Columns {
                c := &Column{
                        table:   t,
                        id:      uint32(i + 1),
                        colName: cSpec.Name,
                        colType: cSpec.Type,
                        maxLen:  cSpec.MaxLen,
                }
                t.cols = append(t.cols, c)
                t.colsByID[c.id] = c
                t.colsByName[c.colName] = c
        }
        t.maxColID = uint32(len(def.Columns))

        pkCol := t.colsByName[def.PKColumn]
        pkIdx := &Index{
                table:    t,
                id:       PKIndexID,
                unique:   true,
                cols:     []*Column{pkCol},
                colsByID: map[uint32]*Column{pkCol.id: pkCol},
        }
        t.indexes = []*Index{pkIdx}
        t.indexesByName = map[string]*Index{pkIdx.Name(): pkIdx}
        t.indexesByColID = map[uint32][]*Index{pkCol.id: {pkIdx}}
        t.primaryIndex = pkIdx

        catlg.tablesByName[def.Name] = t
}

// systemTableRowReader implements RowReader by iterating a
// pre-computed slice of rows. Used for SELECTs against system tables
// registered with a non-nil Scan. The rows are materialised eagerly
// in tableRef.Resolve so Scan sees a live SQLTx, then streamed one at
// a time here.
type systemTableRowReader struct {
        tx         *SQLTx
        tableAlias string
        colsByPos  []ColDescriptor
        colsBySel  map[string]ColDescriptor
        rows       []*Row
        pos        int

        onCloseCallback func()
        closed          bool
}

func newSystemTableRowReader(tx *SQLTx, table *Table, alias string, rows []*Row) (*systemTableRowReader, error) {
        if alias == "" {
                alias = table.name
        }

        colsByPos := make([]ColDescriptor, len(table.cols))
        colsBySel := make(map[string]ColDescriptor, len(table.cols))
        for i, c := range table.cols {
                d := ColDescriptor{
                        Table:  alias,
                        Column: c.colName,
                        Type:   c.colType,
                }
                colsByPos[i] = d
                colsBySel[d.Selector()] = d
        }

        // Populate ValuesBySelector for each row from ValuesByPosition.
        // Scan implementations only need to fill ValuesByPosition, which
        // keeps their call sites short.
        for _, r := range rows {
                if len(r.ValuesByPosition) != len(colsByPos) {
                        return nil, fmt.Errorf("%w: system table %q row has %d values, expected %d",
                                ErrInvalidNumberOfValues, table.name, len(r.ValuesByPosition), len(colsByPos))
                }
                if r.ValuesBySelector == nil {
                        r.ValuesBySelector = make(map[string]TypedValue, len(colsByPos))
                }
                for i, v := range r.ValuesByPosition {
                        r.ValuesBySelector[colsByPos[i].Selector()] = v
                }
        }

        return &systemTableRowReader{
                tx:         tx,
                tableAlias: alias,
                colsByPos:  colsByPos,
                colsBySel:  colsBySel,
                rows:       rows,
        }, nil
}

func (r *systemTableRowReader) Tx() *SQLTx                         { return r.tx }
func (r *systemTableRowReader) TableAlias() string                 { return r.tableAlias }
func (r *systemTableRowReader) Parameters() map[string]interface{} { return nil }
func (r *systemTableRowReader) OrderBy() []ColDescriptor           { return nil }
func (r *systemTableRowReader) ScanSpecs() *ScanSpecs              { return nil }

func (r *systemTableRowReader) Columns(ctx context.Context) ([]ColDescriptor, error) {
        return r.colsByPos, nil
}

func (r *systemTableRowReader) colsBySelector(ctx context.Context) (map[string]ColDescriptor, error) {
        return r.colsBySel, nil
}

func (r *systemTableRowReader) InferParameters(ctx context.Context, params map[string]SQLValueType) error {
        return nil
}

func (r *systemTableRowReader) Read(ctx context.Context) (*Row, error) {
        if r.pos >= len(r.rows) {
                return nil, ErrNoMoreRows
        }
        row := r.rows[r.pos]
        r.pos++
        return row, nil
}

func (r *systemTableRowReader) Close() error {
        if r.closed {
                return ErrAlreadyClosed
        }
        r.closed = true
        if r.onCloseCallback != nil {
                r.onCloseCallback()
        }
        return nil
}

func (r *systemTableRowReader) onClose(cb func()) { r.onCloseCallback = cb }

1	/*
2	Copyright 2026 Codenotary Inc. All rights reserved.
3
4	SPDX-License-Identifier: BUSL-1.1
5	you may not use this file except in compliance with the License.
6	You may obtain a copy of the License at
7
8	https://mariadb.com/bsl11/
9
10	Unless required by applicable law or agreed to in writing, software
11	distributed under the License is distributed on an "AS IS" BASIS,
12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	See the License for the specific language governing permissions and
14	limitations under the License.
15	*/
16
17	package sql
18
19	import (
20	"context"
21	"fmt"
22	"sort"
23	"sync"
24	)
25
26	// SystemTableDef describes a catalog-registered virtual table whose
27	// schema is visible to the SQL engine but whose rows come from Go code
28	// rather than persistent storage. Used to expose PG-compatibility
29	// catalogs (pg_type today; pg_class, pg_attribute, pg_index, etc. in
30	// future phases) without requiring DDL at server start.
31	//
32	// Registered definitions are installed into every Catalog on creation
33	// (see newCatalog) and survive Catalog.Clone(). They are effectively
34	// immutable: the registry is a package-level map populated via
35	// RegisterSystemTable at init() time.
36	type SystemTableDef struct {
37	// Name is the SQL name used to reference the table (e.g. "pg_type").
38	// Must be unique across all registrations.
39	Name string
40
41	// Columns is the table's column schema. Must be non-empty. The
42	// ValuesByPosition slices of Rows returned by Scan must match this
43	// order 1:1.
44	Columns []SystemTableColumn
45
46	// PKColumn is the name of the primary-key column. Must appear in
47	// Columns. The PK index is created as a unique single-column index.
48	// System tables that don't have a natural single-column PK should
49	// synthesise one (typically a stable oid).
50	PKColumn string
51
52	// Scan returns the rows for a query against this table. When nil,
53	// the table appears empty at the SQL engine layer — the historic
54	// behaviour for pg_type, where rows are synthesised by the PG wire
55	// layer instead.
56	//
57	// The returned Rows' ValuesByPosition must match Columns order;
58	// ValuesBySelector is populated by newSystemTableRowReader from
59	// ValuesByPosition, so Scan implementations can leave it nil.
60	Scan func(ctx context.Context, tx SQLTx) ([]Row, error)
61	}
62
63	// SystemTableColumn describes a column in a SystemTableDef.
64	type SystemTableColumn struct {
65	Name string
66	Type SQLValueType
67	MaxLen int
68	}
69
70	var (
71	sysTablesMu sync.RWMutex
72	sysTables = map[string]*SystemTableDef{}
73	)
74
75	// RegisterSystemTable registers a system-table definition. Panics on
76	// duplicate Name or malformed definition — errors here are programmer
77	// bugs, not runtime conditions. Call at package init().
78	func RegisterSystemTable(def *SystemTableDef) {	379✔
79	if def == nil {	380✔
80	panic("sql: RegisterSystemTable: nil definition")	1✔
81	}
82	if def.Name == "" {	379✔
83	panic("sql: RegisterSystemTable: empty Name")	1✔
84	}
85	if len(def.Columns) == 0 {	378✔
86	panic(fmt.Sprintf("sql: RegisterSystemTable(%s): no columns", def.Name))	1✔
87	}
88	if def.PKColumn == "" {	377✔
89	panic(fmt.Sprintf("sql: RegisterSystemTable(%s): empty PKColumn", def.Name))	1✔
90	}
91	pkFound := false	375✔
92	for _, c := range def.Columns {	2,141✔
93	if c.Name == def.PKColumn {	2,140✔
94	pkFound = true	374✔
95	break	374✔
96	}
97	}
98	if !pkFound {	376✔
99	panic(fmt.Sprintf("sql: RegisterSystemTable(%s): PKColumn %q not in Columns",	1✔
100	def.Name, def.PKColumn))	1✔
101	}
102
103	sysTablesMu.Lock()	374✔
104	defer sysTablesMu.Unlock()	374✔
105	if _, dup := sysTables[def.Name]; dup {	375✔
106	panic(fmt.Sprintf("sql: RegisterSystemTable(%s): already registered", def.Name))	1✔
107	}
108	sysTables[def.Name] = def	373✔
109	}
110
111	// registeredSystemTables returns the registry snapshot in
112	// name-sorted order so catalog construction is deterministic across
113	// restarts (affects test stability and any future oid-hash assignment
114	// that depends on iteration order).
115	func registeredSystemTables() []*SystemTableDef {	5,149✔
116	sysTablesMu.RLock()	5,149✔
117	defer sysTablesMu.RUnlock()	5,149✔
118		5,149✔
119	out := make([]*SystemTableDef, 0, len(sysTables))	5,149✔
120	for _, d := range sysTables {	28,735✔
121	out = append(out, d)	23,586✔
122	}	23,586✔
123	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })	90,165✔
124	return out	5,149✔
125	}
126
127	// installSystemTable materialises def as a *Table on catlg. The table
128	// is installed only into tablesByName (not tables/tablesByID) so user
129	// DDL enumeration remains unaffected — matching the historic pg_type
130	// treatment. The PK index is single-column, unique, with id
131	// PKIndexID; this mirrors how user tables build their primary index
132	// via newTable, and keeps SELECT planning (which expects every table
133	// to have a primaryIndex) working uniformly.
134	func installSystemTable(catlg Catalog, def SystemTableDef) {	23,586✔
135	// A nil Scan from the def means "zero rows at the SQL engine	23,586✔
136	// layer" (the historic pg_type contract — rows are fabricated by	23,586✔
137	// the PG wire layer). Install a sentinel no-op Scan so the	23,586✔
138	// systemScan field is always non-nil on system tables; call sites	23,586✔
139	// use that pointer as a "is this a system table?" marker.	23,586✔
140	scan := def.Scan	23,586✔
141	if scan == nil {	23,586✔
NEW 142	scan = func(ctx context.Context, tx SQLTx) ([]Row, error) {	×
NEW 143	return nil, nil	×
NEW 144	}	×
145	}
146
147	t := &Table{	23,586✔
148	catalog: catlg,	23,586✔
149	name: def.Name,	23,586✔
150	cols: make([]*Column, 0, len(def.Columns)),	23,586✔
151	colsByID: make(map[uint32]*Column, len(def.Columns)),	23,586✔
152	colsByName: make(map[string]*Column, len(def.Columns)),	23,586✔
153	systemScan: scan,	23,586✔
154	}	23,586✔
155		23,586✔
156	for i, cSpec := range def.Columns {	260,956✔
157	c := &Column{	237,370✔
158	table: t,	237,370✔
159	id: uint32(i + 1),	237,370✔
160	colName: cSpec.Name,	237,370✔
161	colType: cSpec.Type,	237,370✔
162	maxLen: cSpec.MaxLen,	237,370✔
163	}	237,370✔
164	t.cols = append(t.cols, c)	237,370✔
165	t.colsByID[c.id] = c	237,370✔
166	t.colsByName[c.colName] = c	237,370✔
167	}	237,370✔
168	t.maxColID = uint32(len(def.Columns))	23,586✔
169		23,586✔
170	pkCol := t.colsByName[def.PKColumn]	23,586✔
171	pkIdx := &Index{	23,586✔
172	table: t,	23,586✔
173	id: PKIndexID,	23,586✔
174	unique: true,	23,586✔
175	cols: []*Column{pkCol},	23,586✔
176	colsByID: map[uint32]*Column{pkCol.id: pkCol},	23,586✔
177	}	23,586✔
178	t.indexes = []*Index{pkIdx}	23,586✔
179	t.indexesByName = map[string]*Index{pkIdx.Name(): pkIdx}	23,586✔
180	t.indexesByColID = map[uint32][]*Index{pkCol.id: {pkIdx}}	23,586✔
181	t.primaryIndex = pkIdx	23,586✔
182		23,586✔
183	catlg.tablesByName[def.Name] = t	23,586✔
184	}
185
186	// systemTableRowReader implements RowReader by iterating a
187	// pre-computed slice of rows. Used for SELECTs against system tables
188	// registered with a non-nil Scan. The rows are materialised eagerly
189	// in tableRef.Resolve so Scan sees a live SQLTx, then streamed one at
190	// a time here.
191	type systemTableRowReader struct {
192	tx *SQLTx
193	tableAlias string
194	colsByPos []ColDescriptor
195	colsBySel map[string]ColDescriptor
196	rows []*Row
197	pos int
198
199	onCloseCallback func()
200	closed bool
201	}
202
203	func newSystemTableRowReader(tx SQLTx, table Table, alias string, rows []Row) (systemTableRowReader, error) {	146✔
204	if alias == "" {	146✔
NEW 205	alias = table.name	×
NEW 206	}	×
207
208	colsByPos := make([]ColDescriptor, len(table.cols))	146✔
209	colsBySel := make(map[string]ColDescriptor, len(table.cols))	146✔
210	for i, c := range table.cols {	2,447✔
211	d := ColDescriptor{	2,301✔
212	Table: alias,	2,301✔
213	Column: c.colName,	2,301✔
214	Type: c.colType,	2,301✔
215	}	2,301✔
216	colsByPos[i] = d	2,301✔
217	colsBySel[d.Selector()] = d	2,301✔
218	}	2,301✔
219
220	// Populate ValuesBySelector for each row from ValuesByPosition.
221	// Scan implementations only need to fill ValuesByPosition, which
222	// keeps their call sites short.
223	for _, r := range rows {	976✔
224	if len(r.ValuesByPosition) != len(colsByPos) {	830✔
NEW 225	return nil, fmt.Errorf("%w: system table %q row has %d values, expected %d",	×
NEW 226	ErrInvalidNumberOfValues, table.name, len(r.ValuesByPosition), len(colsByPos))	×
NEW 227	}	×
228	if r.ValuesBySelector == nil {	1,660✔
229	r.ValuesBySelector = make(map[string]TypedValue, len(colsByPos))	830✔
230	}	830✔
231	for i, v := range r.ValuesByPosition {	14,110✔
232	r.ValuesBySelector[colsByPos[i].Selector()] = v	13,280✔
233	}	13,280✔
234	}
235
236	return &systemTableRowReader{	146✔
237	tx: tx,	146✔
238	tableAlias: alias,	146✔
239	colsByPos: colsByPos,	146✔
240	colsBySel: colsBySel,	146✔
241	rows: rows,	146✔
242	}, nil	146✔
243	}
244
245	func (r systemTableRowReader) Tx() SQLTx { return r.tx }	1,719✔
246	func (r *systemTableRowReader) TableAlias() string { return r.tableAlias }	5,030✔
247	func (r *systemTableRowReader) Parameters() map[string]interface{} { return nil }	1,369✔
NEW 248	func (r *systemTableRowReader) OrderBy() []ColDescriptor { return nil }	×
NEW 249	func (r systemTableRowReader) ScanSpecs() ScanSpecs { return nil }	×
250
251	func (r *systemTableRowReader) Columns(ctx context.Context) ([]ColDescriptor, error) {	68✔
252	return r.colsByPos, nil	68✔
253	}	68✔
254
255	func (r *systemTableRowReader) colsBySelector(ctx context.Context) (map[string]ColDescriptor, error) {	119✔
256	return r.colsBySel, nil	119✔
257	}	119✔
258
259	func (r *systemTableRowReader) InferParameters(ctx context.Context, params map[string]SQLValueType) error {	19✔
260	return nil	19✔
261	}	19✔
262
263	func (r systemTableRowReader) Read(ctx context.Context) (Row, error) {	618✔
264	if r.pos >= len(r.rows) {	701✔
265	return nil, ErrNoMoreRows	83✔
266	}	83✔
267	row := r.rows[r.pos]	535✔
268	r.pos++	535✔
269	return row, nil	535✔
270	}
271
272	func (r *systemTableRowReader) Close() error {	146✔
273	if r.closed {	146✔
NEW 274	return ErrAlreadyClosed	×
NEW 275	}	×
276	r.closed = true	146✔
277	if r.onCloseCallback != nil {	218✔
278	r.onCloseCallback()	72✔
279	}	72✔
280	return nil	146✔
281	}
282
283	func (r *systemTableRowReader) onClose(cb func()) { r.onCloseCallback = cb }	72✔

codenotary / immudb / 24841644892

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

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