summaryrefslogtreecommitdiffstats
path: root/sql/sql_statistics.cc
blob: ec36900600ff8faca262b26ce70b4499ac834864 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
1959
1960
1961
1962
1963
1964
1965
1966
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
1979
1980
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999
2000
2001
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
2032
2033
2034
2035
2036
2037
2038
2039
2040
2041
2042
2043
2044
2045
2046
2047
2048
2049
2050
2051
2052
2053
2054
2055
2056
2057
2058
2059
2060
2061
2062
2063
2064
2065
2066
2067
2068
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079
2080
2081
2082
2083
2084
2085
2086
2087
2088
2089
2090
2091
2092
2093
2094
2095
2096
2097
2098
2099
2100
2101
2102
2103
2104
2105
2106
2107
2108
2109
2110
2111
2112
2113
2114
2115
2116
2117
2118
2119
2120
2121
2122
2123
2124
2125
2126
2127
2128
2129
2130
2131
2132
2133
2134
2135
2136
2137
2138
2139
2140
2141
2142
2143
2144
2145
2146
2147
2148
2149
2150
2151
2152
2153
2154
2155
2156
2157
2158
2159
2160
2161
2162
2163
2164
2165
2166
2167
2168
2169
2170
2171
2172
2173
2174
2175
2176
2177
2178
2179
2180
2181
2182
2183
2184
2185
2186
2187
2188
2189
2190
2191
2192
2193
2194
2195
2196
2197
2198
2199
2200
2201
2202
2203
2204
2205
2206
2207
2208
2209
2210
2211
2212
2213
2214
2215
2216
2217
2218
2219
2220
2221
2222
2223
2224
2225
2226
2227
2228
2229
2230
2231
2232
2233
2234
2235
2236
2237
2238
2239
2240
2241
2242
2243
2244
2245
2246
2247
2248
2249
2250
2251
2252
2253
2254
2255
2256
2257
2258
2259
2260
2261
2262
2263
2264
2265
2266
2267
2268
2269
2270
2271
2272
2273
2274
2275
2276
2277
2278
2279
2280
2281
2282
2283
2284
2285
2286
2287
2288
2289
2290
2291
2292
2293
2294
2295
2296
2297
2298
2299
2300
2301
2302
2303
2304
2305
2306
2307
2308
2309
2310
2311
2312
2313
2314
2315
2316
2317
2318
2319
2320
2321
2322
2323
2324
2325
2326
2327
2328
2329
2330
2331
2332
2333
2334
2335
2336
2337
2338
2339
2340
2341
2342
2343
2344
2345
2346
2347
2348
2349
2350
2351
2352
2353
2354
2355
2356
2357
2358
2359
2360
2361
2362
2363
2364
2365
2366
2367
2368
2369
2370
2371
2372
2373
2374
2375
2376
2377
2378
2379
2380
2381
2382
2383
2384
2385
2386
2387
2388
2389
2390
2391
2392
2393
2394
2395
2396
2397
2398
2399
2400
2401
2402
2403
2404
2405
2406
2407
2408
2409
2410
2411
2412
2413
2414
2415
2416
2417
2418
2419
2420
2421
2422
2423
2424
2425
2426
2427
2428
2429
2430
2431
2432
2433
2434
2435
2436
2437
2438
2439
2440
2441
2442
2443
2444
2445
2446
2447
2448
2449
2450
2451
2452
2453
2454
2455
2456
2457
2458
2459
2460
2461
2462
2463
2464
2465
2466
2467
2468
2469
2470
2471
2472
2473
2474
2475
2476
2477
2478
2479
2480
2481
2482
2483
2484
2485
2486
2487
2488
2489
2490
2491
2492
2493
2494
2495
2496
2497
2498
2499
2500
2501
2502
2503
2504
2505
2506
2507
2508
2509
2510
2511
2512
2513
2514
2515
2516
2517
2518
2519
2520
2521
2522
2523
2524
2525
2526
2527
2528
2529
2530
2531
2532
2533
2534
2535
2536
2537
2538
2539
2540
2541
2542
2543
2544
2545
2546
2547
2548
2549
2550
2551
2552
2553
2554
2555
2556
2557
2558
2559
2560
2561
2562
2563
2564
2565
2566
2567
2568
2569
2570
2571
2572
2573
2574
2575
2576
2577
2578
2579
2580
2581
2582
2583
2584
2585
2586
2587
2588
2589
2590
2591
2592
2593
2594
2595
2596
2597
2598
2599
2600
2601
2602
2603
2604
2605
2606
2607
2608
2609
2610
2611
2612
2613
2614
2615
2616
2617
2618
2619
2620
2621
2622
2623
2624
2625
2626
2627
2628
2629
2630
2631
2632
2633
2634
2635
2636
2637
2638
2639
2640
2641
2642
2643
2644
2645
2646
2647
2648
2649
2650
2651
2652
2653
2654
2655
2656
2657
2658
2659
2660
2661
2662
2663
2664
2665
2666
2667
2668
2669
2670
2671
2672
2673
2674
2675
2676
2677
2678
2679
2680
2681
2682
2683
2684
2685
2686
2687
2688
2689
2690
2691
2692
2693
2694
2695
2696
2697
2698
2699
2700
2701
2702
2703
2704
2705
2706
2707
2708
2709
2710
2711
2712
2713
2714
2715
2716
2717
2718
2719
2720
2721
2722
2723
2724
2725
2726
2727
2728
2729
2730
2731
2732
2733
2734
2735
2736
2737
2738
2739
2740
2741
2742
2743
2744
2745
2746
2747
2748
2749
2750
2751
2752
2753
2754
2755
2756
2757
2758
2759
2760
2761
2762
2763
2764
2765
2766
2767
2768
2769
2770
2771
2772
2773
2774
2775
2776
2777
2778
2779
2780
2781
2782
2783
2784
2785
2786
2787
2788
2789
2790
2791
2792
2793
2794
2795
2796
2797
2798
2799
2800
2801
2802
2803
2804
2805
2806
2807
2808
2809
2810
2811
2812
2813
2814
2815
2816
2817
2818
2819
2820
2821
2822
2823
2824
2825
2826
2827
2828
2829
2830
2831
2832
2833
2834
2835
2836
2837
2838
2839
2840
2841
2842
2843
2844
2845
2846
2847
2848
2849
2850
2851
2852
2853
2854
2855
2856
2857
2858
2859
2860
2861
2862
2863
2864
2865
2866
2867
2868
2869
2870
2871
2872
2873
2874
2875
2876
2877
2878
2879
2880
2881
2882
2883
2884
2885
2886
2887
2888
2889
2890
2891
2892
2893
2894
2895
2896
2897
2898
2899
2900
2901
2902
2903
2904
2905
2906
2907
2908
2909
2910
2911
2912
2913
2914
2915
2916
2917
2918
2919
2920
2921
2922
2923
2924
2925
2926
2927
2928
2929
2930
2931
2932
2933
2934
2935
2936
2937
2938
2939
2940
2941
2942
2943
2944
2945
2946
2947
2948
2949
2950
2951
2952
2953
2954
2955
2956
2957
2958
2959
2960
2961
2962
2963
2964
2965
2966
2967
2968
2969
2970
2971
2972
2973
2974
2975
2976
2977
2978
2979
2980
2981
2982
2983
2984
2985
2986
2987
2988
2989
2990
2991
2992
2993
2994
2995
2996
2997
2998
2999
3000
3001
3002
3003
3004
3005
3006
3007
3008
3009
3010
3011
3012
3013
3014
3015
3016
3017
3018
3019
3020
3021
3022
3023
3024
3025
3026
3027
3028
3029
3030
3031
3032
3033
3034
3035
3036
3037
3038
3039
3040
3041
3042
3043
3044
3045
3046
3047
3048
3049
3050
3051
3052
3053
3054
3055
3056
3057
3058
3059
3060
3061
3062
3063
3064
3065
3066
3067
3068
3069
3070
3071
3072
3073
3074
3075
3076
3077
3078
3079
3080
3081
3082
3083
3084
3085
3086
3087
3088
3089
3090
3091
3092
3093
3094
3095
3096
3097
3098
3099
3100
3101
3102
3103
3104
3105
3106
3107
3108
3109
3110
3111
3112
3113
3114
3115
3116
3117
3118
3119
3120
3121
3122
3123
3124
3125
3126
3127
3128
3129
3130
3131
3132
3133
3134
3135
3136
3137
3138
3139
3140
3141
3142
3143
3144
3145
3146
3147
3148
3149
3150
3151
3152
3153
3154
3155
3156
3157
3158
3159
3160
3161
3162
3163
3164
3165
3166
3167
3168
3169
3170
3171
3172
3173
3174
3175
3176
3177
3178
3179
3180
3181
3182
3183
3184
3185
3186
3187
3188
3189
3190
3191
3192
3193
3194
3195
3196
3197
3198
3199
3200
3201
3202
3203
3204
3205
3206
3207
3208
3209
3210
3211
3212
3213
3214
3215
3216
3217
3218
3219
3220
3221
3222
3223
3224
3225
3226
3227
3228
3229
3230
3231
3232
3233
3234
3235
3236
3237
3238
3239
3240
3241
3242
3243
3244
3245
3246
3247
3248
3249
3250
3251
3252
3253
3254
3255
3256
3257
3258
3259
3260
3261
3262
3263
3264
3265
3266
3267
3268
3269
3270
3271
3272
3273
3274
3275
3276
3277
3278
3279
3280
3281
3282
3283
3284
3285
3286
3287
3288
3289
3290
3291
3292
3293
3294
3295
3296
3297
3298
3299
3300
3301
3302
3303
3304
3305
3306
3307
3308
3309
3310
3311
3312
3313
3314
3315
3316
3317
3318
3319
3320
3321
3322
3323
3324
3325
3326
3327
3328
3329
3330
3331
3332
3333
3334
3335
3336
3337
3338
3339
3340
3341
3342
3343
3344
3345
3346
3347
3348
3349
3350
3351
3352
3353
3354
3355
3356
3357
3358
3359
3360
3361
3362
3363
3364
3365
3366
3367
3368
3369
3370
3371
3372
3373
3374
3375
3376
3377
3378
3379
3380
3381
3382
3383
3384
3385
3386
3387
3388
3389
3390
3391
3392
3393
3394
3395
3396
3397
3398
3399
3400
3401
3402
3403
3404
3405
3406
3407
3408
3409
3410
3411
3412
3413
3414
3415
3416
3417
3418
3419
3420
3421
3422
3423
3424
3425
3426
3427
3428
3429
3430
3431
3432
3433
3434
3435
3436
3437
3438
3439
3440
3441
3442
3443
3444
3445
3446
3447
3448
3449
3450
3451
3452
3453
3454
3455
3456
3457
3458
3459
3460
3461
3462
3463
3464
3465
3466
3467
3468
3469
3470
3471
3472
3473
3474
3475
3476
3477
3478
3479
3480
3481
3482
3483
3484
3485
3486
3487
3488
3489
3490
3491
3492
3493
3494
3495
3496
3497
3498
3499
3500
3501
3502
3503
3504
3505
3506
3507
3508
3509
3510
3511
3512
3513
3514
3515
3516
3517
3518
3519
3520
3521
3522
3523
3524
3525
3526
3527
3528
3529
3530
3531
3532
3533
3534
3535
3536
3537
3538
3539
3540
3541
3542
3543
3544
3545
3546
3547
3548
3549
3550
3551
3552
3553
3554
3555
3556
3557
3558
3559
3560
3561
3562
3563
3564
3565
3566
3567
3568
3569
3570
3571
3572
3573
3574
3575
3576
3577
3578
3579
3580
3581
3582
3583
3584
3585
3586
3587
3588
3589
3590
3591
3592
3593
3594
3595
3596
3597
3598
3599
3600
3601
3602
3603
3604
3605
3606
3607
3608
3609
3610
3611
3612
3613
3614
3615
3616
3617
3618
3619
3620
3621
3622
3623
3624
3625
3626
3627
3628
3629
3630
3631
3632
3633
3634
3635
3636
3637
3638
3639
3640
3641
3642
3643
3644
3645
3646
3647
3648
3649
3650
3651
3652
3653
3654
3655
3656
3657
3658
3659
3660
3661
3662
3663
3664
3665
3666
3667
3668
3669
3670
3671
3672
3673
3674
3675
3676
3677
3678
3679
3680
3681
3682
3683
3684
3685
3686
3687
3688
3689
3690
3691
3692
3693
3694
3695
3696
3697
3698
3699
3700
3701
3702
3703
3704
3705
3706
3707
3708
3709
3710
3711
3712
3713
3714
3715
3716
3717
3718
3719
3720
3721
3722
3723
3724
3725
3726
3727
3728
3729
3730
3731
3732
3733
3734
3735
3736
3737
3738
3739
3740
3741
3742
3743
3744
3745
3746
3747
3748
3749
3750
3751
3752
3753
3754
3755
3756
3757
3758
3759
3760
3761
3762
3763
3764
3765
3766
3767
3768
3769
3770
3771
3772
3773
3774
3775
3776
3777
3778
3779
3780
3781
3782
3783
3784
3785
3786
3787
3788
3789
3790
3791
3792
3793
3794
3795
3796
3797
3798
3799
3800
3801
3802
3803
3804
3805
3806
3807
3808
3809
3810
3811
3812
3813
3814
3815
3816
3817
3818
3819
3820
3821
3822
3823
3824
3825
3826
3827
3828
3829
3830
3831
3832
3833
3834
3835
3836
3837
3838
3839
3840
3841
3842
3843
3844
3845
3846
3847
3848
3849
3850
3851
3852
3853
3854
3855
3856
3857
3858
3859
3860
3861
3862
3863
3864
3865
3866
3867
3868
3869
3870
3871
3872
3873
3874
3875
3876
3877
3878
3879
3880
3881
3882
3883
3884
3885
3886
3887
3888
3889
3890
3891
3892
3893
3894
3895
3896
3897
3898
3899
3900
3901
3902
3903
3904
3905
3906
3907
3908
3909
3910
3911
3912
3913
3914
3915
3916
3917
3918
3919
3920
3921
3922
3923
3924
3925
3926
3927
3928
3929
3930
3931
3932
3933
3934
3935
3936
3937
3938
3939
3940
3941
3942
3943
3944
3945
3946
3947
3948
3949
3950
3951
3952
3953
3954
3955
3956
3957
3958
3959
3960
3961
3962
3963
3964
3965
3966
3967
3968
3969
3970
3971
3972
3973
3974
3975
3976
3977
3978
3979
3980
3981
3982
3983
3984
3985
3986
3987
3988
3989
3990
3991
3992
3993
3994
3995
3996
3997
3998
3999
4000
4001
4002
4003
4004
4005
4006
4007
4008
4009
4010
4011
4012
4013
4014
4015
4016
4017
4018
4019
4020
4021
4022
4023
4024
4025
4026
4027
4028
4029
4030
4031
4032
4033
4034
4035
4036
4037
4038
4039
4040
4041
4042
4043
4044
4045
4046
4047
4048
4049
4050
4051
4052
4053
4054
4055
4056
4057
4058
4059
4060
4061
4062
4063
4064
4065
4066
4067
4068
4069
4070
4071
4072
4073
4074
4075
4076
4077
4078
4079
4080
4081
4082
4083
4084
4085
4086
4087
4088
4089
4090
4091
4092
4093
4094
4095
4096
4097
4098
4099
4100
4101
4102
4103
4104
4105
4106
4107
4108
4109
4110
4111
4112
4113
4114
4115
4116
4117
4118
4119
4120
4121
4122
4123
4124
4125
4126
4127
4128
4129
4130
4131
4132
4133
4134
4135
4136
4137
4138
4139
4140
4141
4142
4143
4144
4145
4146
4147
4148
4149
4150
4151
4152
4153
4154
4155
4156
4157
4158
4159
4160
4161
4162
4163
4164
4165
4166
4167
4168
4169
4170
4171
4172
4173
4174
4175
4176
4177
4178
4179
4180
4181
4182
4183
4184
4185
4186
4187
4188
4189
4190
4191
4192
4193
4194
4195
4196
4197
4198
4199
4200
4201
4202
4203
4204
4205
4206
4207
4208
4209
4210
4211
4212
4213
4214
4215
4216
4217
4218
4219
4220
4221
4222
4223
4224
4225
4226
4227
4228
4229
4230
4231
4232
4233
4234
4235
4236
4237
4238
4239
4240
4241
4242
4243
4244
4245
4246
4247
4248
4249
4250
4251
4252
4253
4254
4255
4256
4257
4258
4259
4260
4261
4262
4263
4264
4265
4266
4267
4268
4269
4270
4271
4272
4273
4274
4275
4276
4277
4278
4279
4280
4281
4282
4283
4284
4285
4286
4287
4288
4289
4290
4291
4292
4293
4294
4295
4296
4297
4298
4299
4300
4301
4302
4303
4304
4305
4306
4307
4308
4309
4310
4311
4312
4313
4314
4315
4316
4317
4318
4319
4320
4321
4322
4323
4324
4325
4326
4327
4328
4329
4330
4331
4332
4333
4334
4335
4336
4337
4338
4339
4340
4341
4342
4343
4344
4345
4346
4347
4348
4349
4350
4351
4352
4353
4354
4355
4356
4357
4358
4359
4360
4361
4362
4363
4364
4365
4366
4367
4368
4369
4370
4371
4372
4373
4374
4375
4376
4377
4378
4379
4380
4381
4382
4383
4384
4385
4386
4387
4388
4389
4390
4391
4392
4393
4394
4395
4396
4397
4398
4399
4400
4401
4402
4403
4404
4405
4406
4407
4408
4409
4410
4411
4412
4413
4414
4415
4416
4417
4418
4419
4420
4421
4422
4423
4424
4425
4426
4427
4428
4429
4430
4431
4432
4433
4434
4435
4436
4437
4438
4439
4440
4441
4442
4443
4444
4445
4446
4447
4448
4449
4450
4451
4452
4453
4454
4455
4456
4457
4458
4459
4460
4461
4462
4463
4464
4465
4466
4467
4468
4469
4470
4471
4472
4473
4474
4475
4476
4477
4478
4479
4480
4481
4482
4483
4484
4485
4486
4487
4488
4489
4490
4491
4492
4493
/* Copyright (C) 2009 MySQL AB
   Copyright (c) 2019, 2022, MariaDB Corporation.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA */

/**
  @file

  @brief
  functions to update persitent statistical tables and to read from them

  @defgroup Query_Optimizer  Query Optimizer
  @{
*/

#include "mariadb.h"
#include "sql_base.h"
#include "key.h"
#include "sql_statistics.h"
#include "opt_histogram_json.h"
#include "opt_range.h"
#include "uniques.h"
#include "sql_show.h"
#include "sql_partition.h"
#include "sql_alter.h"                          // RENAME_STAT_PARAMS

#include <vector>
#include <string>

