add fix dup row

Author: elianddb

enginetest/queries/script_queries.go

@@ -120,6 +120,26 @@ type ScriptTestAssertion struct {
 // Unlike other engine tests, ScriptTests must be self-contained. No other tables are created outside the definition of
 // the tests.
 var ScriptTests = []ScriptTest{
+	{
+		// https://github.com/dolthub/dolt/issues/9797
+		Name: "EXISTS subquery returns duplicate rows with PRIMARY KEY",
+		SetUpScript: []string{
+			"CREATE TABLE t(c0 INT, c1 INT, PRIMARY KEY(c0, c1));",
+			"INSERT INTO t VALUES (1, 1);",
+			"INSERT INTO t VALUES (2, 2);",
+			"INSERT INTO t VALUES (2, 3);",
+		},
+		Assertions: []ScriptTestAssertion{
+			{
+				Query: "SELECT * FROM t WHERE EXISTS (SELECT 1 FROM t AS x WHERE x.c0 = t.c0);",
+				Expected: []sql.Row{
+					{1, 1},
+					{2, 2},
+					{2, 3},
+				},
+			},
+		},
+	},
 	{
 		// https://github.com/dolthub/dolt/issues/9794
 		Name: "UPDATE with TRIM function on TEXT column",

sql/analyzer/indexed_joins.go

@@ -472,9 +472,9 @@ func convertSemiToInnerJoin(m *memo.Memo) error {
 		}
 
 		// join and its commute are a new group
-		joinGrp := m.MemoizeInnerJoin(nil, semi.Left, rightGrp, plan.JoinTypeInner, semi.Filter)
+		joinGrp := m.MemoizeInnerJoin(nil, semi.Left, rightGrp, plan.JoinTypeSemi, semi.Filter)
 		// TODO: can't commute if right SubqueryAlias references outside scope (OuterScopeVisibility/IsLateral)
-		m.MemoizeInnerJoin(joinGrp, rightGrp, semi.Left, plan.JoinTypeInner, semi.Filter)
+		m.MemoizeInnerJoin(joinGrp, rightGrp, semi.Left, plan.JoinTypeSemi, semi.Filter)
 
 		// project belongs to the original group
 		leftCols := semi.Left.RelProps.OutputCols()
@@ -1105,7 +1105,7 @@ func addMergeJoins(ctx *sql.Context, m *memo.Memo) error {
 					}
 					if rightIndexMatchesFilters(rIndex, join.Left.RelProps.FuncDeps().Constants(), matchedEqFilters) {
 						jb := join.Copy()
-						if d, ok := jb.Left.First.(*memo.Distinct); ok && lIndex.SqlIdx().IsUnique() {
+									if d, ok := jb.Left.First.(*memo.Distinct); ok && lIndex.SqlIdx().IsUnique() {
 							jb.Left = d.Child
 						}
 						if d, ok := jb.Right.First.(*memo.Distinct); ok && rIndex.SqlIdx().IsUnique() {

sql/analyzer/unnest_exists_subqueries.go

@@ -151,6 +151,7 @@ func unnestExistSubqueries(ctx *sql.Context, scope *plan.Scope, a *Analyzer, fil
 			retFilters = append(retFilters, f)
 			continue
 		}
+		
 
 		// recurse
 		if s.inner != nil {

sql/plan/join.go

@@ -177,7 +177,7 @@ func (i JoinType) IsPartial() bool {
 	switch i {
 	case JoinTypeSemi, JoinTypeAnti, JoinTypeAntiIncludeNulls, JoinTypeSemiHash,
 		JoinTypeAntiHash, JoinTypeAntiHashIncludeNulls, JoinTypeAntiLookup, JoinTypeAntiLookupIncludeNulls,
-		JoinTypeSemiLookup:
+		JoinTypeSemiLookup, JoinTypeSemiMerge, JoinTypeAntiMerge, JoinTypeAntiMergeIncludeNulls:
 		return true
 	default:
 		return false

sql/plan/subquery.go

@@ -18,7 +18,6 @@ import (
 	"fmt"
 	"io"
 	"sync"
-
 	"github.com/dolthub/go-mysql-server/sql"
 	"github.com/dolthub/go-mysql-server/sql/hash"
 	"github.com/dolthub/go-mysql-server/sql/transform"
@@ -431,6 +430,7 @@ func (s *Subquery) HashMultiple(ctx *sql.Context, row sql.Row) (sql.KeyValueCach
 
 // HasResultRow returns whether the subquery has a result set > 0.
 func (s *Subquery) HasResultRow(ctx *sql.Context, row sql.Row) (bool, error) {
+	
 	// First check if the query was cached.
 	s.cacheMu.Lock()
 	cached := s.resultsCached

sql/rowexec/join_iters.go

@@ -367,6 +367,10 @@ func (i *existsIter) Next(ctx *sql.Context) (sql.Row, error) {
 				nextState = esIncRight
 			}
 		case esRet:
+			if i.typ.IsSemi() {
+				// For semi-joins, after returning a match, move to next left row
+				nextState = esIncLeft
+			}
 			return i.removeParentRow(i.primaryRow.Copy()), nil
 		default:
 			return nil, fmt.Errorf("invalid exists join state")

sql/rowexec/merge_join.go

@@ -588,3 +588,4 @@ func (i *mergeJoinIter) Close(ctx *sql.Context) (err error) {
 
 	return err
 }
+

sql/rowexec/rel.go

@@ -282,7 +282,7 @@ func (b *BaseBuilder) buildJoinNode(ctx *sql.Context, n *plan.JoinNode, row sql.
 	case n.Op.IsRange():
 		return newRangeHeapJoinIter(ctx, b, n, row)
 	default:
-		return newJoinIter(ctx, b, n, row)
+			return newJoinIter(ctx, b, n, row)
 	}
 }