Skip to content

Fix scalar aggregation with literals in empty result sets #18477

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
systay merged 7 commits into from
Jul 25, 2025
Merged

Fix scalar aggregation with literals in empty result sets #18477

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
2eb8f48
fix: Fix scalar aggregation with literals in empty result sets
systay Jul 24, 2025
32a2621
test: don't run this test on older branches
systay Jul 24, 2025
1c80255
test: use a noopVcursor instead of nil
systay Jul 24, 2025
c59f31f
fix: code was broken and sad
systay Jul 24, 2025
386ae98
cleanup: wrangler vdiff should use a better vcursor
systay Jul 24, 2025
e944d81
refactor: use the provided default collation
systay Jul 24, 2025
7b9c3c6
fix: use the proper collation
systay Jul 24, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions go/test/endtoend/vtgate/queries/aggregation/aggregation_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,9 @@ func TestAggrWithLimit(t *testing.T) {
mcmp.Exec(fmt.Sprintf("insert into aggr_test(id, val1, val2) values(%d, 'a', %d)", i, r))
}
mcmp.Exec("select val2, count(*) from aggr_test group by val2 order by count(*), val2 limit 10")
if utils.BinaryIsAtLeastAtVersion(23, "vtgate") {
mcmp.Exec("SELECT 1 AS `id`, COUNT(*) FROM (SELECT `id` FROM aggr_test WHERE val1 = 1 LIMIT 100) `t`")
}
}

