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

98.06

/embedded/sql/union_row_reader.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"
        "errors"
        "fmt"

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

type unionRowReader struct {
        rowReaders []RowReader
        currReader int

        cols  []ColDescriptor
        alias string
}

func newUnionRowReader(ctx context.Context, rowReaders []RowReader) (*unionRowReader, error) {
        if len(rowReaders) == 0 {
                return nil, ErrIllegalArguments
        }

        cols, err := rowReaders[0].Columns(ctx)
        if err != nil {
                return nil, err
        }

        for i := 1; i < len(rowReaders); i++ {
                cs, err := rowReaders[i].Columns(ctx)
                if err != nil {
                        return nil, err
                }

                if len(cols) != len(cs) {
                        return nil, fmt.Errorf("%w: each subquery must have same number of columns", ErrColumnMismatchInUnionStmt)
                }

                for c := 0; c < len(cols); c++ {
                        if cols[c].Type != cs[c].Type {
                                return nil, fmt.Errorf("%w: expecting type '%v' for column '%s'", ErrColumnMismatchInUnionStmt, cols[c].Type, cs[c].Column)
                        }
                }
        }

        return &unionRowReader{
                rowReaders: rowReaders,
                cols:       cols,
        }, nil
}

func (ur *unionRowReader) onClose(callback func()) {
        ur.rowReaders[0].onClose(callback)
}

func (ur *unionRowReader) Tx() *SQLTx {
        return ur.rowReaders[0].Tx()
}

func (ur *unionRowReader) TableAlias() string {
        return ur.alias
}

func (ur *unionRowReader) Parameters() map[string]interface{} {
        return ur.rowReaders[0].Parameters()
}

func (ur *unionRowReader) OrderBy() []ColDescriptor {
        return nil
}

func (ur *unionRowReader) ScanSpecs() *ScanSpecs {
        return nil
}

func (ur *unionRowReader) Columns(ctx context.Context) ([]ColDescriptor, error) {
        cols, err := ur.rowReaders[0].Columns(ctx)
        if err != nil {
                return nil, err
        }
        if ur.alias != "" {
                aliasedCols := make([]ColDescriptor, len(cols))
                for i, c := range cols {
                        aliasedCols[i] = ColDescriptor{Table: ur.alias, Column: c.Column, Type: c.Type}
                }
                return aliasedCols, nil
        }
        return cols, nil
}

func (ur *unionRowReader) colsBySelector(ctx context.Context) (map[string]ColDescriptor, error) {
        cols, err := ur.Columns(ctx)
        if err != nil {
                return nil, err
        }
        result := make(map[string]ColDescriptor, len(cols))
        for _, c := range cols {
                result[c.Selector()] = c
        }
        return result, nil
}

func (ur *unionRowReader) InferParameters(ctx context.Context, params map[string]SQLValueType) error {
        for _, r := range ur.rowReaders {
                err := r.InferParameters(ctx, params)
                if err != nil {
                        return err
                }
        }

        return nil
}

func (ur *unionRowReader) Read(ctx context.Context) (*Row, error) {
        for {
                row, err := ur.rowReaders[ur.currReader].Read(ctx)
                if errors.Is(err, store.ErrNoMoreEntries) && ur.currReader+1 < len(ur.rowReaders) {
                        ur.currReader++
                        continue
                }
                if err != nil {
                        return nil, err
                }

                if ur.currReader > 0 || ur.alias != "" {
                        // remap selectors using the alias or first subquery's column names
                        valuesBySelector := make(map[string]TypedValue, len(ur.cols))

                        table := ur.alias
                        for i, c := range ur.cols {
                                col := c
                                if table != "" {
                                        col = ColDescriptor{Table: table, Column: c.Column, Type: c.Type}
                                }
                                valuesBySelector[col.Selector()] = row.ValuesByPosition[i]
                        }

                        row.ValuesBySelector = valuesBySelector
                }

                return row, nil
        }
}