/*
  The system variable 'use_stat_tables' can take one of the
  following values:
  "never", "complementary", "preferably". 
  If the values of the variable 'use_stat_tables' is set to
  "never then any statistical data from  the persistent statistical tables
  is ignored by the optimizer.
  If the value of the variable 'use_stat_tables' is set to
  "complementary" then a particular statistical characteristic is used
  by the optimizer only if the database engine does not provide similar
  statistics. For example, 'nulls_ratio' for table columns  currently 
  are not provided by any engine. So optimizer uses this statistical data
  from the  statistical tables. At the same time it does not use 
  'avg_frequency' for any index prefix from the statistical tables since
  the a similar statistical characteristic 'records_per_key' can be
  requested from the database engine.
  If the value the variable 'use_stat_tables' is set to
  "preferably" the optimizer uses a particular statistical data only if
  it can't be found in the statistical data.
  If an ANALYZE command is executed then it results in collecting
  statistical data for the tables specified by the command and storing
  the collected statistics in the persistent statistical tables only
  when the value of the variable 'use_stat_tables' is not
  equal to "never".
*/

Histogram_base *create_histogram(MEM_ROOT *mem_root, Histogram_type hist_type);

/* Currently there are only 3 persistent statistical tables */
static const uint STATISTICS_TABLES= 3;

/*
  The names of the statistical tables in this array must correspond the
  definitions of the tables in the file ../scripts/mysql_system_tables.sql
*/
static const LEX_CSTRING stat_table_name[STATISTICS_TABLES]=
{
  { STRING_WITH_LEN("table_stats") },
  { STRING_WITH_LEN("column_stats") },
  { STRING_WITH_LEN("index_stats") }
};


TABLE_STATISTICS_CB::TABLE_STATISTICS_CB():
  usage_count(0), table_stats(0),
  stats_available(TABLE_STAT_NO_STATS), histograms_exists_on_disk(0)
{
  init_sql_alloc(PSI_INSTRUMENT_ME, &mem_root, TABLE_ALLOC_BLOCK_SIZE, 0,
                 MYF(0));
}

TABLE_STATISTICS_CB::~TABLE_STATISTICS_CB()
{
  Column_statistics *column_stats= table_stats->column_stats;
  Column_statistics *column_stats_end= column_stats + table_stats->columns;
  DBUG_ASSERT(usage_count == 0);

  /* Free json histograms */
  for (; column_stats < column_stats_end ; column_stats++)
  {
    delete column_stats->histogram;
    /*
      Protect against possible other free in free_statistics_for_table()
    */
    column_stats->histogram= 0;
  }
  free_root(&mem_root, MYF(0));
}


/**
  @details
  The function builds a list of TABLE_LIST elements for system statistical
  tables using array of TABLE_LIST passed as a parameter. 
  The lock type of each element is set to TL_READ if for_write = FALSE,
  otherwise it is set to TL_WRITE.
*/

static
inline void init_table_list_for_stat_tables(TABLE_LIST *tables, bool for_write)
{
  uint i;

  memset((char *) &tables[0], 0, sizeof(TABLE_LIST) * STATISTICS_TABLES);

  for (i= 0; i < STATISTICS_TABLES; i++)
  {
    tables[i].db= MYSQL_SCHEMA_NAME;
    tables[i].table_name= stat_table_name[i];
    tables[i].alias=      stat_table_name[i];
    tables[i].lock_type= for_write ? TL_WRITE : TL_READ;
    if (i < STATISTICS_TABLES - 1)
    tables[i].next_global= tables[i].next_local=
      tables[i].next_name_resolution_table= &tables[i+1];
    if (i != 0)
      tables[i].prev_global= &tables[i-1].next_global;
  }
}

static Table_check_intact_log_error stat_table_intact;

static const
TABLE_FIELD_TYPE table_stat_fields[TABLE_STAT_N_FIELDS] =
{
  {
    { STRING_WITH_LEN("db_name") },
    { STRING_WITH_LEN("varchar(64)") },
    { STRING_WITH_LEN("utf8mb3") }
  },
  {
    { STRING_WITH_LEN("table_name") },
    { STRING_WITH_LEN("varchar(64)") },
    { STRING_WITH_LEN("utf8mb3") }
  },
  {
    { STRING_WITH_LEN("cardinality") },
    { STRING_WITH_LEN("bigint(21)") },
    { NULL, 0 }
  },
};
static const uint table_stat_pk_col[]= {0,1};
static const TABLE_FIELD_DEF
table_stat_def= {TABLE_STAT_N_FIELDS, table_stat_fields, 2, table_stat_pk_col };

static const
TABLE_FIELD_TYPE column_stat_fields[COLUMN_STAT_N_FIELDS] =
{
  {
    { STRING_WITH_LEN("db_name") },
    { STRING_WITH_LEN("varchar(64)") },
    { STRING_WITH_LEN("utf8mb3") }
  },
  {
    { STRING_WITH_LEN("table_name") },
    { STRING_WITH_LEN("varchar(64)") },
    { STRING_WITH_LEN("utf8mb3") }
  },
  {
    { STRING_WITH_LEN("column_name") },
    { STRING_WITH_LEN("varchar(64)") },
    { STRING_WITH_LEN("utf8mb3") }
  },
  {
    { STRING_WITH_LEN("min_value") },
    { STRING_WITH_LEN("varbinary(255)") },
    { NULL, 0 }
  },
  {
    { STRING_WITH_LEN("max_value") },
    { STRING_WITH_LEN("varbinary(255)") },
    { NULL, 0 }
  },
  {
    { STRING_WITH_LEN("nulls_ratio") },
    { STRING_WITH_LEN("decimal(12,4)") },
    { NULL, 0 }
  },
  {
    { STRING_WITH_LEN("avg_length") },
    { STRING_WITH_LEN("decimal(12,4)") },
    { NULL, 0 }
  },
  {
    { STRING_WITH_LEN("avg_frequency") },
    { STRING_WITH_LEN("decimal(12,4)") },
    { NULL, 0 }
  },
  {
    { STRING_WITH_LEN("hist_size") },
    { STRING_WITH_LEN("tinyint(3)") },
    { NULL, 0 }
  },
  {
    { STRING_WITH_LEN("hist_type") },
    { STRING_WITH_LEN("enum('SINGLE_PREC_HB','DOUBLE_PREC_HB','JSON_HB')") },
    { STRING_WITH_LEN("utf8mb3") }
  },
  {
    { STRING_WITH_LEN("histogram") },
    { STRING_WITH_LEN("longblob") },
    { NULL, 0 }
  }
};
static const uint column_stat_pk_col[]= {0,1,2};
static const TABLE_FIELD_DEF
column_stat_def= {COLUMN_STAT_N_FIELDS, column_stat_fields, 3, column_stat_pk_col};

static const
TABLE_FIELD_TYPE index_stat_fields[INDEX_STAT_N_FIELDS] =
{
  {
    { STRING_WITH_LEN("db_name") },
    { STRING_WITH_LEN("varchar(64)") },
    { STRING_WITH_LEN("utf8mb3") }
  },
  {
    { STRING_WITH_LEN("table_name") },
    { STRING_WITH_LEN("varchar(64)") },
    { STRING_WITH_LEN("utf8mb3") }
  },
  {
    { STRING_WITH_LEN("index") },
    { STRING_WITH_LEN("varchar(64)") },
    { STRING_WITH_LEN("utf8mb3") }
  },
  {
    { STRING_WITH_LEN("prefix_arity") },
    { STRING_WITH_LEN("int(11)") },
    { NULL, 0 }
  },
  {
    { STRING_WITH_LEN("avg_frequency") },
    { STRING_WITH_LEN("decimal(12,4)") },
    { NULL, 0 }
  }
};
static const uint index_stat_pk_col[]= {0,1,2,3};
static const TABLE_FIELD_DEF
index_stat_def= {INDEX_STAT_N_FIELDS, index_stat_fields, 4, index_stat_pk_col};


/**
  @brief
  Open all statistical tables and lock them
*/

static int open_stat_tables(THD *thd, TABLE_LIST *tables, bool for_write)
{
  int rc;
  Dummy_error_handler deh; // suppress errors
  DBUG_ASSERT(thd->internal_transaction());

  thd->push_internal_handler(&deh);
  init_table_list_for_stat_tables(tables, for_write);
  init_mdl_requests(tables);
  thd->in_sub_stmt|= SUB_STMT_STAT_TABLES;
  rc= open_system_tables_for_read(thd, tables);
  thd->in_sub_stmt&= ~SUB_STMT_STAT_TABLES;
  thd->pop_internal_handler();


  /* If the number of tables changes, we should revise the check below. */
  compile_time_assert(STATISTICS_TABLES == 3);

  if (!rc &&
      (stat_table_intact.check(tables[TABLE_STAT].table, &table_stat_def) ||
       stat_table_intact.check(tables[COLUMN_STAT].table, &column_stat_def) ||
       stat_table_intact.check(tables[INDEX_STAT].table, &index_stat_def)))
  {
    close_thread_tables(thd);
    rc= 1;
  }

  return rc;
} 


/**
  @brief
  Open a statistical table and lock it

  @details
  This is used by DDLs. When a column or index is dropped or renamed,
  stat tables need to be adjusted accordingly.

  This function should not generate any errors as the callers are not checking
  the result of delete_statistics_for_table()

*/
static inline int open_stat_table_for_ddl(THD *thd, TABLE_LIST *table,
                                          const LEX_CSTRING *stat_tab_name)
{
  table->init_one_table(&MYSQL_SCHEMA_NAME, stat_tab_name, NULL, TL_WRITE);
  Dummy_error_handler error_handler;
  thd->push_internal_handler(&error_handler);
  int res= open_system_tables_for_read(thd, table);
  thd->pop_internal_handler();
  if (res && error_handler.any_error())
  {
    push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
                        ER_CHECK_NO_SUCH_TABLE,
                        "Got error %d when trying to open statistics "
                        "table %`s for updating statistics",
                        error_handler.got_error(), stat_table_name->str);
  }
  return res;
}


/*
  The class Column_statistics_collected is a helper class used to collect
  statistics on a table column. The class is derived directly from
  the class Column_statistics, and, additionally to the fields of the
  latter, it contains the fields to accumulate the results of aggregation
  for the number of nulls in the column and for the size of the column
  values. There is also a container for distinct column values used
  to calculate the average number of records per distinct column value. 
*/ 

class Column_statistics_collected :public Column_statistics
{

private:
  Field *column;  /* The column to collect statistics on */
  ha_rows nulls;  /* To accumulate the number of nulls in the column */ 
  ulonglong column_total_length; /* To accumulate the size of column values */
  Count_distinct_field *count_distinct; /* The container for distinct 
                                           column values */

  bool is_single_pk_col; /* TRUE <-> the only column of the primary key */ 

public:

  inline void init(THD *thd, Field * table_field);
  inline bool add();
  inline bool finish(MEM_ROOT *mem_root, ha_rows rows, double sample_fraction);
  inline void cleanup();
};


/**
  Stat_table is the base class for classes Table_stat, Column_stat and
  Index_stat. The methods of these classes allow us to read statistical
  data from statistical tables, write collected statistical data into
  statistical tables and update statistical data in these  tables
  as well as update access fields belonging to the primary key and
  delete records by prefixes of the primary key.
  Objects of the classes Table_stat, Column_stat  and Index stat are used 
  for reading/writing statistics from/into persistent tables table_stats,
  column_stats and index_stats correspondingly.  These tables are stored in
  the system database 'mysql'.

  Statistics is read and written always for a given database table t. When
  an  object of any of these classes is created a pointer to the TABLE
  structure for this database table is passed as a parameter to the constructor
  of the object. The other parameter is a pointer to the TABLE structure for
  the corresponding statistical table st. So construction of an object to 
  read/write statistical data on table t from/into statistical table st 
  requires both table t and st to be opened.
  In some cases the TABLE structure for table t may be undefined. Then
  the objects of the classes Table_stat, Column_stat  and Index stat are
  created by the alternative constructor that require only the name
  of the table t and the name of the database it belongs to. Currently the
  alternative constructors are used only in the cases when some records
  belonging to the table are to be deleted, or its keys are to be updated   

  Reading/writing statistical data from/into a statistical table is always
  performed by a key.  At the moment there is only one key defined for each
  statistical table and this key is primary.
  The primary key for the table table_stats is built as (db_name, table_name).
  The primary key for the table column_stats is built as (db_name, table_name,
  column_name).
  The primary key for the table index_stats is built as (db_name, table_name,
  index_name, prefix_arity).

  Reading statistical data from a statistical table is performed by the 
  following pattern. First a table dependent method sets the values of the
  the fields that comprise the lookup key. Then, get_stat_values(...) call
  finds the row from the statistical table by the set key. If the row is
  found the values of statistical fields are read from this row and are
  distributed in the internal structures.

  Let's assume the statistical data is read for table t from database db.

  When statistical data is searched in the table table_stats first 
  Table_stat::set_key_fields() should set the fields of db_name and
  table_name. Then get_stat_values looks for a row by the set key value,
  and, if the row is found, reads the value from the column
  table_stats.cardinality into the field read_stat.cardinality of the TABLE
  structure for table t and sets the value of read_stat.cardinality_is_null
  from this structure to FALSE. If the value of the 'cardinality' column
  in the row is null or if no row is found read_stat.cardinality_is_null
  is set to TRUE.

  When statistical data is searched in the table column_stats first
  Column_stat::set_key_fields() should set the fields of db_name, table_name
  and column_name with column_name taken out of the only parameter f of the
  Field* type passed to this method. After this get_stat_values looks
  for a row by the set key value. If the row is found the values of statistical 
  data columns min_value, max_value, nulls_ratio, avg_length, avg_frequency,
  hist_size, hist_type, histogram are read into internal structures. Values
  of nulls_ratio, avg_length, avg_frequency, hist_size, hist_type, histogram
  are read into the corresponding fields of the read_stat  structure from
  the Field object f, while values from min_value and max_value  are copied
  into the min_value and  max_value record buffers attached to the TABLE
   structure for table t.
  If the value of a statistical column in the found row is null, then the
  corresponding flag in the f->read_stat.column_stat_nulls bitmap is set off.
  Otherwise the flag is set on. If no row is found for the column the all flags
  in f->column_stat_nulls are set off.
  
  When statistical data is searched in the table index_stats first
  Index_stat::set_key_fields() has to be called to set the fields of db_name,
  table_name, index_name and prefix_arity. The value of index_name is extracted
  from the first parameter key_info of the KEY* type passed to the method.
  This parameter  specifies the index of interest idx. The second parameter
  passed to the method specifies the arity k of the index prefix for which
  statistical data is to be read. E.g. if the index idx consists of 3
  components (p1,p2,p3) the table  index_stats usually will contain 3 rows for
  this index: the first - for the prefix (p1), the second - for the prefix
  (p1,p2), and the third - for the the prefix (p1,p2,p3). After the key fields
  has been set a call of get_stat_value looks for a row by the set key value.
  If the row is found and the value of the avg_frequency column is not null 
  then this value is assigned to key_info->read_stat.avg_frequency[k].
  Otherwise 0 is assigned to this element. 

  The method Stat_table::update_stat is used to write statistical data
  collected in the internal structures into a statistical table st.
  It is assumed that before any invocation of this method a call of the
  function st.set_key_fields has set the values of the primary key fields
  that serve to locate the row from the statistical table st where the 
  the collected statistical data from internal structures are to be written
  to. The statistical data is written from the counterparts of the
  statistical fields of internal structures into which it would be read
  by the functions get_stat_values. The counterpart fields are used
  only when statistics is collected
  When updating/inserting a row from the statistical table st the method
  Stat_table::update_stat calls the implementation of the pure virtual
  method store_field_values to transfer statistical data from the fields
  of internal structures to the fields of record buffer used for updates
  of the statistical table st.     
*/  
         
class Stat_table 
{

private:

  /* Handler used for the retrieval of the statistical table stat_table */
  handler *stat_file;
  
  uint stat_key_length; /* Length of the key to access stat_table */
  uchar *record[2];     /* Record buffers used to access/update stat_table */


  /* This is a helper function used only by the Stat_table constructors */
  void common_init_stat_table()
  {
    stat_file= stat_table->file;
    /* Currently any statistical table has only one key */
    stat_key_idx= 0;
    stat_key_info= &stat_table->key_info[stat_key_idx];
    stat_key_length= stat_key_info->key_length;
    last_key_length= last_prefix_parts= 0;
    record[0]= stat_table->record[0];
    record[1]= stat_table->record[1];
  }

protected:

  /* Statistical table to read statistics from or to update/delete */
  TABLE *stat_table;
  KEY *stat_key_info;   /* Structure for the index to access stat_table */
  
  /* Table for which statistical data is read / updated */
  const TABLE *table;
  const TABLE_SHARE *table_share;  /* Table share for 'table */
  const LEX_CSTRING *db_name;      /* Name of the database containing 'table' */
  const LEX_CSTRING *table_name;   /* Name of the table 'table' */

  uchar last_key[MAX_KEY_LENGTH];
  uint last_key_length;
  uint last_prefix_parts;

  void store_record_for_lookup()
  {
    DBUG_ASSERT(record[0] == stat_table->record[0]);
  }

  int update_record()
  {
    int err;
    if ((err= stat_file->ha_update_row(record[1], record[0])) &&
         err != HA_ERR_RECORD_IS_THE_SAME)
      return err;
    return 0;
  }

public:

  uint stat_key_idx;    /* The number of the key to access stat_table */
  void store_record_for_update()
  {
    store_record(stat_table, record[1]);
  }

  /**
    @details
    This constructor has to be called by any constructor of the derived
    classes. The constructor 'tunes' the private and protected members of
    the constructed object to the statistical table 'stat_table' with the
    statistical data of our interest and to the table 'tab' for which this
    statistics has been collected.
  */  

  Stat_table(TABLE *stat, const TABLE *tab)
    :stat_table(stat), table(tab)
  {
    table_share= tab->s;
    common_init_stat_table();
    db_name= &table_share->db;
    table_name= &table_share->table_name;
  }


  /**
    @details
    This constructor has to be called by any constructor of the derived
    classes. The constructor 'tunes' the private and protected members of
    the constructed object to the statistical table 'stat_table' with the
    statistical data of our interest and to the table t for which this
    statistics has been collected. The table t is uniquely specified
    by the database name 'db' and the table name 'tab'.
  */  
  
  Stat_table(TABLE *stat, const LEX_CSTRING *db, const LEX_CSTRING *tab)
    :stat_table(stat), table_share(NULL),db_name(db), table_name(tab)
  {
    common_init_stat_table();
  } 


  virtual ~Stat_table() = default;

  /**
    @brief
    Store the given values of fields for database name and table name 
   
    @details
    This is a purely virtual method.
    The implementation for any derived class shall store the given
    values of the database name and table name in the corresponding
    fields of stat_table.
    
    @note
    The method is called by the update_table_name_key_parts function.
  */      

