Stream lookup join unexpected results

[ulya-sidorina]

ut: KqpNewEngine.JoinIdxLookupWithPredicate

sql:

            $input = (
                SELECT Key, Value1
                FROM `Left` WHERE Value2 == "Value1"
            );

            SELECT t1.Key AS Key, t2.Value AS Value
            FROM $input AS t1
            INNER JOIN `Right` AS t2
            ON t1.Value1 = t2.Key AND t1.Key = t2.Key
            ORDER BY Key, Value

AST with enabled flag:

(                                                                                                                                                                                                                                                             
(let $1 (KqpTable '"/Root/Left" '"72057594046644480:12" '"" '1))                                                                                                                                                                                              
(let $2 '('"Key" '"Value1" '"Value2"))                                                                                                                                                                                                                        
(let $3 (KqpRowsSourceSettings $1 $2 '() (Void) '()))                                                                                                                                                                                                         
(let $4 (DqPhyStage '((DqSource (DataSource '"KqpReadRangesSource") $3)) (lambda '($18) (FromFlow (Map (Filter (ToFlow $18) (lambda '($19) (Coalesce (== (Member $19 '"Value2") (String '"Value1")) (Bool 'false)))) (lambda '($20) (block '(
  (let $21 (Member $20 '"Value1"))                                                                                             
  (return '((AsStruct '('"Key" $21)) (AsStruct '('"Key" (Member $20 '"Key")) '('"Value1" $21))))
)))))) '('('"_logical_id" '1225) '('"_id" '"bb8163a9-4b715974-afbd3a07-8aa90bb4"))))
(let $5 (KqpTable '"/Root/Right" '"72057594046644480:13" '"" '1))
(let $6 '('"Key" '"Value"))         
(let $7 (OptionalType (DataType 'Uint64)))
(let $8 '('"Key" $7))                                                                                                                                                                                                                                         
(let $9 (KqpCnStreamLookup (TDqOutput $4 '0) $5 $6 (ListType (TupleType (StructType $8) (StructType $8 '('"Value1" $7)))) '"LookupJoinRows"))
(let $10 (DqPhyStage '($9) (lambda '($22) $22) '('('"_logical_id" '1276) '('"_id" '"e9260410-9e97f7c5-ba4b2118-a136acb"))))
(let $11 (DqCnUnionAll (TDqOutput $10 '0)))                                                                                    
(let $12 (Uint64 '"1001"))                                                                                                                                                                                                                                    
(let $13 (DqPhyStage '($11) (lambda '($23) (block '(                                                                                                                                                                                                          
  (let $24 (KqpIndexLookupJoin $23 'Inner '"t1" '"t2"))                                                                                                                                                                                                       
  (let $25 (Bool 'true))                                                                                                                                                                                                                                      
  (return (TopSort (ExtractMembers $24 '('"t1.Key" '"t2.Value")) $12 '($25 $25) (lambda '($26) '((Member $26 '"t1.Key") (Member $26 '"t2.Value")))))                                                                                                          ))) '('('"_logical_id" '1386) '('"_id" '"e802725f-6aaecfdd-45a75c07-35e7b4cc"))))                                                                                                                                                                             
(let $14 (DqCnMerge (TDqOutput $13 '0) '('('"t1.Key" '"Asc") '('"t2.Value" '"Asc"))))                                                                                                                                                                         (let $15 (DqPhyStage '($14) (lambda '($27) (FromFlow (OrderedMap (Take (ToFlow $27) $12) (lambda '($28) (AsStruct '('"Key" (Member $28 '"t1.Key")) '('"Value" (Member $28 '"t2.Value"))))))) '('('"_logical_id" '1461) '('"_id" '"917decec-46dd40ad-d6e071e2-2
95a4632"))))                                                                                                                                                                                                                                                  (let $16 '($4 $10 $13 $15))                                                                                                    
(let $17 (DqCnResult (TDqOutput $15 '0) $6))                                                                                                                                                                                                                  (return (KqpPhysicalQuery '((KqpPhysicalTx $16 '($17) '() '('('"type" '"data")))) '((KqpTxResultBinding (ListType (StructType $8 '('"Value" (OptionalType (DataType 'String))))) '0 '0)) '('('"type" '"data_query"))))
) 

AST with disabled flag:

(                                                                                                                                                                                                                                                             
(declare %kqp%tx_result_binding_0_0 (ListType (StructType '('"Key" (OptionalType (DataType 'Uint64))) '('"Value1" (OptionalType (DataType 'Uint64))))))                                                
(let $1 (KqpTable '"/Root/Left" '"72057594046644480:12" '"" '1))                                                                                                                                                                                              
(let $2 '('"Key" '"Value1" '"Value2"))                                                                                         
(let $3 (KqpRowsSourceSettings $1 $2 '() (Void) '()))                                                                          
(let $4 (Bool 'false))                                                                                                         
(let $5 (DqPhyStage '((DqSource (DataSource '"KqpReadRangesSource") $3)) (lambda '($31) (FromFlow (Map (Filter (ToFlow $31) (lambda '($32) (Coalesce (== (Member $32 '"Value2") (String '"Value1")) $4))) (lambda '($33) (AsStruct '('"Key" (Member $33 '"Key"
)) '('"Value1" (Member $33 '"Value1"))))))) '('('"_logical_id" '1338) '('"_id" '"8e6d53c6-9c7fa289-eba4f56f-bfb89674"))))
(let $6 (DqCnUnionAll (TDqOutput $5 '0)))
(let $7 (DqPhyStage '($6) (lambda '($34) $34) '('('"_logical_id" '1850) '('"_id" '"ba4d4924-438f6918-9574bac5-7cc81692"))))
(let $8 (DqCnResult (TDqOutput $7 '0) '()))                                                                                                                                                                                                                   
(let $9 '('('"type" '"data")))      
(let $10 (KqpPhysicalTx '($5 $7) '($8) '() $9))                                                                                
(let $11 '"%kqp%tx_result_binding_0_0")                                                                                        
(let $12 (DataType 'Uint64))                                                                                                                                                                                                                                  
(let $13 (OptionalType $12))                                                                                                                                                                                                                                  
(let $14 '('"Key" $13))                                                                                                                                                                                                                                       
(let $15 (StructType $14 '('"Value1" $13)))                                                                                                                                                                                                                   
(let $16 (ListType $15))                                                                                                                                                                                                                                      
(let $17 %kqp%tx_result_binding_0_0)                                                                                                                                                                                                                          
(let $18 '('('"_logical_id" '1390) '('"_id" '"865a1c3f-cb65518a-f1396d4a-f96e3f3e") '('"_partition_mode" '"single")))                                                                                                                                         (let $19 (DqPhyStage '() (lambda '() (Iterator (PartitionByKey $17 (lambda '($35) (Member $35 '"Value1")) (Void) (Void) (lambda '($36) (Map (Filter (FlatMap $36 (lambda '($37) (Take (Map (Filter (Nth $37 '1) (lambda '($38) (Coalesce (== (Member $38 '"Value1") (Member $38 '"Key")) $4))) (lambda '($39) (AsStruct '('"Value1" (Member $39 '"Value1"))))) (Uint64 '1)))) (lambda '($40) (Exists (Member $40 '"Value1")))) (lambda '($41) (AsStruct '('"Key" (Member $41 '"Value1"))))))))) $18))                       
(let $20 (KqpTable '"/Root/Right" '"72057594046644480:13" '"" '1))                                     
(let $21 '('"Key" '"Value"))                                                                                                                                                                                                                                  
(let $22 (KqpCnStreamLookup (TDqOutput $19 '0) $20 $21 (ListType (StructType $14)) '"LookupRows"))
(let $23 (Uint64 '"1001"))                                                                                                                                                                                                                                    
(let $24 (DqPhyStage '($22) (lambda '($42) (block '(                                                                                                                                                                                                          
  (let $43 '('Many 'Hashed 'Compact))                                                                                                                                                                                                                         
  (let $44 (SqueezeToDict (FlatMap (ToFlow $17) (lambda '($46) (block '(                                                                                                                                                                                      
    (let $47 (Member $46 '"Key"))                                                                                                                                                                                                                             
    (let $48 (Member $46 '"Value1"))                                                                                           
    (let $49 (IfPresent (FilterNullElements (Just '($48 $47)) '('0 '1)) (lambda '($50) (Just '($50 $46))) (Nothing (OptionalType (TupleType (TupleType $12 $12) $15)))))
    (return (If (And (Exists $47) (Exists $48)) $49 (EmptyFrom $49)))  
  )))) (lambda '($51) (Nth $51 '0)) (lambda '($52) (Nth $52 '1)) $43))
  (let $45 (Bool 'true)) 
  (return (FromFlow (TopSort (FlatMap $44 (lambda '($53) (block '(
    (let $54 '('"Key" '"Key"))                                                                                                 
    (let $55 '('"Value1" '"Key"))                                                                                              
    (return (MapJoinCore (OrderedFilter (ToFlow $42) (lambda '($56) (Exists (Member $56 '"Key")))) $53 '"Inner" $54 $55 '('"Value" '"t2.Value") '('"Key" '"t1.Key") $54 $55))
  )))) $23 '($45 $45) (lambda '($57) '((Member $57 '"t1.Key") (Member $57 '"t2.Value"))))))
))) '('('"_logical_id" '1680) '('"_id" '"34bff714-7d4911e2-dbf4750a-19799424"))))
(let $25 (DqCnMerge (TDqOutput $24 '0) '('('"t1.Key" '"Asc") '('"t2.Value" '"Asc"))))
(let $26 (DqPhyStage '($25) (lambda '($58) (FromFlow (OrderedMap (Take (ToFlow $58) $23) (lambda '($59) (AsStruct '('"Key" (Member $59 '"t1.Key")) '('"Value" (Member $59 '"t2.Value"))))))) '('('"_logical_id" '1758) '('"_id" '"3bae1460-9de88ae8-6c9981de-f
147d953"))))
(let $27 '($19 $24 $26))
(let $28 (DqCnResult (TDqOutput $26 '0) $21))
(let $29 (KqpTxResultBinding $16 '0 '0))
(let $30 (KqpPhysicalTx $27 '($28) '('($11 $29)) $9))
(return (KqpPhysicalQuery '($10 $30) '((KqpTxResultBinding (ListType (StructType $14 '('"Value" (OptionalType (DataType 'String))))) '1 '0)) '('('"type" '"data_query"))))
)

Stream join doesn't support such join condition ON t1.Value1 = t2.Key AND t1.Key = t2.Key