func (ur *unionRowReader) Close() error {
        merr := multierr.NewMultiErr()

        // Closing in reverse order to ensure the onClose callback
        // is called after the last reader is closed
        for i := len(ur.rowReaders) - 1; i >= 0; i-- {
                err := ur.rowReaders[i].Close()
                merr.Append(err)
        }

        return merr.Reduce()
}

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	"errors"
22	"fmt"
23
24	"github.com/codenotary/immudb/embedded/multierr"
25	"github.com/codenotary/immudb/embedded/store"
26	)
27
28	type unionRowReader struct {
29	rowReaders []RowReader
30	currReader int
31
32	cols []ColDescriptor
33	alias string
34	}
35
36	func newUnionRowReader(ctx context.Context, rowReaders []RowReader) (*unionRowReader, error) {	22✔
37	if len(rowReaders) == 0 {	23✔
38	return nil, ErrIllegalArguments	1✔
39	}	1✔
40
41	cols, err := rowReaders[0].Columns(ctx)	21✔
42	if err != nil {	22✔
43	return nil, err	1✔
44	}	1✔
45
46	for i := 1; i < len(rowReaders); i++ {	39✔
47	cs, err := rowReaders[i].Columns(ctx)	19✔
48	if err != nil {	20✔
49	return nil, err	1✔
50	}	1✔
51
52	if len(cols) != len(cs) {	19✔
53	return nil, fmt.Errorf("%w: each subquery must have same number of columns", ErrColumnMismatchInUnionStmt)	1✔
54	}	1✔
55
56	for c := 0; c < len(cols); c++ {	34✔
57	if cols[c].Type != cs[c].Type {	18✔
58	return nil, fmt.Errorf("%w: expecting type '%v' for column '%s'", ErrColumnMismatchInUnionStmt, cols[c].Type, cs[c].Column)	1✔
59	}	1✔
60	}
61	}
62
63	return &unionRowReader{	17✔
64	rowReaders: rowReaders,	17✔
65	cols: cols,	17✔
66	}, nil	17✔
67	}
68
69	func (ur *unionRowReader) onClose(callback func()) {	13✔
70	ur.rowReaders[0].onClose(callback)	13✔
71	}	13✔
72
73	func (ur unionRowReader) Tx() SQLTx {	63✔
74	return ur.rowReaders[0].Tx()	63✔
75	}	63✔
76
77	func (ur *unionRowReader) TableAlias() string {	84✔
78	return ur.alias	84✔
79	}	84✔
80
81	func (ur *unionRowReader) Parameters() map[string]interface{} {	12✔
82	return ur.rowReaders[0].Parameters()	12✔
83	}	12✔
84
85	func (ur *unionRowReader) OrderBy() []ColDescriptor {	1✔
86	return nil	1✔
87	}	1✔
88
89	func (ur unionRowReader) ScanSpecs() ScanSpecs {	1✔
90	return nil	1✔
91	}	1✔
92
93	func (ur *unionRowReader) Columns(ctx context.Context) ([]ColDescriptor, error) {	18✔
94	cols, err := ur.rowReaders[0].Columns(ctx)	18✔
95	if err != nil {	19✔
96	return nil, err	1✔
97	}	1✔
98	if ur.alias != "" {	29✔
99	aliasedCols := make([]ColDescriptor, len(cols))	12✔
100	for i, c := range cols {	24✔
101	aliasedCols[i] = ColDescriptor{Table: ur.alias, Column: c.Column, Type: c.Type}	12✔
102	}	12✔
103	return aliasedCols, nil	12✔
104	}
105	return cols, nil	5✔
106	}
107
108	func (ur *unionRowReader) colsBySelector(ctx context.Context) (map[string]ColDescriptor, error) {	12✔
109	cols, err := ur.Columns(ctx)	12✔
110	if err != nil {	12✔
NEW 111	return nil, err	×
NEW 112	}	×
113	result := make(map[string]ColDescriptor, len(cols))	12✔
114	for _, c := range cols {	24✔
115	result[c.Selector()] = c	12✔
116	}	12✔
117	return result, nil	12✔
118	}
119
120	func (ur *unionRowReader) InferParameters(ctx context.Context, params map[string]SQLValueType) error {	4✔
121	for _, r := range ur.rowReaders {	10✔
122	err := r.InferParameters(ctx, params)	6✔
123	if err != nil {	7✔
124	return err	1✔
125	}	1✔
126	}
127
128	return nil	3✔
129	}
130
131	func (ur unionRowReader) Read(ctx context.Context) (Row, error) {	86✔
132	for {	183✔
133	row, err := ur.rowReaders[ur.currReader].Read(ctx)	97✔
134	if errors.Is(err, store.ErrNoMoreEntries) && ur.currReader+1 < len(ur.rowReaders) {	108✔
135	ur.currReader++	11✔
136	continue	11✔
137	}
138	if err != nil {	99✔
139	return nil, err	13✔
140	}	13✔
141
142	if ur.currReader > 0 \|\| ur.alias != "" {	119✔
143	// remap selectors using the alias or first subquery's column names	46✔
144	valuesBySelector := make(map[string]TypedValue, len(ur.cols))	46✔
145		46✔
146	table := ur.alias	46✔
147	for i, c := range ur.cols {	94✔
148	col := c	48✔
149	if table != "" {	67✔
150	col = ColDescriptor{Table: table, Column: c.Column, Type: c.Type}	19✔
151	}	19✔
152	valuesBySelector[col.Selector()] = row.ValuesByPosition[i]	48✔
153	}
154
155	row.ValuesBySelector = valuesBySelector	46✔
156	}
157
158	return row, nil	73✔
159	}
160	}
161
162	func (ur *unionRowReader) Close() error {	16✔
163	merr := multierr.NewMultiErr()	16✔
164		16✔
165	// Closing in reverse order to ensure the onClose callback	16✔
166	// is called after the last reader is closed	16✔
167	for i := len(ur.rowReaders) - 1; i >= 0; i-- {	48✔
168	err := ur.rowReaders[i].Close()	32✔
169	merr.Append(err)	32✔
170	}	32✔
171
172	return merr.Reduce()	16✔
173	}

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