Skip to content

Commit

Permalink
planner: Fix the issue that TiDB may dispatch duplicated tasks to TiF…
Browse files Browse the repository at this point in the history
…lash (#32813) (#32839)

ref pingcap/tiflash#4163, close #32814
  • Loading branch information
ti-srebot authored Apr 12, 2022
1 parent 16944f1 commit 7c5373d
Show file tree
Hide file tree
Showing 2 changed files with 55 additions and 1 deletion.
3 changes: 2 additions & 1 deletion planner/core/fragment.go
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,8 @@ func (f *Fragment) init(p PhysicalPlan) error {
}
f.TableScan = x
case *PhysicalExchangeReceiver:
f.singleton = x.children[0].(*PhysicalExchangeSender).ExchangeType == tipb.ExchangeType_PassThrough
// TODO: after we support partial merge, we should check whether all the target exchangeReceiver is same.
f.singleton = f.singleton || x.children[0].(*PhysicalExchangeSender).ExchangeType == tipb.ExchangeType_PassThrough
f.ExchangeReceivers = append(f.ExchangeReceivers, x)
case *PhysicalUnionAll:
return errors.New("unexpected union all detected")
Expand Down
53 changes: 53 additions & 0 deletions planner/core/fragment_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
// Copyright 2020 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// 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 core

import (
"github.com/pingcap/tipb/go-tipb"
"github.com/stretchr/testify/require"

"testing"
)

func TestFragmentInitSingleton(t *testing.T) {
r1, r2 := &PhysicalExchangeReceiver{}, &PhysicalExchangeReceiver{}
r1.SetChildren(&PhysicalExchangeSender{ExchangeType: tipb.ExchangeType_PassThrough})
r2.SetChildren(&PhysicalExchangeSender{ExchangeType: tipb.ExchangeType_Broadcast})
p := &PhysicalHashJoin{}

f := &Fragment{}
p.SetChildren(r1, r1)
err := f.init(p)
require.NoError(t, err)
require.Equal(t, f.singleton, true)

f = &Fragment{}
p.SetChildren(r1, r2)
err = f.init(p)
require.NoError(t, err)
require.Equal(t, f.singleton, true)

f = &Fragment{}
p.SetChildren(r2, r1)
err = f.init(p)
require.NoError(t, err)
require.Equal(t, f.singleton, true)

f = &Fragment{}
p.SetChildren(r2, r2)
err = f.init(p)
require.NoError(t, err)
require.Equal(t, f.singleton, false)
}

0 comments on commit 7c5373d

Please sign in to comment.