 virtual void change_full_table_name(const LEX_CSTRING *db,
                                     const LEX_CSTRING *tab)= 0;

 
  /**
    @brief
    Store statistical data into fields of the statistical table
   
    @details
    This is a purely virtual method.
    The implementation for any derived class shall put the appropriate
    statistical data into the corresponding fields of stat_table.
    
    @note
    The method is called by the update_stat function.
  */      

  virtual void store_stat_fields()= 0;

  
  /**
    @brief
    Find a record in the statistical table by a primary key

    @details
    The function looks for a record in stat_table by its primary key.
    It assumes that the key fields have been already stored in the record
    buffer of stat_table.

    @retval
    FALSE    the record is not found
    @retval
    TRUE     the record is found
  */

  bool find_stat()
  {
    last_key_length= stat_key_length;
    key_copy(last_key, record[0], stat_key_info, stat_key_length);
    return !stat_file->ha_index_read_idx_map(record[0], stat_key_idx, last_key,
                                             HA_WHOLE_KEY, HA_READ_KEY_EXACT);
  }

  void create_key_for_read(uint prefix_parts)
  {
    last_key_length= 0;
    last_prefix_parts= prefix_parts;
    for (uint i= 0; i < prefix_parts; i++)
      last_key_length+= stat_key_info->key_part[i].store_length;
    key_copy(last_key, record[0], stat_key_info, last_key_length);
  }


  /**
    @brief
    Find a record in the statistical table by a key prefix value 

    @details
    The function looks for a record in stat_table by the key value consisting
    of 'prefix_parts' major components for the primary index.  
    It assumes that the key prefix fields have been already stored in the record
    buffer of stat_table.

    @retval
    FALSE    the record is not found
    @retval
    TRUE     the record is found
  */

  bool find_next_stat_for_prefix(uint prefix_parts)
  {
    create_key_for_read(prefix_parts);
    key_part_map prefix_map= (key_part_map) ((1 << prefix_parts) - 1);
    return !stat_file->ha_index_read_idx_map(record[0], stat_key_idx, last_key,
                                         prefix_map, HA_READ_KEY_EXACT);
  }
   
  bool find_next_stat_for_prefix_with_next(uint prefix_parts)
  {
    create_key_for_read(prefix_parts);
    key_part_map prefix_map= (key_part_map) ((1 << prefix_parts) - 1);
    return !stat_file->ha_index_read_map(record[0], last_key,
                                         prefix_map,
                                         HA_READ_KEY_EXACT);
  }

  /*
    Read row with same key parts as last find_next_stat_for_prefix_with_next()
  */

  bool find_stat_with_next()
  {
    key_copy(last_key, record[0], stat_key_info, last_key_length);
    key_part_map prefix_map= (key_part_map) ((1 << last_prefix_parts) - 1);
    return !stat_file->ha_index_read_map(record[0], last_key,
                                         prefix_map, HA_READ_KEY_EXACT);
  }

  /**
    @brief
    Update/insert a record in the statistical table with new statistics

    @details
    The function first looks for a record by its primary key in the statistical
    table stat_table. If the record is found the function updates statistical
    fields of the records. The data for these fields are taken from internal
    structures containing info on the table 'table'. If the record is not
    found the function inserts a new record with the primary key set to the
    search key and the statistical data taken from the internal structures.
    The function assumes that the key fields have been already stored in
    the record buffer of stat_table.

    @retval
    FALSE    success with the update/insert of the record
    @retval
    TRUE     failure with the update/insert of the record

    @note
    The function calls the virtual method store_stat_fields to populate the
    statistical fields of the updated/inserted row with new statistics.
  */

  bool update_stat()
  {
    if (find_stat())
    {    
      bool res;
      store_record_for_update();
      store_stat_fields();
      res= update_record() != 0;
      DBUG_ASSERT(res == 0);
      return res;
    }
    else
    {
      int err;
      store_stat_fields();
      if ((err= stat_file->ha_write_row(record[0])))
      {
        DBUG_ASSERT(0);
	return TRUE;
      }
    }
    return FALSE;
  }

  /**
    @brief
    Update the table name fields in the current record of stat_table

    @details
    The function updates the fields containing database name and table name
    for the last found record in the statistical table stat_table.
    The corresponding names for update is taken from the parameters
    db and tab.
 
    @retval
    FALSE    success with the update of the record
    @retval
    TRUE     failure with the update of the record

    @note
    The function calls the virtual method change_full_table_name  
    to store the new names in the record buffer used for updates.
  */

  bool update_table_name_key_parts(const LEX_CSTRING *db, const LEX_CSTRING *tab)
  {
    store_record_for_update();
    change_full_table_name(db, tab);
    bool rc= update_record() != 0;
    store_record_for_lookup();
    return rc;
  }   


  /** 
    @brief
    Delete the current record of the statistical table stat_table

    @details
    The function deletes the last found record from the statistical
    table stat_table.
 
    @retval
    FALSE    success with the deletion of the record
    @retval
    TRUE     failure with the deletion of the record
  */

  bool delete_stat()
  {
    int err;
    if ((err= stat_file->ha_delete_row(record[0])))
      return TRUE;
    return FALSE;
  }

  void flush()
  {
    stat_file->extra(HA_EXTRA_FLUSH);
  }

  friend class Stat_table_write_iter;
};


/*
  An object of the class Table_stat is created to read statistical
  data on tables from the statistical table table_stats, to update
  table_stats with such statistical data, or to update columns
  of the primary key, or to delete the record by its primary key or
  its prefix. 
  Rows from the statistical table are read and updated always by
  primary key. 
*/

class Table_stat: public Stat_table
{

private:

  Field *db_name_field;     /* Field for the column table_stats.db_name */
  Field *table_name_field;  /* Field for the column table_stats.table_name */

  void common_init_table_stat()
  {  
    db_name_field= stat_table->field[TABLE_STAT_DB_NAME];
    table_name_field= stat_table->field[TABLE_STAT_TABLE_NAME];
  }

  void change_full_table_name(const LEX_CSTRING *db,
                              const LEX_CSTRING *tab) override
  {
    db_name_field->store(db, system_charset_info);
    table_name_field->store(tab, system_charset_info);
  }

public:

  /**
    @details
    The constructor 'tunes' the private and protected members of the
    constructed object for the statistical table table_stats to read/update
    statistics on table 'tab'. The TABLE structure for the table table_stat
    must be passed as a value for the parameter 'stat'.
  */

  Table_stat(TABLE *stat, const TABLE *tab) :Stat_table(stat, tab)
  {
    common_init_table_stat();
  }


  /**
    @details
    The constructor 'tunes' the private and protected members of the
    object constructed for the statistical table table_stat for 
    the future updates/deletes of the record concerning the table 'tab'
    from the database 'db'.
  */

  Table_stat(TABLE *stat, const LEX_CSTRING *db, const LEX_CSTRING *tab)
    :Stat_table(stat, db, tab)
  {
    common_init_table_stat();
  }


  /** 
    @brief
    Set the key fields for the statistical table table_stat

    @details
    The function sets the values of the fields db_name and table_name
    in the record buffer for the statistical table table_stat.
    These fields comprise the primary key for the table.

    @note
    The function is supposed to be called before any use of the  
    method find_stat for an object of the Table_stat class. 
  */

  void set_key_fields()
  {
    db_name_field->store(db_name, system_charset_info);
    table_name_field->store(table_name, system_charset_info);
  }


  /** 
    @brief
    Store statistical data into statistical fields of table_stat

    @details
    This implementation of a purely virtual method sets the value of the
    column 'cardinality' of the statistical table table_stat according to
    the value of the flag write_stat.cardinality_is_null and the value of
    the field write_stat.cardinality' from the TABLE structure for 'table'.
  */    

  void store_stat_fields() override
  {
    Field *stat_field= stat_table->field[TABLE_STAT_CARDINALITY];
    if (table->collected_stats->cardinality_is_null)
      stat_field->set_null();
    else
    {
      stat_field->set_notnull();
      stat_field->store(table->collected_stats->cardinality,true);
    }
  }


  /**
    @brief
    Read statistical data from statistical fields of table_stat

    @details
    Find a record in mysql.table_stat that has statistics for this table.
    We search for record using a PK lookup. The lookup values are in the stat
    table's record buffer, they were put there by Table_stat::set_key_fields.

    The result is stored in *read_stats.
  */

  bool get_stat_values(Table_statistics *read_stats)
  {
    bool res;
    read_stats->cardinality_is_null= TRUE;
    read_stats->cardinality= 0;
    if ((res= find_stat()))
    {
      Field *stat_field= stat_table->field[TABLE_STAT_CARDINALITY];
      if (!stat_field->is_null())
      {
        read_stats->cardinality_is_null= FALSE;
        read_stats->cardinality= stat_field->val_int();
      }
    }
    return res;
  } 
};


/*
  An object of the class Column_stat is created to read statistical data
  on table columns from the statistical table column_stats, to update
  column_stats with such statistical data, or to update columns
  of the primary key, or to delete the record by its primary key or
  its prefix.
  Rows from the statistical table are read and updated always by 
  primary key.
*/ 

class Column_stat: public Stat_table
{

private:

  Field *db_name_field;     /* Field for the column column_stats.db_name */
  Field *table_name_field;  /* Field for the column column_stats.table_name */
  Field *column_name_field; /* Field for the column column_stats.column_name */

  Field *table_field;  /* Field from 'table' to read /update statistics on */

  void common_init_column_stat_table()
  {
    db_name_field= stat_table->field[COLUMN_STAT_DB_NAME];
    table_name_field= stat_table->field[COLUMN_STAT_TABLE_NAME];
    column_name_field= stat_table->field[COLUMN_STAT_COLUMN_NAME];
  } 

  void change_full_table_name(const LEX_CSTRING *db,
                              const LEX_CSTRING *tab) override
  {
    db_name_field->store(db, system_charset_info);
     table_name_field->store(tab, system_charset_info);
  }

public:

  /**
    @details
    The constructor 'tunes' the private and protected members of the
    constructed object for the statistical table column_stats to read/update
    statistics on fields of the table 'tab'. The TABLE structure for the table
    column_stats must be passed as a value for the parameter 'stat'.
  */

  Column_stat(TABLE *stat, const TABLE *tab) :Stat_table(stat, tab)
  {
    common_init_column_stat_table();
  } 


  /**
    @details
    The constructor 'tunes' the private and protected members of the
    object constructed for the statistical table column_stats for 
    the future updates/deletes of the record concerning the table 'tab'
    from the database 'db'. 
  */

  Column_stat(TABLE *stat, const LEX_CSTRING *db, const LEX_CSTRING *tab)
    :Stat_table(stat, db, tab)
  {
    common_init_column_stat_table();
  } 

  /** 
    @brief
    Set table name fields for the statistical table column_stats

    @details
    The function stores the values of the fields db_name and table_name 
    of the statistical table column_stats in the record buffer.
  */

  void set_full_table_name()
  {
    db_name_field->store(db_name, system_charset_info);
    table_name_field->store(table_name, system_charset_info);
  }


  /** 
    @brief
    Set the key fields for the statistical table column_stats

    @param
    col       Field for the 'table' column to read/update statistics on

    @details
    The function stores the values of the fields db_name, table_name and
    column_name in the record buffer for the statistical table column_stats.
    These fields comprise the primary key for the table.
    It also sets table_field to the passed parameter.

    @note
    The function is supposed to be called before any use of the
    method find_stat for an object of the Column_stat class.
  */

  void set_key_fields(Field *col)
  {
    set_full_table_name();
    column_name_field->store(&col->field_name, system_charset_info);
    table_field= col;
  }

  void set_key_fields(LEX_CSTRING *field_name)
  {
    set_full_table_name();
    column_name_field->store(field_name, system_charset_info);
    table_field= 0;                             // Safety
  }


  /** 
    @brief
    Update the table name fields in the current record of stat_table

    @details
    The function updates the primary key fields containing database name,
    table name, and column name for the last found record in the statistical
    table column_stats.

    @retval
    0        success with the update of the record
    @retval
    #        handler error in case of failure
  */

  int update_column_key_part(LEX_CSTRING *col)
  {
    int rc;
    store_record_for_update();
    rc= update_column(col);
    store_record_for_lookup();
    return rc;
  }

  int update_column(LEX_CSTRING *col)
  {
    column_name_field->store(col, system_charset_info);
    return update_record();
  }


  /** 
    @brief
    Store statistical data into statistical fields of column_stats

    @details
    This implementation of a purely virtual method sets the value of the
    columns 'min_value', 'max_value', 'nulls_ratio', 'avg_length',
    'avg_frequency', 'hist_size', 'hist_type' and 'histogram'  of the 
    stistical table columns_stat according to the contents of the bitmap
    write_stat.column_stat_nulls and the values of the fields min_value,
    max_value, nulls_ratio, avg_length, avg_frequency, hist_size, hist_type
    and histogram of the structure write_stat from the Field structure
    for the field 'table_field'.
    The value of the k-th column in the table columns_stat is set to NULL
    if the k-th bit in the bitmap 'column_stat_nulls' is set to 1. 

    @note
    A value from the field min_value/max_value is always converted
    into a varbinary string. If the length of the column 'min_value'/'max_value'
    is less than the length of the string the string is trimmed to fit the
    length of the column. 
  */    

  void store_stat_fields() override
  {
    StringBuffer<MAX_FIELD_WIDTH> val;

    MY_BITMAP *old_map= dbug_tmp_use_all_columns(stat_table,
                                                 &stat_table->read_set);
    for (uint i= COLUMN_STAT_MIN_VALUE; i <= COLUMN_STAT_HISTOGRAM; i++)
    {  
      Field *stat_field= stat_table->field[i];
      Column_statistics *stats= table_field->collected_stats;
      if (stats->is_null(i))
        stat_field->set_null();
      else
      {
        stat_field->set_notnull();
        switch (i) {
        case COLUMN_STAT_MIN_VALUE:
        {
          /*
            TODO varun: After MDEV-22583 is fixed, add a function in Field_bit
            and move this implementation there
          */
          if (table_field->type() == MYSQL_TYPE_BIT)
            stat_field->store(stats->min_value->val_int(),true);
          else
            stats->min_value->store_to_statistical_minmax_field(stat_field, &val);
          break;
        }
        case COLUMN_STAT_MAX_VALUE:
        {
          if (table_field->type() == MYSQL_TYPE_BIT)
            stat_field->store(stats->max_value->val_int(),true);
          else
            stats->max_value->store_to_statistical_minmax_field(stat_field, &val);
          break;
        }
        case COLUMN_STAT_NULLS_RATIO:
          stat_field->store(stats->get_nulls_ratio());
          break;
        case COLUMN_STAT_AVG_LENGTH:
          stat_field->store(stats->get_avg_length());
          break;
        case COLUMN_STAT_AVG_FREQUENCY:
          stat_field->store(stats->get_avg_frequency());
          break; 
        case COLUMN_STAT_HIST_SIZE:
          /*
            This is only here so that one can see the size when selecting
            from the table. It is not used.
          */
          stat_field->store(stats->histogram ?
                            stats->histogram->get_size() : 0);
          break;
        case COLUMN_STAT_HIST_TYPE:
          if (stats->histogram)
            stat_field->store(stats->histogram->get_type() + 1);
          else
            stat_field->set_null();
          break;
        case COLUMN_STAT_HISTOGRAM:
          if (stats->histogram)
            stats->histogram->serialize(stat_field);
          else
            stat_field->set_null();
          break;
        }
      }
    }
    dbug_tmp_restore_column_map(&stat_table->read_set, old_map);
  }


  /** 
    @brief
    Read statistical data from statistical fields of column_stats

    @details
    Find a record in mysql.column_stats that has statistics for this column.
    We search for record using a PK lookup. The lookup values are in the stat
    table's record buffer. Then, if the row is
    found, the function reads the values of the columns 'min_value',
    'max_value', 'nulls_ratio', 'avg_length', 'avg_frequency', 'hist_size' and
    'hist_type" of the  table column_stat and sets the members of *read_stats
    accordingly.
  */

  bool get_stat_values(Column_statistics *read_stats, MEM_ROOT *mem_root,
                       bool want_histograms)
  {
    bool res;
    read_stats->set_all_nulls();

    if (read_stats->min_value)
      read_stats->min_value->set_null();
    if (read_stats->max_value)
      read_stats->max_value->set_null();
    read_stats->histogram= 0;

    if ((res= find_stat()))
    {
      char buff[MAX_FIELD_WIDTH];
      String val(buff, sizeof(buff), &my_charset_bin);
      Histogram_type hist_type= INVALID_HISTOGRAM;

      for (uint i= COLUMN_STAT_MIN_VALUE; i <= COLUMN_STAT_HISTOGRAM; i++)
      {
        Field *stat_field= stat_table->field[i];

        if (!stat_field->is_null() &&
            (i > COLUMN_STAT_MAX_VALUE ||
             (i == COLUMN_STAT_MIN_VALUE && 
              read_stats->min_value) ||
             (i == COLUMN_STAT_MAX_VALUE && 
              read_stats->max_value)))
        {
          read_stats->set_not_null(i);

          switch (i) {
          case COLUMN_STAT_MIN_VALUE:
          {
            Field *field= read_stats->min_value;
            field->set_notnull();
            if (table_field->type() == MYSQL_TYPE_BIT)
              field->store(stat_field->val_int(), true);
            else
              field->store_from_statistical_minmax_field(stat_field, &val,
                                                         mem_root);
            break;
          }
          case COLUMN_STAT_MAX_VALUE:
          {
            Field *field= read_stats->max_value;
            field->set_notnull();
            if (table_field->type() == MYSQL_TYPE_BIT)
              field->store(stat_field->val_int(), true);
            else
              field->store_from_statistical_minmax_field(stat_field, &val,
                                                         mem_root);
            break;
          }
          case COLUMN_STAT_NULLS_RATIO:
            read_stats->set_nulls_ratio(stat_field->val_real());
            break;
          case COLUMN_STAT_AVG_LENGTH:
            read_stats->set_avg_length(stat_field->val_real());
            break;
          case COLUMN_STAT_AVG_FREQUENCY:
            read_stats->set_avg_frequency(stat_field->val_real());
            break;
          case COLUMN_STAT_HIST_SIZE:
            /*
              Ignore the contents of mysql.column_stats.hist_size. We take the
              size from the mysql.column_stats.histogram column, itself.
            */
            break;
          case COLUMN_STAT_HIST_TYPE:
            hist_type= (Histogram_type) (stat_field->val_int() - 1);
            break;
          case COLUMN_STAT_HISTOGRAM:
          {
            Histogram_base *hist= 0;
            read_stats->histogram_exists= 0;
            if (hist_type != INVALID_HISTOGRAM)
            {
              if (want_histograms)
              {
                char buff[MAX_FIELD_WIDTH];
                String val(buff, sizeof(buff), &my_charset_bin), *result;
                result= stat_field->val_str(&val);
                if (result->length())
                {
                  MY_BITMAP *old_sets[2];
                  TABLE *tbl= (TABLE *) table;
                  dbug_tmp_use_all_columns(tbl, old_sets,
                                           &tbl->read_set, &tbl->write_set);

                  if ((hist= create_histogram(mem_root, hist_type)))
                  {
                    if (hist->parse(mem_root, db_name->str, table_name->str,
                                    table->field[table_field->field_index],
                                    result->ptr(), result->length()))
                    {
                      delete hist;
                    }
                    else
                    {
                      read_stats->histogram= hist;
                      read_stats->histogram_exists= 1;
                    }
                  }
                  dbug_tmp_restore_column_maps(&tbl->read_set,
                                               &tbl->write_set,
                                               old_sets);
                }
              }
              else
                read_stats->histogram_exists= 1;
            }
            if (!hist)
              read_stats->set_null(COLUMN_STAT_HISTOGRAM);
            break;
          }
          }
        }
      }
    }
    return res;
  }
};