func TestAggregateTypes(t *testing.T) {
Expand Down
14 changes: 14 additions & 0 deletions go/vt/sqlparser/analyzer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -379,6 +379,20 @@ func IsColName(node Expr) bool {
return ok
}

var errNotStatic = fmt.Errorf("not static")
Comment thread
systay marked this conversation as resolved.
Outdated

// IsConstant returns true if the Expr can be evaluated without input or access to tables.
func IsConstant(node Expr) bool {
err := Walk(func(node SQLNode) (kontinue bool, err error) {
switch node.(type) {
case *ColName, *Subquery:
return false, errNotStatic
}
return true, nil
}, node)
return err == nil
}

// IsValue returns true if the Expr is a string, integral or value arg.
// NULL is not considered to be a value.
func IsValue(node Expr) bool {
Expand Down
144 changes: 100 additions & 44 deletions go/vt/vtgate/engine/aggregations.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,12 @@ import (
// It contains the opcode and input column number.
type AggregateParams struct {
Opcode opcode.AggregateOpcode
Col int

// Input source specification - exactly one of these should be set:
// Col: Column index for simple column references (e.g., SUM(column_name))
// EExpr: Evaluated expression for literals, parameters
Col int
EExpr evalengine.Expr
Comment on lines +42 to +43

@arthurschreiber arthurschreiber Jul 25, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess if EExpr is set, the value in Col is ignored? I'm asking because it says "exactly one should be set", but 0 (the default / empty value fro Col) is a valid value and I was confused for a moment what this means then.


// These are used only for distinct opcodes.
KeyCol int
Expand All @@ -53,15 +58,26 @@ type AggregateParams struct {
CollationEnv *collations.Environment
}

func NewAggregateParam(opcode opcode.AggregateOpcode, col int, alias string, collationEnv *collations.Environment) *AggregateParams {
// NewAggregateParam creates a new aggregate param
func NewAggregateParam(
oc opcode.AggregateOpcode,
col int,
expr evalengine.Expr,
alias string,
collationEnv *collations.Environment,
) *AggregateParams {
if expr != nil && oc != opcode.AggregateConstant {
panic(vterrors.VT13001("expr should be nil"))
Comment on lines +69 to +70

@harshit-gangal harshit-gangal Jul 25, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: can have better error message, defining why.

@systay systay Jul 25, 2025

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is just defensive coding. If this ever happens, we messed up somewhere else. VT13001 means bug - i.e. not expected. not sure which why to write here

}
out := &AggregateParams{
Opcode: opcode,
Opcode: oc,
Col: col,
EExpr: expr,
Alias: alias,
WCol: -1,
CollationEnv: collationEnv,
}
if opcode.NeedsComparableValues() {
if oc.NeedsComparableValues() {
out.KeyCol = col
}
return out
Expand All @@ -73,6 +89,9 @@ func (ap *AggregateParams) WAssigned() bool {

func (ap *AggregateParams) String() string {
keyCol := strconv.Itoa(ap.Col)
if ap.EExpr != nil {
keyCol = sqlparser.String(ap.EExpr)
}
if ap.WAssigned() {
keyCol = fmt.Sprintf("%s|%d", keyCol, ap.WCol)
}
Expand All @@ -89,16 +108,23 @@ func (ap *AggregateParams) String() string {
return fmt.Sprintf("%s%s(%s)", ap.Opcode.String(), dispOrigOp, keyCol)
}

func (ap *AggregateParams) typ(inputType querypb.Type) querypb.Type {
func (ap *AggregateParams) typ(inputType querypb.Type, env *evalengine.ExpressionEnv) querypb.Type {
if ap.EExpr != nil {
value, err := eval(env, ap.EExpr)
if err != nil {
return sqltypes.Unknown
}
return value.Type()
}
if ap.OrigOpcode != opcode.AggregateUnassigned {
return ap.OrigOpcode.SQLType(inputType)
}
return ap.Opcode.SQLType(inputType)
}

type aggregator interface {
add(row []sqltypes.Value) error
finish() sqltypes.Value
add(row []sqltypes.Value, env *evalengine.ExpressionEnv) error
finish(*evalengine.ExpressionEnv) (sqltypes.Value, error)
reset()
}

Expand Down Expand Up @@ -140,7 +166,7 @@ type aggregatorCount struct {
distinct aggregatorDistinct
}

func (a *aggregatorCount) add(row []sqltypes.Value) error {
func (a *aggregatorCount) add(row []sqltypes.Value, _ *evalengine.ExpressionEnv) error {
if row[a.from].IsNull() {
return nil
}
Expand All @@ -151,8 +177,8 @@ func (a *aggregatorCount) add(row []sqltypes.Value) error {
return nil
}

func (a *aggregatorCount) finish() sqltypes.Value {
return sqltypes.NewInt64(a.n)
func (a *aggregatorCount) finish(*evalengine.ExpressionEnv) (sqltypes.Value, error) {
return sqltypes.NewInt64(a.n), nil
}

func (a *aggregatorCount) reset() {
Expand All @@ -164,13 +190,13 @@ type aggregatorCountStar struct {
n int64
}

func (a *aggregatorCountStar) add(_ []sqltypes.Value) error {
func (a *aggregatorCountStar) add([]sqltypes.Value, *evalengine.ExpressionEnv) error {
a.n++
return nil
}

func (a *aggregatorCountStar) finish() sqltypes.Value {
return sqltypes.NewInt64(a.n)
func (a *aggregatorCountStar) finish(*evalengine.ExpressionEnv) (sqltypes.Value, error) {
return sqltypes.NewInt64(a.n), nil
}

func (a *aggregatorCountStar) reset() {
Expand All @@ -186,20 +212,20 @@ type aggregatorMin struct {
aggregatorMinMax
}

func (a *aggregatorMin) add(row []sqltypes.Value) (err error) {
func (a *aggregatorMin) add(row []sqltypes.Value, _ *evalengine.ExpressionEnv) (err error) {
return a.minmax.Min(row[a.from])
}

type aggregatorMax struct {
aggregatorMinMax
}

func (a *aggregatorMax) add(row []sqltypes.Value) (err error) {
func (a *aggregatorMax) add(row []sqltypes.Value, _ *evalengine.ExpressionEnv) (err error) {
return a.minmax.Max(row[a.from])
}

func (a *aggregatorMinMax) finish() sqltypes.Value {
return a.minmax.Result()
func (a *aggregatorMinMax) finish(*evalengine.ExpressionEnv) (sqltypes.Value, error) {
return a.minmax.Result(), nil
}

func (a *aggregatorMinMax) reset() {
Expand All @@ -212,7 +238,7 @@ type aggregatorSum struct {
distinct aggregatorDistinct
}

func (a *aggregatorSum) add(row []sqltypes.Value) error {
func (a *aggregatorSum) add(row []sqltypes.Value, _ *evalengine.ExpressionEnv) error {
if row[a.from].IsNull() {
return nil
}
Expand All @@ -222,8 +248,8 @@ func (a *aggregatorSum) add(row []sqltypes.Value) error {
return a.sum.Add(row[a.from])
}

func (a *aggregatorSum) finish() sqltypes.Value {
return a.sum.Result()
func (a *aggregatorSum) finish(*evalengine.ExpressionEnv) (sqltypes.Value, error) {
return a.sum.Result(), nil
}

func (a *aggregatorSum) reset() {
Expand All @@ -232,28 +258,51 @@ func (a *aggregatorSum) reset() {
}

type aggregatorScalar struct {
from int
current sqltypes.Value
init bool
from int
current sqltypes.Value
hasValue bool
}

func (a *aggregatorScalar) add(row []sqltypes.Value) error {
if !a.init {
func (a *aggregatorScalar) add(row []sqltypes.Value, _ *evalengine.ExpressionEnv) error {
if !a.hasValue {
a.current = row[a.from]
a.init = true
a.hasValue = true
}
return nil
}

func (a *aggregatorScalar) finish() sqltypes.Value {
return a.current
func (a *aggregatorScalar) finish(env *evalengine.ExpressionEnv) (sqltypes.Value, error) {
return a.current, nil
}

func (a *aggregatorScalar) reset() {
a.current = sqltypes.NULL
a.init = false
a.hasValue = false
}

type aggregatorConstant struct {
expr evalengine.Expr
}

func (*aggregatorConstant) add([]sqltypes.Value, *evalengine.ExpressionEnv) error {
return nil
}

func (a *aggregatorConstant) finish(env *evalengine.ExpressionEnv) (sqltypes.Value, error) {
return eval(env, a.expr)
}

func eval(env *evalengine.ExpressionEnv, eexpr evalengine.Expr) (sqltypes.Value, error) {
v, err := env.Evaluate(eexpr)
if err != nil {
return sqltypes.Value{}, err
}

return v.Value(collations.Unknown), nil
Comment thread
systay marked this conversation as resolved.
Outdated
}

func (*aggregatorConstant) reset() {}

type aggregatorGroupConcat struct {
from int
type_ sqltypes.Type
Expand All @@ -263,7 +312,7 @@ type aggregatorGroupConcat struct {
n int
}

func (a *aggregatorGroupConcat) add(row []sqltypes.Value) error {
func (a *aggregatorGroupConcat) add(row []sqltypes.Value, _ *evalengine.ExpressionEnv) error {
if row[a.from].IsNull() {
return nil
}
Expand All @@ -275,11 +324,11 @@ func (a *aggregatorGroupConcat) add(row []sqltypes.Value) error {
return nil
}

func (a *aggregatorGroupConcat) finish() sqltypes.Value {
func (a *aggregatorGroupConcat) finish(*evalengine.ExpressionEnv) (sqltypes.Value, error) {
if a.n == 0 {
return sqltypes.NULL
return sqltypes.NULL, nil
}
return sqltypes.MakeTrusted(a.type_, a.concat)
return sqltypes.MakeTrusted(a.type_, a.concat), nil
}

func (a *aggregatorGroupConcat) reset() {
Expand All @@ -292,7 +341,7 @@ type aggregatorGtid struct {
shards []*binlogdatapb.ShardGtid
}

func (a *aggregatorGtid) add(row []sqltypes.Value) error {
func (a *aggregatorGtid) add(row []sqltypes.Value, _ *evalengine.ExpressionEnv) error {
a.shards = append(a.shards, &binlogdatapb.ShardGtid{
Keyspace: row[a.from-1].ToString(),
Shard: row[a.from+1].ToString(),
Expand All @@ -301,9 +350,9 @@ func (a *aggregatorGtid) add(row []sqltypes.Value) error {
return nil
}

func (a *aggregatorGtid) finish() sqltypes.Value {
func (a *aggregatorGtid) finish(*evalengine.ExpressionEnv) (sqltypes.Value, error) {
gtid := binlogdatapb.VGtid{ShardGtids: a.shards}
return sqltypes.NewVarChar(gtid.String())
return sqltypes.NewVarChar(gtid.String()), nil
}

func (a *aggregatorGtid) reset() {
Expand All @@ -312,21 +361,25 @@ func (a *aggregatorGtid) reset() {

type aggregationState []aggregator

func (a aggregationState) add(row []sqltypes.Value) error {
func (a aggregationState) add(row []sqltypes.Value, env *evalengine.ExpressionEnv) error {
for _, st := range a {
if err := st.add(row); err != nil {
if err := st.add(row, env); err != nil {
return err
}
}
return nil
}

func (a aggregationState) finish() (row []sqltypes.Value) {
row = make([]sqltypes.Value, 0, len(a))
func (a aggregationState) finish(env *evalengine.ExpressionEnv) ([]sqltypes.Value, error) {
row := make([]sqltypes.Value, 0, len(a))
for _, st := range a {
row = append(row, st.finish())
v, err := st.finish(env)
if err != nil {
return nil, err
}
row = append(row, v)
}
return
return row, nil
}

func (a aggregationState) reset() {
Expand Down Expand Up @@ -354,13 +407,13 @@ func isComparable(typ sqltypes.Type) bool {
return false
}

func newAggregation(fields []*querypb.Field, aggregates []*AggregateParams) (aggregationState, []*querypb.Field, error) {
func newAggregation(fields []*querypb.Field, aggregates []*AggregateParams, env *evalengine.ExpressionEnv) (aggregationState, []*querypb.Field, error) {
fields = slice.Map(fields, func(from *querypb.Field) *querypb.Field { return from.CloneVT() })

agstate := make([]aggregator, len(fields))
for _, aggr := range aggregates {
sourceType := fields[aggr.Col].Type
targetType := aggr.typ(sourceType)
targetType := aggr.typ(sourceType, env)

var ag aggregator
var distinct = -1
Expand Down Expand Up @@ -444,6 +497,9 @@ func newAggregation(fields []*querypb.Field, aggregates []*AggregateParams) (agg
separator: separator,
}

case opcode.AggregateConstant:
ag = &aggregatorConstant{expr: aggr.EExpr}

default:
panic("BUG: unexpected Aggregation opcode")
}
Expand Down
6 changes: 5 additions & 1 deletion go/vt/vtgate/engine/cached_size.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading
Loading