From c9ffdb452e70334c40694767e9404fd1508af0d0 Mon Sep 17 00:00:00 2001 From: James Cor Date: Mon, 11 Aug 2025 16:15:44 -0700 Subject: [PATCH 1/4] avoid rows.append --- sql/iters/rel_iters.go | 2 + sql/rowexec/dml_iters.go | 3 +- sql/rowexec/insert.go | 2 +- sql/rowexec/join_iters.go | 83 ++++++++++++++++++++++------------ sql/rowexec/range_heap_iter.go | 39 +++++++++------- sql/rowexec/rel.go | 2 +- sql/rows.go | 2 +- 7 files changed, 83 insertions(+), 50 deletions(-) diff --git a/sql/iters/rel_iters.go b/sql/iters/rel_iters.go index 357f24b107..778310b378 100644 --- a/sql/iters/rel_iters.go +++ b/sql/iters/rel_iters.go @@ -167,6 +167,8 @@ type JsonTableCol struct { pos int finished bool // exhausted all rows in data currSib int + + resultRow sql.Row } // IsSibling returns if the jsonTableCol contains multiple columns diff --git a/sql/rowexec/dml_iters.go b/sql/rowexec/dml_iters.go index 583c7c2dcf..4e298c1dc5 100644 --- a/sql/rowexec/dml_iters.go +++ b/sql/rowexec/dml_iters.go @@ -754,7 +754,8 @@ func (u *updateSourceIter) Next(ctx *sql.Context) (sql.Row, error) { newRow = newRow[len(newRow)-expectedSchemaLen:] } - return oldRow.Append(newRow), nil + row := append(oldRow, newRow...) + return row, nil } func (u *updateSourceIter) Close(ctx *sql.Context) error { diff --git a/sql/rowexec/insert.go b/sql/rowexec/insert.go index 28b1c2cec9..34c375e8af 100644 --- a/sql/rowexec/insert.go +++ b/sql/rowexec/insert.go @@ -221,7 +221,7 @@ func (i *insertIter) Next(ctx *sql.Context) (returnRow sql.Row, returnErr error) func (i *insertIter) handleOnDuplicateKeyUpdate(ctx *sql.Context, oldRow, newRow sql.Row) (returnRow sql.Row, returnErr error) { var err error - updateAcc := append(oldRow, newRow...) + updateAcc := append(oldRow, newRow...) // TODO: how is this safe? var evalRow sql.Row for _, updateExpr := range i.updateExprs { // this SET indexes into LHS, but the can diff --git a/sql/rowexec/join_iters.go b/sql/rowexec/join_iters.go index a39c5f0ff3..a706dad2ee 100644 --- a/sql/rowexec/join_iters.go +++ b/sql/rowexec/join_iters.go @@ -54,14 +54,22 @@ func newJoinIter(ctx *sql.Context, b sql.NodeExecBuilder, j *plan.JoinNode, row span.End() return nil, err } + + parentLen := len(row) + + primaryRow := make(sql.Row, parentLen+len(j.Left().Schema())) + copy(primaryRow, row) + return sql.NewSpanIter(span, &joinIter{ - parentRow: row, + primaryRow: primaryRow, + loadPrimaryRow: true, primary: l, secondaryProvider: j.Right(), cond: j.Filter, joinType: j.Op, - rowSize: len(row) + len(j.Left().Schema()) + len(j.Right().Schema()), + rowSize: parentLen + len(j.Left().Schema()) + len(j.Right().Schema()), scopeLen: j.ScopeLen, + parentLen: parentLen, b: b, }), nil } @@ -69,7 +77,6 @@ func newJoinIter(ctx *sql.Context, b sql.NodeExecBuilder, j *plan.JoinNode, row // joinIter is an iterator that iterates over every row in the primary table and performs an index lookup in // the secondary table for each value type joinIter struct { - parentRow sql.Row primary sql.RowIter primaryRow sql.Row secondaryProvider sql.Node @@ -77,21 +84,24 @@ type joinIter struct { cond sql.Expression joinType plan.JoinType + loadPrimaryRow bool + foundMatch bool rowSize int scopeLen int + parentLen int b sql.NodeExecBuilder } func (i *joinIter) loadPrimary(ctx *sql.Context) error { - if i.primaryRow == nil { + if i.loadPrimaryRow { r, err := i.primary.Next(ctx) if err != nil { return err } - - i.primaryRow = i.parentRow.Append(r) + copy(i.primaryRow[i.parentLen:], r) i.foundMatch = false + i.loadPrimaryRow = false } return nil @@ -118,7 +128,7 @@ func (i *joinIter) loadSecondary(ctx *sql.Context) (sql.Row, error) { if err != nil { return nil, err } - i.primaryRow = nil + i.loadPrimaryRow = true return nil, io.EOF } return nil, err @@ -138,14 +148,14 @@ func (i *joinIter) Next(ctx *sql.Context) (sql.Row, error) { if err != nil { if errors.Is(err, io.EOF) { if !i.foundMatch && i.joinType.IsLeftOuter() { - i.primaryRow = nil + i.loadPrimaryRow = true row := i.buildRow(primary, nil) return i.removeParentRow(row), nil } continue } else if errors.Is(err, plan.ErrEmptyCachedResult) { if !i.foundMatch && i.joinType.IsLeftOuter() { - i.primaryRow = nil + i.loadPrimaryRow = true row := i.buildRow(primary, nil) return i.removeParentRow(row), nil } @@ -167,7 +177,7 @@ func (i *joinIter) Next(ctx *sql.Context) (sql.Row, error) { if err != nil { return nil, err } - i.primaryRow = nil + i.loadPrimaryRow = true continue } @@ -181,8 +191,8 @@ func (i *joinIter) Next(ctx *sql.Context) (sql.Row, error) { } func (i *joinIter) removeParentRow(r sql.Row) sql.Row { - copy(r[i.scopeLen:], r[len(i.parentRow):]) - r = r[:len(r)-len(i.parentRow)+i.scopeLen] + copy(r[i.scopeLen:], r[i.parentLen:]) + r = r[:len(r)-i.parentLen+i.scopeLen] return r } @@ -216,37 +226,50 @@ func (i *joinIter) Close(ctx *sql.Context) (err error) { func newExistsIter(ctx *sql.Context, b sql.NodeExecBuilder, j *plan.JoinNode, row sql.Row) (sql.RowIter, error) { leftIter, err := b.Build(ctx, j.Left(), row) - if err != nil { return nil, err } + + parentLen := len(row) + + rowSize := parentLen + len(j.Left().Schema()) + len(j.Right().Schema()) + fullRow := make(sql.Row, rowSize) + copy(fullRow, row) + + primaryRow := make(sql.Row, parentLen+len(j.Left().Schema())) + copy(primaryRow, row) + return &existsIter{ - parentRow: row, + b: b, typ: j.Op, primary: leftIter, + primaryRow: primaryRow, + fullRow: fullRow, + parentLen: parentLen, secondaryProvider: j.Right(), cond: j.Filter, scopeLen: j.ScopeLen, - rowSize: len(row) + len(j.Left().Schema()) + len(j.Right().Schema()), + rowSize: rowSize, nullRej: !(j.Filter != nil && plan.IsNullRejecting(j.Filter)), - b: b, }, nil } type existsIter struct { + b sql.NodeExecBuilder typ plan.JoinType - primary sql.RowIter secondaryProvider sql.Node cond sql.Expression + primary sql.RowIter primaryRow sql.Row + fullRow sql.Row + + parentLen int + scopeLen int + rowSize int - parentRow sql.Row - scopeLen int - rowSize int nullRej bool rightIterNonEmpty bool - b sql.NodeExecBuilder } type existsState uint8 @@ -261,9 +284,7 @@ const ( ) func (i *existsIter) Next(ctx *sql.Context) (sql.Row, error) { - var row sql.Row var right sql.Row - var left sql.Row var rIter sql.RowIter var err error @@ -282,8 +303,9 @@ func (i *existsIter) Next(ctx *sql.Context) (sql.Row, error) { if err != nil { return nil, err } - left = i.parentRow.Append(r) - rIter, err = i.b.Build(ctx, i.secondaryProvider, left) + copy(i.primaryRow[i.parentLen:], r) + //i.primaryRow = i.parentRow.Append(r) + rIter, err = i.b.Build(ctx, i.secondaryProvider, i.primaryRow) if err != nil { return nil, err } @@ -319,8 +341,9 @@ func (i *existsIter) Next(ctx *sql.Context) (sql.Row, error) { nextState = esRet } case esCompare: - row = i.buildRow(left, right) - res, err := sql.EvaluateCondition(ctx, i.cond, row) + copy(i.fullRow[i.parentLen:], i.primaryRow[i.parentLen:]) + copy(i.fullRow[len(i.primaryRow):], right) + res, err := sql.EvaluateCondition(ctx, i.cond, i.fullRow) if err != nil { return nil, err } @@ -351,7 +374,7 @@ func (i *existsIter) Next(ctx *sql.Context) (sql.Row, error) { nextState = esIncRight } case esRet: - return i.removeParentRow(left), nil + return i.removeParentRow(i.primaryRow.Copy()), nil default: return nil, fmt.Errorf("invalid exists join state") } @@ -366,8 +389,8 @@ func isTrueLit(e sql.Expression) bool { } func (i *existsIter) removeParentRow(r sql.Row) sql.Row { - copy(r[i.scopeLen:], r[len(i.parentRow):]) - r = r[:len(r)-len(i.parentRow)+i.scopeLen] + copy(r[i.scopeLen:], r[i.parentLen:]) + r = r[:len(r)-i.parentLen+i.scopeLen] return r } diff --git a/sql/rowexec/range_heap_iter.go b/sql/rowexec/range_heap_iter.go index 4ddbd7e3e2..279b26380e 100644 --- a/sql/rowexec/range_heap_iter.go +++ b/sql/rowexec/range_heap_iter.go @@ -43,16 +43,20 @@ func newRangeHeapJoinIter(ctx *sql.Context, b sql.NodeExecBuilder, j *plan.JoinN return nil, errors.New("right side of join must be a range heap") } + primaryRow := make(sql.Row, len(row)+len(j.Left().Schema())) + return sql.NewSpanIter(span, &rangeHeapJoinIter{ - parentRow: row, - primary: l, - cond: j.Filter, - joinType: j.Op, - rowSize: len(row) + len(j.Left().Schema()) + len(j.Right().Schema()), - scopeLen: j.ScopeLen, - b: b, - rangeHeapPlan: rhp, - ctx: ctx, + parentRow: row, + primary: l, + primaryRow: primaryRow, + loadPrimaryRow: true, + cond: j.Filter, + joinType: j.Op, + rowSize: len(row) + len(j.Left().Schema()) + len(j.Right().Schema()), + scopeLen: j.ScopeLen, + b: b, + rangeHeapPlan: rhp, + ctx: ctx, }), nil } @@ -66,6 +70,8 @@ type rangeHeapJoinIter struct { cond sql.Expression joinType plan.JoinType + loadPrimaryRow bool + foundMatch bool rowSize int scopeLen int @@ -82,14 +88,15 @@ type rangeHeapJoinIter struct { } func (iter *rangeHeapJoinIter) loadPrimary(ctx *sql.Context) error { - if iter.primaryRow == nil { + if iter.loadPrimaryRow { r, err := iter.primary.Next(ctx) if err != nil { return err } - iter.primaryRow = iter.parentRow.Append(r) + copy(iter.primaryRow[len(iter.parentRow):], r) iter.foundMatch = false + iter.loadPrimaryRow = false err = iter.initializeHeap(ctx, iter.b, iter.primaryRow) if err != nil { @@ -121,7 +128,7 @@ func (iter *rangeHeapJoinIter) loadSecondary(ctx *sql.Context) (sql.Row, error) if err != nil { return nil, err } - iter.primaryRow = nil + iter.loadPrimaryRow = true return nil, io.EOF } return nil, err @@ -141,14 +148,14 @@ func (iter *rangeHeapJoinIter) Next(ctx *sql.Context) (sql.Row, error) { if err != nil { if errors.Is(err, io.EOF) { if !iter.foundMatch && iter.joinType.IsLeftOuter() { - iter.primaryRow = nil + iter.loadPrimaryRow = true row := iter.buildRow(primary, nil) return iter.removeParentRow(row), nil } continue } else if errors.Is(err, plan.ErrEmptyCachedResult) { if !iter.foundMatch && iter.joinType.IsLeftOuter() { - iter.primaryRow = nil + iter.loadPrimaryRow = true row := iter.buildRow(primary, nil) return iter.removeParentRow(row), nil } @@ -171,7 +178,7 @@ func (iter *rangeHeapJoinIter) Next(ctx *sql.Context) (sql.Row, error) { if err != nil { return nil, err } - iter.primaryRow = nil + iter.loadPrimaryRow = true continue } @@ -306,7 +313,7 @@ func compareNullsFirst(ctx *sql.Context, comparisonType sql.Type, a, b interface return comparisonType.Compare(ctx, a, b) } -func (iter rangeHeapJoinIter) Len() int { return len(iter.activeRanges) } +func (iter *rangeHeapJoinIter) Len() int { return len(iter.activeRanges) } func (iter *rangeHeapJoinIter) Less(i, j int) bool { lhs := iter.activeRanges[i][iter.rangeHeapPlan.MaxColumnIndex] diff --git a/sql/rowexec/rel.go b/sql/rowexec/rel.go index 42942213f9..d5ea5107e0 100644 --- a/sql/rowexec/rel.go +++ b/sql/rowexec/rel.go @@ -103,7 +103,7 @@ func (b *BaseBuilder) buildValues(ctx *sql.Context, n *plan.Values, row sql.Row) } } - rows[i] = sql.NewRow(vals...) + rows[i] = vals } return sql.RowsToRowIter(rows...), nil diff --git a/sql/rows.go b/sql/rows.go index faf3a738bf..bd89fd7cd4 100644 --- a/sql/rows.go +++ b/sql/rows.go @@ -29,7 +29,7 @@ type Row []interface{} // NewRow creates a row from the given values. func NewRow(values ...interface{}) Row { - row := make([]interface{}, len(values)) + row := make(Row, len(values)) copy(row, values) return row } From df035ce9f1a2e1053369ccba939dd991e5cb26b6 Mon Sep 17 00:00:00 2001 From: James Cor Date: Tue, 12 Aug 2025 11:57:40 -0700 Subject: [PATCH 2/4] more optimizations --- sql/rowexec/join_iters.go | 152 ++++++++++++++++----------------- sql/rowexec/range_heap_iter.go | 83 +++++++++--------- 2 files changed, 116 insertions(+), 119 deletions(-) diff --git a/sql/rowexec/join_iters.go b/sql/rowexec/join_iters.go index a706dad2ee..a5a735aa66 100644 --- a/sql/rowexec/join_iters.go +++ b/sql/rowexec/join_iters.go @@ -30,6 +30,26 @@ import ( "github.com/dolthub/go-mysql-server/sql/transform" ) +// joinIter is an iterator that iterates over every row in the primary table and performs an index lookup in +// the secondary table for each value +type joinIter struct { + b sql.NodeExecBuilder + joinType plan.JoinType + cond sql.Expression + + primary sql.RowIter + primaryRow sql.Row + loadPrimaryRow bool + + secondaryProvider sql.Node + secondary sql.RowIter + + foundMatch bool + rowSize int + scopeLen int + parentLen int +} + func newJoinIter(ctx *sql.Context, b sql.NodeExecBuilder, j *plan.JoinNode, row sql.Row) (sql.RowIter, error) { var leftName, rightName string if leftTable, ok := j.Left().(sql.Nameable); ok { @@ -61,36 +81,20 @@ func newJoinIter(ctx *sql.Context, b sql.NodeExecBuilder, j *plan.JoinNode, row copy(primaryRow, row) return sql.NewSpanIter(span, &joinIter{ - primaryRow: primaryRow, - loadPrimaryRow: true, - primary: l, - secondaryProvider: j.Right(), - cond: j.Filter, - joinType: j.Op, - rowSize: parentLen + len(j.Left().Schema()) + len(j.Right().Schema()), - scopeLen: j.ScopeLen, - parentLen: parentLen, - b: b, - }), nil -} + b: b, + joinType: j.Op, + cond: j.Filter, -// joinIter is an iterator that iterates over every row in the primary table and performs an index lookup in -// the secondary table for each value -type joinIter struct { - primary sql.RowIter - primaryRow sql.Row - secondaryProvider sql.Node - secondary sql.RowIter - cond sql.Expression - joinType plan.JoinType + primary: l, + primaryRow: primaryRow, + loadPrimaryRow: true, - loadPrimaryRow bool + secondaryProvider: j.Right(), - foundMatch bool - rowSize int - scopeLen int - parentLen int - b sql.NodeExecBuilder + rowSize: parentLen + len(j.Left().Schema()) + len(j.Right().Schema()), + scopeLen: j.ScopeLen, + parentLen: parentLen, + }), nil } func (i *joinIter) loadPrimary(ctx *sql.Context) error { @@ -103,14 +107,12 @@ func (i *joinIter) loadPrimary(ctx *sql.Context) error { i.foundMatch = false i.loadPrimaryRow = false } - return nil } func (i *joinIter) loadSecondary(ctx *sql.Context) (sql.Row, error) { if i.secondary == nil { rowIter, err := i.b.Build(ctx, i.secondaryProvider, i.primaryRow) - if err != nil { return nil, err } @@ -153,13 +155,13 @@ func (i *joinIter) Next(ctx *sql.Context) (sql.Row, error) { return i.removeParentRow(row), nil } continue - } else if errors.Is(err, plan.ErrEmptyCachedResult) { + } + if errors.Is(err, plan.ErrEmptyCachedResult) { if !i.foundMatch && i.joinType.IsLeftOuter() { i.loadPrimaryRow = true row := i.buildRow(primary, nil) return i.removeParentRow(row), nil } - return nil, io.EOF } return nil, err @@ -199,10 +201,8 @@ func (i *joinIter) removeParentRow(r sql.Row) sql.Row { // buildRow builds the result set row using the rows from the primary and secondary tables func (i *joinIter) buildRow(primary, secondary sql.Row) sql.Row { row := make(sql.Row, i.rowSize) - copy(row, primary) copy(row[len(primary):], secondary) - return row } @@ -255,15 +255,16 @@ func newExistsIter(ctx *sql.Context, b sql.NodeExecBuilder, j *plan.JoinNode, ro } type existsIter struct { - b sql.NodeExecBuilder - typ plan.JoinType - secondaryProvider sql.Node - cond sql.Expression + b sql.NodeExecBuilder + typ plan.JoinType + cond sql.Expression primary sql.RowIter primaryRow sql.Row fullRow sql.Row + secondaryProvider sql.Node + parentLen int scopeLen int rowSize int @@ -304,7 +305,6 @@ func (i *existsIter) Next(ctx *sql.Context) (sql.Row, error) { return nil, err } copy(i.primaryRow[i.parentLen:], r) - //i.primaryRow = i.parentRow.Append(r) rIter, err = i.b.Build(ctx, i.secondaryProvider, i.primaryRow) if err != nil { return nil, err @@ -397,10 +397,8 @@ func (i *existsIter) removeParentRow(r sql.Row) sql.Row { // buildRow builds the result set row using the rows from the primary and secondary tables func (i *existsIter) buildRow(primary, secondary sql.Row) sql.Row { row := make(sql.Row, i.rowSize) - copy(row, primary) copy(row[len(primary):], secondary) - return row } @@ -415,7 +413,6 @@ func (i *existsIter) Close(ctx *sql.Context) (err error) { func newFullJoinIter(ctx *sql.Context, b sql.NodeExecBuilder, j *plan.JoinNode, row sql.Row) (sql.RowIter, error) { leftIter, err := b.Build(ctx, j.Left(), row) - if err != nil { return nil, err } @@ -549,7 +546,8 @@ func (i *fullJoinIter) Next(ctx *sql.Context) (sql.Row, error) { continue } // (null, right) only if we haven't matched right - ret := i.buildRow(make(sql.Row, i.leftLen), rightRow) + ret := make(sql.Row, i.rowSize) + copy(ret[i.leftLen:], rightRow) return i.removeParentRow(ret), nil } } @@ -563,10 +561,8 @@ func (i *fullJoinIter) removeParentRow(r sql.Row) sql.Row { // buildRow builds the result set row using the rows from the primary and secondary tables func (i *fullJoinIter) buildRow(primary, secondary sql.Row) sql.Row { row := make(sql.Row, i.rowSize) - copy(row, primary) copy(row[len(primary):], secondary) - return row } @@ -586,6 +582,19 @@ func (i *fullJoinIter) Close(ctx *sql.Context) (err error) { return err } +type crossJoinIterator struct { + l sql.RowIter + r sql.RowIter + rp sql.Node + b sql.NodeExecBuilder + + primaryRow sql.Row + + rowSize int + scopeLen int + parentLen int +} + func newCrossJoinIter(ctx *sql.Context, b sql.NodeExecBuilder, j *plan.JoinNode, row sql.Row) (sql.RowIter, error) { var left, right string if leftTable, ok := j.Left().(sql.Nameable); ok { @@ -611,54 +620,44 @@ func newCrossJoinIter(ctx *sql.Context, b sql.NodeExecBuilder, j *plan.JoinNode, return nil, err } - return sql.NewSpanIter(span, &crossJoinIterator{ - b: b, - parentRow: row, - l: l, - rp: j.Right(), - rowSize: len(row) + len(j.Left().Schema()) + len(j.Right().Schema()), - scopeLen: j.ScopeLen, - }), nil -} + parentLen := len(row) -type crossJoinIterator struct { - l sql.RowIter - r sql.RowIter - rp sql.Node - b sql.NodeExecBuilder + primaryRow := make(sql.Row, parentLen+len(j.Left().Schema())) + copy(primaryRow, row) - parentRow sql.Row + return sql.NewSpanIter(span, &crossJoinIterator{ + b: b, + l: l, + rp: j.Right(), - rowSize int - scopeLen int + primaryRow: primaryRow, - leftRow sql.Row + rowSize: len(row) + len(j.Left().Schema()) + len(j.Right().Schema()), + scopeLen: j.ScopeLen, + parentLen: parentLen, + }), nil } func (i *crossJoinIterator) Next(ctx *sql.Context) (sql.Row, error) { for { - if i.leftRow == nil { + if i.r == nil { r, err := i.l.Next(ctx) if err != nil { return nil, err } + copy(i.primaryRow[i.parentLen:], r) - i.leftRow = i.parentRow.Append(r) - } - - if i.r == nil { - iter, err := i.b.Build(ctx, i.rp, i.leftRow) + // TODO: do we need to keep rebuilding? there's no condition, so i.leftRow shouldn't produce a new iter right? + iter, err := i.b.Build(ctx, i.rp, i.primaryRow) if err != nil { return nil, err } - i.r = iter } rightRow, err := i.r.Next(ctx) if err == io.EOF { i.r = nil - i.leftRow = nil continue } @@ -666,17 +665,16 @@ func (i *crossJoinIterator) Next(ctx *sql.Context) (sql.Row, error) { return nil, err } - var row sql.Row - row = append(row, i.leftRow...) - row = append(row, rightRow...) - + row := make(sql.Row, i.rowSize) + copy(row, i.primaryRow) + copy(row[len(i.primaryRow):], rightRow) return i.removeParentRow(row), nil } } func (i *crossJoinIterator) removeParentRow(r sql.Row) sql.Row { - copy(r[i.scopeLen:], r[len(i.parentRow):]) - r = r[:len(r)-len(i.parentRow)+i.scopeLen] + copy(r[i.scopeLen:], r[i.parentLen:]) + r = r[:len(r)-i.parentLen+i.scopeLen] return r } @@ -684,7 +682,6 @@ func (i *crossJoinIterator) Close(ctx *sql.Context) (err error) { if i.l != nil { err = i.l.Close(ctx) } - if i.r != nil { if err == nil { err = i.r.Close(ctx) @@ -692,7 +689,6 @@ func (i *crossJoinIterator) Close(ctx *sql.Context) (err error) { i.r.Close(ctx) } } - return err } diff --git a/sql/rowexec/range_heap_iter.go b/sql/rowexec/range_heap_iter.go index 279b26380e..12182bbfcf 100644 --- a/sql/rowexec/range_heap_iter.go +++ b/sql/rowexec/range_heap_iter.go @@ -13,6 +13,31 @@ import ( "github.com/dolthub/go-mysql-server/sql/plan" ) +// joinIter is an iterator that iterates over every row in the primary table and performs an index lookup in +// the secondary table for each value +type rangeHeapJoinIter struct { + ctx *sql.Context + err error + b sql.NodeExecBuilder + joinType plan.JoinType + cond sql.Expression + + primary sql.RowIter + primaryRow sql.Row + loadPrimaryRow bool + + secondary sql.RowIter + foundMatch bool + rowSize int + scopeLen int + parentLen int + + rangeHeapPlan *plan.RangeHeap + childRowIter sql.RowIter + pendingRow sql.Row + activeRanges []sql.Row +} + func newRangeHeapJoinIter(ctx *sql.Context, b sql.NodeExecBuilder, j *plan.JoinNode, row sql.Row) (sql.RowIter, error) { var leftName, rightName string if leftTable, ok := j.Left().(sql.Nameable); ok { @@ -43,48 +68,27 @@ func newRangeHeapJoinIter(ctx *sql.Context, b sql.NodeExecBuilder, j *plan.JoinN return nil, errors.New("right side of join must be a range heap") } - primaryRow := make(sql.Row, len(row)+len(j.Left().Schema())) + parentLen := len(row) + + primaryRow := make(sql.Row, parentLen+len(j.Left().Schema())) + copy(primaryRow, row) return sql.NewSpanIter(span, &rangeHeapJoinIter{ - parentRow: row, + ctx: ctx, + b: b, + joinType: j.Op, + cond: j.Filter, + primary: l, primaryRow: primaryRow, loadPrimaryRow: true, - cond: j.Filter, - joinType: j.Op, - rowSize: len(row) + len(j.Left().Schema()) + len(j.Right().Schema()), - scopeLen: j.ScopeLen, - b: b, - rangeHeapPlan: rhp, - ctx: ctx, - }), nil -} -// joinIter is an iterator that iterates over every row in the primary table and performs an index lookup in -// the secondary table for each value -type rangeHeapJoinIter struct { - parentRow sql.Row - primary sql.RowIter - primaryRow sql.Row - secondary sql.RowIter - cond sql.Expression - joinType plan.JoinType - - loadPrimaryRow bool - - foundMatch bool - rowSize int - scopeLen int - b sql.NodeExecBuilder + rowSize: len(row) + len(j.Left().Schema()) + len(j.Right().Schema()), + scopeLen: j.ScopeLen, + parentLen: parentLen, - rangeHeapPlan *plan.RangeHeap - childRowIter sql.RowIter - pendingRow sql.Row - - activeRanges []sql.Row - err error - - ctx *sql.Context + rangeHeapPlan: rhp, + }), nil } func (iter *rangeHeapJoinIter) loadPrimary(ctx *sql.Context) error { @@ -94,7 +98,7 @@ func (iter *rangeHeapJoinIter) loadPrimary(ctx *sql.Context) error { return err } - copy(iter.primaryRow[len(iter.parentRow):], r) + copy(iter.primaryRow[iter.parentLen:], r) iter.foundMatch = false iter.loadPrimaryRow = false @@ -110,7 +114,6 @@ func (iter *rangeHeapJoinIter) loadPrimary(ctx *sql.Context) error { func (iter *rangeHeapJoinIter) loadSecondary(ctx *sql.Context) (sql.Row, error) { if iter.secondary == nil { rowIter, err := iter.getActiveRanges(ctx, iter.b, iter.primaryRow) - if err != nil { return nil, err } @@ -192,18 +195,16 @@ func (iter *rangeHeapJoinIter) Next(ctx *sql.Context) (sql.Row, error) { } func (iter *rangeHeapJoinIter) removeParentRow(r sql.Row) sql.Row { - copy(r[iter.scopeLen:], r[len(iter.parentRow):]) - r = r[:len(r)-len(iter.parentRow)+iter.scopeLen] + copy(r[iter.scopeLen:], r[iter.parentLen:]) + r = r[:len(r)-iter.parentLen+iter.scopeLen] return r } // buildRow builds the result set row using the rows from the primary and secondary tables func (iter *rangeHeapJoinIter) buildRow(primary, secondary sql.Row) sql.Row { row := make(sql.Row, iter.rowSize) - copy(row, primary) copy(row[len(primary):], secondary) - return row } From 67a1fc3f68d933c569b7792fc7437d2b4c077ad0 Mon Sep 17 00:00:00 2001 From: James Cor Date: Tue, 12 Aug 2025 12:11:09 -0700 Subject: [PATCH 3/4] clean up --- sql/rowexec/insert.go | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/sql/rowexec/insert.go b/sql/rowexec/insert.go index 34c375e8af..70238f4ce5 100644 --- a/sql/rowexec/insert.go +++ b/sql/rowexec/insert.go @@ -221,7 +221,7 @@ func (i *insertIter) Next(ctx *sql.Context) (returnRow sql.Row, returnErr error) func (i *insertIter) handleOnDuplicateKeyUpdate(ctx *sql.Context, oldRow, newRow sql.Row) (returnRow sql.Row, returnErr error) { var err error - updateAcc := append(oldRow, newRow...) // TODO: how is this safe? + updateAcc := append(oldRow, newRow...) var evalRow sql.Row for _, updateExpr := range i.updateExprs { // this SET indexes into LHS, but the can @@ -233,13 +233,11 @@ func (i *insertIter) handleOnDuplicateKeyUpdate(ctx *sql.Context, oldRow, newRow if !ok { return nil, err } - val = convertDataAndWarn(ctx, i.schema, newRow, idx, err) } else { return nil, err } } - updateAcc = val.(sql.Row) } // project LHS only From 982897fce5987b5c7d1f8f58b310050a0a73d83d Mon Sep 17 00:00:00 2001 From: James Cor Date: Tue, 12 Aug 2025 13:52:08 -0700 Subject: [PATCH 4/4] feedback --- sql/iters/rel_iters.go | 2 -- sql/rowexec/join_iters.go | 1 - 2 files changed, 3 deletions(-) diff --git a/sql/iters/rel_iters.go b/sql/iters/rel_iters.go index 778310b378..357f24b107 100644 --- a/sql/iters/rel_iters.go +++ b/sql/iters/rel_iters.go @@ -167,8 +167,6 @@ type JsonTableCol struct { pos int finished bool // exhausted all rows in data currSib int - - resultRow sql.Row } // IsSibling returns if the jsonTableCol contains multiple columns diff --git a/sql/rowexec/join_iters.go b/sql/rowexec/join_iters.go index a5a735aa66..30d7de4100 100644 --- a/sql/rowexec/join_iters.go +++ b/sql/rowexec/join_iters.go @@ -647,7 +647,6 @@ func (i *crossJoinIterator) Next(ctx *sql.Context) (sql.Row, error) { } copy(i.primaryRow[i.parentLen:], r) - // TODO: do we need to keep rebuilding? there's no condition, so i.leftRow shouldn't produce a new iter right? iter, err := i.b.Build(ctx, i.rp, i.primaryRow) if err != nil { return nil, err