bool Histogram_binary::parse(MEM_ROOT *mem_root, const char*, const char*,
                             Field*,
                             const char *hist_data, size_t hist_data_len)
{
  size= hist_data_len; // 'size' holds the size of histogram in bytes
  if (!(values= (uchar*)alloc_root(mem_root, hist_data_len)))
    return true;

  memcpy(values, hist_data, hist_data_len);
  return false;
}

/*
  Save the histogram data info a table field.
*/
void Histogram_binary::serialize(Field *field)
{
  field->store((char*)values, size, &my_charset_bin);
}

void Histogram_binary::init_for_collection(MEM_ROOT *mem_root,
                                           Histogram_type htype_arg,
                                           ulonglong size_arg)
{
  type= htype_arg;
  values= (uchar*)alloc_root(mem_root, (size_t)size_arg);
  size= (uint8) size_arg;
}


/*
  An object of the class Index_stat is created to read statistical
  data on tables from the statistical table table_stat, to update
  index_stats with such statistical data, or to update columns
  of the primary key, or to delete the record by its primary key or
  its prefix. 
  Rows from the statistical table are read and updated always by
  primary key. 
*/ 

class Index_stat: public Stat_table
{

private:

  Field *db_name_field;      /* Field for the column index_stats.db_name */
  Field *table_name_field;   /* Field for the column index_stats.table_name */
  Field *index_name_field;   /* Field for the column index_stats.table_name */
  Field *prefix_arity_field; /* Field for the column index_stats.prefix_arity */
  const KEY *table_key_info; /* Info on the index to read/update statistics on */
  uint prefix_arity; /* Number of components of the index prefix of interest */

  void common_init_index_stat_table()
  {
    db_name_field= stat_table->field[INDEX_STAT_DB_NAME];
    table_name_field= stat_table->field[INDEX_STAT_TABLE_NAME];
    index_name_field= stat_table->field[INDEX_STAT_INDEX_NAME];
    prefix_arity_field= stat_table->field[INDEX_STAT_PREFIX_ARITY];
  } 

  void change_full_table_name(const LEX_CSTRING *db,
                              const LEX_CSTRING *tab) override
  {
     db_name_field->store(db, system_charset_info);
     table_name_field->store(tab, system_charset_info);
  }

public:


  /**
    @details
    The constructor 'tunes' the private and protected members of the
    constructed object for the statistical table index_stats to read/update
    statistics on prefixes of different indexes of the table 'tab'.
    The TABLE structure for the table index_stats must be passed as a value
    for the parameter 'stat'.
  */

  Index_stat(TABLE *stat, const TABLE *tab) :Stat_table(stat, tab)
  {
    common_init_index_stat_table();
  }


  /**
    @details
    The constructor 'tunes' the private and protected members of the
    object constructed for the statistical table index_stats for 
    the future updates/deletes of the record concerning the table 'tab'
    from the database 'db'. 
  */

  Index_stat(TABLE *stat, const LEX_CSTRING *db, const LEX_CSTRING *tab)
    :Stat_table(stat, db, tab)
  {
    common_init_index_stat_table();
  }


  /**
    @brief
    Set table name fields for the statistical table index_stats

    @details
    The function stores the values of the fields db_name and table_name 
    of the statistical table index_stats in the record buffer.
  */

  void set_full_table_name()
  {
    db_name_field->store(db_name, system_charset_info);
    table_name_field->store(table_name, system_charset_info);
  }

  inline void set_index_name(const LEX_CSTRING *name)
  {
    index_name_field->store(name, system_charset_info);
  }

  /** 
    @brief
    Set the key fields of index_stats used to access records for index prefixes

    @param
    index_info   Info for the index of 'table' to read/update statistics on

    @details
    The function sets the values of the fields db_name, table_name and
    index_name in the record buffer for the statistical table index_stats. 
    It also sets table_key_info to the passed parameter.

    @note
    The function is supposed to be called before any use of the method
    find_next_stat_for_prefix for an object of the Index_stat class.
  */

  void set_index_prefix_key_fields(const KEY *index_info)
  {
    set_full_table_name();
    set_index_name(&index_info->name);
    table_key_info= index_info;
  }


  /** 
    @brief
    Set the key fields for the statistical table index_stats

    @param
    index_info   Info for the index of 'table' to read/update statistics on
    @param
    index_prefix_arity Number of components in the index prefix of interest

    @details
    The function sets the values of the fields db_name, table_name and
    index_name, prefix_arity in the record buffer for the statistical
    table index_stats. These fields comprise the primary key for the table. 

    @note
    The function is supposed to be called before any use of the  
    method find_stat for an object of the Index_stat class.
  */

  void set_key_fields(KEY *index_info, uint index_prefix_arity)
  {
    set_index_prefix_key_fields(index_info);
    prefix_arity= index_prefix_arity; 
    prefix_arity_field->store(index_prefix_arity, TRUE);  
  }


  int update_index_name(const LEX_CSTRING *name)
  {
    index_name_field->store(name, system_charset_info);
    return update_record();
  }


  int read_next()
  {
    return stat_table->file->ha_index_next_same(stat_table->record[0],
                                                last_key,
                                                last_key_length);
  }

  /** 
    @brief
    Store statistical data into statistical fields of table index_stats

    @details
    This implementation of a purely virtual method sets the value of the
    column 'avg_frequency' of the statistical table index_stats according to
    the value of write_stat.avg_frequency[Index_stat::prefix_arity]
    from the KEY_INFO structure 'table_key_info'.
    If the value of write_stat. avg_frequency[Index_stat::prefix_arity] is
    equal  to 0, the value of the column is set to NULL.
  */    

  void store_stat_fields() override
  {
    Field *stat_field= stat_table->field[INDEX_STAT_AVG_FREQUENCY];
    double avg_frequency=
      table_key_info->collected_stats->get_avg_frequency(prefix_arity-1);
    if (avg_frequency == 0)
      stat_field->set_null();
    else
    {
      stat_field->set_notnull();
      stat_field->store(avg_frequency);
    }
  }


  /** 
    @brief
    Read statistical data from statistical fields of index_stats

    @details
    Find a record in mysql.index_stats that has statistics for the index prefix
    of interest (the prefix length is in this->prefix_arity).
    We search for record using a PK lookup. The lookup values are in the stat
    table's record buffer.

    The result is stored in read_stats->avg_frequency[this->prefix_arity].
    If mysql.index_stats doesn't have the value or has SQL NULL, we store the
    value of 0.
  */

  bool get_stat_values(Index_statistics *read_stats)
  {
    double avg_frequency= 0;
    bool res;
    if ((res= find_stat()))
    {
      Field *stat_field= stat_table->field[INDEX_STAT_AVG_FREQUENCY];
      if (!stat_field->is_null())
        avg_frequency= stat_field->val_real();
    }
    read_stats->set_avg_frequency(prefix_arity-1, avg_frequency);
    return res;
  }
};


/*
  An iterator to enumerate statistics table rows which allows to modify
  the rows while reading them.

  Used by RENAME TABLE handling to assign new dbname.tablename to statistic
  rows.
*/
class Stat_table_write_iter
{
  Stat_table *owner;
  IO_CACHE io_cache;
  uchar *rowid_buf;
  uint rowid_size;

public:
  Stat_table_write_iter(Stat_table *stat_table_arg)
   : owner(stat_table_arg), rowid_buf(NULL),
     rowid_size(owner->stat_file->ref_length)
  {
     my_b_clear(&io_cache);
  }

  /*
    Initialize the iterator. It will return rows with n_keyparts matching the
    curernt values.

    @return  false - OK
             true  - Error
  */
  bool init(uint n_keyparts)
  {
    if (!(rowid_buf= (uchar*)my_malloc(PSI_INSTRUMENT_ME, rowid_size, MYF(0))))
      return true;

    if (open_cached_file(&io_cache, mysql_tmpdir, TEMP_PREFIX,
                         1024, MYF(MY_WME)))
      return true;

    handler *h= owner->stat_file;
    uchar key[MAX_KEY_LENGTH];
    uint prefix_len= 0;
    for (uint i= 0; i < n_keyparts; i++)
      prefix_len += owner->stat_key_info->key_part[i].store_length;

    key_copy(key, owner->record[0], owner->stat_key_info,
             prefix_len);
    key_part_map prefix_map= (key_part_map) ((1 << n_keyparts) - 1);
    h->ha_index_init(owner->stat_key_idx, false);
    int res= h->ha_index_read_map(owner->record[0], key, prefix_map,
                                  HA_READ_KEY_EXACT);
    if (res)
    {
      reinit_io_cache(&io_cache, READ_CACHE, 0L, 0, 0);
      /* "Key not found" is not considered an error */
      return (res == HA_ERR_KEY_NOT_FOUND)? false: true;
    }

    do {
      h->position(owner->record[0]);
      my_b_write(&io_cache, h->ref, rowid_size);

    } while (!h->ha_index_next_same(owner->record[0], key, prefix_len));

    /* Prepare for reading */
    reinit_io_cache(&io_cache, READ_CACHE, 0L, 0, 0);
    h->ha_index_or_rnd_end();
    if (h->ha_rnd_init(false))
      return true;

    return false;
  }

  /*
     Read the next row.

     @return
        false   OK
        true    No more rows or error.
  */
  bool get_next_row()
  {
    if (!my_b_inited(&io_cache) || my_b_read(&io_cache, rowid_buf, rowid_size))
      return true; /* No more data */

    handler *h= owner->stat_file;
    /*
      We should normally be able to find the row that we have rowid for. If we
      don't, let's consider this an error.
    */
    int res= h->ha_rnd_pos(owner->record[0], rowid_buf);

    return (res==0)? false : true;
  }

  void cleanup()
  {
    if (rowid_buf)
      my_free(rowid_buf);
    rowid_buf= NULL;
    owner->stat_file->ha_index_or_rnd_end();
    close_cached_file(&io_cache);
    my_b_clear(&io_cache);
  }

  ~Stat_table_write_iter()
  {
    /* Ensure that cleanup has been run */
    DBUG_ASSERT(rowid_buf == 0);
  }
};

class Histogram_binary_builder : public Histogram_builder
{
  Field *min_value;        /* pointer to the minimal value for the field   */
  Field *max_value;        /* pointer to the maximal value for the field   */
  Histogram_binary *histogram;  /* the histogram location                  */
  uint hist_width;         /* the number of points in the histogram        */
  double bucket_capacity;  /* number of rows in a bucket of the histogram  */ 
  uint curr_bucket;        /* number of the current bucket to be built     */

public:
  Histogram_binary_builder(Field *col, uint col_len, ha_rows rows)
    : Histogram_builder(col, col_len, rows)
  {
    Column_statistics *col_stats= col->collected_stats;
    min_value= col_stats->min_value;
    max_value= col_stats->max_value;
    histogram= (Histogram_binary*)col_stats->histogram;
    hist_width= histogram->get_width();
    bucket_capacity= (double) records / (hist_width + 1);
    curr_bucket= 0;
  }

  int next(void *elem, element_count elem_cnt) override
  {
    counters.next(elem, elem_cnt);
    ulonglong count= counters.get_count();
    if (curr_bucket == hist_width)
      return 0;
    if (count > bucket_capacity * (curr_bucket + 1))
    {
      column->store_field_value((uchar *) elem, col_length);
      histogram->set_value(curr_bucket,
                           column->pos_in_interval(min_value, max_value));
      curr_bucket++;
      while (curr_bucket != hist_width &&
             count > bucket_capacity * (curr_bucket + 1))
      {
        histogram->set_prev_value(curr_bucket);
        curr_bucket++;
      }
    }
    return 0;
  }
  void finalize() override {}
};


Histogram_builder *Histogram_binary::create_builder(Field *col, uint col_len,
                                                    ha_rows rows)
{
  return new Histogram_binary_builder(col, col_len, rows);
}


Histogram_base *create_histogram(MEM_ROOT *mem_root, Histogram_type hist_type)
{
  Histogram_base *res= NULL;
  switch (hist_type) {
  case SINGLE_PREC_HB:
  case DOUBLE_PREC_HB:
    res= new (mem_root) Histogram_binary(hist_type);
    break;
  case JSON_HB:
    res= new (mem_root) Histogram_json_hb();
    break;
  default:
    DBUG_ASSERT(0);
  }
  return res;
}


C_MODE_START

static int histogram_build_walk(void *elem, element_count elem_cnt, void *arg)
{
  Histogram_builder *hist_builder= (Histogram_builder *) arg;
  return hist_builder->next(elem, elem_cnt);
}

int basic_stats_collector_walk(void *elem, element_count count,
                               void *arg)
{
  ((Basic_stats_collector*)arg)->next(elem, count);
  return 0;
}

C_MODE_END
/*
  The class Count_distinct_field is a helper class used to calculate
  the number of distinct values for a column. The class employs the
  Unique class for this purpose.
  The class Count_distinct_field is used only by the function
  collect_statistics_for_table to calculate the values for 
  column avg_frequency of the statistical table column_stats.
*/
    
class Count_distinct_field: public Sql_alloc
{
protected:

  /* Field for which the number of distinct values is to be find out */
  Field *table_field;  
  Unique *tree;       /* The helper object to contain distinct values */
  uint tree_key_length; /* The length of the keys for the elements of 'tree */

  ulonglong distincts;
  ulonglong distincts_single_occurence;

public:
  
  Count_distinct_field() = default;

  /**
    @param
    field               Field for which the number of distinct values is 
                        to be find out
    @param
    max_heap_table_size The limit for the memory used by the RB tree container
                        of the constructed Unique object 'tree' 

    @details
    The constructor sets the values of 'table_field' and 'tree_key_length',
    and then calls the 'new' operation to create a Unique object for 'tree'.
    The type of 'field' and the value max_heap_table_size of determine the set
    of the parameters to be passed to the constructor of the Unique object. 
  */  

  Count_distinct_field(Field *field, size_t max_heap_table_size)
  {
    table_field= field;
    tree_key_length= field->pack_length();

    tree= new Unique((qsort_cmp2) simple_str_key_cmp, (void*) field,
                     tree_key_length, max_heap_table_size, 1);
  }

  virtual ~Count_distinct_field()
  {
    delete tree;
    tree= NULL;
  }

  /* 
    @brief
    Check whether the Unique object tree has been successfully created
  */
  bool exists()
  {
    return (tree != NULL);
  }

  /*
    @brief
    Add the value of 'field' to the container of the Unique object 'tree'
  */
  virtual bool add()
  {
    table_field->mark_unused_memory_as_defined();
    return tree->unique_add(table_field->ptr);
  }

  /*
    @brief
    Calculate the number of elements accumulated in the container of 'tree'
  */
  void walk_tree()
  {
    Basic_stats_collector stats_collector;
    tree->walk(table_field->table, basic_stats_collector_walk,
               (void*)&stats_collector );
    distincts= stats_collector.get_count_distinct();
    distincts_single_occurence= stats_collector.get_count_single_occurence();
  }

  /*
    @brief
    Calculate a histogram of the tree
  */
  bool walk_tree_with_histogram(ha_rows rows)
  {
    Histogram_base *hist= table_field->collected_stats->histogram;
    Histogram_builder *hist_builder=
       hist->create_builder(table_field, tree_key_length, rows);

    if (tree->walk(table_field->table, histogram_build_walk,
                   (void*)hist_builder))
    {
      delete hist_builder;
      return true; // Error
    }
    hist_builder->finalize();
    distincts= hist_builder->counters.get_count_distinct();
    distincts_single_occurence= hist_builder->counters.
                                  get_count_single_occurence();
    delete hist_builder;
    return false;
  }

  ulonglong get_count_distinct()
  {
    return distincts;
  }

  ulonglong get_count_distinct_single_occurence()
  {
    return distincts_single_occurence;
  }

  /*
    @brief
    Get the pointer to the histogram built for table_field
  */
  Histogram_base *get_histogram()
  {
    return table_field->collected_stats->histogram;
  }
};


static
int simple_ulonglong_key_cmp(void* arg, uchar* key1, uchar* key2)
{
  ulonglong *val1= (ulonglong *) key1;
  ulonglong *val2= (ulonglong *) key2;
  return *val1 > *val2 ? 1 : *val1 == *val2 ? 0 : -1; 
}
  

/* 
  The class Count_distinct_field_bit is derived from the class 
  Count_distinct_field to be used only for fields of the MYSQL_TYPE_BIT type.
  The class provides a different implementation for the method add 
*/

class Count_distinct_field_bit: public Count_distinct_field
{
public:

  Count_distinct_field_bit(Field *field, size_t max_heap_table_size)
  {
    table_field= field;
    tree_key_length= sizeof(ulonglong);

    tree= new Unique((qsort_cmp2) simple_ulonglong_key_cmp,
                     (void*) &tree_key_length,
                     tree_key_length, max_heap_table_size, 1);
  }

  bool add()
  {
    longlong val= table_field->val_int();   
    return tree->unique_add(&val);
  }
};


/* 
  The class Index_prefix_calc is a helper class used to calculate the values
  for the column 'avg_frequency' of the statistical table index_stats.
  For any table t from the database db and any k-component prefix of the
  index i for this table the row from index_stats with the primary key
  (db,t,i,k) must contain in the column 'avg_frequency' either NULL or 
  the number that is the ratio of N and V, where N is the number of index
  entries without NULL values in the first k components of the index i,
  and V is the number of distinct tuples composed of the first k components
  encountered among these index entries.  
  Currently the objects of this class are used only by the function
  collect_statistics_for_index. 
*/

class Index_prefix_calc: public Sql_alloc
{

private:

  /* Info for the index i for whose prefix 'avg_frequency' is calculated */
  KEY *index_info;  
  /* The maximum number of the components in the prefixes of interest */   
  uint prefixes; 
  bool empty;  

