From dfe7e3582b63ea7298beab956c7c2d2972f3f75f Mon Sep 17 00:00:00 2001 From: aptend Date: Tue, 11 Aug 2026 10:54:50 +0800 Subject: [PATCH] fix(plan): harden sampled stats and join estimates --- pkg/pb/statsinfo/statsinfo.pb.go | 459 +++++++++++++++++++--- pkg/sql/compile/compile.go | 18 + pkg/sql/compile/compile_test.go | 26 ++ pkg/sql/compile/operator.go | 18 + pkg/sql/compile/operator_test.go | 42 ++ pkg/sql/plan/shuffle.go | 217 +++++++--- pkg/sql/plan/shuffle_test.go | 204 +++++++++- pkg/sql/plan/stats.go | 590 ++++++++++++++++++++++++++-- pkg/sql/plan/stats_test.go | 476 ++++++++++++++++++++++ pkg/vm/engine/disttae/stats.go | 142 ++++++- pkg/vm/engine/disttae/stats_test.go | 118 ++++++ pkg/vm/engine/types.go | 9 +- proto/statsinfo.proto | 3 + 13 files changed, 2169 insertions(+), 153 deletions(-) diff --git a/pkg/pb/statsinfo/statsinfo.pb.go b/pkg/pb/statsinfo/statsinfo.pb.go index 09ee6c6e4e4da..7bb7d4be954d6 100644 --- a/pkg/pb/statsinfo/statsinfo.pb.go +++ b/pkg/pb/statsinfo/statsinfo.pb.go @@ -294,6 +294,10 @@ type StatsInfo struct { ApproxObjectNumber int64 `protobuf:"varint,10,opt,name=ApproxObjectNumber,proto3" json:"ApproxObjectNumber,omitempty"` TableCnt float64 `protobuf:"fixed64,11,opt,name=TableCnt,proto3" json:"TableCnt,omitempty"` TableName string `protobuf:"bytes,12,opt,name=TableName,proto3" json:"TableName,omitempty"` + // int64 TimeSecond = 13; + SampleRatio float64 `protobuf:"fixed64,14,opt,name=SampleRatio,proto3" json:"SampleRatio,omitempty"` + MinMaxValidMap map[string]bool `protobuf:"bytes,15,rep,name=MinMaxValidMap,proto3" json:"MinMaxValidMap,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"varint,2,opt,name=value,proto3"` + MinMaxCompleteMap map[string]bool `protobuf:"bytes,16,rep,name=MinMaxCompleteMap,proto3" json:"MinMaxCompleteMap,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"varint,2,opt,name=value,proto3"` } func (m *StatsInfo) Reset() { *m = StatsInfo{} } @@ -413,6 +417,27 @@ func (m *StatsInfo) GetTableName() string { return "" } +func (m *StatsInfo) GetSampleRatio() float64 { + if m != nil { + return m.SampleRatio + } + return 0 +} + +func (m *StatsInfo) GetMinMaxValidMap() map[string]bool { + if m != nil { + return m.MinMaxValidMap + } + return nil +} + +func (m *StatsInfo) GetMinMaxCompleteMap() map[string]bool { + if m != nil { + return m.MinMaxCompleteMap + } + return nil +} + type StatsInfoKey struct { DatabaseID uint64 `protobuf:"varint,1,opt,name=DatabaseID,proto3" json:"DatabaseID,omitempty"` TableID uint64 `protobuf:"varint,2,opt,name=TableID,proto3" json:"TableID,omitempty"` @@ -539,6 +564,8 @@ func init() { proto.RegisterType((*StatsInfo)(nil), "statsinfo.StatsInfo") proto.RegisterMapType((map[string]uint64)(nil), "statsinfo.StatsInfo.DataTypeMapEntry") proto.RegisterMapType((map[string]float64)(nil), "statsinfo.StatsInfo.MaxValMapEntry") + proto.RegisterMapType((map[string]bool)(nil), "statsinfo.StatsInfo.MinMaxCompleteMapEntry") + proto.RegisterMapType((map[string]bool)(nil), "statsinfo.StatsInfo.MinMaxValidMapEntry") proto.RegisterMapType((map[string]float64)(nil), "statsinfo.StatsInfo.MinValMapEntry") proto.RegisterMapType((map[string]float64)(nil), "statsinfo.StatsInfo.NdvMapEntry") proto.RegisterMapType((map[string]uint64)(nil), "statsinfo.StatsInfo.NullCntMapEntry") @@ -551,64 +578,68 @@ func init() { func init() { proto.RegisterFile("statsinfo.proto", fileDescriptor_a3f8e561c9795adb) } var fileDescriptor_a3f8e561c9795adb = []byte{ - // 901 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x56, 0x4d, 0x6f, 0xe3, 0x44, - 0x18, 0x8e, 0x63, 0x27, 0x8d, 0x5f, 0xa7, 0x1f, 0x1a, 0x55, 0xdd, 0x51, 0x85, 0x82, 0x09, 0x20, - 0x99, 0x15, 0x9b, 0x88, 0x70, 0x59, 0x96, 0x0f, 0xa9, 0xdd, 0x00, 0x1b, 0x76, 0xd3, 0x45, 0x93, - 0xb2, 0x07, 0x38, 0x8d, 0xb3, 0x93, 0xd4, 0xd4, 0xb1, 0x2d, 0xdb, 0x29, 0x49, 0x6f, 0xfc, 0x03, - 0x0e, 0xfc, 0x00, 0x7e, 0xce, 0x1e, 0x7b, 0xe4, 0x84, 0x50, 0x7b, 0xe0, 0x3f, 0x70, 0x42, 0xf3, - 0x8e, 0x9d, 0x38, 0xc5, 0xea, 0xaa, 0xa7, 0xce, 0xf3, 0xce, 0xf3, 0x3c, 0x33, 0xf3, 0xf6, 0x99, - 0x89, 0x61, 0x37, 0x49, 0x79, 0x9a, 0x78, 0xc1, 0x24, 0xec, 0x44, 0x71, 0x98, 0x86, 0xc4, 0x5c, - 0x15, 0x0e, 0x1f, 0x4d, 0xbd, 0xf4, 0x6c, 0xee, 0x76, 0xc6, 0xe1, 0xac, 0x3b, 0x0d, 0xa7, 0x61, - 0x17, 0x19, 0xee, 0x7c, 0x82, 0x08, 0x01, 0x8e, 0x94, 0xb2, 0xfd, 0x8f, 0x06, 0xd6, 0xe8, 0x6c, - 0x3e, 0x99, 0xf8, 0xe2, 0x99, 0xe0, 0x11, 0x79, 0x08, 0xc6, 0x0b, 0x31, 0x49, 0xa9, 0x66, 0x6b, - 0x8e, 0xd5, 0x3b, 0xe8, 0xac, 0x57, 0x2a, 0xb0, 0x18, 0x72, 0xc8, 0xc7, 0x50, 0x63, 0xde, 0xf4, - 0x2c, 0xa5, 0xd5, 0x3b, 0xc9, 0x8a, 0x44, 0xf6, 0x40, 0x7f, 0x2e, 0x96, 0x54, 0xb7, 0x35, 0x47, - 0x63, 0x72, 0x48, 0xf6, 0xa1, 0xf6, 0x8a, 0xfb, 0x73, 0x41, 0x0d, 0xac, 0x29, 0x40, 0x0e, 0xa0, - 0xfe, 0x4c, 0xa0, 0x6d, 0xcd, 0xd6, 0x1c, 0x9d, 0x65, 0x88, 0xec, 0x40, 0x75, 0x74, 0x49, 0xeb, - 0x58, 0xab, 0x8e, 0x2e, 0xa5, 0xfa, 0x64, 0xee, 0xfb, 0x09, 0xdd, 0xc2, 0x92, 0x02, 0x84, 0xc2, - 0x16, 0x13, 0x17, 0x22, 0x4e, 0x04, 0x6d, 0xd8, 0x9a, 0xd3, 0x60, 0x39, 0x6c, 0xff, 0x5b, 0x85, - 0x66, 0xb6, 0x2d, 0xc6, 0x83, 0xa9, 0x20, 0xef, 0x80, 0x39, 0x48, 0x46, 0x69, 0x7c, 0xba, 0x8c, - 0x04, 0x9e, 0xb7, 0xc1, 0xd6, 0x85, 0x6c, 0xb9, 0xea, 0x6a, 0xb9, 0x87, 0x60, 0x9c, 0xc6, 0x42, - 0xe0, 0xfe, 0xef, 0x68, 0x8c, 0xe4, 0xc8, 0xa3, 0x0e, 0xbd, 0x20, 0x3b, 0x96, 0x1c, 0x62, 0x85, - 0x2f, 0xf0, 0x44, 0xb2, 0xc2, 0x17, 0x84, 0x80, 0x31, 0xf4, 0x82, 0x84, 0xd6, 0x6d, 0xdd, 0x69, - 0x32, 0x1c, 0x63, 0x8d, 0x2f, 0xe4, 0x89, 0x54, 0x8d, 0x2f, 0xb0, 0xc6, 0xc2, 0x5f, 0x12, 0xda, - 0xb0, 0x75, 0x47, 0x67, 0x38, 0x5e, 0x1f, 0xdd, 0xc4, 0x62, 0x76, 0xf4, 0x03, 0xa8, 0x0f, 0xf9, - 0xe2, 0x85, 0x08, 0x28, 0xa8, 0xc6, 0x29, 0x24, 0xd9, 0xdf, 0xf8, 0x7c, 0x9a, 0x50, 0xcb, 0xd6, - 0x9d, 0x06, 0x53, 0x40, 0x36, 0xea, 0xe5, 0x85, 0x88, 0x7d, 0x1e, 0xd1, 0x26, 0xee, 0x2a, 0x87, - 0x72, 0xe6, 0x87, 0xc0, 0x9b, 0x84, 0xf1, 0x8c, 0x6e, 0xab, 0x99, 0x0c, 0xca, 0x15, 0x98, 0x48, - 0xe6, 0x7e, 0x4a, 0x77, 0x6c, 0xdd, 0xd1, 0x58, 0x86, 0x88, 0x0d, 0xd6, 0x88, 0xcf, 0x22, 0xd9, - 0xd8, 0xd4, 0x0b, 0xe9, 0x2e, 0xaa, 0x8a, 0xa5, 0xf6, 0xaf, 0x26, 0x98, 0x23, 0xd9, 0xb1, 0x41, - 0x30, 0x09, 0xc9, 0x63, 0xa8, 0x9f, 0xbc, 0xbe, 0x18, 0xf2, 0x88, 0x6a, 0xb6, 0xee, 0x58, 0x3d, - 0xbb, 0xd8, 0xcd, 0x9c, 0xd5, 0x51, 0x94, 0xaf, 0x83, 0x34, 0x5e, 0xb2, 0x8c, 0x4f, 0x8e, 0xc0, - 0x1c, 0x7a, 0xc1, 0x2b, 0xee, 0x4b, 0x71, 0x15, 0xc5, 0xef, 0x97, 0x8a, 0x57, 0x2c, 0xa5, 0x5f, - 0xab, 0xd0, 0x82, 0x2f, 0x32, 0x0b, 0xfd, 0x2e, 0x8b, 0x9c, 0x95, 0x5b, 0xe4, 0x98, 0x7c, 0x0b, - 0x56, 0x9f, 0xa7, 0x5c, 0xe6, 0x44, 0x9a, 0x18, 0x68, 0xf2, 0x61, 0xa9, 0x49, 0x81, 0xa7, 0x6c, - 0x8a, 0x4a, 0xd2, 0x07, 0x90, 0xff, 0xbb, 0xa7, 0x41, 0x2a, 0x7d, 0x6a, 0xe8, 0xf3, 0x41, 0x79, - 0x33, 0x56, 0x34, 0x65, 0x53, 0xd0, 0x91, 0xcf, 0x61, 0x6b, 0xe4, 0x5d, 0xe2, 0x56, 0xea, 0x68, - 0xf1, 0x5e, 0xa9, 0x45, 0xc6, 0x51, 0xfa, 0x5c, 0x41, 0x46, 0xb0, 0x5b, 0xbc, 0x15, 0xd2, 0x64, - 0x0b, 0x4d, 0x3e, 0x2a, 0x37, 0xd9, 0xe4, 0x2a, 0xb3, 0xdb, 0x0e, 0x32, 0x10, 0xc7, 0x7e, 0x38, - 0x3e, 0x3f, 0x99, 0xcf, 0x5c, 0x11, 0xe3, 0x4d, 0xd4, 0x59, 0xb1, 0x44, 0x7a, 0xb0, 0x7f, 0x34, - 0x1e, 0xcf, 0x63, 0x9e, 0x8a, 0x97, 0xee, 0xcf, 0x62, 0x9c, 0x66, 0x54, 0x13, 0xa9, 0xa5, 0x73, - 0xa4, 0x03, 0xe4, 0x28, 0x8a, 0xe2, 0x70, 0xb1, 0xa1, 0x50, 0x61, 0x2f, 0x99, 0x21, 0x87, 0xd0, - 0x38, 0xe5, 0xae, 0x2f, 0x9e, 0x06, 0x29, 0xb5, 0x30, 0x93, 0x2b, 0x2c, 0x2f, 0x3f, 0x8e, 0x4f, - 0xf8, 0x4c, 0xe0, 0x05, 0x30, 0xd9, 0xba, 0x70, 0xf8, 0x19, 0x58, 0x85, 0xf4, 0xc9, 0xdb, 0x7b, - 0x2e, 0x96, 0xf8, 0x46, 0x98, 0x4c, 0x0e, 0xe5, 0x9d, 0xba, 0xc0, 0xa7, 0xab, 0xaa, 0x9e, 0x2e, - 0x04, 0x4f, 0xaa, 0x8f, 0xb5, 0xc3, 0x2f, 0x60, 0x67, 0x33, 0x7b, 0xf7, 0x56, 0x6f, 0xc4, 0xee, - 0x5e, 0xea, 0xaf, 0x60, 0xef, 0x76, 0xde, 0xde, 0xa6, 0x37, 0x8a, 0xfa, 0x2f, 0x61, 0xf7, 0x56, - 0xce, 0xee, 0x25, 0x7f, 0x02, 0xcd, 0x62, 0xc6, 0xee, 0xa5, 0xfd, 0x09, 0xf6, 0xcb, 0xa2, 0x55, - 0xe2, 0xf1, 0xa8, 0xe8, 0x61, 0xf5, 0x1e, 0xfc, 0xff, 0x25, 0x46, 0x87, 0x82, 0xf9, 0x77, 0x46, - 0x63, 0x7b, 0x6f, 0xa7, 0xfd, 0xbb, 0x06, 0xcd, 0x55, 0x90, 0xe5, 0xef, 0x4f, 0x0b, 0x40, 0xb6, - 0xcb, 0xe5, 0x89, 0x18, 0xf4, 0x71, 0x09, 0x83, 0x15, 0x2a, 0xf2, 0x21, 0xc4, 0x48, 0x0c, 0xfa, - 0xd9, 0x7e, 0x73, 0x28, 0xcf, 0x71, 0x34, 0x1e, 0x0f, 0x5e, 0xe3, 0xaf, 0xc1, 0x36, 0x53, 0x60, - 0x33, 0x53, 0xc6, 0xad, 0x4c, 0xc9, 0xc7, 0xb3, 0xef, 0xe2, 0x54, 0x0d, 0xa7, 0x32, 0xd4, 0x3e, - 0x86, 0xed, 0xe2, 0xae, 0x12, 0xf2, 0x09, 0x18, 0xf2, 0x6f, 0xf6, 0x36, 0x3e, 0x28, 0xbb, 0x86, - 0xcf, 0xc5, 0xf2, 0xd8, 0x78, 0xf3, 0xd7, 0xbb, 0x15, 0x86, 0xd4, 0xe3, 0xef, 0xdf, 0x5c, 0xb7, - 0xb4, 0xab, 0xeb, 0x96, 0xf6, 0xf7, 0x75, 0xab, 0xf2, 0xdb, 0x4d, 0xab, 0xf2, 0xc7, 0x4d, 0x4b, - 0xbb, 0xba, 0x69, 0x55, 0xfe, 0xbc, 0x69, 0x55, 0x7e, 0xec, 0x15, 0x3e, 0x09, 0x66, 0x3c, 0x8d, - 0xbd, 0x45, 0x18, 0x7b, 0x53, 0x2f, 0xc8, 0x41, 0x20, 0xba, 0xd1, 0xf9, 0xb4, 0x1b, 0xb9, 0xdd, - 0xd5, 0x52, 0x6e, 0x1d, 0x3f, 0x0f, 0x3e, 0xfd, 0x2f, 0x00, 0x00, 0xff, 0xff, 0x25, 0xda, 0x97, - 0xc9, 0x6b, 0x08, 0x00, 0x00, + // 973 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x56, 0x4f, 0x6f, 0xe3, 0x44, + 0x14, 0xaf, 0x63, 0x37, 0x4d, 0x9e, 0xd3, 0xa4, 0x0c, 0x55, 0xd6, 0xaa, 0x50, 0x30, 0x01, 0x24, + 0xb3, 0xb0, 0x89, 0x08, 0x97, 0x65, 0xf9, 0x23, 0xa5, 0x0d, 0xb0, 0x61, 0x37, 0xdd, 0xd5, 0xa4, + 0xac, 0x04, 0x9c, 0x26, 0xe9, 0x24, 0x35, 0x75, 0x6c, 0xcb, 0x76, 0x4a, 0xd2, 0x4f, 0xc1, 0x81, + 0x0f, 0xc0, 0xc7, 0xd9, 0xe3, 0x1e, 0xe1, 0x82, 0x50, 0x7b, 0xe0, 0x3b, 0x70, 0x42, 0xf3, 0xc6, + 0x49, 0xa6, 0xa9, 0xb7, 0xab, 0x9c, 0x32, 0xbf, 0x37, 0xbf, 0xdf, 0x6f, 0x66, 0x9e, 0xdf, 0x9b, + 0x09, 0x54, 0xe2, 0x84, 0x25, 0xb1, 0xeb, 0x8f, 0x82, 0x46, 0x18, 0x05, 0x49, 0x40, 0x8a, 0xcb, + 0xc0, 0xc1, 0x83, 0xb1, 0x9b, 0x9c, 0x4d, 0x07, 0x8d, 0x61, 0x30, 0x69, 0x8e, 0x83, 0x71, 0xd0, + 0x44, 0xc6, 0x60, 0x3a, 0x42, 0x84, 0x00, 0x47, 0x52, 0x59, 0xff, 0x57, 0x03, 0xb3, 0x7f, 0x36, + 0x1d, 0x8d, 0x3c, 0xfe, 0x98, 0xb3, 0x90, 0xdc, 0x07, 0xe3, 0x29, 0x1f, 0x25, 0x96, 0x66, 0x6b, + 0x8e, 0xd9, 0xaa, 0x36, 0x56, 0x2b, 0x29, 0x2c, 0x8a, 0x1c, 0xf2, 0x09, 0x6c, 0x53, 0x77, 0x7c, + 0x96, 0x58, 0xb9, 0x3b, 0xc9, 0x92, 0x44, 0xf6, 0x40, 0x7f, 0xc2, 0xe7, 0x96, 0x6e, 0x6b, 0x8e, + 0x46, 0xc5, 0x90, 0xec, 0xc3, 0xf6, 0x0b, 0xe6, 0x4d, 0xb9, 0x65, 0x60, 0x4c, 0x02, 0x52, 0x85, + 0xfc, 0x63, 0x8e, 0xb6, 0xdb, 0xb6, 0xe6, 0xe8, 0x34, 0x45, 0xa4, 0x0c, 0xb9, 0xfe, 0xa5, 0x95, + 0xc7, 0x58, 0xae, 0x7f, 0x29, 0xd4, 0xc7, 0x53, 0xcf, 0x8b, 0xad, 0x1d, 0x0c, 0x49, 0x40, 0x2c, + 0xd8, 0xa1, 0xfc, 0x82, 0x47, 0x31, 0xb7, 0x0a, 0xb6, 0xe6, 0x14, 0xe8, 0x02, 0xd6, 0xff, 0xcb, + 0x41, 0x29, 0xdd, 0x16, 0x65, 0xfe, 0x98, 0x93, 0x77, 0xa0, 0xd8, 0x8d, 0xfb, 0x49, 0x74, 0x32, + 0x0f, 0x39, 0x9e, 0xb7, 0x40, 0x57, 0x81, 0x74, 0xb9, 0xdc, 0x72, 0xb9, 0xfb, 0x60, 0x9c, 0x44, + 0x9c, 0xe3, 0xfe, 0xef, 0x48, 0x8c, 0xe0, 0x88, 0xa3, 0xf6, 0x5c, 0x3f, 0x3d, 0x96, 0x18, 0x62, + 0x84, 0xcd, 0xf0, 0x44, 0x22, 0xc2, 0x66, 0x84, 0x80, 0xd1, 0x73, 0xfd, 0xd8, 0xca, 0xdb, 0xba, + 0x53, 0xa2, 0x38, 0xc6, 0x18, 0x9b, 0x89, 0x13, 0xc9, 0x18, 0x9b, 0x61, 0x8c, 0x06, 0xbf, 0xc6, + 0x56, 0xc1, 0xd6, 0x1d, 0x9d, 0xe2, 0x78, 0x75, 0xf4, 0x22, 0x06, 0xd3, 0xa3, 0x57, 0x21, 0xdf, + 0x63, 0xb3, 0xa7, 0xdc, 0xb7, 0x40, 0x26, 0x4e, 0x22, 0xc1, 0xfe, 0xd6, 0x63, 0xe3, 0xd8, 0x32, + 0x6d, 0xdd, 0x29, 0x50, 0x09, 0x44, 0xa2, 0x9e, 0x5d, 0xf0, 0xc8, 0x63, 0xa1, 0x55, 0xc2, 0x5d, + 0x2d, 0xa0, 0x98, 0xf9, 0xc1, 0x77, 0x47, 0x41, 0x34, 0xb1, 0x76, 0xe5, 0x4c, 0x0a, 0xc5, 0x0a, + 0x94, 0xc7, 0x53, 0x2f, 0xb1, 0xca, 0xb6, 0xee, 0x68, 0x34, 0x45, 0xc4, 0x06, 0xb3, 0xcf, 0x26, + 0xa1, 0x48, 0x6c, 0xe2, 0x06, 0x56, 0x05, 0x55, 0x6a, 0xa8, 0xfe, 0x97, 0x09, 0xc5, 0xbe, 0xc8, + 0x58, 0xd7, 0x1f, 0x05, 0xe4, 0x21, 0xe4, 0x8f, 0x4f, 0x2f, 0x7a, 0x2c, 0xb4, 0x34, 0x5b, 0x77, + 0xcc, 0x96, 0xad, 0x66, 0x73, 0xc1, 0x6a, 0x48, 0xca, 0x37, 0x7e, 0x12, 0xcd, 0x69, 0xca, 0x27, + 0x6d, 0x28, 0xf6, 0x5c, 0xff, 0x05, 0xf3, 0x84, 0x38, 0x87, 0xe2, 0xf7, 0x33, 0xc5, 0x4b, 0x96, + 0xd4, 0xaf, 0x54, 0x68, 0xc1, 0x66, 0xa9, 0x85, 0x7e, 0x97, 0xc5, 0x82, 0xb5, 0xb0, 0x58, 0x60, + 0xf2, 0x1d, 0x98, 0x1d, 0x96, 0x30, 0x51, 0x27, 0xc2, 0xc4, 0x40, 0x93, 0x0f, 0x33, 0x4d, 0x14, + 0x9e, 0xb4, 0x51, 0x95, 0xa4, 0x03, 0x20, 0xbe, 0xdd, 0x91, 0x9f, 0x08, 0x9f, 0x6d, 0xf4, 0xf9, + 0x20, 0x3b, 0x19, 0x4b, 0x9a, 0xb4, 0x51, 0x74, 0xe4, 0x0b, 0xd8, 0xe9, 0xbb, 0x97, 0xb8, 0x95, + 0x3c, 0x5a, 0xbc, 0x97, 0x69, 0x91, 0x72, 0xa4, 0x7e, 0xa1, 0x20, 0x7d, 0xa8, 0xa8, 0x5d, 0x21, + 0x4c, 0x76, 0xd0, 0xe4, 0xa3, 0x6c, 0x93, 0x9b, 0x5c, 0x69, 0xb6, 0xee, 0x20, 0x0a, 0xe2, 0xd0, + 0x0b, 0x86, 0xe7, 0xc7, 0xd3, 0xc9, 0x80, 0x47, 0xd8, 0x89, 0x3a, 0x55, 0x43, 0xa4, 0x05, 0xfb, + 0xed, 0xe1, 0x70, 0x1a, 0xb1, 0x84, 0x3f, 0x1b, 0xfc, 0xc2, 0x87, 0x49, 0x4a, 0x2d, 0x22, 0x35, + 0x73, 0x8e, 0x34, 0x80, 0xb4, 0xc3, 0x30, 0x0a, 0x66, 0x37, 0x14, 0xb2, 0xd8, 0x33, 0x66, 0xc8, + 0x01, 0x14, 0x4e, 0xd8, 0xc0, 0xe3, 0x47, 0x7e, 0x62, 0x99, 0x58, 0x93, 0x4b, 0x2c, 0x9a, 0x1f, + 0xc7, 0xc7, 0x6c, 0xc2, 0xb1, 0x01, 0x8a, 0x74, 0x15, 0x58, 0x2f, 0xe8, 0xf2, 0xad, 0x82, 0x26, + 0xcf, 0xa1, 0xdc, 0x73, 0x7d, 0x59, 0x12, 0xee, 0xa9, 0xc8, 0x5a, 0x05, 0xb3, 0xe6, 0xbc, 0xae, + 0x1a, 0x15, 0xaa, 0x4c, 0xda, 0x9a, 0x9e, 0xfc, 0x08, 0x6f, 0xc9, 0xc8, 0x51, 0x20, 0x96, 0x49, + 0xf0, 0x53, 0xec, 0xa1, 0xe9, 0xc7, 0x77, 0x98, 0x2a, 0x6c, 0xe9, 0x7b, 0xdb, 0xe5, 0xe0, 0x73, + 0x30, 0x95, 0x66, 0x12, 0x97, 0xd1, 0x39, 0x9f, 0xe3, 0x95, 0x57, 0xa4, 0x62, 0x28, 0xae, 0x88, + 0x0b, 0xbc, 0x89, 0x73, 0xf2, 0x26, 0x46, 0xf0, 0x28, 0xf7, 0x50, 0x3b, 0xf8, 0x12, 0xcf, 0xa9, + 0xf4, 0xc1, 0xc6, 0xea, 0x1b, 0x5d, 0xb4, 0x91, 0xfa, 0x6b, 0xd8, 0x5b, 0x6f, 0x9f, 0x37, 0xe9, + 0x0d, 0x55, 0xff, 0x15, 0x54, 0xd6, 0xda, 0x66, 0x23, 0xf9, 0x23, 0x28, 0xa9, 0x2d, 0xb3, 0x91, + 0xf6, 0x67, 0xd8, 0xcf, 0xea, 0x94, 0x0c, 0x8f, 0x07, 0xaa, 0x87, 0xd9, 0xba, 0x77, 0xfb, 0x61, + 0x41, 0x07, 0xd5, 0xbc, 0x0d, 0x6f, 0x67, 0x14, 0xd4, 0x9b, 0xf6, 0x57, 0x50, 0x2d, 0x3a, 0x50, + 0xcd, 0x2e, 0x9f, 0x4d, 0x5c, 0xbe, 0x37, 0x0a, 0xbb, 0x7b, 0xe5, 0xfa, 0xef, 0x1a, 0x94, 0x96, + 0x55, 0x29, 0xde, 0xf5, 0x1a, 0x80, 0xf8, 0x6e, 0x03, 0x16, 0xf3, 0x6e, 0x07, 0x9d, 0x0c, 0xaa, + 0x44, 0xc4, 0x03, 0x83, 0xad, 0xd6, 0xed, 0xa4, 0x89, 0x5b, 0x40, 0xb1, 0x54, 0x7b, 0x38, 0xec, + 0x9e, 0xe2, 0x2b, 0xbb, 0x4b, 0x25, 0xb8, 0xd9, 0xab, 0xc6, 0x7a, 0xaf, 0x56, 0x21, 0xdf, 0x19, + 0xe0, 0xd4, 0x36, 0x4e, 0xa5, 0xa8, 0x7e, 0x08, 0xbb, 0xea, 0xae, 0x62, 0xf2, 0x29, 0x18, 0xe2, + 0x37, 0x7d, 0x73, 0xee, 0x65, 0xf5, 0xd4, 0x13, 0x3e, 0x3f, 0x34, 0x5e, 0xfe, 0xfd, 0xee, 0x16, + 0x45, 0xea, 0xe1, 0xf3, 0x97, 0x57, 0x35, 0xed, 0xd5, 0x55, 0x4d, 0xfb, 0xe7, 0xaa, 0xb6, 0xf5, + 0xdb, 0x75, 0x6d, 0xeb, 0x8f, 0xeb, 0x9a, 0xf6, 0xea, 0xba, 0xb6, 0xf5, 0xe7, 0x75, 0x6d, 0xeb, + 0xa7, 0x96, 0xf2, 0x57, 0x6b, 0xc2, 0x92, 0xc8, 0x9d, 0x05, 0x91, 0x3b, 0x76, 0xfd, 0x05, 0xf0, + 0x79, 0x33, 0x3c, 0x1f, 0x37, 0xc3, 0x41, 0x73, 0xb9, 0xd4, 0x20, 0x8f, 0x7f, 0xbb, 0x3e, 0xfb, + 0x3f, 0x00, 0x00, 0xff, 0xff, 0x15, 0x7b, 0x08, 0x2c, 0xc3, 0x09, 0x00, 0x00, } func (m *ShuffleHeap) Marshal() (dAtA []byte, err error) { @@ -879,6 +910,58 @@ func (m *StatsInfo) MarshalToSizedBuffer(dAtA []byte) (int, error) { _ = i var l int _ = l + if len(m.MinMaxCompleteMap) > 0 { + for k := range m.MinMaxCompleteMap { + v := m.MinMaxCompleteMap[k] + baseI := i + i-- + if v { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x10 + i -= len(k) + copy(dAtA[i:], k) + i = encodeVarintStatsinfo(dAtA, i, uint64(len(k))) + i-- + dAtA[i] = 0xa + i = encodeVarintStatsinfo(dAtA, i, uint64(baseI-i)) + i-- + dAtA[i] = 0x1 + i-- + dAtA[i] = 0x82 + } + } + if len(m.MinMaxValidMap) > 0 { + for k := range m.MinMaxValidMap { + v := m.MinMaxValidMap[k] + baseI := i + i-- + if v { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x10 + i -= len(k) + copy(dAtA[i:], k) + i = encodeVarintStatsinfo(dAtA, i, uint64(len(k))) + i-- + dAtA[i] = 0xa + i = encodeVarintStatsinfo(dAtA, i, uint64(baseI-i)) + i-- + dAtA[i] = 0x7a + } + } + if m.SampleRatio != 0 { + i -= 8 + encoding_binary.LittleEndian.PutUint64(dAtA[i:], uint64(math.Float64bits(float64(m.SampleRatio)))) + i-- + dAtA[i] = 0x71 + } if len(m.TableName) > 0 { i -= len(m.TableName) copy(dAtA[i:], m.TableName) @@ -1328,6 +1411,25 @@ func (m *StatsInfo) ProtoSize() (n int) { if l > 0 { n += 1 + l + sovStatsinfo(uint64(l)) } + if m.SampleRatio != 0 { + n += 9 + } + if len(m.MinMaxValidMap) > 0 { + for k, v := range m.MinMaxValidMap { + _ = k + _ = v + mapEntrySize := 1 + len(k) + sovStatsinfo(uint64(len(k))) + 1 + 1 + n += mapEntrySize + 1 + sovStatsinfo(uint64(mapEntrySize)) + } + } + if len(m.MinMaxCompleteMap) > 0 { + for k, v := range m.MinMaxCompleteMap { + _ = k + _ = v + mapEntrySize := 1 + len(k) + sovStatsinfo(uint64(len(k))) + 1 + 1 + n += mapEntrySize + 2 + sovStatsinfo(uint64(mapEntrySize)) + } + } return n } @@ -3053,6 +3155,247 @@ func (m *StatsInfo) Unmarshal(dAtA []byte) error { } m.TableName = string(dAtA[iNdEx:postIndex]) iNdEx = postIndex + case 14: + if wireType != 1 { + return fmt.Errorf("proto: wrong wireType = %d for field SampleRatio", wireType) + } + var v uint64 + if (iNdEx + 8) > l { + return io.ErrUnexpectedEOF + } + v = uint64(encoding_binary.LittleEndian.Uint64(dAtA[iNdEx:])) + iNdEx += 8 + m.SampleRatio = float64(math.Float64frombits(v)) + case 15: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field MinMaxValidMap", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowStatsinfo + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthStatsinfo + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthStatsinfo + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.MinMaxValidMap == nil { + m.MinMaxValidMap = make(map[string]bool) + } + var mapkey string + var mapvalue bool + for iNdEx < postIndex { + entryPreIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowStatsinfo + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + if fieldNum == 1 { + var stringLenmapkey uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowStatsinfo + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLenmapkey |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLenmapkey := int(stringLenmapkey) + if intStringLenmapkey < 0 { + return ErrInvalidLengthStatsinfo + } + postStringIndexmapkey := iNdEx + intStringLenmapkey + if postStringIndexmapkey < 0 { + return ErrInvalidLengthStatsinfo + } + if postStringIndexmapkey > l { + return io.ErrUnexpectedEOF + } + mapkey = string(dAtA[iNdEx:postStringIndexmapkey]) + iNdEx = postStringIndexmapkey + } else if fieldNum == 2 { + var mapvaluetemp int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowStatsinfo + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + mapvaluetemp |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + mapvalue = bool(mapvaluetemp != 0) + } else { + iNdEx = entryPreIndex + skippy, err := skipStatsinfo(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthStatsinfo + } + if (iNdEx + skippy) > postIndex { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + m.MinMaxValidMap[mapkey] = mapvalue + iNdEx = postIndex + case 16: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field MinMaxCompleteMap", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowStatsinfo + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthStatsinfo + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthStatsinfo + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.MinMaxCompleteMap == nil { + m.MinMaxCompleteMap = make(map[string]bool) + } + var mapkey string + var mapvalue bool + for iNdEx < postIndex { + entryPreIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowStatsinfo + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + if fieldNum == 1 { + var stringLenmapkey uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowStatsinfo + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLenmapkey |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLenmapkey := int(stringLenmapkey) + if intStringLenmapkey < 0 { + return ErrInvalidLengthStatsinfo + } + postStringIndexmapkey := iNdEx + intStringLenmapkey + if postStringIndexmapkey < 0 { + return ErrInvalidLengthStatsinfo + } + if postStringIndexmapkey > l { + return io.ErrUnexpectedEOF + } + mapkey = string(dAtA[iNdEx:postStringIndexmapkey]) + iNdEx = postStringIndexmapkey + } else if fieldNum == 2 { + var mapvaluetemp int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowStatsinfo + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + mapvaluetemp |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + mapvalue = bool(mapvaluetemp != 0) + } else { + iNdEx = entryPreIndex + skippy, err := skipStatsinfo(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthStatsinfo + } + if (iNdEx + skippy) > postIndex { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + m.MinMaxCompleteMap[mapkey] = mapvalue + iNdEx = postIndex default: iNdEx = preIndex skippy, err := skipStatsinfo(dAtA[iNdEx:]) diff --git a/pkg/sql/compile/compile.go b/pkg/sql/compile/compile.go index fbc539ad2f0a0..97a2e2caf96a2 100644 --- a/pkg/sql/compile/compile.go +++ b/pkg/sql/compile/compile.go @@ -3575,6 +3575,7 @@ func (c *Compile) compileTableScan(node *plan.Node) ([]*Scope, error) { if err != nil { return nil, err } + finalizeTableScanShuffleForCNCount(node, len(nodes)) ss := make([]*Scope, 0, len(nodes)) currentFirstFlag := c.anal.isFirst @@ -3597,6 +3598,23 @@ func (c *Compile) compileTableScan(node *plan.Node) ([]*Scope, error) { return ss, nil } +// finalizeTableScanShuffleForCNCount makes the coordinator the single owner of +// the cross-CN object-partitioning decision. ShuffleRangeReEval cannot produce +// one boundary per CN when the quantile list is too short; encode hash in the +// existing plan field before scopes are serialized so old and new CNs cannot +// choose different owners for the same object. +func finalizeTableScanShuffleForCNCount(node *plan.Node, cnCount int) { + if node == nil || node.Stats == nil || node.Stats.HashmapStats == nil || + node.Stats.HashmapStats.ShuffleType != plan.ShuffleType_Range || + len(node.Stats.HashmapStats.Ranges) == 0 || cnCount <= len(node.Stats.HashmapStats.Ranges)/2 { + return + } + node.Stats.HashmapStats.ShuffleType = plan.ShuffleType_Hash + node.Stats.HashmapStats.Ranges = nil + node.Stats.HashmapStats.ShuffleColMin = 0 + node.Stats.HashmapStats.ShuffleColMax = 0 +} + func (c *Compile) compileTableScanWithNode(node *plan.Node, engNode engine.Node, firstFlag bool) (*Scope, error) { s := newScope(Remote) s.NodeInfo = engNode diff --git a/pkg/sql/compile/compile_test.go b/pkg/sql/compile/compile_test.go index ee54408bc7621..208cfb1bc3dc6 100644 --- a/pkg/sql/compile/compile_test.go +++ b/pkg/sql/compile/compile_test.go @@ -923,6 +923,32 @@ func TestCompileShuffleGroupGatesOrderedAggregateByProtocolVersion(t *testing.T) "legacy shuffle aggregates remain safe on protocol v5") } +func TestFinalizeTableScanShuffleForCNCount(t *testing.T) { + newNode := func(ranges []float64) *plan.Node { + return &plan.Node{Stats: &plan.Stats{HashmapStats: &plan.HashMapStats{ + ShuffleType: plan.ShuffleType_Range, + ShuffleColMin: 1, + ShuffleColMax: 100, + Ranges: ranges, + }}} + } + + short := newNode([]float64{1, 50, 100}) + finalizeTableScanShuffleForCNCount(short, 2) + require.Equal(t, plan.ShuffleType_Hash, short.Stats.HashmapStats.ShuffleType) + require.Nil(t, short.Stats.HashmapStats.Ranges) + require.Zero(t, short.Stats.HashmapStats.ShuffleColMin) + require.Zero(t, short.Stats.HashmapStats.ShuffleColMax) + + enough := newNode([]float64{1, 25, 50, 100}) + finalizeTableScanShuffleForCNCount(enough, 2) + require.Equal(t, plan.ShuffleType_Range, enough.Stats.HashmapStats.ShuffleType) + + boundsOnly := newNode(nil) + finalizeTableScanShuffleForCNCount(boundsOnly, 512) + require.Equal(t, plan.ShuffleType_Range, boundsOnly.Stats.HashmapStats.ShuffleType) +} + func TestCompileShuffleGroupUsesDistributedPathWhenInputScopesNotSingle(t *testing.T) { c := newCompileForShuffleGroupTest(t) aggNode, nodes := newShuffleGroupTestNodes(16) diff --git a/pkg/sql/compile/operator.go b/pkg/sql/compile/operator.go index 8b7f2fdefcbd2..cad48a8ad0eb4 100644 --- a/pkg/sql/compile/operator.go +++ b/pkg/sql/compile/operator.go @@ -1928,6 +1928,7 @@ func constructShuffleOperatorForJoin(bucketNum int32, node *plan.Node, left bool case types.T_uint64, types.T_uint32, types.T_uint16, types.T_varchar, types.T_char, types.T_text, types.T_bit, types.T_datalink: arg.ShuffleRangeUint64 = plan2.ShuffleRangeReEvalUnsigned(node.Stats.HashmapStats.Ranges, int(arg.BucketNum), node.Stats.HashmapStats.Nullcnt, int64(node.Stats.TableCnt)) } + fallbackRangeShuffleToHash(arg, node) if left && len(node.RuntimeFilterProbeList) > 0 { arg.RuntimeFilterSpec = plan2.DeepCopyRuntimeFilterSpec(node.RuntimeFilterProbeList[0]) } @@ -1953,9 +1954,26 @@ func constructShuffleArgForGroup(bucketNum int32, node *plan.Node) *shuffle.Shuf case types.T_uint64, types.T_uint32, types.T_uint16, types.T_varchar, types.T_char, types.T_text, types.T_bit, types.T_datalink: arg.ShuffleRangeUint64 = plan2.ShuffleRangeReEvalUnsigned(node.Stats.HashmapStats.Ranges, int(arg.BucketNum), node.Stats.HashmapStats.Nullcnt, int64(node.Stats.TableCnt)) } + fallbackRangeShuffleToHash(arg, node) return arg } +// Quantile boundaries are optional at plan time because bucket count is only +// known during compilation. If re-evaluation cannot produce one boundary per +// bucket, hash shuffle is the only safe fallback: sampled stats deliberately do +// not provide global min/max, whose zero values would otherwise create extreme +// range skew. +func fallbackRangeShuffleToHash(arg *shuffle.Shuffle, node *plan.Node) { + if arg.ShuffleType != int32(plan.ShuffleType_Range) || + node == nil || node.Stats == nil || node.Stats.HashmapStats == nil || + len(node.Stats.HashmapStats.Ranges) == 0 { + return + } + if len(arg.ShuffleRangeInt64) == 0 && len(arg.ShuffleRangeUint64) == 0 { + arg.ShuffleType = int32(plan.ShuffleType_Hash) + } +} + // cross-cn dispath will send same batch to all register func constructDispatch(idx int, target []*Scope, source *Scope, node *plan.Node, left bool) *dispatch.Dispatch { hasRemote, arg := constructDispatchLocalAndRemote(idx, target, source) diff --git a/pkg/sql/compile/operator_test.go b/pkg/sql/compile/operator_test.go index 0f532527c3e63..5bdd66d71335f 100644 --- a/pkg/sql/compile/operator_test.go +++ b/pkg/sql/compile/operator_test.go @@ -802,6 +802,48 @@ func TestRangeShuffleJoinSingleBucketSkewedBatch(t *testing.T) { require.Equal(t, rowCount, rows) } +func TestRangeShuffleFallsBackToHashWhenQuantilesDoNotCoverBuckets(t *testing.T) { + makeNode := func(ranges []float64) *plan.Node { + left := &plan.Expr{ + Typ: plan.Type{Id: int32(types.T_int64)}, + Expr: &plan.Expr_Col{Col: &plan.ColRef{ColPos: 0}}, + } + right := &plan.Expr{ + Typ: plan.Type{Id: int32(types.T_int64)}, + Expr: &plan.Expr_Col{Col: &plan.ColRef{ColPos: 0}}, + } + return &plan.Node{ + OnList: []*plan.Expr{{Expr: &plan.Expr_F{F: &plan.Function{Args: []*plan.Expr{left, right}}}}}, + GroupBy: []*plan.Expr{left}, + Stats: &plan.Stats{ + TableCnt: 1_000_000, + HashmapStats: &plan.HashMapStats{ + ShuffleType: plan.ShuffleType_Range, + ShuffleColIdx: 0, + Ranges: ranges, + }, + }, + } + } + + collectorRanges := make([]float64, 1023) + for i := range collectorRanges { + collectorRanges[i] = float64(i) + } + node := makeNode(collectorRanges) + require.Equal(t, int32(plan.ShuffleType_Hash), constructShuffleOperatorForJoin(512, node, true).ShuffleType) + require.Equal(t, int32(plan.ShuffleType_Hash), constructShuffleArgForGroup(512, node).ShuffleType) + require.Equal(t, int32(plan.ShuffleType_Range), constructShuffleOperatorForJoin(511, node, true).ShuffleType) + + shortNode := makeNode([]float64{1, 25, 50, 75, 100}) + require.Equal(t, int32(plan.ShuffleType_Hash), constructShuffleOperatorForJoin(4, shortNode, true).ShuffleType) + + boundsOnly := makeNode(nil) + boundsOnly.Stats.HashmapStats.ShuffleColMin = 1 + boundsOnly.Stats.HashmapStats.ShuffleColMax = 100 + require.Equal(t, int32(plan.ShuffleType_Range), constructShuffleOperatorForJoin(512, boundsOnly, true).ShuffleType) +} + func TestGetPercentileConfig(t *testing.T) { mp, err := mpool.NewMPool("test_pct_config", 0, mpool.NoFixed) require.NoError(t, err) diff --git a/pkg/sql/plan/shuffle.go b/pkg/sql/plan/shuffle.go index 8042588de0649..22d1e2081f1b6 100644 --- a/pkg/sql/plan/shuffle.go +++ b/pkg/sql/plan/shuffle.go @@ -108,16 +108,24 @@ func SimpleInt64HashToRange(i uint64, upperLimit uint64) uint64 { return hashtable.Int64HashWithFixedSeed(i) % upperLimit } -func shuffleByZonemap(rsp *engine.RangesShuffleParam, zm objectio.ZoneMap, bucketNum int) uint64 { - if !rsp.Init { - rsp.Init = true - switch zm.GetType() { - case types.T_int64, types.T_int32, types.T_int16: - rsp.ShuffleRangeInt64 = ShuffleRangeReEvalSigned(rsp.Node.Stats.HashmapStats.Ranges, bucketNum, rsp.Node.Stats.HashmapStats.Nullcnt, int64(rsp.Node.Stats.TableCnt)) - case types.T_uint64, types.T_uint32, types.T_uint16, types.T_varchar, types.T_char, types.T_text, types.T_bit, types.T_datalink: - rsp.ShuffleRangeUint64 = ShuffleRangeReEvalUnsigned(rsp.Node.Stats.HashmapStats.Ranges, bucketNum, rsp.Node.Stats.HashmapStats.Nullcnt, int64(rsp.Node.Stats.TableCnt)) - } +func initRangesShuffleParam(rsp *engine.RangesShuffleParam, typ types.T, bucketNum int) { + if rsp.Init && rsp.ShuffleRangeBuckets == bucketNum { + return } + rsp.Init = true + rsp.ShuffleRangeBuckets = bucketNum + rsp.ShuffleRangeInt64 = nil + rsp.ShuffleRangeUint64 = nil + switch typ { + case types.T_int64, types.T_int32, types.T_int16: + rsp.ShuffleRangeInt64 = ShuffleRangeReEvalSigned(rsp.Node.Stats.HashmapStats.Ranges, bucketNum, rsp.Node.Stats.HashmapStats.Nullcnt, int64(rsp.Node.Stats.TableCnt)) + case types.T_uint64, types.T_uint32, types.T_uint16, types.T_varchar, types.T_char, types.T_text, types.T_bit, types.T_datalink: + rsp.ShuffleRangeUint64 = ShuffleRangeReEvalUnsigned(rsp.Node.Stats.HashmapStats.Ranges, bucketNum, rsp.Node.Stats.HashmapStats.Nullcnt, int64(rsp.Node.Stats.TableCnt)) + } +} + +func shuffleByZonemap(rsp *engine.RangesShuffleParam, zm objectio.ZoneMap, bucketNum int) uint64 { + initRangesShuffleParam(rsp, zm.GetType(), bucketNum) var shuffleIDX uint64 if len(rsp.ShuffleRangeUint64) > 0 { @@ -132,15 +140,7 @@ func shuffleByZonemap(rsp *engine.RangesShuffleParam, zm objectio.ZoneMap, bucke func shuffleByValueExtractedFromZonemap(rsp *engine.RangesShuffleParam, zm objectio.ZoneMap, bucketNum int) uint64 { t := types.T(rsp.Node.Stats.HashmapStats.ShuffleColIdx) // actually this is specially used for sort key column type - if !rsp.Init { - rsp.Init = true - switch t { - case types.T_int64, types.T_int32, types.T_int16: - rsp.ShuffleRangeInt64 = ShuffleRangeReEvalSigned(rsp.Node.Stats.HashmapStats.Ranges, bucketNum, rsp.Node.Stats.HashmapStats.Nullcnt, int64(rsp.Node.Stats.TableCnt)) - case types.T_uint64, types.T_uint32, types.T_uint16, types.T_varchar, types.T_char, types.T_text, types.T_bit, types.T_datalink: - rsp.ShuffleRangeUint64 = ShuffleRangeReEvalUnsigned(rsp.Node.Stats.HashmapStats.Ranges, bucketNum, rsp.Node.Stats.HashmapStats.Nullcnt, int64(rsp.Node.Stats.TableCnt)) - } - } + initRangesShuffleParam(rsp, t, bucketNum) var shuffleIDX uint64 if len(rsp.ShuffleRangeUint64) > 0 { @@ -155,6 +155,11 @@ func shuffleByValueExtractedFromZonemap(rsp *engine.RangesShuffleParam, zm objec func CalcRangeShuffleIDXForObj(rsp *engine.RangesShuffleParam, objstats *objectio.ObjectStats, bucketNum int) uint64 { zm := objstats.SortKeyZoneMap() + if len(rsp.Node.TableDef.Pkey.Names) == 1 { + initRangesShuffleParam(rsp, zm.GetType(), bucketNum) + } else { + initRangesShuffleParam(rsp, types.T(rsp.Node.Stats.HashmapStats.ShuffleColIdx), bucketNum) + } if !zm.IsInited() { // an object with all null will send to shuffleIDX 0 return 0 @@ -166,6 +171,19 @@ func CalcRangeShuffleIDXForObj(rsp *engine.RangesShuffleParam, objstats *objecti } } +// sampledRangeFallbackBounds returns plan-level bounds for the legacy min/max +// range path. They are distribution anchors, not whole-table extrema: SQL range +// selectivity must continue to require complete min/max provenance. Encoding +// the anchors in the plan keeps object ownership identical across CN versions +// when there are too few quantiles for the runtime bucket count. +func sampledRangeFallbackBounds(typ types.T, ranges []float64) (int64, int64, bool) { + if len(ranges) < 2 || !shuffleRangeValueSafe(typ, ranges[0]) || + !shuffleRangeValueSafe(typ, ranges[len(ranges)-1]) || ranges[0] >= ranges[len(ranges)-1] { + return 0, 0, false + } + return int64(ranges[0]), int64(ranges[len(ranges)-1]), true +} + func ShouldSkipObjByShuffle(rsp *engine.RangesShuffleParam, objstats *objectio.ObjectStats) bool { if rsp == nil || rsp.CNCNT <= 1 || rsp.Node == nil { return false @@ -307,12 +325,18 @@ func GetRangeShuffleIndexForValuesExtractedFromZMUnsignedSlice(val []uint64, zm } func GetRangeShuffleIndexSignedMinMax(minVal, maxVal, currentVal int64, upplerLimit uint64) uint64 { + if upplerLimit == 0 { + return 0 + } if currentVal <= minVal { return 0 } else if currentVal >= maxVal { return upplerLimit - 1 } else { step := uint64(maxVal-minVal) / upplerLimit + if step == 0 { + return 0 + } ret := uint64(currentVal-minVal) / step if ret >= upplerLimit { return upplerLimit - 1 @@ -322,12 +346,18 @@ func GetRangeShuffleIndexSignedMinMax(minVal, maxVal, currentVal int64, upplerLi } func GetRangeShuffleIndexUnsignedMinMax(minVal, maxVal, currentVal uint64, upplerLimit uint64) uint64 { + if upplerLimit == 0 { + return 0 + } if currentVal <= minVal { return 0 } else if currentVal >= maxVal { return upplerLimit - 1 } else { step := (maxVal - minVal) / upplerLimit + if step == 0 { + return 0 + } ret := (currentVal - minVal) / step if ret >= upplerLimit { return upplerLimit - 1 @@ -571,16 +601,38 @@ func determineNonReusableShuffleType( return } s := w.GetStats() + colStats := validateColumnStats(s, tableDef, colName) + colID, ok := findColumnPosition(tableDef, colName) + if !ok { + return + } + typ := types.T(tableDef.Cols[colID].Typ.Id) + shuffleRange := s.ShuffleRangeMap[colName] + ranges := shouldUseShuffleRanges(shuffleRange, colStats.shuffleBoundsSafe) + rangesSafe := shuffleRangesSafe(typ, shuffleRange, ranges) if node.NodeType == plan.Node_AGG { if shouldUseHashShuffle(s.ShuffleRangeMap[colName]) { return } } + if !colStats.shuffleBoundsSafe && !rangesSafe { + return + } node.Stats.HashmapStats.ShuffleType = plan.ShuffleType_Range - node.Stats.HashmapStats.ShuffleColMin = int64(s.MinValMap[colName]) - node.Stats.HashmapStats.ShuffleColMax = int64(s.MaxValMap[colName]) - node.Stats.HashmapStats.Ranges = shouldUseShuffleRanges(s.ShuffleRangeMap[colName], colName) - node.Stats.HashmapStats.Nullcnt = int64(s.NullCntMap[colName]) + if colStats.shuffleBoundsSafe { + node.Stats.HashmapStats.ShuffleColMin = int64(colStats.minVal) + node.Stats.HashmapStats.ShuffleColMax = int64(colStats.maxVal) + } else if minVal, maxVal, ok := sampledRangeFallbackBounds(typ, ranges); ok { + node.Stats.HashmapStats.ShuffleColMin = minVal + node.Stats.HashmapStats.ShuffleColMax = maxVal + } else { + resetShuffleStrategy(node.Stats.HashmapStats) + return + } + if rangesSafe { + node.Stats.HashmapStats.Ranges = ranges + } + node.Stats.HashmapStats.Nullcnt = int64(colStats.nullCnt) } // to determine if join need to go shuffle @@ -588,6 +640,29 @@ func determineShuffleForJoin(node *plan.Node, builder *QueryBuilder) { determineShuffleForJoinWithColRefMode(node, builder, false) } +// shuffleJoinBuildSizeForAdmission keeps the point estimate used by join +// ordering separate from the memory-risk estimate used to admit shuffle. A +// residual FILTER is currently estimated with a fixed heuristic rather than +// column statistics, so its input cardinality is the conservative build-size +// estimate for this physical decision. +func shuffleJoinBuildSizeForAdmission(node *plan.Node, builder *QueryBuilder, afterRemap bool) float64 { + buildSize := node.Stats.HashmapStats.HashmapSize + if afterRemap || node.IsRightJoin || len(node.Children) != 2 { + return buildSize + } + + build := builder.qry.Nodes[node.Children[1]] + if build.NodeType != plan.Node_FILTER || len(build.Children) != 1 { + return buildSize + } + input := builder.qry.Nodes[build.Children[0]] + if input.Stats != nil && input.Stats.Outcnt >= threshHoldForHashShuffle && + input.Stats.Outcnt > buildSize { + return input.Stats.Outcnt + } + return buildSize +} + func isSupportedShuffleJoinKeyType(typ int32) bool { switch types.T(typ) { case types.T_int64, types.T_int32, types.T_int16, @@ -599,9 +674,9 @@ func isSupportedShuffleJoinKeyType(typ int32) bool { } } -func shuffleJoinCandidateSurvivesRecheck(node *plan.Node, ndv float64) bool { +func shuffleJoinCandidateSurvivesRecheck(node *plan.Node, ndv, admissionBuildSize float64) bool { hashmapStats := node.Stats.HashmapStats - if hashmapStats.ShuffleType == plan.ShuffleType_Hash && hashmapStats.HashmapSize < threshHoldForHashShuffle { + if hashmapStats.ShuffleType == plan.ShuffleType_Hash && admissionBuildSize < threshHoldForHashShuffle { return false } if hashmapStats.ShuffleType == plan.ShuffleType_Range && hashmapStats.Ranges == nil && @@ -633,7 +708,8 @@ func planShuffleJoinCandidate( afterRemap bool, candidateIdx int32, previousHashmapStats *plan.HashMapStats, -) (plan.HashMapStats, bool) { + admissionBuildSize float64, +) (plan.HashMapStats, bool, bool) { candidateNode := *node candidateStats := *node.Stats // HashmapSize and HashOnPK describe the join itself. All shuffle-strategy @@ -652,7 +728,7 @@ func planShuffleJoinCandidate( // Expressions cannot reuse an aggregate column partition. Shortcut the // same known-low NDV guard used by the final check. if condition.Ndv >= 0 && condition.Ndv < ShuffleThreshHoldOfNDV { - return candidateHashmapStats, false + return candidateHashmapStats, false, false } } else { child, reusable := reusableShuffleChild(leftHashCol, &candidateNode, builder, afterRemap) @@ -662,7 +738,7 @@ func planShuffleJoinCandidate( // Reuse is the only exception to the low-NDV guard. Check it once, // before looking up range statistics for a candidate that cannot win. if condition.Ndv >= 0 && condition.Ndv < ShuffleThreshHoldOfNDV { - return candidateHashmapStats, false + return candidateHashmapStats, false, false } if !afterRemap || !restoreRangeStrategyAfterRemap( &candidateHashmapStats, previousHashmapStats, candidateIdx, @@ -671,7 +747,16 @@ func planShuffleJoinCandidate( } } } - return candidateHashmapStats, shuffleJoinCandidateSurvivesRecheck(&candidateNode, condition.Ndv) + pointEligible := shuffleJoinCandidateSurvivesRecheck( + &candidateNode, condition.Ndv, node.Stats.HashmapStats.HashmapSize, + ) + riskEligible := pointEligible + if !pointEligible && condition.Ndv >= ShuffleThreshHoldOfNDV { + riskEligible = shuffleJoinCandidateSurvivesRecheck( + &candidateNode, condition.Ndv, admissionBuildSize, + ) + } + return candidateHashmapStats, pointEligible, riskEligible } // selectShuffleJoinCondition keeps the first condition that the current plan @@ -686,9 +771,12 @@ func selectShuffleJoinCondition( leftTags, rightTags map[int32]bool, afterRemap bool, previousHashmapStats *plan.HashMapStats, -) (int, plan.HashMapStats) { + admissionBuildSize float64, +) (int, plan.HashMapStats, bool) { firstSupportedIdx := -1 var firstSupportedStats plan.HashMapStats + firstRiskEligibleIdx := -1 + var firstRiskEligibleStats plan.HashMapStats for i, condition := range onList { fn := condition.GetF() @@ -712,20 +800,27 @@ func selectShuffleJoinCondition( continue } - candidateStats, eligible := planShuffleJoinCandidate( + candidateStats, pointEligible, riskEligible := planShuffleJoinCandidate( node, builder, condition, leftHashCol, rightHashCol, afterRemap, - int32(i), previousHashmapStats, + int32(i), previousHashmapStats, admissionBuildSize, ) if firstSupportedIdx == -1 { firstSupportedIdx = i firstSupportedStats = candidateStats } - if eligible { - return i, candidateStats + if pointEligible { + return i, candidateStats, true + } + if riskEligible && firstRiskEligibleIdx == -1 { + firstRiskEligibleIdx = i + firstRiskEligibleStats = candidateStats } } - return firstSupportedIdx, firstSupportedStats + if firstRiskEligibleIdx != -1 { + return firstRiskEligibleIdx, firstRiskEligibleStats, true + } + return firstSupportedIdx, firstSupportedStats, false } // determineShuffleForJoinWithColRefMode plans join shuffle either before or @@ -797,13 +892,15 @@ func determineShuffleForJoinWithColRefMode(node *plan.Node, builder *QueryBuilde if node.JoinType == plan.Node_MARK && !markJoinSupportsShuffle(node, builder, leftTags, rightTags, afterRemap) { return } - idx, candidateHashmapStats := selectShuffleJoinCondition( + admissionBuildSize := shuffleJoinBuildSizeForAdmission(node, builder, afterRemap) + idx, candidateHashmapStats, candidateEligible := selectShuffleJoinCondition( node, builder, node.OnList, leftTags, rightTags, afterRemap, - previousHashmapStats, + previousHashmapStats, admissionBuildSize, ) if idx == -1 { return } + admittedBuildSize := node.Stats.HashmapStats.HashmapSize if node.IsRightJoin { if node.Stats.HashmapStats.HashmapSize < threshHoldForRightJoinShuffle { return @@ -812,7 +909,12 @@ func determineShuffleForJoinWithColRefMode(node *plan.Node, builder *QueryBuilde leftchild := builder.qry.Nodes[node.Children[0]] rightchild := builder.qry.Nodes[node.Children[1]] factor := math.Pow((leftchild.Stats.Outcnt / rightchild.Stats.Outcnt), 0.4) - if node.Stats.HashmapStats.HashmapSize < threshHoldForShuffleJoin*factor { + threshold := threshHoldForShuffleJoin * factor + if admittedBuildSize < threshold && candidateEligible && + node.OnList[idx].Ndv >= ShuffleThreshHoldOfNDV { + admittedBuildSize = admissionBuildSize + } + if admittedBuildSize < threshold { return } } @@ -854,7 +956,7 @@ func determineShuffleForJoinWithColRefMode(node *plan.Node, builder *QueryBuilde //recheck shuffle plan if node.Stats.HashmapStats.Shuffle { - if !shuffleJoinCandidateSurvivesRecheck(node, node.OnList[idx].Ndv) { + if !shuffleJoinCandidateSurvivesRecheck(node, node.OnList[idx].Ndv, admittedBuildSize) { node.Stats.HashmapStats.Shuffle = false } @@ -1261,22 +1363,40 @@ func determineShuffleForScan(node *plan.Node, builder *QueryBuilder) { } s := w.GetStats() - if s.NdvMap[firstSortColName] < ShuffleThreshHoldOfNDV { + colStats := validateColumnStats(s, node.TableDef, firstSortColName) + if !colStats.ndvKnown || colStats.ndv < ShuffleThreshHoldOfNDV { return } firstSortColID, ok := node.TableDef.Name2ColIndex[firstSortColName] if !ok { return } - switch types.T(node.TableDef.Cols[firstSortColID].Typ.Id) { + typ := types.T(node.TableDef.Cols[firstSortColID].Typ.Id) + shuffleRange := s.ShuffleRangeMap[firstSortColName] + ranges := shouldUseShuffleRanges(shuffleRange, colStats.shuffleBoundsSafe) + rangesSafe := shuffleRangesSafe(typ, shuffleRange, ranges) + if !colStats.shuffleBoundsSafe && !rangesSafe { + return + } + switch typ { case types.T_int64, types.T_int32, types.T_int16, types.T_uint64, types.T_uint32, types.T_uint16, types.T_char, types.T_varchar, types.T_text: node.Stats.HashmapStats.ShuffleType = plan.ShuffleType_Range node.Stats.HashmapStats.ShuffleColIdx = node.TableDef.Cols[firstSortColID].Typ.Id // actually this is specially used for sort key column type - node.Stats.HashmapStats.ShuffleColMin = int64(s.MinValMap[firstSortColName]) - node.Stats.HashmapStats.ShuffleColMax = int64(s.MaxValMap[firstSortColName]) - node.Stats.HashmapStats.Ranges = shouldUseShuffleRanges(s.ShuffleRangeMap[firstSortColName], firstSortColName) - node.Stats.HashmapStats.Nullcnt = int64(s.NullCntMap[firstSortColName]) + if colStats.shuffleBoundsSafe { + node.Stats.HashmapStats.ShuffleColMin = int64(colStats.minVal) + node.Stats.HashmapStats.ShuffleColMax = int64(colStats.maxVal) + } else if minVal, maxVal, ok := sampledRangeFallbackBounds(typ, ranges); ok { + node.Stats.HashmapStats.ShuffleColMin = minVal + node.Stats.HashmapStats.ShuffleColMax = maxVal + } else { + resetShuffleStrategy(node.Stats.HashmapStats) + return + } + if rangesSafe { + node.Stats.HashmapStats.Ranges = ranges + } + node.Stats.HashmapStats.Nullcnt = int64(colStats.nullCnt) } } @@ -1348,11 +1468,16 @@ func shouldUseHashShuffle(s *pb.ShuffleRange) bool { return false } -func shouldUseShuffleRanges(s *pb.ShuffleRange, colname string) []float64 { +func shouldUseShuffleRanges(s *pb.ShuffleRange, completeBoundsSafe bool) []float64 { if s == nil || math.IsNaN(s.Uniform) || s.Result == nil { return nil } - if s.Uniform < uniformThreshold { + // Complete min/max can partition a uniform domain directly. When those + // bounds are unavailable (for example, object-sampled stats), the sampled + // quantiles are the only range-shuffle boundary with explicit provenance. + // Use them even for a uniform distribution instead of silently degrading to + // hash shuffle. + if !completeBoundsSafe || s.Uniform < uniformThreshold { return s.Result } return nil diff --git a/pkg/sql/plan/shuffle_test.go b/pkg/sql/plan/shuffle_test.go index 2d1daf1389536..2e109895ef9d8 100644 --- a/pkg/sql/plan/shuffle_test.go +++ b/pkg/sql/plan/shuffle_test.go @@ -19,6 +19,7 @@ import ( "context" "encoding/binary" "fmt" + "math" "math/rand" "testing" "unsafe" @@ -247,6 +248,147 @@ func TestShouldSkipObjByShuffle(t *testing.T) { ShouldSkipObjByShuffle(rsp, stats) } +func TestObjectRangeShuffleUsesPlanBoundsWhenQuantilesDoNotCoverCNs(t *testing.T) { + row := types.RandomRowid() + objectID := row.BorrowObjectID() + stats := objectio.NewObjectStatsWithObjectID(objectID, false, false, true) + minVal, maxVal := int64(1), int64(100) + zm := index2.NewZM(types.T_int64, 0) + index2.UpdateZM(zm, types.EncodeInt64(&minVal)) + index2.UpdateZM(zm, types.EncodeInt64(&maxVal)) + objectio.SetObjectStatsSortKeyZoneMap(stats, zm) + + makeNode := func(ranges []float64, composite bool) *plan.Node { + pkeyNames := []string{"k"} + if composite { + pkeyNames = []string{"k1", "k2"} + } + node := &plan.Node{ + TableDef: &plan.TableDef{Pkey: &plan.PrimaryKeyDef{Names: pkeyNames}}, + Stats: DefaultStats(), + } + node.Stats.TableCnt = 1_000_000 + node.Stats.HashmapStats.ShuffleType = plan.ShuffleType_Range + node.Stats.HashmapStats.ShuffleColIdx = int32(types.T_int64) + node.Stats.HashmapStats.ShuffleColMin = 1 + node.Stats.HashmapStats.ShuffleColMax = 100 + if len(ranges) >= 2 && ranges[0] < ranges[len(ranges)-1] { + node.Stats.HashmapStats.ShuffleColMin = int64(ranges[0]) + node.Stats.HashmapStats.ShuffleColMax = int64(ranges[len(ranges)-1]) + } + node.Stats.HashmapStats.Ranges = ranges + return node + } + + for _, composite := range []bool{false, true} { + t.Run(fmt.Sprintf("short/composite=%v", composite), func(t *testing.T) { + node := makeNode([]float64{1, 50, 100}, composite) + objectStats := stats + wantOwner := GetRangeShuffleIndexForZM(1, 100, zm, 2) + if composite { + objectStats = objectio.NewObjectStatsWithObjectID(objectID, false, false, true) + compositeZM := index2.NewZM(types.T_varchar, 0) + packer := types.NewPacker() + packer.EncodeInt64(1) + index2.UpdateZM(compositeZM, packer.Bytes()) + packer = types.NewPacker() + packer.EncodeInt64(100) + index2.UpdateZM(compositeZM, packer.Bytes()) + objectio.SetObjectStatsSortKeyZoneMap(objectStats, compositeZM) + wantOwner = GetRangeShuffleIndexForExtractedZM(1, 100, compositeZM, 2, types.T_int64) + } + owners := 0 + for cnidx := int32(0); cnidx < 2; cnidx++ { + rsp := &engine.RangesShuffleParam{Node: node, CNCNT: 2, CNIDX: cnidx} + if !ShouldSkipObjByShuffle(rsp, objectStats) { + owners++ + require.Equal(t, uint64(cnidx), wantOwner) + } + } + require.Equal(t, 1, owners) + }) + } + + collectorRanges := make([]float64, 1023) + for i := range collectorRanges { + collectorRanges[i] = float64(i) + } + collectorNode := makeNode(collectorRanges, false) + collectorRsp := &engine.RangesShuffleParam{Node: collectorNode} + require.Equal(t, GetRangeShuffleIndexForZM(0, 1022, zm, 512), + CalcRangeShuffleIDXForObj(collectorRsp, stats, 512)) + + boundsNode := makeNode(nil, false) + boundsNode.Stats.HashmapStats.ShuffleColMin = 1 + boundsNode.Stats.HashmapStats.ShuffleColMax = 100 + boundsRsp := &engine.RangesShuffleParam{Node: boundsNode} + require.Equal(t, GetRangeShuffleIndexForZM(1, 100, zm, 2), + CalcRangeShuffleIDXForObj(boundsRsp, stats, 2)) +} + +func TestRangeShuffleMinMaxHandlesMoreBucketsThanValues(t *testing.T) { + require.Equal(t, uint64(0), GetRangeShuffleIndexSignedMinMax(1, 3, 2, 8)) + require.Equal(t, uint64(0), GetRangeShuffleIndexUnsignedMinMax(1, 3, 2, 8)) + require.Equal(t, uint64(0), GetRangeShuffleIndexSignedMinMax(1, 3, 2, 0)) + require.Equal(t, uint64(0), GetRangeShuffleIndexUnsignedMinMax(1, 3, 2, 0)) +} + +func TestSampledRangeFallbackBounds(t *testing.T) { + minVal, maxVal, ok := sampledRangeFallbackBounds(types.T_int64, []float64{1, 25, 100}) + require.True(t, ok) + require.Equal(t, int64(1), minVal) + require.Equal(t, int64(100), maxVal) + + _, _, ok = sampledRangeFallbackBounds(types.T_int64, []float64{7, 7}) + require.False(t, ok) + _, _, ok = sampledRangeFallbackBounds(types.T_int64, []float64{math.NaN(), 7}) + require.False(t, ok) +} + +func TestDetermineShuffleForScanPublishesSampledRangeFallbackBounds(t *testing.T) { + tableDef := &plan.TableDef{ + TblId: 1, + Name: "t", + Cols: []*plan.ColDef{{ + Name: "k", + Typ: plan.Type{Id: int32(types.T_int64)}, + }}, + Name2ColIndex: map[string]int32{"k": 0}, + Pkey: &plan.PrimaryKeyDef{PkeyColName: "k", Names: []string{"k"}}, + } + stats := NewStatsInfo() + stats.TableCnt = 1_000_000 + stats.SampleRatio = 0.1 + stats.AccurateObjectNumber = 1_000 + stats.NdvMap["k"] = 500_000 + stats.MinValMap["k"] = 1 + stats.MaxValMap["k"] = 1_000_000 + stats.MinMaxValidMap["k"] = true + shuffleRange := NewShuffleRange(false) + shuffleRange.SampleRatio = 0.1 + shuffleRange.Result = []float64{10, 250_000, 900_000} + stats.ShuffleRangeMap["k"] = shuffleRange + statsCache := NewStatsCache() + statsCache.Set(1, stats) + ctx := &statsCacheCompilerContext{ + MockCompilerContext: &MockCompilerContext{ctx: context.Background()}, + statsCache: statsCache, + } + builder := NewQueryBuilder(plan.Query_SELECT, ctx, false, false) + + node := &plan.Node{NodeType: plan.Node_TABLE_SCAN, TableDef: tableDef, Stats: DefaultStats()} + determineShuffleForScan(node, builder) + require.Equal(t, plan.ShuffleType_Range, node.Stats.HashmapStats.ShuffleType) + require.Equal(t, int64(10), node.Stats.HashmapStats.ShuffleColMin) + require.Equal(t, int64(900_000), node.Stats.HashmapStats.ShuffleColMax) + require.Equal(t, shuffleRange.Result, node.Stats.HashmapStats.Ranges) + + shuffleRange.Result = []float64{7, 7} + node = &plan.Node{NodeType: plan.Node_TABLE_SCAN, TableDef: tableDef, Stats: DefaultStats()} + determineShuffleForScan(node, builder) + require.Equal(t, plan.ShuffleType_Hash, node.Stats.HashmapStats.ShuffleType) +} + func TestShouldSkipAppendableObjByShuffleKeepsDefaultLocalBehavior(t *testing.T) { row := types.RandomRowid() stats := objectio.NewObjectStatsWithObjectID(row.BorrowObjectID(), true, false, true) @@ -934,6 +1076,62 @@ func TestDetermineShuffleForJoinNormalizesReversedConditionAfterRemap(t *testing require.Equal(t, int32(1), condition.GetF().Args[1].GetCol().RelPos) } +func TestDetermineShuffleForJoinUsesResidualFilterInputForAdmission(t *testing.T) { + tests := []struct { + name string + secondTag int32 + inputRows float64 + joinKeyNDV float64 + wantAdmissionRows float64 + wantShuffle bool + }{ + {name: "large single relation residual filter uses input risk", secondTag: 2, inputRows: 60_000_000, joinKeyNDV: 100_000, wantAdmissionRows: 60_000_000, wantShuffle: true}, + {name: "input risk admits known high ndv hash candidate", secondTag: 2, inputRows: 30_000_000, joinKeyNDV: 100_000, wantAdmissionRows: 30_000_000, wantShuffle: true}, + {name: "input risk does not promote unknown ndv hash candidate", secondTag: 2, inputRows: 30_000_000, joinKeyNDV: -1, wantAdmissionRows: 30_000_000, wantShuffle: false}, + {name: "unknown ndv preserves point admitted hash candidate", secondTag: 2, inputRows: 80_000_000, joinKeyNDV: -1, wantAdmissionRows: 80_000_000, wantShuffle: true}, + {name: "small cross relation filter stays resident", secondTag: 3, inputRows: 1_000_000, joinKeyNDV: 100_000, wantAdmissionRows: 50_000, wantShuffle: false}, + {name: "large cross relation filter uses input risk", secondTag: 3, inputRows: 60_000_000, joinKeyNDV: 100_000, wantAdmissionRows: 60_000_000, wantShuffle: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + predicate, err := BindFuncExprImplByPlanExpr(context.Background(), "!=", []*plan.Expr{ + {Typ: plan.Type{Id: int32(types.T_int64)}, Expr: &plan.Expr_Col{Col: &plan.ColRef{RelPos: 2, ColPos: 1}}}, + {Typ: plan.Type{Id: int32(types.T_int64)}, Expr: &plan.Expr_Col{Col: &plan.ColRef{RelPos: tt.secondTag, ColPos: 1}}}, + }) + require.NoError(t, err) + + probe := makeShuffleJoinTestChild(1, 10_000_000_000) + buildInput := makeShuffleJoinTestChild(2, tt.inputRows) + buildInput.BindingTags = []int32{2, 3} + filter := &plan.Node{ + NodeType: plan.Node_FILTER, + Children: []int32{1}, + FilterList: []*plan.Expr{predicate}, + Stats: DefaultStats(), + } + join := &plan.Node{ + NodeType: plan.Node_JOIN, + JoinType: plan.Node_INNER, + Children: []int32{0, 2}, + OnList: []*plan.Expr{makeShuffleJoinEquality(t, types.T_int64, 100_000, 1, 2, 0)}, + Stats: DefaultStats(), + } + builder := &QueryBuilder{qry: &plan.Query{Nodes: []*plan.Node{probe, buildInput, filter, join}}} + + ReCalcNodeStats(2, builder, false, false, false) + ReCalcNodeStats(3, builder, false, false, false) + join.OnList[0].Ndv = tt.joinKeyNDV + require.Equal(t, 0.05, filter.Stats.Selectivity) + require.Equal(t, tt.inputRows*0.05, join.Stats.HashmapStats.HashmapSize) + require.Equal(t, tt.wantAdmissionRows, shuffleJoinBuildSizeForAdmission(join, builder, false)) + + determineShuffleForJoin(join, builder) + require.Equal(t, tt.wantShuffle, join.Stats.HashmapStats.Shuffle) + }) + } +} + func TestDetermineShuffleForJoinSkipsCandidateRejectedByFinalRecheck(t *testing.T) { tests := []struct { name string @@ -1072,9 +1270,13 @@ func TestSelectShuffleJoinConditionAdversarialPermutations(t *testing.T) { }}, } - idx, _ := selectShuffleJoinCondition(node, builder, conditions, leftTags, rightTags, false, nil) + idx, _, eligible := selectShuffleJoinCondition( + node, builder, conditions, leftTags, rightTags, false, nil, + node.Stats.HashmapStats.HashmapSize, + ) require.NotEqual(t, -1, idx) + require.True(t, eligible) require.Same(t, reusable, conditions[idx]) return } diff --git a/pkg/sql/plan/stats.go b/pkg/sql/plan/stats.go index fca5e4c8a88b7..f43eb2a28bb7d 100644 --- a/pkg/sql/plan/stats.go +++ b/pkg/sql/plan/stats.go @@ -173,12 +173,303 @@ func NewStatsInfo() *pb.StatsInfo { NullCntMap: make(map[string]uint64), SizeMap: make(map[string]uint64), ShuffleRangeMap: make(map[string]*pb.ShuffleRange), + MinMaxValidMap: make(map[string]bool), + MinMaxCompleteMap: make(map[string]bool), BlockNumber: 0, ApproxObjectNumber: 0, TableCnt: 0, } } +type validatedColumnStats struct { + tableCnt float64 + nullCnt float64 + ndv float64 + ndvKnown bool + minVal float64 + maxVal float64 + minMaxKnown bool + minMaxComplete bool + rangeEstimateSafe bool + shuffleBoundsSafe bool +} + +func validTableCnt(tableCnt float64) float64 { + if tableCnt < 0 || math.IsNaN(tableCnt) || math.IsInf(tableCnt, 0) { + return 0 + } + return tableCnt +} + +func isSingleColumnPrimaryKey(tableDef *plan.TableDef, colName string) bool { + return tableDef != nil && tableDef.Pkey != nil && + len(tableDef.Pkey.Names) == 1 && tableDef.Pkey.Names[0] == colName && + colName != catalog.FakePrimaryKeyColName +} + +func findColumnPosition(tableDef *plan.TableDef, colName string) (int32, bool) { + if tableDef == nil { + return 0, false + } + if colPos, ok := tableDef.Name2ColIndex[colName]; ok && colPos >= 0 && int(colPos) < len(tableDef.Cols) { + return colPos, true + } + for colPos, colDef := range tableDef.Cols { + if colDef.Name == colName { + return int32(colPos), true + } + } + return 0, false +} + +// discreteDomainCardinality returns a provable upper bound on NDV from a +// numeric min/max pair. Values wider than float64's exact integer range are +// deliberately rejected: StatsInfo stores min/max as float64, so narrowing +// such a domain could turn representation loss into a cardinality underestimate. +func discreteDomainCardinality(typ types.T, minVal, maxVal float64) (float64, bool) { + if math.IsNaN(minVal) || math.IsNaN(maxVal) || math.IsInf(minVal, 0) || + math.IsInf(maxVal, 0) || minVal > maxVal { + return 0, false + } + + switch typ { + case types.T_bool: + if minVal < 0 || maxVal > 1 { + return 0, false + } + case types.T_int8: + if minVal < math.MinInt8 || maxVal > math.MaxInt8 { + return 0, false + } + case types.T_int16: + if minVal < math.MinInt16 || maxVal > math.MaxInt16 { + return 0, false + } + case types.T_int32, types.T_date: + if minVal < math.MinInt32 || maxVal > math.MaxInt32 { + return 0, false + } + case types.T_uint8: + if minVal < 0 || maxVal > math.MaxUint8 { + return 0, false + } + case types.T_uint16, types.T_year, types.T_enum: + if minVal < 0 || maxVal > math.MaxUint16 { + return 0, false + } + case types.T_uint32: + if minVal < 0 || maxVal > math.MaxUint32 { + return 0, false + } + case types.T_int64, types.T_uint64, types.T_bit, + types.T_time, types.T_datetime, types.T_timestamp: + const maxExactInteger = float64(1 << 53) + if math.Abs(minVal) >= maxExactInteger || math.Abs(maxVal) >= maxExactInteger || + (typ == types.T_uint64 || typ == types.T_bit) && minVal < 0 { + return 0, false + } + default: + return 0, false + } + + if math.Trunc(minVal) != minVal || math.Trunc(maxVal) != maxVal { + return 0, false + } + domain := maxVal - minVal + 1 + if domain < 1 || domain >= float64(1<<53) || math.IsInf(domain, 0) { + return 0, false + } + return domain, true +} + +func validSampleRatio(sampleRatio float64) float64 { + if sampleRatio <= 0 || sampleRatio > 1 || math.IsNaN(sampleRatio) || math.IsInf(sampleRatio, 0) { + return 0 + } + return sampleRatio +} + +func exactDiscreteValue(typ types.T, value float64) bool { + _, ok := discreteDomainCardinality(typ, value, value) + return ok +} + +func rangeEstimateSafe(typ types.T, minVal, maxVal float64) bool { + switch typ { + case types.T_char, types.T_varchar, types.T_text, types.T_datalink: + // String extrema are stored as an encoded uint64 in float64. The encoding + // is useful for sampled shuffle quantiles, but is not an exact SQL domain. + return false + case types.T_float32, types.T_float64, types.T_decimal64, types.T_decimal128: + return true + default: + return exactDiscreteValue(typ, minVal) && exactDiscreteValue(typ, maxVal) + } +} + +func shuffleRangeValueSafe(typ types.T, value float64) bool { + if math.IsNaN(value) || math.IsInf(value, 0) { + return false + } + switch typ { + case types.T_int16: + return value >= math.MinInt16 && value <= math.MaxInt16 + case types.T_int32: + return value >= math.MinInt32 && value <= math.MaxInt32 + case types.T_int64: + return math.Abs(value) < float64(1<<53) + case types.T_uint16: + return value >= 0 && value <= math.MaxUint16 + case types.T_uint32: + return value >= 0 && value <= math.MaxUint32 + case types.T_uint64: + return value >= 0 && value < float64(1<<53) + default: + return false + } +} + +func shuffleRangesSafe(typ types.T, shuffleRange *pb.ShuffleRange, ranges []float64) bool { + if shuffleRange == nil || validSampleRatio(shuffleRange.SampleRatio) == 0 || len(ranges) == 0 { + return false + } + for i, value := range ranges { + if !shuffleRangeValueSafe(typ, value) { + return false + } + if i > 0 && value < ranges[i-1] { + return false + } + } + return true +} + +// validateColumnStats is the single planner-side boundary for raw column +// statistics. It removes impossible combinations without claiming that a +// sampled value is accurate. A sampled min/max pair is not a hard domain bound; +// completeness must be explicit in StatsInfo before range consumers may use it. +// Missing or invalid NDV remains unknown. +func validateColumnStats( + s *pb.StatsInfo, + tableDef *plan.TableDef, + colName string, +) validatedColumnStats { + var ret validatedColumnStats + if s == nil || tableDef == nil { + return ret + } + colPos, ok := findColumnPosition(tableDef, colName) + if !ok { + return ret + } + + ret.tableCnt = validTableCnt(s.TableCnt) + ret.nullCnt = math.Min(float64(s.NullCntMap[colName]), ret.tableCnt) + if isSingleColumnPrimaryKey(tableDef, colName) { + ret.nullCnt = 0 + } + nonNullCnt := ret.tableCnt - ret.nullCnt + + var minOK, maxOK bool + ret.minVal, minOK = s.MinValMap[colName] + ret.maxVal, maxOK = s.MaxValMap[colName] + ret.minMaxKnown = s.MinMaxValidMap[colName] && minOK && maxOK && + !math.IsNaN(ret.minVal) && !math.IsNaN(ret.maxVal) && + !math.IsInf(ret.minVal, 0) && !math.IsInf(ret.maxVal, 0) && ret.minVal <= ret.maxVal + ret.minMaxComplete = ret.minMaxKnown && s.MinMaxCompleteMap[colName] + if ret.minMaxComplete { + typ := types.T(tableDef.Cols[colPos].Typ.Id) + ret.rangeEstimateSafe = rangeEstimateSafe(typ, ret.minVal, ret.maxVal) + ret.shuffleBoundsSafe = exactDiscreteValue(typ, ret.minVal) && exactDiscreteValue(typ, ret.maxVal) + } + if isSingleColumnPrimaryKey(tableDef, colName) && ret.minMaxComplete { + if domain, domainOK := discreteDomainCardinality( + types.T(tableDef.Cols[colPos].Typ.Id), ret.minVal, ret.maxVal, + ); domainOK && domain < nonNullCnt { + // Row count plus a single-column primary key proves this complete + // domain impossible. Keep the exact PK NDV and discard stale bounds. + ret.minMaxKnown = false + ret.minMaxComplete = false + ret.rangeEstimateSafe = false + ret.shuffleBoundsSafe = false + } + } + + if nonNullCnt == 0 { + ret.ndvKnown = true + return ret + } + if isSingleColumnPrimaryKey(tableDef, colName) { + ret.ndv = nonNullCnt + ret.ndvKnown = true + return ret + } + + rawNDV, ndvOK := s.NdvMap[colName] + if !ndvOK || rawNDV <= 0 || math.IsNaN(rawNDV) || math.IsInf(rawNDV, 0) { + return ret + } + ret.ndv = math.Min(math.Max(rawNDV, 1), nonNullCnt) + ret.ndvKnown = true + if ret.minMaxComplete { + if domain, domainOK := discreteDomainCardinality( + types.T(tableDef.Cols[colPos].Typ.Id), ret.minVal, ret.maxVal, + ); domainOK { + ret.ndv = math.Min(ret.ndv, domain) + } + } + return ret +} + +// sanitizeStatsInfo canonicalizes collected statistics before publication. +// Planner consumers still validate on read so older cached or patched stats +// cannot bypass the same invariants. +func sanitizeStatsInfo(tableDef *plan.TableDef, s *pb.StatsInfo) { + if s == nil || tableDef == nil { + return + } + if s.NdvMap == nil { + s.NdvMap = make(map[string]float64) + } + if s.MinValMap == nil { + s.MinValMap = make(map[string]float64) + } + if s.MaxValMap == nil { + s.MaxValMap = make(map[string]float64) + } + if s.NullCntMap == nil { + s.NullCntMap = make(map[string]uint64) + } + if s.MinMaxValidMap == nil { + s.MinMaxValidMap = make(map[string]bool) + } + if s.MinMaxCompleteMap == nil { + s.MinMaxCompleteMap = make(map[string]bool) + } + s.TableCnt = validTableCnt(s.TableCnt) + s.SampleRatio = validSampleRatio(s.SampleRatio) + for _, col := range tableDef.Cols { + if col.Name == catalog.Row_ID { + continue + } + stats := validateColumnStats(s, tableDef, col.Name) + s.NullCntMap[col.Name] = uint64(stats.nullCnt) + if stats.ndvKnown { + s.NdvMap[col.Name] = stats.ndv + } else { + delete(s.NdvMap, col.Name) + } + if !stats.minMaxKnown { + delete(s.MinValMap, col.Name) + delete(s.MaxValMap, col.Name) + delete(s.MinMaxValidMap, col.Name) + delete(s.MinMaxCompleteMap, col.Name) + } else if !stats.minMaxComplete { + delete(s.MinMaxCompleteMap, col.Name) + } + } +} + type TableStatsInfo struct { ColumnZMs []objectio.ZoneMap DataTypes []types.Type @@ -195,6 +486,7 @@ type TableStatsInfo struct { AccurateObjectNumber int64 ApproxObjectNumber int64 TableRowCount float64 // Total row count in the table + SampleRatio float64 // Fraction of objects whose column metadata was read } func NewTableStatsInfo(lenCols int) *TableStatsInfo { @@ -216,7 +508,7 @@ func AdjustNDV(info *TableStatsInfo, tableDef *TableDef, s *pb.StatsInfo) { if info.AccurateObjectNumber > 1 { for i, coldef := range tableDef.Cols[:len(tableDef.Cols)-1] { if info.ColumnNDVs[i] > s.TableCnt { - info.ColumnNDVs[i] = s.TableCnt * 0.99 // to avoid a bug + info.ColumnNDVs[i] = s.TableCnt } colName := coldef.Name rate := info.ColumnNDVs[i] / info.TableRowCount @@ -292,10 +584,9 @@ func AdjustNDV(info *TableStatsInfo, tableDef *TableDef, s *pb.StatsInfo) { for i, coldef := range tableDef.Cols[:len(tableDef.Cols)-1] { colName := coldef.Name s.NdvMap[colName] = info.ColumnNDVs[i] - if s.NdvMap[colName] > s.TableCnt { - s.NdvMap[colName] = s.TableCnt * 0.99 - } } + s.SampleRatio = info.SampleRatio + sanitizeStatsInfo(tableDef, s) } func UpdateStatsInfo(info *TableStatsInfo, tableDef *plan.TableDef, s *pb.StatsInfo) { @@ -308,6 +599,13 @@ func UpdateStatsInfo(info *TableStatsInfo, tableDef *plan.TableDef, s *pb.StatsI s.BlockNumber = info.BlockNumber s.TableCnt = info.TableRowCount s.TableName = tableDef.Name + s.SampleRatio = info.SampleRatio + if s.MinMaxValidMap == nil { + s.MinMaxValidMap = make(map[string]bool) + } + if s.MinMaxCompleteMap == nil { + s.MinMaxCompleteMap = make(map[string]bool) + } for i, coldef := range tableDef.Cols[:len(tableDef.Cols)-1] { colName := coldef.Name @@ -315,14 +613,15 @@ func UpdateStatsInfo(info *TableStatsInfo, tableDef *plan.TableDef, s *pb.StatsI s.NullCntMap[colName] = uint64(info.NullCnts[i]) s.SizeMap[colName] = uint64(info.ColumnSize[i]) - // When ZoneMap is not inited we cannot decode min/max from it; set them to 0 and skip the type switch. + // When ZoneMap is not inited we cannot decode min/max from it; leave them unknown and skip the type switch. // We must NOT continue here: the ShuffleRange block below must still run so that a ShuffleRange // produced by collect (e.g. from NDV accumulation in later objects) is written to s.ShuffleRangeMap. // Otherwise, "collect has ShuffleRanges[i] but ZoneMap never inited" would leave ShuffleRangeMap empty. - if !info.ColumnZMs[i].IsInited() { - s.MinValMap[colName] = 0 - s.MaxValMap[colName] = 0 - } else { + delete(s.MinValMap, colName) + delete(s.MaxValMap, colName) + delete(s.MinMaxValidMap, colName) + delete(s.MinMaxCompleteMap, colName) + if info.ColumnZMs[i].IsInited() { switch info.DataTypes[i].Oid { case types.T_bit: s.MinValMap[colName] = float64(types.DecodeUint64(info.ColumnZMs[i].GetMinBuf())) @@ -398,11 +697,21 @@ func UpdateStatsInfo(info *TableStatsInfo, tableDef *plan.TableDef, s *pb.StatsI } } + _, minOK := s.MinValMap[colName] + _, maxOK := s.MaxValMap[colName] + if minOK && maxOK { + s.MinMaxValidMap[colName] = true + if validSampleRatio(info.SampleRatio) == 1 { + s.MinMaxCompleteMap[colName] = true + } + } else { + delete(s.MinMaxValidMap, colName) + delete(s.MinMaxCompleteMap, colName) + } if info.ShuffleRanges[i] != nil { // Allow writing ShuffleRange when we have one from collect and other conditions are met. - // When ZoneMap is not inited we set min=max=0, so min!=max is false; we relax by allowing - // write when !IsInited() so that collect-produced ShuffleRanges are not dropped. + // A collect-produced ShuffleRange remains useful even when the ZoneMap is not initialized. canFill := (s.MinValMap[colName] != s.MaxValMap[colName] || !info.ColumnZMs[i].IsInited()) && s.TableCnt > ShuffleThreshHoldOfNDV*2 && info.ColumnNDVs[i] >= ShuffleThreshHoldOfNDV && @@ -435,9 +744,13 @@ func isHighNdvCols(cols []int32, tableDef *TableDef, builder *QueryBuilder) bool s := w.GetStats() var totalNDV float64 = 1 for i := range cols { - totalNDV *= s.NdvMap[tableDef.Cols[cols[i]].Name] + colStats := validateColumnStats(s, tableDef, tableDef.Cols[cols[i]].Name) + if !colStats.ndvKnown { + return false + } + totalNDV = math.Min(colStats.tableCnt, totalNDV*colStats.ndv) } - return totalNDV > s.TableCnt*highNDVcolumnThreshHold + return totalNDV > validTableCnt(s.TableCnt)*highNDVcolumnThreshHold } func (builder *QueryBuilder) getColNDVRatio(cols []int32, tableDef *TableDef) float64 { @@ -456,9 +769,13 @@ func (builder *QueryBuilder) getColNDVRatio(cols []int32, tableDef *TableDef) fl s := w.GetStats() var totalNDV float64 = 1 for i := range cols { - totalNDV *= s.NdvMap[tableDef.Cols[cols[i]].Name] + colStats := validateColumnStats(s, tableDef, tableDef.Cols[cols[i]].Name) + if !colStats.ndvKnown { + return 0 + } + totalNDV = math.Min(colStats.tableCnt, totalNDV*colStats.ndv) } - result := safeRatio(totalNDV, s.TableCnt, 0) + result := safeRatio(totalNDV, validTableCnt(s.TableCnt), 0) if result > 1 { result = 1 } @@ -508,7 +825,15 @@ func (builder *QueryBuilder) getColNdv(col *plan.ColRef) float64 { if w == nil || w.GetStats() == nil { return -1 } - return w.GetStats().NdvMap[col.Name] + tableDef, ok := builder.tag2Table[col.RelPos] + if !ok { + return -1 + } + stats := validateColumnStats(w.GetStats(), tableDef, col.Name) + if !stats.ndvKnown { + return -1 + } + return stats.ndv } //func (builder *QueryBuilder) getColOverlap(col *plan.ColRef) float64 { @@ -527,12 +852,15 @@ func getNullSelectivity(arg *plan.Expr, builder *QueryBuilder, isnull bool) floa if w == nil || w.GetStats() == nil { break } - s := w.GetStats() - nullCnt := float64(s.NullCntMap[col.Name]) + tableDef, ok := builder.tag2Table[col.RelPos] + if !ok { + break + } + s := validateColumnStats(w.GetStats(), tableDef, col.Name) if isnull { - return safeRatio(nullCnt, s.TableCnt, 0.1) + return safeRatio(s.nullCnt, s.tableCnt, 0.1) } else { - return 1 - safeRatio(nullCnt, s.TableCnt, 0.1) + return 1 - safeRatio(s.nullCnt, s.tableCnt, 0.1) } } @@ -627,6 +955,189 @@ func getExprNdv(expr *plan.Expr, builder *QueryBuilder) float64 { return -1 } +type equiJoinColumnStats struct { + ndv float64 + nonNullRows float64 +} + +func subtreeHasBindingTag(nodeID, tag int32, builder *QueryBuilder) bool { + if builder == nil || builder.qry == nil || nodeID < 0 || int(nodeID) >= len(builder.qry.Nodes) { + return false + } + for _, subtreeTag := range builder.enumerateTags(nodeID) { + if subtreeTag == tag { + return true + } + } + return false +} + +// joinColumnNullLineageSafe rejects deterministic cases where a base-table +// null fraction no longer describes the child column. Filters on the key can +// select only NULL values, and outer/single joins can inject new NULL values. +// Other predicates still use the optimizer's existing independence assumption. +func joinColumnNullLineageSafe(nodeID int32, col *plan.ColRef, builder *QueryBuilder) bool { + if builder == nil || builder.qry == nil || col == nil || + nodeID < 0 || int(nodeID) >= len(builder.qry.Nodes) { + return false + } + node := builder.qry.Nodes[nodeID] + if node == nil { + return false + } + for _, filter := range node.FilterList { + if refsColumn(filter, col.RelPos, col.ColPos) { + return false + } + } + for _, filter := range node.BlockFilterList { + if refsColumn(filter, col.RelPos, col.ColPos) { + return false + } + } + + if node.NodeType == plan.Node_JOIN && len(node.Children) == 2 { + leftHasCol := subtreeHasBindingTag(node.Children[0], col.RelPos, builder) + rightHasCol := subtreeHasBindingTag(node.Children[1], col.RelPos, builder) + switch node.JoinType { + case plan.Node_LEFT: + if rightHasCol { + return false + } + case plan.Node_RIGHT: + if leftHasCol { + return false + } + case plan.Node_OUTER: + if leftHasCol || rightHasCol { + return false + } + case plan.Node_SINGLE: + if (!node.IsRightJoin && rightHasCol) || (node.IsRightJoin && leftHasCol) { + return false + } + } + } + + foundSource := len(node.BindingTags) > 0 + for _, childID := range node.Children { + if !subtreeHasBindingTag(childID, col.RelPos, builder) { + continue + } + foundSource = true + if !joinColumnNullLineageSafe(childID, col, builder) { + return false + } + } + return foundSource +} + +func availableEquiJoinColumnStats( + builder *QueryBuilder, + childID int32, + col *plan.ColRef, + childRows float64, +) (equiJoinColumnStats, bool) { + if builder == nil || col == nil || childRows <= 0 || math.IsNaN(childRows) || math.IsInf(childRows, 0) { + return equiJoinColumnStats{}, false + } + if !joinColumnNullLineageSafe(childID, col, builder) { + return equiJoinColumnStats{}, false + } + w := builder.getStatsInfoByCol(col) + if w == nil || w.GetStats() == nil { + return equiJoinColumnStats{}, false + } + tableDef, ok := builder.tag2Table[col.RelPos] + if !ok { + return equiJoinColumnStats{}, false + } + s := w.GetStats() + if _, nullCntKnown := s.NullCntMap[col.Name]; !nullCntKnown { + return equiJoinColumnStats{}, false + } + stats := validateColumnStats(s, tableDef, col.Name) + if !stats.ndvKnown || stats.ndv <= 0 || stats.tableCnt <= 0 { + return equiJoinColumnStats{}, false + } + nonNullFraction := safeRatio(stats.tableCnt-stats.nullCnt, stats.tableCnt, 0) + sampleRatio := validSampleRatio(s.SampleRatio) + if sampleRatio == 0 { + return equiJoinColumnStats{}, false + } + if sampleRatio < 1 { + sampledObjects := float64(s.AccurateObjectNumber) * sampleRatio + if sampledObjects <= 0 || math.IsNaN(sampledObjects) || math.IsInf(sampledObjects, 0) { + return equiJoinColumnStats{}, false + } + // The rule of three gives an approximate 95% upper bound for an unseen + // event. Apply it at object granularity so sampled NullCnt=0 is not + // mistaken for a complete proof that the column is non-NULL. + nonNullFraction = math.Max(0, nonNullFraction-math.Min(1, 3/sampledObjects)) + } + nonNullRows := childRows * nonNullFraction + if nonNullRows <= 0 || math.IsNaN(nonNullRows) || math.IsInf(nonNullRows, 0) { + return equiJoinColumnStats{}, false + } + ndv := math.Min(stats.ndv, nonNullRows) + if ndv <= 0 || math.IsNaN(ndv) || math.IsInf(ndv, 0) { + return equiJoinColumnStats{}, false + } + return equiJoinColumnStats{ndv: ndv, nonNullRows: nonNullRows}, true +} + +// availableSingleEquiJoinCardinality returns the standard uniform estimate +// for a single direct-column equality. Equality does not match NULL, so both +// inputs are reduced by their known non-NULL fractions before applying NDV. +// Multi-key joins require tuple NDV/correlation statistics and are excluded. +func availableSingleEquiJoinCardinality( + node *plan.Node, + builder *QueryBuilder, + leftStats, rightStats *Stats, +) (float64, bool) { + if node == nil || builder == nil || builder.compCtx == nil || builder.qry == nil || + leftStats == nil || rightStats == nil || len(node.Children) != 2 || len(node.OnList) != 1 { + return 0, false + } + fn := node.OnList[0].GetF() + if fn == nil || !IsEqualFunc(fn.Func.GetObj()) || len(fn.Args) != 2 { + return 0, false + } + leftCol := fn.Args[0].GetCol() + rightCol := fn.Args[1].GetCol() + if leftCol == nil || rightCol == nil { + return 0, false + } + + leftTags := make(map[int32]bool) + for _, tag := range builder.enumerateTags(node.Children[0]) { + leftTags[tag] = true + } + rightTags := make(map[int32]bool) + for _, tag := range builder.enumerateTags(node.Children[1]) { + rightTags[tag] = true + } + leftFirst := leftTags[leftCol.RelPos] && !rightTags[leftCol.RelPos] && + rightTags[rightCol.RelPos] && !leftTags[rightCol.RelPos] + if !leftFirst { + rightFirst := leftTags[rightCol.RelPos] && !rightTags[rightCol.RelPos] && + rightTags[leftCol.RelPos] && !leftTags[leftCol.RelPos] + if !rightFirst { + return 0, false + } + leftCol, rightCol = rightCol, leftCol + } + + left, leftOK := availableEquiJoinColumnStats(builder, node.Children[0], leftCol, leftStats.Outcnt) + right, rightOK := availableEquiJoinColumnStats(builder, node.Children[1], rightCol, rightStats.Outcnt) + if !leftOK || !rightOK { + return 0, false + } + joinNDV := math.Max(left.ndv, right.ndv) + cardinality := left.nonNullRows * right.nonNullRows / joinNDV + return cardinality, cardinality >= 0 && !math.IsNaN(cardinality) && !math.IsInf(cardinality, 0) +} + func estimateEqualitySelectivity(expr *plan.Expr, builder *QueryBuilder, s *pb.StatsInfo) float64 { // only filter like func(col)=1 or col=? can estimate outcnt // and only 1 colRef is allowd in the filter. otherwise, no good method to calculate @@ -873,9 +1384,26 @@ func estimateNonEqualitySelectivity(expr *plan.Expr, funcName string, builder *Q return 0.01 } } - s := w.GetStats() - if colRef != nil && len(literals) > 0 { - typ := types.T(s.DataTypeMap[colRef.Name]) + if colRef == nil || len(literals) == 0 { + return 0.1 + } + tableDef, ok := builder.tag2Table[colRef.RelPos] + if !ok { + return 0.1 + } + validated := validateColumnStats(w.GetStats(), tableDef, colRef.Name) + if len(literals) > 0 { + if !validated.rangeEstimateSafe { + return 0.1 + } + colPos, ok := findColumnPosition(tableDef, colRef.Name) + if !ok { + return 0.1 + } + // The table schema is authoritative. StatsInfo.DataTypeMap may be absent + // in old caches or stale after a schema change, and must not change how + // otherwise valid min/max values are interpreted. + typ := types.T(tableDef.Cols[colPos].Typ.Id) switch colFnName { case "": @@ -883,15 +1411,15 @@ func estimateNonEqualitySelectivity(expr *plan.Expr, funcName string, builder *Q // Decimal literals store internal scaled values, need proper conversion if typ == types.T_decimal64 || typ == types.T_decimal128 { return calcSelectivityByMinMaxForDecimal( - funcName, s.MinValMap[colRef.Name], s.MaxValMap[colRef.Name], expr) + funcName, validated.minVal, validated.maxVal, expr) } return calcSelectivityByMinMax( - funcName, s.MinValMap[colRef.Name], s.MaxValMap[colRef.Name], typ, literals) + funcName, validated.minVal, validated.maxVal, typ, literals) case "year": switch typ { case types.T_date: - minVal := types.Date(s.MinValMap[colRef.Name]) - maxVal := types.Date(s.MaxValMap[colRef.Name]) + minVal := types.Date(validated.minVal) + maxVal := types.Date(validated.maxVal) return calcSelectivityByMinMax(funcName, float64(minVal.Year()), float64(maxVal.Year()), litType, literals) case types.T_datetime: // TODO @@ -1149,6 +1677,12 @@ func ReCalcNodeStats(nodeID int32, builder *QueryBuilder, recursive bool, leafNo switch node.JoinType { case plan.Node_INNER: outcnt := leftStats.Outcnt * rightStats.Outcnt / ndv + // The legacy fallback assumes the smaller input is effectively unique. + // A single direct equality with available NDV/null statistics can expose + // a possible many-to-many join. Never lower the established estimate. + if ndvOutcnt, ok := availableSingleEquiJoinCardinality(node, builder, leftStats, rightStats); ok && ndvOutcnt > outcnt { + outcnt = ndvOutcnt + } if !isCrossJoin { outcnt *= selectivity } @@ -1705,7 +2239,7 @@ func calcScanStats(node *plan.Node, builder *QueryBuilder) *plan.Stats { } stats := new(plan.Stats) - stats.TableCnt = s.TableCnt + stats.TableCnt = validTableCnt(s.TableCnt) var blockSel float64 = 1 var preservedCompositeFilters []*plan.Expr if builder.optimizerHints == nil || builder.optimizerHints.blockFilter != 2 { diff --git a/pkg/sql/plan/stats_test.go b/pkg/sql/plan/stats_test.go index 5bcac8788bd09..2848c064c6494 100644 --- a/pkg/sql/plan/stats_test.go +++ b/pkg/sql/plan/stats_test.go @@ -202,6 +202,265 @@ func TestSafeStatsRatiosAvoidNonFiniteSelectivity(t *testing.T) { }) } +func TestSanitizeStatsInfoInvariants(t *testing.T) { + newTableDef := func(typ types.T, primaryKey bool) *planpb.TableDef { + tableDef := &planpb.TableDef{ + Name: "t", + Cols: []*planpb.ColDef{ + {Name: "c", Typ: planpb.Type{Id: int32(typ)}}, + {Name: catalog.Row_ID}, + }, + } + if primaryKey { + tableDef.Pkey = &planpb.PrimaryKeyDef{PkeyColName: "c", Names: []string{"c"}} + } + return tableDef + } + newStats := func(tableCnt float64, nullCnt uint64, ndv, minVal, maxVal float64) *pb.StatsInfo { + return &pb.StatsInfo{ + TableCnt: tableCnt, + SampleRatio: 1, + NdvMap: map[string]float64{"c": ndv}, + NullCntMap: map[string]uint64{"c": nullCnt}, + MinValMap: map[string]float64{"c": minVal}, + MaxValMap: map[string]float64{"c": maxVal}, + MinMaxValidMap: map[string]bool{"c": true}, + MinMaxCompleteMap: map[string]bool{"c": true}, + } + } + + t.Run("discrete domain caps impossible ndv", func(t *testing.T) { + s := newStats(1000, 100, 2000, 1, 10) + tableDef := newTableDef(types.T_int32, false) + sanitizeStatsInfo(tableDef, s) + require.Equal(t, 10.0, s.NdvMap["c"]) + validated := validateColumnStats(s, tableDef, "c") + require.True(t, validated.rangeEstimateSafe) + require.True(t, validated.shuffleBoundsSafe) + }) + + t.Run("sampled min max is not a hard domain bound", func(t *testing.T) { + s := newStats(1000, 0, 500, 1, 10) + s.SampleRatio = 0.1 + delete(s.MinMaxCompleteMap, "c") + tableDef := newTableDef(types.T_int32, false) + sanitizeStatsInfo(tableDef, s) + require.Equal(t, 500.0, s.NdvMap["c"]) + validated := validateColumnStats(s, tableDef, "c") + require.True(t, validated.minMaxKnown) + require.False(t, validated.minMaxComplete) + require.False(t, validated.rangeEstimateSafe) + require.False(t, validated.shuffleBoundsSafe) + }) + + t.Run("legacy zero pair is unknown without validity metadata", func(t *testing.T) { + s := newStats(1000, 0, 500, 0, 0) + s.AccurateObjectNumber = 1 + s.MinMaxValidMap = nil + s.MinMaxCompleteMap = nil + sanitizeStatsInfo(newTableDef(types.T_int32, false), s) + require.Equal(t, 500.0, s.NdvMap["c"]) + _, minOK := s.MinValMap["c"] + _, maxOK := s.MaxValMap["c"] + require.False(t, minOK) + require.False(t, maxOK) + }) + + t.Run("primary key rejects contradictory complete domain", func(t *testing.T) { + s := newStats(1000, 0, 10, 1, 10) + tableDef := newTableDef(types.T_int64, true) + sanitizeStatsInfo(tableDef, s) + require.Equal(t, 1000.0, s.NdvMap["c"]) + require.NotContains(t, s.MinValMap, "c") + require.NotContains(t, s.MaxValMap, "c") + require.NotContains(t, s.MinMaxValidMap, "c") + require.NotContains(t, s.MinMaxCompleteMap, "c") + }) + + t.Run("non-null rows cap ndv", func(t *testing.T) { + s := newStats(1000, 999, 10, 1, 100) + sanitizeStatsInfo(newTableDef(types.T_int32, false), s) + require.Equal(t, 1.0, s.NdvMap["c"]) + }) + + t.Run("invalid range becomes unknown without fabricating ndv", func(t *testing.T) { + s := newStats(1000, 100, 950, 10, 1) + sanitizeStatsInfo(newTableDef(types.T_int32, false), s) + require.Equal(t, 900.0, s.NdvMap["c"]) + _, minOK := s.MinValMap["c"] + _, maxOK := s.MaxValMap["c"] + require.False(t, minOK) + require.False(t, maxOK) + }) + + t.Run("invalid ndv remains unknown", func(t *testing.T) { + s := newStats(1000, 0, math.NaN(), 1, 10) + sanitizeStatsInfo(newTableDef(types.T_int32, false), s) + _, ok := s.NdvMap["c"] + require.False(t, ok) + }) + + t.Run("single column primary key is exact", func(t *testing.T) { + s := newStats(1000, 100, math.NaN(), 1, 1000) + sanitizeStatsInfo(newTableDef(types.T_int64, true), s) + require.Equal(t, 1000.0, s.NdvMap["c"]) + require.Equal(t, uint64(0), s.NullCntMap["c"]) + }) + + t.Run("wide integer range is not narrowed through lossy float min max", func(t *testing.T) { + minVal := float64(1<<53) + 2 + s := newStats(1000, 0, 500, minVal, minVal+8) + tableDef := newTableDef(types.T_int64, false) + sanitizeStatsInfo(tableDef, s) + require.Equal(t, 500.0, s.NdvMap["c"]) + validated := validateColumnStats(s, tableDef, "c") + require.True(t, validated.minMaxComplete) + require.False(t, validated.rangeEstimateSafe) + require.False(t, validated.shuffleBoundsSafe) + }) + + t.Run("encoded string min max is not treated as a discrete domain", func(t *testing.T) { + s := newStats(1000, 0, 500, 1, 10) + tableDef := newTableDef(types.T_varchar, false) + sanitizeStatsInfo(tableDef, s) + require.Equal(t, 500.0, s.NdvMap["c"]) + validated := validateColumnStats(s, tableDef, "c") + require.False(t, validated.rangeEstimateSafe) + require.False(t, validated.shuffleBoundsSafe) + }) + + t.Run("invalid table count collapses dependent counts safely", func(t *testing.T) { + s := newStats(math.NaN(), 10, 10, 1, 10) + sanitizeStatsInfo(newTableDef(types.T_int32, false), s) + require.Equal(t, 0.0, s.TableCnt) + require.Equal(t, uint64(0), s.NullCntMap["c"]) + require.Equal(t, 0.0, s.NdvMap["c"]) + }) +} + +func TestAdjustNDVUsesMinMaxOnlyWithCompleteObjectCoverage(t *testing.T) { + tableDef := &planpb.TableDef{ + Name: "t", + Cols: []*planpb.ColDef{ + {Name: "c", Typ: planpb.Type{Id: int32(types.T_int32)}}, + {Name: catalog.Row_ID}, + }, + } + + for _, tc := range []struct { + name string + sampleRatio float64 + wantNDV float64 + }{ + {name: "full metadata", sampleRatio: 1, wantNDV: 10}, + {name: "sampled metadata", sampleRatio: 0.1, wantNDV: 500}, + } { + t.Run(tc.name, func(t *testing.T) { + info := &TableStatsInfo{ + ColumnNDVs: []float64{500}, + TableRowCount: 1000, + SampleRatio: tc.sampleRatio, + } + s := NewStatsInfo() + s.TableCnt = 1000 + s.AccurateObjectNumber = 1000 + s.MinValMap["c"] = 1 + s.MaxValMap["c"] = 10 + s.MinMaxValidMap["c"] = true + if tc.sampleRatio == 1 { + s.MinMaxCompleteMap["c"] = true + } + + AdjustNDV(info, tableDef, s) + + require.Equal(t, tc.wantNDV, s.NdvMap["c"]) + }) + } +} + +func TestRangeSelectivityRequiresCompleteMinMax(t *testing.T) { + stats := NewStatsInfo() + stats.TableCnt = 1000 + stats.NdvMap["c"] = 100 + stats.MinValMap["c"] = 1 + stats.MaxValMap["c"] = 100 + stats.MinMaxValidMap["c"] = true + stats.MinMaxCompleteMap["c"] = true + stats.DataTypeMap["c"] = uint64(types.T_int64) + statsCache := NewStatsCache() + statsCache.Set(1, stats) + ctx := &statsCacheCompilerContext{ + MockCompilerContext: &MockCompilerContext{ctx: context.Background()}, + statsCache: statsCache, + } + builder := NewQueryBuilder(planpb.Query_SELECT, ctx, false, false) + builder.tag2Table[0] = &planpb.TableDef{ + TblId: 1, + Cols: []*planpb.ColDef{{Name: "c", Typ: planpb.Type{Id: int32(types.T_int64)}}}, + } + expr := &planpb.Expr{Expr: &planpb.Expr_F{F: &planpb.Function{ + Func: &planpb.ObjectRef{ObjName: ">"}, + Args: []*planpb.Expr{ + {Typ: planpb.Type{Id: int32(types.T_int64)}, Expr: &planpb.Expr_Col{Col: &planpb.ColRef{RelPos: 0, Name: "c"}}}, + {Typ: planpb.Type{Id: int32(types.T_int64)}, Expr: &planpb.Expr_Lit{Lit: &planpb.Literal{Value: &planpb.Literal_I64Val{I64Val: 50}}}}, + }, + }}} + + stats.SampleRatio = 0.1 + delete(stats.MinMaxCompleteMap, "c") + require.Equal(t, 0.1, estimateNonEqualitySelectivity(expr, ">", builder)) + + stats.SampleRatio = 1 + stats.MinMaxCompleteMap["c"] = true + require.InDelta(t, 0.5, estimateNonEqualitySelectivity(expr, ">", builder), 0.02) + + delete(stats.DataTypeMap, "c") // schema, not cached metadata, defines the type + require.InDelta(t, 0.5, estimateNonEqualitySelectivity(expr, ">", builder), 0.02) + + stats.MinMaxValidMap = nil // mixed-version/legacy StatsInfo + require.Equal(t, 0.1, estimateNonEqualitySelectivity(expr, ">", builder)) +} + +func TestShuffleRangesRequireRepresentableSampledValues(t *testing.T) { + rangeInfo := &pb.ShuffleRange{SampleRatio: 0.1} + require.True(t, shuffleRangesSafe(types.T_int64, rangeInfo, []float64{1, 50.5, 100})) + require.False(t, shuffleRangesSafe(types.T_int64, rangeInfo, []float64{50, 1})) + require.False(t, shuffleRangesSafe(types.T_int64, rangeInfo, []float64{float64(1 << 53)})) + require.False(t, shuffleRangesSafe(types.T_varchar, rangeInfo, []float64{1, 50, 100})) + require.False(t, shuffleRangesSafe(types.T_int64, &pb.ShuffleRange{}, []float64{1, 50, 100})) +} + +func TestSampledUniformShuffleUsesQuantileRangesWithoutCompleteBounds(t *testing.T) { + ranges := []float64{1, 50, 100} + rangeInfo := &pb.ShuffleRange{ + SampleRatio: 0.1, + Uniform: 0.999, + Result: ranges, + } + + require.Equal(t, ranges, shouldUseShuffleRanges(rangeInfo, false)) + require.Nil(t, shouldUseShuffleRanges(rangeInfo, true)) + + rangeInfo.Uniform = uniformThreshold - 0.01 + require.Equal(t, ranges, shouldUseShuffleRanges(rangeInfo, true)) +} + +func TestStatsInfoMinMaxMetadataProtoRoundTrip(t *testing.T) { + stats := NewStatsInfo() + stats.SampleRatio = 0.25 + stats.MinMaxValidMap["c"] = true + stats.MinMaxCompleteMap["c"] = false + + data, err := stats.Marshal() + require.NoError(t, err) + var decoded pb.StatsInfo + require.NoError(t, decoded.Unmarshal(data)) + require.Equal(t, 0.25, decoded.SampleRatio) + require.True(t, decoded.MinMaxValidMap["c"]) + require.Contains(t, decoded.MinMaxCompleteMap, "c") + require.False(t, decoded.MinMaxCompleteMap["c"]) +} + func TestStatsSelectivityClampAvoidsNonFiniteJoin(t *testing.T) { t.Run("not over year equality stays in range", func(t *testing.T) { builder := newStatsTestBuilderWithNDV("d", 1) @@ -1048,6 +1307,7 @@ func TestUpdateStatsInfo_Decimal64_NegativeValues(t *testing.T) { // Verify results minVal := statsInfo.MinValMap["balance"] maxVal := statsInfo.MaxValMap["balance"] + require.True(t, statsInfo.MinMaxValidMap["balance"]) // The key assertion: min should be less than max require.Less(t, minVal, maxVal, "Min value should be less than max value") @@ -1190,6 +1450,12 @@ func TestUpdateStatsInfo_ShuffleRangeWrittenWhenZoneMapNotInited(t *testing.T) { // Before the fix: continue skipped the ShuffleRange block, so ShuffleRangeMap stayed empty. require.NotNil(t, s.ShuffleRangeMap["id"], "ShuffleRange must be written when ZoneMap is not inited but ShuffleRange is present and other canFill conditions are met") + _, minOK := s.MinValMap["id"] + _, maxOK := s.MaxValMap["id"] + _, validOK := s.MinMaxValidMap["id"] + require.False(t, minOK, "uninitialized ZoneMap must leave min unknown") + require.False(t, maxOK, "uninitialized ZoneMap must leave max unknown") + require.False(t, validOK, "uninitialized ZoneMap must leave min/max invalid") require.Nil(t, info.ShuffleRanges[0], "UpdateStatsInfo nils out info.ShuffleRanges after copying to s") } @@ -1797,3 +2063,213 @@ func TestCompareStatsIsStrictWeakOrdering(t *testing.T) { } } } + +func TestInnerJoinCardinalityUsesAvailableSingleKeyStatsWithoutLoweringFallback(t *testing.T) { + type testCase struct { + name string + leftRows float64 + rightRows float64 + leftNDV float64 + rightNDV float64 + leftNulls uint64 + rightNulls uint64 + omitNulls bool + sampleRatio float64 + objectCount int64 + conditions int + want float64 + } + tests := []testCase{ + { + name: "repeated single key raises underestimated join output", + leftRows: 1_000_000, rightRows: 1_000_000, + leftNDV: 10_000, rightNDV: 10_000, conditions: 1, + want: 100_000_000, + }, + { + name: "asymmetric repeated key uses both side ndv", + leftRows: 1_000_000, rightRows: 100_000, + leftNDV: 1_000, rightNDV: 1_000, conditions: 1, + want: 100_000_000, + }, + { + name: "unique sized key preserves old estimate", + leftRows: 1_000_000, rightRows: 100_000, + leftNDV: 1_000_000, rightNDV: 100_000, conditions: 1, + want: 1_000_000, + }, + { + name: "child rows below base ndv preserves old estimate", + leftRows: 100, rightRows: 1_000_000, + leftNDV: 10_000, rightNDV: 10_000, conditions: 1, + want: 1_000_000, + }, + { + name: "null heavy keys do not inflate the numerator", + leftRows: 1_000_000, rightRows: 1_000_000, + leftNDV: 1, rightNDV: 1, leftNulls: 999_000, rightNulls: 999_000, conditions: 1, + want: 1_000_000, + }, + { + name: "asymmetric null fractions use non null rows", + leftRows: 1_000_000, rightRows: 100_000, + leftNDV: 1_000, rightNDV: 1_000, leftNulls: 900_000, conditions: 1, + want: 10_000_000, + }, + { + name: "missing null count preserves old estimate", + leftRows: 1_000_000, rightRows: 1_000_000, + leftNDV: 10_000, rightNDV: 10_000, omitNulls: true, conditions: 1, + want: 1_000_000, + }, + { + name: "sampled zero null count keeps an uncertainty margin", + leftRows: 1_000_000, rightRows: 1_000_000, + leftNDV: 10_000, rightNDV: 10_000, sampleRatio: 0.1, objectCount: 1_000, conditions: 1, + want: 94_090_000, + }, + { + name: "missing ndv preserves old estimate", + leftRows: 1_000_000, rightRows: 100_000, + leftNDV: 0, rightNDV: 0, conditions: 1, + want: 1_000_000, + }, + { + name: "multiple keys preserve old estimate without tuple ndv", + leftRows: 1_000_000, rightRows: 1_000_000, + leftNDV: 1_000, rightNDV: 1_000, conditions: 2, + want: 1_000_000, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + statsCache := NewStatsCache() + makeTableStats := func(rows, ndv float64, nulls uint64) *pb.StatsInfo { + stats := NewStatsInfo() + stats.TableCnt = rows + stats.SampleRatio = tt.sampleRatio + stats.AccurateObjectNumber = tt.objectCount + if stats.SampleRatio == 0 { + stats.SampleRatio = 1 + } + if stats.AccurateObjectNumber == 0 { + stats.AccurateObjectNumber = 1 + } + if ndv > 0 { + stats.NdvMap["k1"] = ndv + stats.NdvMap["k2"] = ndv + } + if !tt.omitNulls { + stats.NullCntMap["k1"] = nulls + stats.NullCntMap["k2"] = nulls + } + return stats + } + statsCache.Set(1, makeTableStats(tt.leftRows, tt.leftNDV, tt.leftNulls)) + statsCache.Set(2, makeTableStats(tt.rightRows, tt.rightNDV, tt.rightNulls)) + ctx := &statsCacheCompilerContext{ + MockCompilerContext: &MockCompilerContext{ctx: context.Background()}, + statsCache: statsCache, + } + builder := NewQueryBuilder(planpb.Query_SELECT, ctx, false, false) + + makeTableDef := func(tableID uint64) *planpb.TableDef { + return &planpb.TableDef{ + TblId: tableID, + Cols: []*planpb.ColDef{ + {Name: "k1", Typ: planpb.Type{Id: int32(types.T_int64)}}, + {Name: "k2", Typ: planpb.Type{Id: int32(types.T_int64)}}, + }, + Name2ColIndex: map[string]int32{"k1": 0, "k2": 1}, + } + } + leftDef := makeTableDef(1) + rightDef := makeTableDef(2) + builder.tag2Table[1] = leftDef + builder.tag2Table[2] = rightDef + left := &planpb.Node{ + NodeType: planpb.Node_TABLE_SCAN, BindingTags: []int32{1}, TableDef: leftDef, + Stats: &planpb.Stats{TableCnt: tt.leftRows, Outcnt: tt.leftRows, Cost: tt.leftRows, Selectivity: 1, BlockNum: 1}, + } + right := &planpb.Node{ + NodeType: planpb.Node_TABLE_SCAN, BindingTags: []int32{2}, TableDef: rightDef, + Stats: &planpb.Stats{TableCnt: tt.rightRows, Outcnt: tt.rightRows, Cost: tt.rightRows, Selectivity: 1, BlockNum: 1}, + } + join := &planpb.Node{ + NodeType: planpb.Node_JOIN, JoinType: planpb.Node_INNER, + Children: []int32{0, 1}, Stats: DefaultStats(), + } + for i := 0; i < tt.conditions; i++ { + name := "k1" + if i == 1 { + name = "k2" + } + predicate, err := BindFuncExprImplByPlanExpr(context.Background(), "=", []*planpb.Expr{ + {Typ: planpb.Type{Id: int32(types.T_int64)}, Expr: &planpb.Expr_Col{Col: &planpb.ColRef{RelPos: 1, ColPos: int32(i), Name: name}}}, + {Typ: planpb.Type{Id: int32(types.T_int64)}, Expr: &planpb.Expr_Col{Col: &planpb.ColRef{RelPos: 2, ColPos: int32(i), Name: name}}}, + }) + require.NoError(t, err) + join.OnList = append(join.OnList, predicate) + } + builder.qry.Nodes = []*planpb.Node{left, right, join} + + ReCalcNodeStats(2, builder, false, false, false) + + require.Equal(t, tt.want, join.Stats.Outcnt) + }) + } +} + +func TestJoinColumnNullLineageSafety(t *testing.T) { + col := &planpb.ColRef{RelPos: 1, ColPos: 0, Name: "k"} + scan := func(tag int32) *planpb.Node { + return &planpb.Node{NodeType: planpb.Node_TABLE_SCAN, BindingTags: []int32{tag}} + } + + t.Run("filter on join key is unsafe", func(t *testing.T) { + builder := &QueryBuilder{qry: &planpb.Query{Nodes: []*planpb.Node{ + scan(1), + { + NodeType: planpb.Node_FILTER, + Children: []int32{0}, + FilterList: []*planpb.Expr{{ + Expr: &planpb.Expr_Col{Col: col}, + }}, + }, + }}} + require.False(t, joinColumnNullLineageSafe(1, col, builder)) + }) + + t.Run("filter on another column keeps existing independence assumption", func(t *testing.T) { + builder := &QueryBuilder{qry: &planpb.Query{Nodes: []*planpb.Node{ + scan(1), + { + NodeType: planpb.Node_FILTER, + Children: []int32{0}, + FilterList: []*planpb.Expr{{ + Expr: &planpb.Expr_Col{Col: &planpb.ColRef{RelPos: 1, ColPos: 1, Name: "v"}}, + }}, + }, + }}} + require.True(t, joinColumnNullLineageSafe(1, col, builder)) + }) + + t.Run("left join null extension is unsafe", func(t *testing.T) { + builder := &QueryBuilder{qry: &planpb.Query{Nodes: []*planpb.Node{ + scan(3), + scan(1), + {NodeType: planpb.Node_JOIN, JoinType: planpb.Node_LEFT, Children: []int32{0, 1}}, + }}} + require.False(t, joinColumnNullLineageSafe(2, col, builder)) + }) + + t.Run("inner join lineage remains usable", func(t *testing.T) { + builder := &QueryBuilder{qry: &planpb.Query{Nodes: []*planpb.Node{ + scan(1), + scan(3), + {NodeType: planpb.Node_JOIN, JoinType: planpb.Node_INNER, Children: []int32{0, 1}}, + }}} + require.True(t, joinColumnNullLineageSafe(2, col, builder)) + }) +} diff --git a/pkg/vm/engine/disttae/stats.go b/pkg/vm/engine/disttae/stats.go index ff99f1216c8dc..cbdf9cc46ab41 100644 --- a/pkg/vm/engine/disttae/stats.go +++ b/pkg/vm/engine/disttae/stats.go @@ -708,9 +708,10 @@ func (gs *GlobalStats) GetBaseObjectCnt(key pb.StatsInfoKey) int64 { // ShuffleRangePartialUpdate contains fields that can be independently updated in ShuffleRange type ShuffleRangePartialUpdate struct { - Overlap *float64 `json:"overlap,omitempty"` - Uniform *float64 `json:"uniform,omitempty"` - Result []float64 `json:"result,omitempty"` + Overlap *float64 `json:"overlap,omitempty"` + Uniform *float64 `json:"uniform,omitempty"` + Result []float64 `json:"result,omitempty"` + SampleRatio *float64 `json:"sample_ratio,omitempty"` } // PatchArgs defines arguments for patch command @@ -719,6 +720,7 @@ type PatchArgs struct { TableCnt *float64 `json:"table_cnt,omitempty"` BlockNumber *int64 `json:"block_number,omitempty"` AccurateObjectNumber *int64 `json:"accurate_object_number,omitempty"` + SampleRatio *float64 `json:"sample_ratio,omitempty"` // Column-level stats (merge mode) NdvMap map[string]float64 `json:"ndv_map,omitempty"` @@ -726,6 +728,10 @@ type PatchArgs struct { MaxValMap map[string]float64 `json:"max_val_map,omitempty"` NullCntMap map[string]uint64 `json:"null_cnt_map,omitempty"` SizeMap map[string]uint64 `json:"size_map,omitempty"` + // MinMaxCompleteMap explicitly marks a column's min/max pair as a + // whole-table bound. A paired min/max patch defaults to complete for + // backward compatibility; callers can set false for sampled bounds. + MinMaxCompleteMap map[string]bool `json:"min_max_complete_map,omitempty"` // ShuffleRange partial updates (fine-grained control per column) // Each column can have its Overlap/Uniform/Result fields updated independently @@ -743,10 +749,64 @@ func (gs *GlobalStats) PatchStats(key pb.StatsInfoKey, patch *PatchArgs) error { stats := gs.mu.statsInfoMap[key] if stats == nil { - // Create new stats if not exists stats = plan2.NewStatsInfo() - gs.mu.statsInfoMap[key] = stats } + if patch.SampleRatio != nil && (*patch.SampleRatio <= 0 || *patch.SampleRatio > 1 || + math.IsNaN(*patch.SampleRatio) || math.IsInf(*patch.SampleRatio, 0)) { + return moerr.NewInvalidInputNoCtxf("invalid stats sample ratio %v", *patch.SampleRatio) + } + + // Validate every partial update before mutating the cached object so a + // rejected patch is atomic. + for col, update := range patch.ShuffleRangeMap { + if update == nil { + continue + } + if update.SampleRatio != nil { + ratio := *update.SampleRatio + if ratio <= 0 || ratio > 1 || math.IsNaN(ratio) || math.IsInf(ratio, 0) { + return moerr.NewInvalidInputNoCtxf( + "invalid shuffle range sample ratio %v for column %q", ratio, col) + } + } + for i, value := range update.Result { + if math.IsNaN(value) || math.IsInf(value, 0) || i > 0 && value < update.Result[i-1] { + return moerr.NewInvalidInputNoCtxf("invalid shuffle range result for column %q", col) + } + } + } + + touchedMinMax := make(map[string]struct{}, len(patch.MinValMap)+len(patch.MaxValMap)+len(patch.MinMaxCompleteMap)) + for col := range patch.MinValMap { + touchedMinMax[col] = struct{}{} + } + for col := range patch.MaxValMap { + touchedMinMax[col] = struct{}{} + } + for col := range patch.MinMaxCompleteMap { + touchedMinMax[col] = struct{}{} + } + for col := range touchedMinMax { + minVal, minOK := stats.MinValMap[col] + patchedMin, minPatched := patch.MinValMap[col] + if minPatched { + minVal, minOK = patchedMin, true + } + maxVal, maxOK := stats.MaxValMap[col] + patchedMax, maxPatched := patch.MaxValMap[col] + if maxPatched { + maxVal, maxOK = patchedMax, true + } + explicitComplete, hasExplicitComplete := patch.MinMaxCompleteMap[col] + provenanceKnown := stats.MinMaxValidMap[col] || minPatched && maxPatched || + hasExplicitComplete && explicitComplete + if !minOK || !maxOK || !provenanceKnown || math.IsNaN(minVal) || math.IsNaN(maxVal) || + math.IsInf(minVal, 0) || math.IsInf(maxVal, 0) || minVal > maxVal { + return moerr.NewInvalidInputNoCtxf( + "min/max patch for column %q requires a complete valid pair", col) + } + } + gs.mu.statsInfoMap[key] = stats // Apply table-level stats if patch.TableCnt != nil { @@ -758,6 +818,9 @@ func (gs *GlobalStats) PatchStats(key pb.StatsInfoKey, patch *PatchArgs) error { if patch.AccurateObjectNumber != nil { stats.AccurateObjectNumber = *patch.AccurateObjectNumber } + if patch.SampleRatio != nil { + stats.SampleRatio = *patch.SampleRatio + } // Apply column-level stats (merge mode) for col, v := range patch.NdvMap { @@ -766,17 +829,53 @@ func (gs *GlobalStats) PatchStats(key pb.StatsInfoKey, patch *PatchArgs) error { } stats.NdvMap[col] = v } - for col, v := range patch.MinValMap { - if stats.MinValMap == nil { - stats.MinValMap = make(map[string]float64) - } - stats.MinValMap[col] = v + if stats.MinValMap == nil { + stats.MinValMap = make(map[string]float64) + } + if stats.MaxValMap == nil { + stats.MaxValMap = make(map[string]float64) + } + if stats.MinMaxValidMap == nil { + stats.MinMaxValidMap = make(map[string]bool) + } + if stats.MinMaxCompleteMap == nil { + stats.MinMaxCompleteMap = make(map[string]bool) } - for col, v := range patch.MaxValMap { - if stats.MaxValMap == nil { - stats.MaxValMap = make(map[string]float64) + for col := range touchedMinMax { + wasValid := stats.MinMaxValidMap[col] + wasComplete := stats.MinMaxCompleteMap[col] + minVal, minPatched := patch.MinValMap[col] + maxVal, maxPatched := patch.MaxValMap[col] + if minPatched { + stats.MinValMap[col] = minVal + } + if maxPatched { + stats.MaxValMap[col] = maxVal + } + _, minOK := stats.MinValMap[col] + _, maxOK := stats.MaxValMap[col] + valid := minOK && maxOK && (wasValid || minPatched && maxPatched) + if explicit, ok := patch.MinMaxCompleteMap[col]; ok && explicit && minOK && maxOK { + valid = true + } + if valid { + stats.MinMaxValidMap[col] = true + } else { + delete(stats.MinMaxValidMap, col) + } + + complete := wasComplete + if minPatched && maxPatched { + complete = true + } + if explicit, ok := patch.MinMaxCompleteMap[col]; ok { + complete = explicit + } + if valid && complete { + stats.MinMaxCompleteMap[col] = true + } else { + delete(stats.MinMaxCompleteMap, col) } - stats.MaxValMap[col] = v } for col, v := range patch.NullCntMap { if stats.NullCntMap == nil { @@ -793,6 +892,9 @@ func (gs *GlobalStats) PatchStats(key pb.StatsInfoKey, patch *PatchArgs) error { // Apply ShuffleRange partial updates (fine-grained) for col, update := range patch.ShuffleRangeMap { + if update == nil { + continue + } if stats.ShuffleRangeMap == nil { stats.ShuffleRangeMap = make(map[string]*pb.ShuffleRange) } @@ -813,6 +915,12 @@ func (gs *GlobalStats) PatchStats(key pb.StatsInfoKey, patch *PatchArgs) error { } if update.Result != nil { sr.Result = update.Result + if update.SampleRatio == nil { + sr.SampleRatio = 1 + } + } + if update.SampleRatio != nil { + sr.SampleRatio = *update.SampleRatio } } @@ -1355,6 +1463,7 @@ func collectTableStats( if exactObjectNumber > 0 { actualSamplingRatio = float64(sampledObjectCount) / float64(exactObjectNumber) } + info.SampleRatio = actualSamplingRatio for _, r := range info.ShuffleRanges { if r != nil { r.SampleRatio = actualSamplingRatio @@ -1367,8 +1476,9 @@ func collectTableStats( for i := range info.ColumnSize { info.ColumnSize[i] = int64(float64(info.ColumnSize[i]) * rowScaleFactor) info.NullCnts[i] = int64(float64(info.NullCnts[i]) * rowScaleFactor) - // NDV: scale up by inverse of sampling ratio, cap at row count - upper := info.TableRowCount * 0.99 + // NDV: scale up by inverse of sampling ratio, cap at row count. + // Equality with row count is valid for a unique column. + upper := info.TableRowCount info.ColumnNDVs[i] = math.Min(info.ColumnNDVs[i]*rowScaleFactor, upper) } } diff --git a/pkg/vm/engine/disttae/stats_test.go b/pkg/vm/engine/disttae/stats_test.go index cc6c0a1504c8b..b14de5534b8f2 100644 --- a/pkg/vm/engine/disttae/stats_test.go +++ b/pkg/vm/engine/disttae/stats_test.go @@ -18,6 +18,7 @@ import ( "context" "encoding/binary" "fmt" + "math" goruntime "runtime" "sync" "sync/atomic" @@ -47,6 +48,123 @@ type mockStatsKeyRouter struct { target string } +func TestPatchStatsMaintainsMinMaxProvenance(t *testing.T) { + newGlobalStats := func() *GlobalStats { + gs := new(GlobalStats) + gs.mu.statsInfoMap = make(map[statsinfo.StatsInfoKey]*statsinfo.StatsInfo) + gs.mu.cond = sync.NewCond(&gs.mu) + return gs + } + key := statsinfo.StatsInfoKey{TableID: 42} + + t.Run("paired bounds and ranges are authoritative by default", func(t *testing.T) { + gs := newGlobalStats() + err := gs.PatchStats(key, &PatchArgs{ + MinValMap: map[string]float64{"k": 1}, + MaxValMap: map[string]float64{"k": 100}, + ShuffleRangeMap: map[string]*ShuffleRangePartialUpdate{ + "k": {Result: []float64{1, 50, 100}}, + }, + }) + require.NoError(t, err) + stats := gs.mu.statsInfoMap[key] + require.True(t, stats.MinMaxValidMap["k"]) + require.True(t, stats.MinMaxCompleteMap["k"]) + require.Equal(t, 1.0, stats.ShuffleRangeMap["k"].SampleRatio) + }) + + t.Run("sampled bounds remain valid but incomplete", func(t *testing.T) { + gs := newGlobalStats() + ratio := 0.1 + err := gs.PatchStats(key, &PatchArgs{ + SampleRatio: &ratio, + MinValMap: map[string]float64{"k": 1}, + MaxValMap: map[string]float64{"k": 100}, + MinMaxCompleteMap: map[string]bool{"k": false}, + }) + require.NoError(t, err) + stats := gs.mu.statsInfoMap[key] + require.True(t, stats.MinMaxValidMap["k"]) + require.False(t, stats.MinMaxCompleteMap["k"]) + require.Equal(t, ratio, stats.SampleRatio) + }) + + t.Run("range-only patch carries explicit provenance", func(t *testing.T) { + gs := newGlobalStats() + err := gs.PatchStats(key, &PatchArgs{ + ShuffleRangeMap: map[string]*ShuffleRangePartialUpdate{ + "k": {Result: []float64{1, 50, 100}}, + }, + }) + require.NoError(t, err) + require.Equal(t, 1.0, gs.mu.statsInfoMap[key].ShuffleRangeMap["k"].SampleRatio) + }) + + t.Run("one patched legacy bound is rejected atomically", func(t *testing.T) { + gs := newGlobalStats() + legacy := plan2.NewStatsInfo() + legacy.MinValMap["k"] = 0 + legacy.MaxValMap["k"] = 0 + gs.mu.statsInfoMap[key] = legacy + err := gs.PatchStats(key, &PatchArgs{ + MaxValMap: map[string]float64{"k": 100}, + }) + require.Error(t, err) + require.Equal(t, 0.0, legacy.MaxValMap["k"]) + require.False(t, legacy.MinMaxValidMap["k"]) + require.False(t, legacy.MinMaxCompleteMap["k"]) + }) + + t.Run("sequential bounds cannot create ambiguous provenance", func(t *testing.T) { + gs := newGlobalStats() + require.Error(t, gs.PatchStats(key, &PatchArgs{ + MinValMap: map[string]float64{"k": 1}, + })) + require.NotContains(t, gs.mu.statsInfoMap, key) + require.Error(t, gs.PatchStats(key, &PatchArgs{ + MaxValMap: map[string]float64{"k": 100}, + })) + require.NotContains(t, gs.mu.statsInfoMap, key) + }) + + t.Run("invalid ratio-only update is rejected atomically", func(t *testing.T) { + gs := newGlobalStats() + gs.mu.statsInfoMap[key] = plan2.NewStatsInfo() + invalidRatio := 2.0 + tableCnt := 100.0 + err := gs.PatchStats(key, &PatchArgs{ + TableCnt: &tableCnt, + ShuffleRangeMap: map[string]*ShuffleRangePartialUpdate{ + "k": {SampleRatio: &invalidRatio}, + }, + }) + require.Error(t, err) + require.Equal(t, 0.0, gs.mu.statsInfoMap[key].TableCnt) + }) + + for _, tc := range []struct { + name string + minVal float64 + maxVal float64 + }{ + {name: "nan", minVal: math.NaN(), maxVal: 10}, + {name: "infinite", minVal: 1, maxVal: math.Inf(1)}, + {name: "reversed", minVal: 10, maxVal: 1}, + } { + t.Run("invalid bounds "+tc.name, func(t *testing.T) { + gs := newGlobalStats() + tableCnt := 100.0 + err := gs.PatchStats(key, &PatchArgs{ + TableCnt: &tableCnt, + MinValMap: map[string]float64{"k": tc.minVal}, + MaxValMap: map[string]float64{"k": tc.maxVal}, + }) + require.Error(t, err) + require.NotContains(t, gs.mu.statsInfoMap, key) + }) + } +} + func (r *mockStatsKeyRouter) Target(statsinfo.StatsInfoKey) string { return r.target } func (r *mockStatsKeyRouter) AddItem(gossip.CommonItem) {} diff --git a/pkg/vm/engine/types.go b/pkg/vm/engine/types.go index 5e6b7ef0237eb..404893c5e3ffe 100644 --- a/pkg/vm/engine/types.go +++ b/pkg/vm/engine/types.go @@ -1044,10 +1044,11 @@ type RangesShuffleParam struct { IsLocalCN bool // ShuffleByObjectID assigns IVF persisted and appendable objects to the // same physical CN owner. - ShuffleByObjectID bool - ShuffleRangeUint64 []uint64 - ShuffleRangeInt64 []int64 - Init bool + ShuffleByObjectID bool + ShuffleRangeUint64 []uint64 + ShuffleRangeInt64 []int64 + Init bool + ShuffleRangeBuckets int } type RangesParam struct { diff --git a/proto/statsinfo.proto b/proto/statsinfo.proto index bc7bf03b880a4..a60174493d1f4 100644 --- a/proto/statsinfo.proto +++ b/proto/statsinfo.proto @@ -67,6 +67,9 @@ message StatsInfo { string TableName = 12; reserved 13; // int64 TimeSecond = 13; + double SampleRatio = 14; + map MinMaxValidMap = 15; + map MinMaxCompleteMap = 16; } message StatsInfoKey {