public Statistics getTableStatistics(
            OptimizerContext session,
            Table table,
            List<ColumnRefOperator> columns,
            List<PartitionKey> partitionKeys) {
        Statistics.Builder builder = Statistics.builder();
        HiveMetaStoreTable hmsTbl = (HiveMetaStoreTable) table;
        if (hmsTbl.isUnPartitioned()) {
            HivePartitionStats tableStats = hmsOps.getTableStatistics(hmsTbl.getDbName(), hmsTbl.getTableName());
            return createUnpartitionedStats(tableStats, columns, builder, table);
        }

        int sampleSize = getSamplePartitionSize(session);
        List<String> partitionColumnNames = hmsTbl.getPartitionColumnNames();
        List<String> partitionNames = partitionKeys.stream()
                .peek(partitionKey -> checkState(partitionKey.getKeys().size() == partitionColumnNames.size(),
                        "columns size is " + partitionColumnNames.size() +
                                " but values size is " + partitionKey.getKeys().size()))
                .map(partitionKey -> toHivePartitionName(partitionColumnNames, partitionKey))
                .collect(Collectors.toList());

        List<String> sampledPartitionNames = getPartitionsSample(partitionNames, sampleSize);
        Map<String, HivePartitionStats> partitionStatistics = hmsOps.getPartitionStatistics(table, sampledPartitionNames);

        double avgRowNumPerPartition = -1;
        double totalRowNums = -1;
        avgRowNumPerPartition = getPerPartitionRowAvgNums(partitionStatistics.values());

        if (avgRowNumPerPartition <= 0) {
            builder.setOutputRowCount(getEstimatedRowCount(table, partitionKeys));
            return builder.build();
        }

        totalRowNums = avgRowNumPerPartition * partitionKeys.size();
        builder.setOutputRowCount(totalRowNums);

        for (ColumnRefOperator columnRefOperator : columns) {
            Column column = table.getColumn(columnRefOperator.getName());
            if (partitionColumnNames.contains(columnRefOperator.getName())) {
                builder.addColumnStatistic(columnRefOperator, createPartitionColumnStatistics(
                        column, partitionKeys, partitionStatistics, partitionColumnNames, avgRowNumPerPartition, totalRowNums));
            } else {
                builder.addColumnStatistic(columnRefOperator, createDataColumnStatistics(
                        column, totalRowNums, partitionStatistics.values()));
            }
        }

        return builder.build();
    }
@Test
    public void testGetTableStatistics() throws AnalysisException {
        HiveTable hiveTable = (HiveTable) hmsOps.getTable("db1", "table1");
        ColumnRefOperator partColumnRefOperator = new ColumnRefOperator(0, Type.INT, "col1", true);
        ColumnRefOperator dataColumnRefOperator = new ColumnRefOperator(1, Type.INT, "col2", true);
        PartitionKey hivePartitionKey1 = PartitionUtil.createPartitionKey(
                Lists.newArrayList("1"), hiveTable.getPartitionColumns());
        PartitionKey hivePartitionKey2 = PartitionUtil.createPartitionKey(
                Lists.newArrayList("2"), hiveTable.getPartitionColumns());
        Statistics statistics = statisticsProvider.getTableStatistics(
                optimizerContext, hiveTable, Lists.newArrayList(partColumnRefOperator, dataColumnRefOperator),
                Lists.newArrayList(hivePartitionKey1, hivePartitionKey2));
        Assert.assertEquals(1, statistics.getOutputRowCount(), 0.001);
        Assert.assertEquals(0, statistics.getColumnStatistics().size());

        cachingHiveMetastore.getPartitionStatistics(hiveTable, Lists.newArrayList("col1=1", "col1=2"));
        statistics = statisticsProvider.getTableStatistics(
                optimizerContext, hiveTable, Lists.newArrayList(partColumnRefOperator, dataColumnRefOperator),
                Lists.newArrayList(hivePartitionKey1, hivePartitionKey2));
        Assert.assertEquals(100, statistics.getOutputRowCount(), 0.001);
        Map<ColumnRefOperator, ColumnStatistic> columnStatistics = statistics.getColumnStatistics();
        Assert.assertEquals(2, statistics.getColumnStatistics().size());
        ColumnStatistic partitionColumnStats = columnStatistics.get(partColumnRefOperator);
        Assert.assertEquals(1, partitionColumnStats.getMinValue(), 0.001);
        Assert.assertEquals(2, partitionColumnStats.getMaxValue(), 0.001);
        Assert.assertEquals(0, partitionColumnStats.getNullsFraction(), 0.001);
        Assert.assertEquals(4, partitionColumnStats.getAverageRowSize(), 0.001);
        Assert.assertEquals(2, partitionColumnStats.getDistinctValuesCount(), 0.001);

        ColumnStatistic dataColumnStats = columnStatistics.get(dataColumnRefOperator);
        Assert.assertEquals(0, dataColumnStats.getMinValue(), 0.001);
        Assert.assertEquals(0.03, dataColumnStats.getNullsFraction(), 0.001);
        Assert.assertEquals(4, dataColumnStats.getAverageRowSize(), 0.001);
        Assert.assertEquals(5, dataColumnStats.getDistinctValuesCount(), 0.001);
    }