  /* This structure is created for every k components of the index i */
  class Prefix_calc_state
  {
  public:
    /* 
      The number of the scanned index entries without nulls 
      in the first k components
    */
    ulonglong entry_count;
    /* 
      The number if the scanned index entries without nulls with 
      the last encountered k-component prefix
    */
    ulonglong prefix_count;
    /* The values of the last encountered k-component prefix */
    Cached_item *last_prefix;
  };

  /* 
    Array of structures used to calculate 'avg_frequency' for different
    prefixes of the index i
  */   
  Prefix_calc_state *calc_state;
    
public:

  bool is_single_comp_pk;
  bool is_partial_fields_present;

  Index_prefix_calc(THD *thd, TABLE *table, KEY *key_info)
    : index_info(key_info), prefixes(0), empty(true),
    calc_state(NULL), is_single_comp_pk(false), is_partial_fields_present(false)
  {
    uint i;
    Prefix_calc_state *state;
    uint key_parts= table->actual_n_key_parts(key_info);

    uint pk= table->s->primary_key;
    if ((uint) (table->key_info - key_info) == pk &&
        table->key_info[pk].user_defined_key_parts == 1)
    {
      prefixes= 1;
      is_single_comp_pk= TRUE;
      return;
    }
        
    if ((calc_state=
         (Prefix_calc_state *) thd->alloc(sizeof(Prefix_calc_state)*key_parts)))
    {
      uint keyno= (uint)(key_info-table->key_info);
      for (i= 0, state= calc_state; i < key_parts; i++, state++)
      {
        /* 
          Do not consider prefixes containing a component that is only part
          of the field. This limitation is set to avoid fetching data when
          calculating the values of 'avg_frequency' for prefixes.
	*/   
        if (!key_info->key_part[i].field->part_of_key.is_set(keyno))
        {
          is_partial_fields_present= TRUE;
          break;
        }

        if (!(state->last_prefix=
              new (thd->mem_root)
              Cached_item_field(thd, key_info->key_part[i].field)))
          break;
        state->entry_count= state->prefix_count= 0;
        prefixes++;
      }
    }
  }


  /** 
    @breif
    Change the elements of calc_state after reading the next index entry

    @details
    This function is to be called at the index scan each time the next
    index entry has been read into the record buffer.
    For each of the index prefixes the function checks whether nulls
    are encountered in any of the k components of the prefix.
    If this is not the case the value of calc_state[k-1].entry_count
    is incremented by 1. Then the function checks whether the value of
    any of these k components has changed. If so, the value of 
    calc_state[k-1].prefix_count is incremented by 1. 
  */

  void add()
  {
    uint i;
    Prefix_calc_state *state;
    uint first_changed= prefixes;
    for (i= prefixes, state= calc_state+prefixes-1; i; i--, state--)
    {
      if (state->last_prefix->cmp())
        first_changed= i-1;
    }
    if (empty)
    {
      first_changed= 0;
      empty= FALSE;
    }
    for (i= 0, state= calc_state; i < prefixes; i++, state++)
    {
      if (state->last_prefix->null_value)
        break;
      if (i >= first_changed)
        state->prefix_count++;
      state->entry_count++;
    }   
  }

  /**
    @brief
    Calculate the values of avg_frequency for all prefixes of an index

    @details
    This function is to be called after the index scan to count the number
    of distinct index prefixes has been done. The function calculates
    the value of avg_frequency for the index prefix with k components
    as calc_state[k-1].entry_count/calc_state[k-1].prefix_count.
    If calc_state[k-1].prefix_count happens to be 0, the value of
    avg_frequency[k-1] is set to 0, i.e. is considered as unknown.
  */

  void get_avg_frequency()
  {
    uint i;
    Prefix_calc_state *state;

    if (is_single_comp_pk)
    {
      index_info->collected_stats->set_avg_frequency(0, 1.0);
      return;
    }

    for (i= 0, state= calc_state; i < prefixes; i++, state++)
    {
      if (i < prefixes)
      {
        double val= state->prefix_count == 0 ?
	            0 : (double) state->entry_count / state->prefix_count;                     
        index_info->collected_stats->set_avg_frequency(i, val);
      }
    }
  }       
};


/**
  @brief 
  Create fields for min/max values to collect column statistics

  @param thd    The thread handle
  @param table  Table the fields are created for
  @param fields Fields for which we want to have statistics

  @details
  The function first allocates record buffers to store min/max values
  for 'table's fields. Then for each table field f it creates Field structures
  that points to these buffers rather that to the record buffer as the
  Field object for f does. The pointers of the created fields are placed
  in the collected_stats structure of the Field object for f.
  The function allocates the buffers for min/max values in the table
  memory. 

  @note 
  The buffers allocated when min/max values are used to read statistics
  from the persistent statistical tables differ from those buffers that
  are used when statistics on min/max values for column is collected
  as they are allocated in different mem_roots.
  The same is true for the fields created for min/max values.  
*/      

static
void create_min_max_statistical_fields_for_table(THD *thd, TABLE *table,
                                                 MY_BITMAP *fields)
{
  uint rec_buff_length= table->s->rec_buff_length;

  if ((table->collected_stats->min_max_record_buffers=
       (uchar *) alloc_root(thd->mem_root, 2*rec_buff_length)))
  {
    uchar *record= table->collected_stats->min_max_record_buffers;
    memset(record, 0,  2*rec_buff_length);

    for (uint i=0; i < 2; i++, record+= rec_buff_length)
    {
      for (Field **field_ptr= table->field; *field_ptr; field_ptr++)
      {
        Field *fld;
        Field *table_field= *field_ptr;
        my_ptrdiff_t diff= record-table->record[0];
        if (!bitmap_is_set(fields, table_field->field_index))
          continue; 
        if (!(fld= table_field->clone(thd->mem_root, table, diff)))
          continue;
        if (i == 0)
          table_field->collected_stats->min_value= fld;
        else
          table_field->collected_stats->max_value= fld;
      }
    }
  }
}


/**
  @brief 
  Create fields for min/max values to read column statistics

  @param
  thd         Thread handler
  @param
  table_share Table share the fields are created for
  @param
  stats_cb    TABLE_STATISTICS_CB object whose mem_root is used for allocations

  @details
  The function first allocates record buffers to store min/max values for
  fields in the table. For each field f it creates Field structures
  that points to these buffers rather that to the record buffer as the
  Field object for f does. The pointers of the created fields are placed
  in the read_stats structure of the Field object for f.
  The function allocates the buffers for min/max values in the stats_cb
  memory.

  @note 
  The buffers allocated when min/max values are used to collect statistics
  from the persistent statistical tables differ from those buffers that
  are used when statistics on min/max values for column is read as they
  are allocated in different mem_roots.
  The same is true for the fields created for min/max values.  
*/

static void
create_min_max_statistical_fields(THD *thd,
                                  const TABLE_SHARE *table_share,
                                  TABLE_STATISTICS_CB *stats_cb)
{
  Table_statistics *stats= stats_cb->table_stats;

  if (stats->min_max_record_buffers)
    return;
   
  uint rec_buff_length= table_share->rec_buff_length;

  if ((stats->min_max_record_buffers=
         (uchar *) alloc_root(&stats_cb->mem_root, 2*rec_buff_length)))
  {
    uchar *record= stats->min_max_record_buffers;
    memset(record, 0,  2*rec_buff_length);

    for (uint i=0; i < 2; i++, record+= rec_buff_length)
    {
      Column_statistics *column_stats= stats_cb->table_stats->column_stats;
      for (Field **field_ptr= table_share->field;
           *field_ptr;
           field_ptr++, column_stats++)
      {
        Field *fld;
        Field *table_field= *field_ptr;
        my_ptrdiff_t diff= record - table_share->default_values;
        if (!(fld= table_field->clone(&stats_cb->mem_root, NULL, diff)))
          continue;
        if (i == 0)
          column_stats->min_value= fld;
        else
          column_stats->max_value= fld;
      }
    }
  }

}


/**
  @brief 
  Allocate memory for the table's statistical data to be collected

  @param thd          The thread handle
  @param table        Table for which we should allocate statistical data
  @param stat_fields  Fields for which we want to have statistics

  @note
  The function allocates the memory for the statistical data on 'table' with
  the intention to collect the data there. The memory is allocated for
  the statistics on the table, on the table's columns, and on the table's
  indexes. The memory is allocated in the thd's mem_root.

  @retval
  0      If the memory for all statistical data has been successfully allocated
  @retval
  1      Otherwise

  @note 
  Each thread allocates its own memory to collect statistics on the table
  It allows us, for example, to collect statistics on the different indexes
  of the same table in parallel. 
*/      

int alloc_statistics_for_table(THD* thd, TABLE *table, MY_BITMAP *stat_fields)
{ 
  Field **field_ptr;
  uint fields= bitmap_bits_set(stat_fields);
  uint keys= table->s->keys;
  uint key_parts= table->s->ext_key_parts;
  uint hist_size= thd->variables.histogram_size;
  Table_statistics *table_stats;
  Column_statistics_collected *column_stats;
  Index_statistics *index_stats;
  ulonglong *idx_avg_frequency;
  uchar *histogram;
  DBUG_ENTER("alloc_statistics_for_table");

  if (!multi_alloc_root(thd->mem_root,
                        &table_stats, sizeof(*table_stats),
                        &column_stats, sizeof(*column_stats) * fields,
                        &index_stats, sizeof(*index_stats) * keys,
                        &idx_avg_frequency,
                        sizeof(*idx_avg_frequency) * key_parts,
                        &histogram, hist_size * fields,
                        NullS))
    DBUG_RETURN(1);

  if (hist_size > 0)
    bzero(histogram, hist_size * fields);
  else
    histogram= 0;

  table->collected_stats= table_stats;
  table_stats->column_stats= column_stats;
  table_stats->index_stats= index_stats;
  table_stats->idx_avg_frequency= idx_avg_frequency;
  
  bzero((void*) column_stats, sizeof(Column_statistics) * fields);

  for (field_ptr= table->field; *field_ptr; field_ptr++)
  {
    if (bitmap_is_set(stat_fields, (*field_ptr)->field_index))
    {
      column_stats->histogram = NULL;
      (*field_ptr)->collected_stats= column_stats++;
    }
    else
      (*field_ptr)->collected_stats= 0;
  }

  memset(idx_avg_frequency, 0, sizeof(ulonglong) * key_parts);

  KEY *key_info, *end;
  for (key_info= table->key_info, end= key_info + table->s->keys;
       key_info < end; 
       key_info++, index_stats++)
  {
    key_info->collected_stats= index_stats;
    key_info->collected_stats->init_avg_frequency(idx_avg_frequency);
    idx_avg_frequency+= key_info->ext_key_parts;
  }
  /*
    idx_avg_frequency can be less than
    table_stats->idx_avg_frequency + key_parts
    in the case of LONG_UNIQUE_HASH_FIELD as these has a hidden
    ext_key_part which is counted in table_share->ext_keyparts but not
    in keyinfo->ext_key_parts.
  */
  DBUG_ASSERT(idx_avg_frequency <= table_stats->idx_avg_frequency + key_parts);

  create_min_max_statistical_fields_for_table(thd, table, stat_fields);

  DBUG_RETURN(0);
}

/*
  Free the "local" statistics for table allocated during getting statistics
*/

void free_statistics_for_table(TABLE *table)
{
  for (Field **field_ptr= table->field; *field_ptr; field_ptr++)
  {
    delete (*field_ptr)->collected_stats;
    (*field_ptr)->collected_stats= 0;
  }
}

/**
  @brief 
  Allocate memory for the statistical data used by a table share

  @param
  thd         Thread handler
  @param
  table_share Table share for which the memory for statistical data is allocated
  @param
  stats_cb    TABLE_STATISTICS_CB object for storing the statistical data

  @note
  The function allocates the memory for the statistical data on a table in the
  table's share memory with the intention to read the statistics there from
  the system persistent statistical tables mysql.table_stat, mysql.column_stats,
  mysql.index_stats. The memory is allocated for the statistics on the table,
  on the tables's columns, and on the table's indexes. The memory is allocated
  in the table_share's mem_root.

  @retval
  0     If the memory for all statistical data has been successfully allocated  
  @retval
  1     Otherwise

  @note
  The situation when more than one thread try to allocate memory for 
  statistical data is rare. It happens under the following scenario:
  1. One thread executes a query over table t with the system variable 
    'use_stat_tables' set to 'never'.
  2. After this the second thread sets 'use_stat_tables' to 'preferably'
     and executes a query over table t.    
  3. Simultaneously the third thread sets 'use_stat_tables' to 'preferably'
     and executes a query over table t. 
  Here the second and the third threads try to allocate the memory for
  statistical data at the same time. The precautions are taken to
  guarantee the correctness of the allocation.
*/

static int
alloc_engine_independent_statistics(THD *thd, const TABLE_SHARE *table_share,
                                 TABLE_STATISTICS_CB *stats_cb)
{
  Table_statistics *table_stats= stats_cb->table_stats;
  uint fields= table_share->fields;
  uint keys= table_share->keys;
  uint key_parts= table_share->ext_key_parts;
  Index_statistics *index_stats;
  ulonglong *idx_avg_frequency;
  DBUG_ENTER("alloc_engine_independent_statistics");

  Column_statistics *column_stats;
  if (!multi_alloc_root(&stats_cb->mem_root,
                        &table_stats, sizeof(Table_statistics),
                        &column_stats, sizeof(Column_statistics) * fields,
                        &index_stats, sizeof(Index_statistics) * keys,
                        &idx_avg_frequency,
                        sizeof(*idx_avg_frequency) * key_parts,
                        NullS))
    DBUG_RETURN(1);

  /* Zero variables but not the gaps between them */
  bzero(table_stats, sizeof(Table_statistics));
  bzero((void*) column_stats, sizeof(Column_statistics) * fields);
  bzero(index_stats, sizeof(Index_statistics) * keys);
  bzero(idx_avg_frequency, sizeof(idx_avg_frequency) * key_parts);

  stats_cb->table_stats= table_stats;
  table_stats->columns= table_share->fields;
  table_stats->column_stats= column_stats;
  table_stats->index_stats= index_stats;
  table_stats->idx_avg_frequency= idx_avg_frequency;

  create_min_max_statistical_fields(thd, table_share, stats_cb);

  for (KEY *key_info= table_share->key_info, *end= key_info + keys;
       key_info < end;
       key_info++, index_stats++)
  {
    index_stats->init_avg_frequency(idx_avg_frequency);
    idx_avg_frequency+= key_info->ext_key_parts;
  }
  DBUG_ASSERT(idx_avg_frequency <= table_stats->idx_avg_frequency + key_parts);
  DBUG_RETURN(0);
}


/**
  @brief
  Initialize the aggregation fields to collect statistics on a column

  @param
  thd            Thread handler
  @param
  table_field    Column to collect statistics for
*/

inline
void Column_statistics_collected::init(THD *thd, Field *table_field)
{
  size_t max_heap_table_size= (size_t)thd->variables.max_heap_table_size;
  TABLE *table= table_field->table;
  uint pk= table->s->primary_key;
  
  is_single_pk_col= FALSE;

  if (pk != MAX_KEY && table->key_info[pk].user_defined_key_parts == 1 &&
      table->key_info[pk].key_part[0].fieldnr == table_field->field_index + 1)
    is_single_pk_col= TRUE;  
  
  column= table_field;

  set_all_nulls();

  nulls= 0;
  column_total_length= 0;
  count_distinct= NULL;
  if (!is_single_pk_col && !(table_field->flags & BLOB_FLAG))
  {
    count_distinct=
      table_field->type() == MYSQL_TYPE_BIT ?
      new (thd->mem_root) Count_distinct_field_bit(table_field,
                                                   max_heap_table_size) :
      new (thd->mem_root) Count_distinct_field(table_field,
                                               max_heap_table_size);
    if (count_distinct && !count_distinct->exists())
    {
      /* Allocation failed */
      delete count_distinct;
      count_distinct= NULL;
    }
  }
}


/**
  @brief
  Perform aggregation for a row when collecting statistics on a column

  @param
  rowno     The order number of the row
*/

inline
bool Column_statistics_collected::add()
{

  bool err= 0;
  if (column->is_null())
    nulls++;
  else
  {
    column_total_length+= column->value_length();
    if (min_value && column->update_min(min_value,
                                        is_null(COLUMN_STAT_MIN_VALUE)))
      set_not_null(COLUMN_STAT_MIN_VALUE);
    if (max_value && column->update_max(max_value,
                                        is_null(COLUMN_STAT_MAX_VALUE)))
      set_not_null(COLUMN_STAT_MAX_VALUE);
    if (count_distinct)
      err= count_distinct->add();
  } 
  return err;
}


/**
  @brief
  Get the results of aggregation when collecting the statistics on a column
  
  @param
  rows          The total number of rows in the table 
*/

inline
bool Column_statistics_collected::finish(MEM_ROOT *mem_root, ha_rows rows,
                                         double sample_fraction)
{
  double val;

  if (rows)
  {
     val= (double) nulls / rows;
     set_nulls_ratio(val);
     set_not_null(COLUMN_STAT_NULLS_RATIO);
  }
  if (rows - nulls)
  {
     val= (double) column_total_length / (rows - nulls);
     set_avg_length(val);
     set_not_null(COLUMN_STAT_AVG_LENGTH);
  }
  if (count_distinct)
  {
    uint hist_size= current_thd->variables.histogram_size;
    Histogram_type hist_type= 
      (Histogram_type) (current_thd->variables.histogram_type);
    bool have_histogram= false;
    if (hist_size != 0 && hist_type != INVALID_HISTOGRAM)
    {
      histogram= create_histogram(mem_root, hist_type);
      histogram->init_for_collection(mem_root, hist_type, hist_size);

      if (count_distinct->walk_tree_with_histogram(rows - nulls))
      {
        delete histogram;
        histogram= NULL;
        delete count_distinct;
        count_distinct= NULL;
        return true; // Error
      }
      have_histogram= true;
    }
    else
    {
      /* Compute cardinality statistics */
      count_distinct->walk_tree();
    }

    ulonglong distincts= count_distinct->get_count_distinct();
    ulonglong distincts_single_occurence=
      count_distinct->get_count_distinct_single_occurence();

    if (distincts)
    {
      /*
       We use the unsmoothed first-order jackknife estimator" to estimate
       the number of distinct values.
       With a sufficient large percentage of rows sampled (80%), we revert back
       to computing the avg_frequency off of the raw data.
      */
      if (sample_fraction > 0.8)
        val= (double) (rows - nulls) / distincts;
      else
      {
        if (nulls == 1)
          distincts_single_occurence+= 1;
        if (nulls)
          distincts+= 1;
        double fraction_single_occurence=
          static_cast<double>(distincts_single_occurence) / rows;
        double total_number_of_rows= rows / sample_fraction;
        double estimate_total_distincts= total_number_of_rows /
                (distincts /
                 (1.0 - (1.0 - sample_fraction) * fraction_single_occurence));
        val = std::fmax(estimate_total_distincts * (rows - nulls) / rows, 1.0);
      }

      set_avg_frequency(val);
      set_not_null(COLUMN_STAT_AVG_FREQUENCY);
    }
    else
      have_histogram= false;

    set_not_null(COLUMN_STAT_HIST_SIZE);
    if (have_histogram && distincts && histogram)
    {
      set_not_null(COLUMN_STAT_HIST_TYPE);
      set_not_null(COLUMN_STAT_HISTOGRAM);
    }
    delete count_distinct;
    count_distinct= NULL;
  }
  else if (is_single_pk_col)
  {
    val= 1.0;
    set_avg_frequency(val); 
    set_not_null(COLUMN_STAT_AVG_FREQUENCY);
  }
  return false;
}


