public static Optional<MapStatistics> mergeMapStatistics(List<ColumnStatistics> stats, Object2LongMap<DwrfProto.KeyInfo> keySizes)
    {
        Map<DwrfProto.KeyInfo, List<ColumnStatistics>> columnStatisticsByKey = new LinkedHashMap<>();
        long nonNullValueCount = 0;

        for (ColumnStatistics columnStatistics : stats) {
            if (columnStatistics.getNumberOfValues() > 0) {
                MapStatistics partialStatistics = columnStatistics.getMapStatistics();
                if (partialStatistics == null) {
                    // there are non-null values but no statistics, so we can not say anything about the data
                    return Optional.empty();
                }

                // collect column stats for each key for merging later
                for (MapStatisticsEntry entry : partialStatistics.getEntries()) {
                    List<ColumnStatistics> allKeyStats = columnStatisticsByKey.computeIfAbsent(entry.getKey(), (k) -> new ArrayList<>());
                    allKeyStats.add(entry.getColumnStatistics());
                }
                nonNullValueCount += columnStatistics.getNumberOfValues();
            }
        }

        // merge all column stats for each key
        MapColumnStatisticsBuilder mapStatisticsBuilder = new MapColumnStatisticsBuilder(true);
        for (Map.Entry<DwrfProto.KeyInfo, List<ColumnStatistics>> entry : columnStatisticsByKey.entrySet()) {
            DwrfProto.KeyInfo key = entry.getKey();
            Long keySize = keySizes != null ? keySizes.getLong(key) : null;
            ColumnStatistics mergedColumnStatistics = mergeColumnStatistics(entry.getValue(), keySize, null);
            mapStatisticsBuilder.addMapStatistics(key, mergedColumnStatistics);
        }
        mapStatisticsBuilder.increaseValueCount(nonNullValueCount);

        return mapStatisticsBuilder.buildMapStatistics();
    }
@Test(dataProvider = "keySupplier")
    public void testMergeMapStatistics(KeyInfo[] keys)
    {
        // merge two stats with keys: [k0,k1] and [k1,k2]
        // column statistics for k1 should be merged together
        MapColumnStatisticsBuilder builder1 = new MapColumnStatisticsBuilder(true);
        builder1.addMapStatistics(keys[0], new IntegerColumnStatistics(3L, null, null, null, new IntegerStatistics(1L, 2L, 3L)));
        builder1.addMapStatistics(keys[1], new IntegerColumnStatistics(5L, null, null, null, new IntegerStatistics(10L, 20L, 30L)));
        builder1.increaseValueCount(8);
        ColumnStatistics columnStatistics1 = builder1.buildColumnStatistics();

        MapColumnStatisticsBuilder builder2 = new MapColumnStatisticsBuilder(true);
        builder2.addMapStatistics(keys[1], new IntegerColumnStatistics(7L, null, null, null, new IntegerStatistics(25L, 95L, 100L)));
        builder2.addMapStatistics(keys[2], new IntegerColumnStatistics(9L, null, null, null, new IntegerStatistics(12L, 22L, 32L)));
        builder2.increaseValueCount(16);
        ColumnStatistics columnStatistics2 = builder2.buildColumnStatistics();

        MapStatistics mergedMapStatistics = MapColumnStatisticsBuilder.mergeMapStatistics(ImmutableList.of(columnStatistics1, columnStatistics2), null).get();
        assertMergedMapStatistics(keys, mergedMapStatistics);
    }