/**
  @brief
  Clean up auxiliary structures used for aggregation
*/

inline
void Column_statistics_collected::cleanup()
{
  if (count_distinct)
  { 
    delete count_distinct;
    count_distinct= NULL;
  }
}


/**
  @brief
  Collect statistical data on an index

  @param 
  table       The table the index belongs to
  index       The number of this index in the table

  @details
  The function collects the value of 'avg_frequency' for the prefixes
  on an index from 'table'. The index is specified by its number.
  If the scan is successful the calculated statistics is saved in the
  elements of the array write_stat.avg_frequency of the KEY_INFO structure
  for the index. The statistics for the prefix with k components is saved
  in the element number k-1.

  @retval
  0         If the statistics has been successfully collected  
  @retval
  1         Otherwise

  @note
  The function collects statistics for the index prefixes for one index
  scan during which no data is fetched from the table records. That's why
  statistical data for prefixes that contain part of a field is not
  collected.
  The function employs an object of the helper class Index_prefix_calc to
  count for each index prefix the number of index entries without nulls and
  the number of distinct entries among them.
 
*/

static
int collect_statistics_for_index(THD *thd, TABLE *table, uint index)
{
  int rc= 0;
  KEY *key_info= &table->key_info[index];
  DBUG_ENTER("collect_statistics_for_index");

  /* No statistics for FULLTEXT indexes. */
  if (key_info->flags & (HA_FULLTEXT|HA_SPATIAL))
    DBUG_RETURN(rc);

  Index_prefix_calc index_prefix_calc(thd, table, key_info);

  DEBUG_SYNC(table->in_use, "statistics_collection_start1");
  DEBUG_SYNC(table->in_use, "statistics_collection_start2");

  if (index_prefix_calc.is_single_comp_pk)
  {
    index_prefix_calc.get_avg_frequency();
    DBUG_RETURN(rc);
  }

  /*
    Request "only index read" in case of absence of fields which are
    partially in the index to avoid problems with partitioning (for example)
    which want to get whole field value.
  */
  if (!index_prefix_calc.is_partial_fields_present)
    table->file->ha_start_keyread(index);
  table->file->ha_index_init(index, TRUE);
  rc= table->file->ha_index_first(table->record[0]);
  while (rc != HA_ERR_END_OF_FILE)
  {
    if (thd->killed)
      break;

    if (rc)
      break;
    index_prefix_calc.add();
    rc= table->file->ha_index_next(table->record[0]);
  }
  table->file->ha_end_keyread();
  table->file->ha_index_end();

  rc= (rc == HA_ERR_END_OF_FILE && !thd->killed) ? 0 : 1;

  if (!rc)
    index_prefix_calc.get_avg_frequency();

  DBUG_RETURN(rc);
}


/**
  @brief 
  Collect statistical data for a table

  @param
  thd         The thread handle
  @param
  table       The table to collect statistics on

  @details
  The function collects data for various statistical characteristics on
  the table 'table'. These data is saved in the internal fields that could
  be reached from 'table'. The data is prepared to be saved in the persistent
  statistical table by the function update_statistics_for_table.
  The collected statistical values are not placed in the same fields that
  keep the statistical data used by the optimizer. Therefore, at any time,
  there is no collision between the statistics being collected and the one
  used by the optimizer to look for optimal query execution plans for other
  clients.

  @retval
  0         If the statistics has been successfully collected  
  @retval
  1         Otherwise

  @note
  The function first collects statistical data for statistical characteristics
  to be saved in the statistical tables table_stat and column_stats. To do this
  it performs a full table scan of 'table'. At this scan the function collects
  statistics on each column of the table and count the total number of the
  scanned rows. To calculate the value of 'avg_frequency' for a column the
  function constructs an object of the helper class Count_distinct_field
  (or its derivation). Currently this class cannot count the number of
  distinct values for blob columns. So the value of 'avg_frequency' for
  blob columns is always null.
  After the full table scan the function calls collect_statistics_for_index
  for each table index. The latter performs full index scan for each index.

  @note
  Currently the statistical data is collected indiscriminately for all
  columns/indexes of 'table', for all statistical characteristics.
  TODO. Collect only specified statistical characteristics for specified
  columns/indexes.

  @note
  Currently the process of collecting statistical data is not optimized.
  For example, 'avg_frequency' for a column could be copied from the
  'avg_frequency' collected for an index if this column is used as the
  first component of the index. Min and min values for this column could
  be extracted from the index as well.       
*/

int collect_statistics_for_table(THD *thd, TABLE *table)
{
  int rc;
  Field **field_ptr;
  Field *table_field;
  ha_rows rows= 0;
  handler *file=table->file;
  double sample_fraction= thd->variables.sample_percentage / 100;
  const ha_rows MIN_THRESHOLD_FOR_SAMPLING= 50000;
  DBUG_ENTER("collect_statistics_for_table");

  table->collected_stats->cardinality_is_null= TRUE;
  table->collected_stats->cardinality= 0;

  if (thd->variables.sample_percentage == 0)
  {
    if (file->records() < MIN_THRESHOLD_FOR_SAMPLING)
    {
      sample_fraction= 1;
    }
    else
    {
      sample_fraction= std::fmin(
                  (MIN_THRESHOLD_FOR_SAMPLING + 4096 *
                   log(200 * file->records())) / file->records(), 1);
    }
  }

  for (field_ptr= table->field; *field_ptr; field_ptr++)
  {
    table_field= *field_ptr;   
    if (!table_field->collected_stats)
      continue; 
    table_field->collected_stats->init(thd, table_field);
  }

  restore_record(table, s->default_values);

  /* Perform a full table scan to collect statistics on 'table's columns */
  if (!(rc= file->ha_rnd_init(TRUE)))
  {
    DEBUG_SYNC(table->in_use, "statistics_collection_start");

    while ((rc= file->ha_rnd_next(table->record[0])) != HA_ERR_END_OF_FILE)
    {
      if (thd->killed)
        break;

      if (rc)
        break;

      if (thd_rnd(thd) <= sample_fraction)
      {
        for (field_ptr= table->field; *field_ptr; field_ptr++)
        {
          table_field= *field_ptr;
          if (!table_field->collected_stats)
            continue;
          if ((rc= table_field->collected_stats->add()))
            break;
        }
        if (rc)
          break;
        rows++;
      }
    }
    file->ha_rnd_end();
  }
  rc= (rc == HA_ERR_END_OF_FILE && !thd->killed) ? 0 : 1;

  /* 
    Calculate values for all statistical characteristics on columns and
    and for each field f of 'table' save them in the write_stat structure
    from the Field object for f. 
  */
  if (!rc)
  {
    table->collected_stats->cardinality_is_null= FALSE;
    table->collected_stats->cardinality=
      static_cast<ha_rows>(rows / sample_fraction);
  }

  bitmap_clear_all(table->write_set);
  for (field_ptr= table->field; *field_ptr; field_ptr++)
  {
    table_field= *field_ptr;
    if (!table_field->collected_stats)
      continue;
    bitmap_set_bit(table->write_set, table_field->field_index); 
    if (!rc)
      rc= table_field->collected_stats->finish(thd->mem_root, rows,
                                               sample_fraction);
    else
      table_field->collected_stats->cleanup();
  }
  bitmap_clear_all(table->write_set);

  if (!rc)
  {
    uint key;
    key_map::Iterator it(table->keys_in_use_for_query);

    MY_BITMAP *save_read_set= table->read_set;
    table->read_set= &table->tmp_set;
    bitmap_set_all(table->read_set);
     
    /* Collect statistics for indexes */
    while ((key= it++) != key_map::Iterator::BITMAP_END)
    {
      if ((rc= collect_statistics_for_index(thd, table, key)))
        break;
    }

    table->read_set= save_read_set;
  }

  DBUG_RETURN(rc);          
}


/**
  @brief
  Update statistics for a table in the persistent statistical tables

  @param thd    The thread handle
  @param table  The table to collect statistics on

  @details
  For each statistical table st the function looks for the rows from this
  table that contain statistical data on 'table'. If rows with given 
  statistical characteristics exist they are updated with the new statistical
  values taken from internal structures for 'table'. Otherwise new rows
  with these statistical characteristics are added into st.
  It is assumed that values stored in the statistical tables are found and
  saved by the function collect_statistics_for_table. 

  @retval
  0         If all statistical tables has been successfully updated  
  @retval
  1         Otherwise

  @note
  The function is called when executing the ANALYZE actions for 'table'.
  The function first unlocks the opened table the statistics on which has
  been collected, but does not closes it, so all collected statistical data
  remains in internal structures for 'table'. Then the function opens the
  statistical tables and writes the statistical data for 'table'into them.
  It is not allowed just to open statistical tables for writing when some
  other tables are locked for reading.
  After the statistical tables have been opened they are updated one by one
  with the new statistics on 'table'. Objects of the helper classes
  Table_stat, Column_stat and Index_stat are employed for this. 
  After having been updated the statistical system tables are closed.     
*/

int update_statistics_for_table(THD *thd, TABLE *table)
{
  TABLE_LIST tables[STATISTICS_TABLES];
  uint i;
  int err;
  enum_binlog_format save_binlog_format;
  int rc= 0;
  TABLE *stat_table;
  DBUG_ENTER("update_statistics_for_table");

  DEBUG_SYNC(thd, "statistics_update_start");

  start_new_trans new_trans(thd);

  if (open_stat_tables(thd, tables, TRUE))
  {
    new_trans.restore_old_transaction();
    DBUG_RETURN(0);
  }
   
  /*
    Ensure that no one is reading satistics while we are writing them
    This ensures that statistics is always read consistently
  */
  mysql_mutex_lock(&table->s->LOCK_statistics);

  save_binlog_format= thd->set_current_stmt_binlog_format_stmt();

  /* Update the statistical table table_stats */
  stat_table= tables[TABLE_STAT].table;
  Table_stat table_stat(stat_table, table);
  restore_record(stat_table, s->default_values);
  table_stat.set_key_fields();
  err= table_stat.update_stat();
  if (err)
    rc= 1;

  /* Update the statistical table colum_stats */
  stat_table= tables[COLUMN_STAT].table;
  Column_stat column_stat(stat_table, table);
  for (Field **field_ptr= table->field; *field_ptr; field_ptr++)
  {
    Field *table_field= *field_ptr;
    if (!table_field->collected_stats)
      continue;
    restore_record(stat_table, s->default_values);
    column_stat.set_key_fields(table_field);
    err= column_stat.update_stat();
    if (err && !rc)
      rc= 1;
  }

  /* Update the statistical table index_stats */
  stat_table= tables[INDEX_STAT].table;
  uint key;
  key_map::Iterator it(table->keys_in_use_for_query);
  Index_stat index_stat(stat_table, table);

  while ((key= it++) != key_map::Iterator::BITMAP_END)
  {
    KEY *key_info= table->key_info+key;
    uint key_parts= table->actual_n_key_parts(key_info);
    for (i= 0; i < key_parts; i++)
    {
      restore_record(stat_table, s->default_values);
      index_stat.set_key_fields(key_info, i+1);
      err= index_stat.update_stat();
      if (err && !rc)
        rc= 1;
    }
  }

  tables[TABLE_STAT].table->file->extra(HA_EXTRA_FLUSH);
  tables[COLUMN_STAT].table->file->extra(HA_EXTRA_FLUSH);
  tables[INDEX_STAT].table->file->extra(HA_EXTRA_FLUSH);

  thd->restore_stmt_binlog_format(save_binlog_format);
  if (thd->commit_whole_transaction_and_close_tables())
    rc= 1;

  mysql_mutex_unlock(&table->s->LOCK_statistics);
  new_trans.restore_old_transaction();
  DBUG_RETURN(rc);
}


/**
  @brief
  Read statistics for a table from the persistent statistical tables

  @param
  thd          The thread handle
  @param
  table        The table to read statistics on.
  @param
  stat_tables  The array of TABLE_LIST objects for statistical tables
  @param
  force_reload Flag to require reloading the statistics from the tables
               even if it has been already loaded

  @details
  For each statistical table the function looks for the rows from this
  table that contain statistical data on 'table'. If such rows is found
  the data from statistical columns of it is read into the appropriate
  fields of internal structures for 'table'. Later at the query processing
  this data are supposed to be used by the optimizer. 
  The parameter stat_tables should point to an array of TABLE_LIST
  objects for all statistical tables linked into a list. All statistical
  tables are supposed to be opened.  
  The function is called by read_statistics_for_tables_if_needed().

  @retval
  pointer to object  If data has been successfully read for the table
  @retval
  0                  Otherwise

  @note
  Objects of the helper classes Table_stat, Column_stat and Index_stat
  are employed to read statistical data from the statistical tables. 
  now.        
*/

static
TABLE_STATISTICS_CB*
read_statistics_for_table(THD *thd, TABLE *table,
                          TABLE_LIST *stat_tables, bool force_reload,
                          bool want_histograms)
{
  uint i;
  TABLE *stat_table;
  Field *table_field;
  Field **field_ptr;
  KEY *key_info, *key_info_end;
  TABLE_SHARE *table_share= table->s;
  DBUG_ENTER("read_statistics_for_table");

  if (!force_reload && table_share->stats_cb &&
      (!want_histograms || !table_share->histograms_exists()))
  {
    if (table->stats_cb == table_share->stats_cb)
      DBUG_RETURN(table->stats_cb);		// Use current
    table->update_engine_independent_stats();	// Copy table_share->stats_cb
    DBUG_RETURN(table->stats_cb);
  }

  /*
    Read data into a new TABLE_STATISTICS_CB object and replace
    TABLE_SHARE::stats_cb with this new one once the reading is finished
  */
  TABLE_STATISTICS_CB *new_stats_cb;
  if (!(new_stats_cb= new TABLE_STATISTICS_CB))
    DBUG_RETURN(0);                           /* purecov: inspected */

  if (alloc_engine_independent_statistics(thd, table_share, new_stats_cb))
  {
    /* purecov: begin inspected */
    delete new_stats_cb;
    DBUG_RETURN(0);
    /* purecov: end */
  }

  /* Don't write warnings for internal field conversions */
  Check_level_instant_set check_level_save(thd, CHECK_FIELD_IGNORE);

  /* Read statistics from the statistical table table_stats */
  Table_statistics *read_stats= new_stats_cb->table_stats;
  stat_table= stat_tables[TABLE_STAT].table;
  Table_stat table_stat(stat_table, table);
  table_stat.set_key_fields();
  if (table_stat.get_stat_values(new_stats_cb->table_stats))
    new_stats_cb->stats_available|= TABLE_STAT_TABLE;

  /* Read statistics from the statistical table column_stats */
  stat_table= stat_tables[COLUMN_STAT].table;
  Column_stat column_stat(stat_table, table);
  Column_statistics *column_statistics= new_stats_cb->table_stats->column_stats;
  for (field_ptr= table_share->field;
       *field_ptr;
       field_ptr++, column_statistics++)
  {
    table_field= *field_ptr;
    column_stat.set_key_fields(table_field);
    if (column_stat.get_stat_values(column_statistics,
                                    &new_stats_cb->mem_root,
                                    want_histograms))
        new_stats_cb->stats_available|= TABLE_STAT_COLUMN;
    if (column_statistics->histogram_exists)
    {
      new_stats_cb->histograms_exists_on_disk= 1;
      if (column_statistics->histogram)
        new_stats_cb->stats_available|= TABLE_STAT_HISTOGRAM;
    }
  }

  /* Read statistics from the statistical table index_stats */
  stat_table= stat_tables[INDEX_STAT].table;
  Index_stat index_stat(stat_table, table);
  Index_statistics *index_statistics= new_stats_cb->table_stats->index_stats;
  for (key_info= table_share->key_info,
       key_info_end= key_info + table_share->keys;
       key_info < key_info_end; key_info++, index_statistics++)
  {
    uint key_parts= key_info->ext_key_parts;
    bool found= 0;
    for (i= 0; i < key_parts; i++)
    {
      index_stat.set_key_fields(key_info, i+1);
      found|= index_stat.get_stat_values(index_statistics);
    }
    if (found)
      new_stats_cb->stats_available|= TABLE_STAT_INDEX;

    key_part_map ext_key_part_map= key_info->ext_key_part_map;
    if (key_info->user_defined_key_parts != key_info->ext_key_parts &&
        index_statistics->get_avg_frequency(key_info->user_defined_key_parts) == 0)
    {
      KEY *pk_key_info= table_share->key_info + table_share->primary_key;
      uint k= key_info->user_defined_key_parts;
      uint pk_parts= pk_key_info->user_defined_key_parts;
      ha_rows n_rows= read_stats->cardinality;
      double k_dist= n_rows / index_statistics->get_avg_frequency(k-1);
      uint m= 0;
      Index_statistics *pk_read_stats= (new_stats_cb->table_stats->index_stats +
                                        table_share->primary_key);
      for (uint j= 0; j < pk_parts; j++)
      {
        if (!(ext_key_part_map & 1 << j))
	{
          for (uint l= k; l < k + m; l++)
	  {
            double avg_frequency= pk_read_stats->get_avg_frequency(j-1);
            set_if_smaller(avg_frequency, 1);
            double val= (pk_read_stats->get_avg_frequency(j) /
                         avg_frequency);
	    index_statistics->set_avg_frequency (l, val);
          }
        }
        else
	{
	  double avg_frequency= pk_read_stats->get_avg_frequency(j);
	  index_statistics->set_avg_frequency(k + m, avg_frequency);
	  m++;
        }    
      }      
      for (uint l= k; l < k + m; l++)
      {
        double avg_frequency= index_statistics->get_avg_frequency(l);
        if (avg_frequency == 0 || read_stats->cardinality_is_null)
          avg_frequency= 1;
        else if (avg_frequency > 1)
	{
          avg_frequency/= k_dist;
          set_if_bigger(avg_frequency, 1);
	}
        index_statistics->set_avg_frequency(l, avg_frequency);
      }
    }
  }
  DBUG_RETURN(new_stats_cb);
}


/**
  @brief
  Read statistics for tables from a table list if it is needed

  @param
  thd         The thread handle
  @param
  tables      The tables list for whose tables to read statistics

  @details
  The function first checks whether for any of the tables opened and locked
  for a statement statistics from statistical tables is needed to be read.
  Then, if so, it opens system statistical tables for read and reads
  the statistical data from them for those tables from the list for which it
  makes sense. Then the function closes system statistical tables.

  @retval
  0       Statistics for tables was successfully read  
  @retval
  1       Otherwise
*/

int read_statistics_for_tables_if_needed(THD *thd, TABLE_LIST *tables)
{
  switch (thd->lex->sql_command) {
  case SQLCOM_SELECT:
  case SQLCOM_INSERT:
  case SQLCOM_INSERT_SELECT:
  case SQLCOM_UPDATE:
  case SQLCOM_UPDATE_MULTI:
  case SQLCOM_DELETE:
  case SQLCOM_DELETE_MULTI:
  case SQLCOM_REPLACE:
  case SQLCOM_REPLACE_SELECT:
  case SQLCOM_CREATE_TABLE:
  case SQLCOM_SET_OPTION:
  case SQLCOM_DO:
    return read_statistics_for_tables(thd, tables, 0);
  default:
    return 0;
  }
}


/*
  Update TABLE field and key objects with pointers to
  the current statistical data in table->stats_cb
*/


void TABLE_STATISTICS_CB::update_stats_in_table(TABLE *table)
{
  DBUG_ASSERT(table->stats_cb == this);

  /*
    Table_statistics doesn't need to be updated: set_statistics_for_table()
    sets TABLE::used_stat_records from table->stats_cb.table_stats.cardinality
  */

  KEY *key_info= table->key_info;
  KEY *key_info_end= key_info + table->s->keys;
  Index_statistics *index_stats= table_stats->index_stats;

  for ( ; key_info < key_info_end; key_info++, index_stats++)
    key_info->read_stats= index_stats;

  Field **field_ptr= table->field;
  Column_statistics *column_stats= table_stats->column_stats;

  for ( ; *field_ptr; field_ptr++, column_stats++)
    (*field_ptr)->read_stats= column_stats;
  /* Mark that stats are now usable */
  table->stats_is_read= (table->stats_cb->stats_available !=
                         TABLE_STAT_NO_STATS);
}


int
read_statistics_for_tables(THD *thd, TABLE_LIST *tables, bool force_reload)
{
  int rc= 0;
  TABLE_LIST stat_tables[STATISTICS_TABLES];
  bool found_stat_table= false;
  bool statistics_for_tables_is_needed= false;
  bool want_histograms= thd->variables.optimizer_use_condition_selectivity > 3;
  DBUG_ENTER("read_statistics_for_tables");

  if (thd->bootstrap || thd->variables.use_stat_tables == NEVER || !tables)
    DBUG_RETURN(0);

  for (TABLE_LIST *tl= tables; tl; tl= tl->next_global)
  {
    TABLE *table= tl->table;
    TABLE_SHARE *table_share;

    /* Skip tables that can't have statistics. */
    if (tl->is_view_or_derived() || !table || !(table_share= table->s))
      continue;
    /* Skip temporary tables */
    if (table_share->tmp_table != NO_TMP_TABLE)
      continue;

    if (table_share->table_category == TABLE_CATEGORY_USER)
    {
      /* Force reloading means we always read all stats tables. */
      if (force_reload || !table_share->stats_cb)
      {
        statistics_for_tables_is_needed= true;
        continue;
      }

      /* Stats versions don't match, take a reference under a mutex. */
      if (table->stats_cb != table_share->stats_cb)
      {
        table->update_engine_independent_stats();
        table->stats_cb->update_stats_in_table(table);
      }
      /*
         We need to read histograms if they exist but have not yet been
         loaded into memory.
      */
      if (want_histograms &&
          table->stats_cb->histograms_exists() &&
          !(table->stats_cb->stats_available & TABLE_STAT_HISTOGRAM))
      {
        statistics_for_tables_is_needed= true;
      }
    }
    else if (is_stat_table(&tl->db, &tl->alias))
      found_stat_table= true;
  }

  DEBUG_SYNC(thd, "statistics_read_start");

  /*
    Do not read statistics for any query that explicity involves
    statistical tables, failure to to do so we may end up
    in a deadlock.
  */
  if (found_stat_table || !statistics_for_tables_is_needed)
    DBUG_RETURN(0);

  start_new_trans new_trans(thd);

  if (open_stat_tables(thd, stat_tables, FALSE))
  {
    rc= 1;
    goto end;
  }

  for (TABLE_LIST *tl= tables; tl; tl= tl->next_global)
  {
    TABLE *table= tl->table;
    TABLE_SHARE *table_share;

    /* Skip tables that can't have statistics. */
    if (tl->is_view_or_derived() || !table || !(table_share= table->s) ||
        table_share->tmp_table != NO_TMP_TABLE ||
        table_share->table_category != TABLE_CATEGORY_USER)
      continue;

    if (force_reload || !table_share->stats_cb ||
        table->stats_cb != table_share->stats_cb ||
        (want_histograms && table->stats_cb->histograms_exists() &&
         !(table->stats_cb->stats_available & TABLE_STAT_HISTOGRAM)))
    {
      TABLE_STATISTICS_CB *stats_cb;
      DEBUG_SYNC(thd, "read_statistics_for_table_start1");
      DEBUG_SYNC(thd, "read_statistics_for_table_start2");

      /*
        The following lock is here to ensure that if a lot of threads are
        accessing the table at the same time after a ANALYZE TABLE,
        only one thread is loading the data from the the stats tables
        and the others threads are reusing the loaded data.
      */
      mysql_mutex_lock(&table_share->LOCK_statistics);
      if (!(stats_cb= read_statistics_for_table(thd, table, stat_tables,
                                                force_reload, want_histograms)))
      {
        /* purecov: begin inspected */
        mysql_mutex_unlock(&table_share->LOCK_statistics);
        continue;
        /* purecov: end */
      }

      if (stats_cb->unused())
      {
        /* New object created, update share to use it */
        table_share->update_engine_independent_stats(stats_cb);
        table->update_engine_independent_stats();
      }
      mysql_mutex_unlock(&table_share->LOCK_statistics);
      table->stats_cb->update_stats_in_table(table);
    }
  }

  thd->commit_whole_transaction_and_close_tables();

end:
  new_trans.restore_old_transaction();
  DBUG_RETURN(rc);
}


/**
  @brief
  Delete statistics on a table from all statistical tables

  @param
  thd         The thread handle
  @param
  db          The name of the database the table belongs to
  @param
  tab         The name of the table whose statistics is to be deleted

  @details
  The function delete statistics on the table called 'tab' of the database
  'db' from all statistical tables: table_stats, column_stats, index_stats.

  @retval
  0         If all deletions are successful or we couldn't open statistics table
  @retval
  1         Otherwise

  @note
  The function is called when executing the statement DROP TABLE 'tab'.
*/

int delete_statistics_for_table(THD *thd, const LEX_CSTRING *db,
                                const LEX_CSTRING *tab)
{
  int err;
  enum_binlog_format save_binlog_format;
  TABLE *stat_table;
  TABLE_LIST tables[STATISTICS_TABLES];
  Open_tables_backup open_tables_backup;
  int rc= 0;
  DBUG_ENTER("delete_statistics_for_table");

  start_new_trans new_trans(thd);

  if (open_stat_tables(thd, tables, TRUE))
  {
    new_trans.restore_old_transaction();
    DBUG_RETURN(0);
  }

  save_binlog_format= thd->set_current_stmt_binlog_format_stmt();

  /* Delete statistics on table from the statistical table index_stats */
  stat_table= tables[INDEX_STAT].table;
  Index_stat index_stat(stat_table, db, tab);
  index_stat.set_full_table_name();
  while (index_stat.find_next_stat_for_prefix(2))
  {
    err= index_stat.delete_stat();
    if (err & !rc)
      rc= 1;
  }

  /* Delete statistics on table from the statistical table column_stats */
  stat_table= tables[COLUMN_STAT].table;
  Column_stat column_stat(stat_table, db, tab);
  column_stat.set_full_table_name();
  while (column_stat.find_next_stat_for_prefix(2))
  {
    err= column_stat.delete_stat();
    if (err & !rc)
      rc= 1;
  }
   
  /* Delete statistics on table from the statistical table table_stats */
  stat_table= tables[TABLE_STAT].table;
  Table_stat table_stat(stat_table, db, tab);
  table_stat.set_key_fields();
  if (table_stat.find_stat())
  {
    err= table_stat.delete_stat();
    if (err & !rc)
      rc= 1;
  }

  err= del_global_table_stat(thd, db, tab);
  if (err & !rc)
      rc= 1;

  tables[TABLE_STAT].table->file->extra(HA_EXTRA_FLUSH);
  tables[COLUMN_STAT].table->file->extra(HA_EXTRA_FLUSH);
  tables[INDEX_STAT].table->file->extra(HA_EXTRA_FLUSH);

  thd->restore_stmt_binlog_format(save_binlog_format);
  thd->commit_whole_transaction_and_close_tables();

  new_trans.restore_old_transaction();
  DBUG_RETURN(rc);
}


/**
  @brief
  Delete statistics on a column of the specified table

  @param thd         The thread handle
  @param tab         The table the column belongs to
  @param col         The field of the column whose statistics is to be deleted

  @details
  The function delete statistics on the column 'col' belonging to the table 
  'tab' from the statistical table column_stats. 

  @retval 0  If all deletions are successful or we couldn't open statistics table
  @retval 1  Otherwise

  @note
  The function is called when dropping a table column  or when changing
  the definition of this column.
*/

int delete_statistics_for_column(THD *thd, TABLE *tab, Field *col)
{
  int err;
  enum_binlog_format save_binlog_format;
  TABLE *stat_table;
  TABLE_LIST tables;
  int rc= 0;
  DBUG_ENTER("delete_statistics_for_column");
   
  start_new_trans new_trans(thd);

  if (open_stat_table_for_ddl(thd, &tables, &stat_table_name[1]))
  {
    new_trans.restore_old_transaction();
    DBUG_RETURN(0);                             // Not an error
  }

  save_binlog_format= thd->set_current_stmt_binlog_format_stmt();

  stat_table= tables.table;
  Column_stat column_stat(stat_table, tab);
  column_stat.set_key_fields(col);
  if (column_stat.find_stat())
  {
    err= column_stat.delete_stat();
    if (err)
      rc= 1;
  }

  column_stat.flush();
  thd->restore_stmt_binlog_format(save_binlog_format);
  if (thd->commit_whole_transaction_and_close_tables())
    rc= 1;

  new_trans.restore_old_transaction();
  DBUG_RETURN(rc);
}


/**
   Generate tempoary column or index name for renames
*/

static LEX_CSTRING *generate_tmp_name(LEX_CSTRING *to, uint counter)
{
  char *res=int10_to_str(counter, strmov((char*) to->str, "#sql_tmp_name#"),
                         10);
  /*
    Include an end zero in the tmp name to avoid any possible conflict
    with existing column names.
   */
  to->length= (size_t) (res - to->str) + 1;
  return to;
}


/**
  Rename a set of columns in the statistical table column_stats

  @param thd         The thread handle
  @param tab         The table the column belongs to
  @param fields      List of fields and names to be renamed

  @details
  The function replaces the names of the columns in fields that belongs
  to the table 'tab' in the statistical table column_stats.

  @retval 0   If update was successful, tmp table or could not open stat table
  @retval -1  Commit failed
  @retval >0  Error number from engine

  @note
  The function is called when executing any statement that renames a column,
  but does not change the column definition.
*/

int rename_columns_in_stat_table(THD *thd, TABLE *tab,
                                 List<Alter_info::RENAME_COLUMN_STAT_PARAMS>
                                 *fields)
{
  int err;
  enum_binlog_format save_binlog_format;
  TABLE *stat_table;
  TABLE_LIST tables;
  int rc= 0;
  uint duplicate_counter= 0;
  uint org_elements= fields->elements+1;
  List_iterator<Alter_info::RENAME_COLUMN_STAT_PARAMS> it(*fields);
  char tmp_name_buffer[32];
  LEX_CSTRING tmp_name= {tmp_name_buffer, 0};
  DBUG_ENTER("rename_column_in_stat_tables");

  if (tab->s->tmp_table != NO_TMP_TABLE)
    DBUG_RETURN(0);

  start_new_trans new_trans(thd);

  if (open_stat_table_for_ddl(thd, &tables, &stat_table_name[1]))
  {
    new_trans.restore_old_transaction();
    DBUG_RETURN(0);
  }

  save_binlog_format= thd->set_current_stmt_binlog_format_stmt();

  /* Rename column in the statistical table table_stat */

  stat_table= tables.table;

  /* Loop until fields is empty or previous round did nothing */
  while (!fields->is_empty() && fields->elements != org_elements)
  {
    Alter_info::RENAME_COLUMN_STAT_PARAMS *field;
    org_elements= fields->elements;
    it.rewind();
    while ((field= it++))
    {
      Column_stat column_stat(stat_table, tab);
      LEX_CSTRING *from_name;
      from_name= (!field->duplicate_counter ?
                  &field->field->field_name :
                  generate_tmp_name(&tmp_name,
                                    field->duplicate_counter));
      column_stat.set_key_fields(from_name);
      if (column_stat.find_stat())
      {
        err= column_stat.update_column_key_part(field->name);
        if (likely(err != HA_ERR_FOUND_DUPP_KEY))
          it.remove();
        else if (!field->duplicate_counter)
        {
          /*
            This is probably an ALTER TABLE of type rename a->b, b->a
            Rename the column to a temporary name
          */
          LEX_CSTRING *new_name=
            generate_tmp_name(&tmp_name, ++duplicate_counter);
          field->duplicate_counter= duplicate_counter;

          if ((err= column_stat.update_column(new_name)))
          {
            if (likely(err != HA_ERR_FOUND_DUPP_KEY))
            {
              DBUG_ASSERT(0);
              it.remove();                      // Unknown error, ignore column
            }
            else
            {
              /*
                The only way this could happen is if the table has a column
                with same name as the temporary column name, probably from a
                failed alter table.
                Remove the conflicting row and update it again.
              */
              if (!column_stat.find_stat())
                DBUG_ASSERT(0);
              else if (column_stat.delete_stat())
                DBUG_ASSERT(0);
              else
              {
                column_stat.set_key_fields(from_name);
                if (!column_stat.find_stat())
                  DBUG_ASSERT(0);
                else if (column_stat.update_column_key_part(&tmp_name))
                  DBUG_ASSERT(0);
              }
            }
          }
        }
      }
      else /* column_stat.find_stat() */
      {
        /* Statistics for the field did not exists */
        it.remove();
      }
    }
  }

  if (!fields->is_empty())
  {
    /*
      All unhandled renamed fields has now a temporary name.
      Remove all conflicing rows and rename the temporary name to
      the final name.
    */

    Alter_info::RENAME_COLUMN_STAT_PARAMS *field;
    it.rewind();
    while ((field= it++))
    {
      Column_stat column_stat(stat_table, tab);
      DBUG_ASSERT(field->duplicate_counter);

      /* Remove the conflicting row */
      column_stat.set_key_fields(field->name);
      if (column_stat.find_stat())
      {
        int err __attribute__((unused));
        err= column_stat.delete_stat();
        DBUG_ASSERT(err == 0);
      }

      /* Restore saved row with old statistics to new name */
      column_stat.
        set_key_fields(generate_tmp_name(&tmp_name,
                                         field->duplicate_counter));
      if (column_stat.find_stat())
      {
        int err __attribute__((unused));
        err= column_stat.update_column_key_part(field->name);
        DBUG_ASSERT(err == 0);
      }
      else
      {
        DBUG_ASSERT(0);
      }
    }
  }

  stat_table->file->extra(HA_EXTRA_FLUSH);
  thd->restore_stmt_binlog_format(save_binlog_format);
  if (thd->commit_whole_transaction_and_close_tables())
    rc= -1;

  new_trans.restore_old_transaction();
  DBUG_RETURN(rc);
}


/**
  @brief
  Delete statistics on an index of the specified table

  @param thd         The thread handle
  @param tab         The table the index belongs to
  @param key_info    The descriptor of the index whose statistics is to be
                     deleted
  @param ext_prefixes_only  Delete statistics only on the index prefixes
                     extended by the components of the primary key

  @details
  The function delete statistics on the index  specified by 'key_info'
  defined on the table 'tab' from the statistical table index_stats.

  @retval 0  If all deletions are successful or we couldn't open statistics
             table
  @retval 1  Otherwise

  @note
  The function is called when dropping an index, or dropping/changing the
   definition of a column used in the definition of the index. 
*/

int delete_statistics_for_index(THD *thd, TABLE *tab, KEY *key_info,
                                bool ext_prefixes_only)
{
  int err;
  enum_binlog_format save_binlog_format;
  TABLE *stat_table;
  TABLE_LIST tables;
  int rc= 0;
  DBUG_ENTER("delete_statistics_for_index");
   
  start_new_trans new_trans(thd);

  if (open_stat_table_for_ddl(thd, &tables, &stat_table_name[2]))
  {
    new_trans.restore_old_transaction();
    DBUG_RETURN(0);                             // Not an error
  }

  save_binlog_format= thd->set_current_stmt_binlog_format_stmt();

  stat_table= tables.table;
  Index_stat index_stat(stat_table, tab);
  if (!ext_prefixes_only)
  {
    index_stat.set_index_prefix_key_fields(key_info);
    while (index_stat.find_next_stat_for_prefix(3))
    {
      err= index_stat.delete_stat();
      if (err && !rc)
        rc= 1;
    }
  }
  else
  {
    for (uint i= key_info->user_defined_key_parts; i < key_info->ext_key_parts; i++)
    {
      index_stat.set_key_fields(key_info, i+1);
      if (index_stat.find_next_stat_for_prefix(4))
      {
        err= index_stat.delete_stat();
        if (err && !rc)
          rc= 1;
      }
    }
  }

  err= del_global_index_stat(thd, tab, key_info);
  if (err && !rc)
    rc= 1;

  /* Make change permanent and avoid 'table is marked as crashed' errors */
  index_stat.flush();

  thd->restore_stmt_binlog_format(save_binlog_format);
  if (thd->commit_whole_transaction_and_close_tables())
    rc= 1;

  new_trans.restore_old_transaction();
  DBUG_RETURN(rc);
}


/**
  Rename a set of indexes in the statistical table index_stats

  @param thd         The thread handle
  @param tab         The table the indexes belongs to
  @param fields      List of indexes to be renamed

  @details
  The function replaces the names of the indexe in fields that belongs
  to the table 'tab' in the statistical table index_stats.

  @retval 0   If update was successful, tmp table or could not open stat table
  @retval -1  Commit failed
  @retval >0  Error number from engine

  @note
  The function is called when executing any statement that renames a column,
  but does not change the column definition.
*/

int rename_indexes_in_stat_table(THD *thd, TABLE *tab,
                                 List<Alter_info::RENAME_INDEX_STAT_PARAMS>
                                 *indexes)
{
  int err;
  enum_binlog_format save_binlog_format;
  TABLE *stat_table;
  TABLE_LIST tables;
  int rc= 0;
  uint duplicate_counter= 0;
  List_iterator<Alter_info::RENAME_INDEX_STAT_PARAMS> it(*indexes);
  Alter_info::RENAME_INDEX_STAT_PARAMS *index;
  char tmp_name_buffer[32];
  LEX_CSTRING tmp_name= {tmp_name_buffer, 0};
  DBUG_ENTER("rename_indexes_in_stat_tables");

  if (tab->s->tmp_table != NO_TMP_TABLE)
    DBUG_RETURN(0);

  start_new_trans new_trans(thd);

  if (open_stat_table_for_ddl(thd, &tables, &stat_table_name[2]))
  {
    new_trans.restore_old_transaction();
    DBUG_RETURN(0);
  }

  save_binlog_format= thd->set_current_stmt_binlog_format_stmt();

  /* Rename index in the statistical table index_stat */

  stat_table= tables.table;

  /*
    Loop over all indexes and rename to new name or temp name in case of
    conflicts
  */

  while ((index= it++))
  {
    Index_stat index_stat(stat_table, tab);
    uint found= 0;

    /* We have to make a loop as one index may have many entries */
    for (;;)
    {
      index_stat.set_index_prefix_key_fields(index->key);
      if (!index_stat.find_next_stat_for_prefix(3))
        break;
      index_stat.store_record_for_update();
      err= index_stat.update_index_name(index->name);

      if (unlikely(err == HA_ERR_FOUND_DUPP_KEY))
      {
        /*
          This is probably an ALTER TABLE of type rename a->b, b->a
          Rename the column to a temporary name
        */
        if (!found++)
          ++duplicate_counter;
        index->duplicate_counter= duplicate_counter;
        index->usage_count++;
        if ((err= index_stat.update_index_name(generate_tmp_name(&tmp_name, duplicate_counter))))
        {
          if (err != HA_ERR_FOUND_DUPP_KEY)
          {
            DBUG_ASSERT(0);
          }
          else
          {
            /*
              The only way this could happen is if the table has an index
              with same name as the temporary column index, probably from a
              failed alter table.
              Remove the conflicting row and update it again.
            */
            if (!index_stat.find_stat())
              DBUG_ASSERT(0);
            else if (index_stat.delete_stat())
              DBUG_ASSERT(0);
            else
            {
              index_stat.set_index_prefix_key_fields(index->key);
              if (!index_stat.find_stat())
                DBUG_ASSERT(0);
              else
              {
                index_stat.store_record_for_update();
                if (index_stat.update_index_name(&tmp_name))
                  DBUG_ASSERT(0);
              }
            }
          }
        }
      }
    }
    if (!found)
      it.remove();                              // All renames succeded
  }

  if (!indexes->is_empty())
  {
    /*
      All unhandled renamed index has now a temporary name.
      Remove all conflicing rows and rename the temporary name to
      the final name.
    */

    Alter_info::RENAME_INDEX_STAT_PARAMS *index;
    it.rewind();
    Index_stat index_stat(stat_table, tab);
    stat_table->file->ha_index_init(index_stat.stat_key_idx, 0);

    while ((index= it++))
    {
      int err __attribute__((unused));

      /* Remove the conflicting rows */
      index_stat.set_index_prefix_key_fields(index->key);
      index_stat.set_index_name(index->name);

      if (index_stat.find_next_stat_for_prefix_with_next(3))
      {
        do
        {
          err= index_stat.delete_stat();
          DBUG_ASSERT(err == 0);
        }
        while (index_stat.read_next() == 0);
      }

      /* Restore saved row with old statistics to new name */
      index_stat.set_index_name(generate_tmp_name(&tmp_name,
                                                  index->duplicate_counter));
      if (!index_stat.find_stat_with_next())
        DBUG_ASSERT(0);
      else
      {
        uint updated= 0;
        do
        {
          index_stat.store_record_for_update();
          err= index_stat.update_index_name(index->name);
          DBUG_ASSERT(err == 0);
        } while (++updated < index->usage_count && index_stat.read_next() == 0);
      }
    }
    stat_table->file->ha_index_end();
  }

  stat_table->file->extra(HA_EXTRA_FLUSH);
  thd->restore_stmt_binlog_format(save_binlog_format);
  if (thd->commit_whole_transaction_and_close_tables())
    rc= -1;

  new_trans.restore_old_transaction();
  DBUG_RETURN(rc);
}


/**
  @brief
  Rename a table in all statistical tables

  @param
  thd         The thread handle
  @param
  db          The name of the database the table belongs to
  @param
  tab         The name of the table to be renamed in statistical tables
  @param
  new_tab     The new name of the table

  @details
  The function replaces the name of the table 'tab' from the database 'db' 
  for 'new_tab' in all all statistical tables: table_stats, column_stats,
  index_stats.

  @retval
  0         If all updates of the table name are successful
  @retval
  1         Otherwise

  @note
  The function is called when executing any statement that renames a table
*/

int rename_table_in_stat_tables(THD *thd, const LEX_CSTRING *db,
                                const LEX_CSTRING *tab,
                                const LEX_CSTRING *new_db,
                                const LEX_CSTRING *new_tab)
{
  int err;
  enum_binlog_format save_binlog_format;
  TABLE *stat_table;
  TABLE_LIST tables[STATISTICS_TABLES];
  int rc= 0;
  DBUG_ENTER("rename_table_in_stat_tables");
   
  start_new_trans new_trans(thd);

  if (open_stat_tables(thd, tables, TRUE))
  {
    new_trans.restore_old_transaction();
    DBUG_RETURN(0);
  }

  save_binlog_format= thd->set_current_stmt_binlog_format_stmt();

  /* Rename table in the statistical table index_stats */
  stat_table= tables[INDEX_STAT].table;
  Index_stat index_stat(stat_table, db, tab);
  index_stat.set_full_table_name();

  Stat_table_write_iter index_iter(&index_stat);
  if (index_iter.init(2))
    rc= 1;
  while (!index_iter.get_next_row())
  {
    err= index_stat.update_table_name_key_parts(new_db, new_tab);
    if (err & !rc)
      rc= 1;
    index_stat.set_full_table_name();
  }
  index_iter.cleanup();

  /* Rename table in the statistical table column_stats */
  stat_table= tables[COLUMN_STAT].table;
  Column_stat column_stat(stat_table, db, tab);
  column_stat.set_full_table_name();
  Stat_table_write_iter column_iter(&column_stat);
  if (column_iter.init(2))
    rc= 1;
  while (!column_iter.get_next_row())
  {
    err= column_stat.update_table_name_key_parts(new_db, new_tab);
    if (err & !rc)
      rc= 1;
    column_stat.set_full_table_name();
  }
  column_iter.cleanup();
   
  /* Rename table in the statistical table table_stats */
  stat_table= tables[TABLE_STAT].table;
  Table_stat table_stat(stat_table, db, tab);
  table_stat.set_key_fields();
  if (table_stat.find_stat())
  {
    err= table_stat.update_table_name_key_parts(new_db, new_tab);
    if (err & !rc)
      rc= 1;
  }

  tables[TABLE_STAT].table->file->extra(HA_EXTRA_FLUSH);
  tables[COLUMN_STAT].table->file->extra(HA_EXTRA_FLUSH);
  tables[INDEX_STAT].table->file->extra(HA_EXTRA_FLUSH);

  thd->restore_stmt_binlog_format(save_binlog_format);
  if (thd->commit_whole_transaction_and_close_tables())
    rc= 1;

  new_trans.restore_old_transaction();
  DBUG_RETURN(rc);
}


/**
  @brief
  Set statistics for a table that will be used by the optimizer 

  @param
  thd         The thread handle
  @param
  table       The table to set statistics for 

  @details
  Depending on the value of thd->variables.use_stat_tables 
  the function performs the settings for the table that will control
  from where the statistical data used by the optimizer will be taken.
*/

void set_statistics_for_table(THD *thd, TABLE *table)
{
  TABLE_STATISTICS_CB *stats_cb= table->stats_cb;

  Table_statistics *read_stats= stats_cb ? stats_cb->table_stats : 0;
  table->used_stat_records= 
    (!check_eits_preferred(thd) ||
     !table->stats_is_read || !read_stats || read_stats->cardinality_is_null) ?
    table->file->stats.records : read_stats->cardinality;

  /*
    For partitioned table, EITS statistics is based on data from all partitions.

    On the other hand, Partition Pruning figures which partitions will be
    accessed and then computes the estimate of rows in used_partitions.

    Use the estimate from Partition Pruning as it is typically more precise.
    Ideally, EITS should provide per-partition statistics but this is not
    implemented currently.
  */
#ifdef WITH_PARTITION_STORAGE_ENGINE
    if (table->part_info)
      table->used_stat_records= table->file->stats.records;
#endif

  KEY *key_info, *key_info_end;
  for (key_info= table->key_info, key_info_end= key_info+table->s->keys;
       key_info < key_info_end; key_info++)
  {
    key_info->is_statistics_from_stat_tables=
      (check_eits_preferred(thd) &&
       table->stats_is_read &&
       key_info->read_stats->avg_frequency_is_inited() &&
       key_info->read_stats->get_avg_frequency(0) > 0.5);
  }
}


/**
  @brief
  Get the average frequency for a column 

  @param
  field       The column whose average frequency is required

  @retval
  The required average frequency
*/

double get_column_avg_frequency(Field * field)
{
  double res;
  TABLE *table= field->table;

  /* 
    Statistics is shared by table instances and  is accessed through
    the table share. If table->s->field is not set for 'table', then
    no column statistics is available for the table .
  */
  if (!table->s->field)
  {
    res= (double)table->stat_records();
    return res;
  }
 
  Column_statistics *col_stats= field->read_stats;

  if (!col_stats)
    res= (double)table->stat_records();
  else
    res= col_stats->get_avg_frequency();
  return res;
} 


/**
  @brief
  Estimate the number of rows in a column range using data from stat tables 

  @param
  field       The column whose range cardinality is to be estimated
  @param
  min_endp    The left end of the range whose cardinality is required 
  @param
  max_endp    The right end of the range whose cardinality is required 
  @param
  range_flag  The range flags

  @details
  The function gets an estimate of the number of rows in a column range
  using the statistical data from the table column_stats.

  @retval
  - The required estimate of the rows in the column range
  - If there is some kind of error, this function should return DBL_MAX (and
    not HA_POS_ERROR as that is an integer constant).

*/

double get_column_range_cardinality(Field *field,
                                    key_range *min_endp,
                                    key_range *max_endp,
                                    uint range_flag)
{
  double res;
  TABLE *table= field->table;
  Column_statistics *col_stats= field->read_stats;
  double tab_records= (double)table->stat_records();

  if (!col_stats)
    return tab_records;
  /*
    Use statistics for a table only when we have actually read
    the statistics from the stat tables. For example due to
    chances of getting a deadlock we disable reading statistics for
    a table.
  */

  if (!table->stats_is_read)
    return tab_records;

  THD *thd= table->in_use;
  double col_nulls= tab_records * col_stats->get_nulls_ratio();

  double col_non_nulls= tab_records - col_nulls;

  bool nulls_incl= field->null_ptr && min_endp && min_endp->key[0] &&
                   !(range_flag & NEAR_MIN);

  if (col_non_nulls < 1)
  {
    if (nulls_incl)
      res= col_nulls;
    else
      res= 0;
  }
  else if (min_endp && max_endp && min_endp->length == max_endp->length &&
           !memcmp(min_endp->key, max_endp->key, min_endp->length))
  { 
    if (nulls_incl)
    {
      /* This is null single point range */
      res= col_nulls;
    }
    else
    {
      double avg_frequency= col_stats->get_avg_frequency();
      res= avg_frequency;   
      if (avg_frequency > 1.0 + 0.000001 && 
          col_stats->min_max_values_are_provided())
      {
        Histogram_base *hist = col_stats->histogram;
        if (hist && hist->is_usable(thd))
        {
          res= col_non_nulls * 
	       hist->point_selectivity(field, min_endp,
                                       avg_frequency / col_non_nulls);
        }
      }
      else if (avg_frequency == 0.0)
      {
        /* This actually means there is no statistics data */
        res= tab_records;
      }
    }
  }  
  else 
  {
    if (col_stats->min_max_values_are_provided())
    {
      Histogram_base *hist= col_stats->histogram;
      double avg_frequency= col_stats->get_avg_frequency();
      double sel;
      if (hist && hist->is_usable(thd))
      {
        sel= hist->range_selectivity(field, min_endp, max_endp,
                                     avg_frequency / col_non_nulls);
        res= col_non_nulls * sel;
      }
      else
      {
        double min_mp_pos, max_mp_pos;
        if (min_endp && !(field->null_ptr && min_endp->key[0]))
        {
          store_key_image_to_rec(field, (uchar *) min_endp->key,
                                 field->key_length());
          min_mp_pos=
              field->pos_in_interval(col_stats->min_value, col_stats->max_value);
        }
        else
          min_mp_pos= 0.0;
        if (max_endp)
        {
          store_key_image_to_rec(field, (uchar *) max_endp->key,
                                 field->key_length());
          max_mp_pos=
              field->pos_in_interval(col_stats->min_value, col_stats->max_value);
        }
        else
          max_mp_pos= 1.0;

        sel = (max_mp_pos - min_mp_pos);
        res= col_non_nulls * sel;
        set_if_bigger(res, avg_frequency);
      }
    }
    else
      res= col_non_nulls;
    if (nulls_incl)
      res+= col_nulls;
  }
  return res;
}

/*
  Estimate selectivity of "col=const" using a histogram
  
  @param field    the field to estimate its selectivity.

  @param endpoint The constant

  @param avg_sel  Average selectivity of condition "col=const" in this table.
                  It is calcuated as (#non_null_values / #distinct_values).
  
  @return
     Expected condition selectivity (a number between 0 and 1)

  @notes 
     [re_zero_length_buckets] If a bucket with zero value-length is in the
     middle of the histogram, we will not have min==max. Example: suppose, 
     pos_value=0x12, and the histogram is:

           #n  #n+1 #n+2                 
      ... 0x10 0x12 0x12 0x14 ...
                      |
                      +------------- bucket with zero value-length
    
      Here, we will get min=#n+1, max=#n+2, and use the multi-bucket formula.
     
      The problem happens at the histogram ends. if pos_value=0, and the
      histogram is:

      0x00 0x10 ...

      then min=0, max=0. This means pos_value is contained within bucket #0,
      but on the other hand, histogram data says that the bucket has only one
      value.
*/

double Histogram_binary::point_selectivity(Field *field, key_range *endpoint,
                                           double avg_sel)
{
  double sel;
  Column_statistics *col_stats= field->read_stats;
  store_key_image_to_rec(field, (uchar *) endpoint->key,
                         field->key_length());
  double pos= field->pos_in_interval(col_stats->min_value,
                                     col_stats->max_value);
  /* Find the bucket that contains the value 'pos'. */
  uint min= find_bucket(pos, TRUE);
  uint pos_value= (uint) (pos * prec_factor());

  /* Find how many buckets this value occupies */
  uint max= min;
  while (max + 1 < get_width() && get_value(max + 1) == pos_value)
    max++;
  
  /*
    A special case: we're looking at a single bucket, and that bucket has
    zero value-length. Use the multi-bucket formula (attempt to use
    single-bucket formula will cause divison by zero).

    For more details see [re_zero_length_buckets] above.
  */
  if (max == min && get_value(max) == ((max==0)? 0 : get_value(max-1)))
    max++;

  if (max > min)
  {
    /*
      The value occupies multiple buckets. Use start_bucket ... end_bucket as
      selectivity.
    */
    double bucket_sel= 1.0/(get_width() + 1);  
    sel= bucket_sel * (max - min + 1);
  }
  else
  {
    /*
      The value 'pos' fits within one single histogram bucket.

      We also have avg_sel which is per-table average selectivity of col=const.
      If there are popular values, this may be larger than one bucket, so 
      cap the returned number by the selectivity of one bucket.
    */
    double avg_bucket_width= 1.0 / (get_width() + 1);

    sel= MY_MIN(avg_bucket_width, avg_sel);
  }
  return sel;
}


double Histogram_binary::range_selectivity(Field *field,
                                           key_range *min_endp,
                                           key_range *max_endp,
                                           double avg_sel)
{
  double sel, min_mp_pos, max_mp_pos;
  Column_statistics *col_stats= field->read_stats;

  if (min_endp && !(field->null_ptr && min_endp->key[0]))
  {
    store_key_image_to_rec(field, (uchar *) min_endp->key,
                           field->key_length());
    min_mp_pos=
        field->pos_in_interval(col_stats->min_value, col_stats->max_value);
  }
  else
    min_mp_pos= 0.0;
  if (max_endp)
  {
    store_key_image_to_rec(field, (uchar *) max_endp->key,
                           field->key_length());
    max_mp_pos=
        field->pos_in_interval(col_stats->min_value, col_stats->max_value);
  }
  else
    max_mp_pos= 1.0;

  double bucket_sel= 1.0 / (get_width() + 1);
  uint min= find_bucket(min_mp_pos, TRUE);
  uint max= find_bucket(max_mp_pos, FALSE);
  sel= bucket_sel * (max - min + 1);

  set_if_bigger(sel, avg_sel);
  return sel;
}

/*
  Check whether the table is one of the persistent statistical tables.
*/
bool is_stat_table(const LEX_CSTRING *db, LEX_CSTRING *table)
{
  DBUG_ASSERT(db->str && table->str);

  if (!my_strcasecmp(table_alias_charset, db->str, MYSQL_SCHEMA_NAME.str))
  {
    for (uint i= 0; i < STATISTICS_TABLES; i ++)
    {
      if (!my_strcasecmp(table_alias_charset, table->str, stat_table_name[i].str))
        return true;
    }
  }
  return false;
}

/*
  Check wheter we can use EITS statistics for a field or not

  TRUE : Use EITS for the columns
  FALSE: Otherwise
*/

bool is_eits_usable(Field *field)
{
  Column_statistics* col_stats= field->read_stats;
  
  // check if column_statistics was allocated for this field
  if (!col_stats || !field->table->stats_is_read)
    return false;

  /*
    (1): checks if we have EITS statistics for a particular column
    (2): Don't use EITS for GEOMETRY columns
    (3): Disabling reading EITS statistics for columns involved in the
         partition list of a table. We assume the selecticivity for
         such columns would be handled during partition pruning.
  */

  return !col_stats->no_stat_values_provided() &&        //(1)
    field->type() != MYSQL_TYPE_GEOMETRY &&              //(2)
#ifdef WITH_PARTITION_STORAGE_ENGINE
    (!field->table->part_info ||
     !field->table->part_info->field_in_partition_expr(field)) &&     //(3)
#endif
    true;
}