@Override
    public TableStatistics getTableStatistics(
            ConnectorSession session,
            SchemaTableName table,
            Map<String, ColumnHandle> columns,
            Map<String, Type> columnTypes,
            List<HivePartition> partitions)
    {
        if (!isStatisticsEnabled(session)) {
            return TableStatistics.empty();
        }
        if (partitions.isEmpty()) {
            return createZeroStatistics(columns, columnTypes);
        }
        int sampleSize = getPartitionStatisticsSampleSize(session);
        List<HivePartition> partitionsSample = getPartitionsSample(partitions, sampleSize);
        try {
            Map<String, PartitionStatistics> statisticsSample = statisticsProvider.getPartitionsStatistics(session, table, partitionsSample);
            validatePartitionStatistics(table, statisticsSample);
            return getTableStatistics(columns, columnTypes, partitions, statisticsSample);
        }
        catch (PrestoException e) {
            if (e.getErrorCode().equals(HIVE_CORRUPTED_COLUMN_STATISTICS.toErrorCode()) && isIgnoreCorruptedStatistics(session)) {
                log.error(e);
                return TableStatistics.empty();
            }
            throw e;
        }
    }
@Test
    public void testGetTableStatisticsValidationFailure()
    {
        PartitionStatistics corruptedStatistics = PartitionStatistics.builder()
                .setBasicStatistics(new HiveBasicStatistics(-1, 0, 0, 0))
                .build();
        String partitionName = "p1=string1/p2=1234";
        MetastoreHiveStatisticsProvider statisticsProvider = new MetastoreHiveStatisticsProvider((session, table, hivePartitions) -> ImmutableMap.of(partitionName, corruptedStatistics), quickStatsProvider);
        TestingConnectorSession session = new TestingConnectorSession(new HiveSessionProperties(
                new HiveClientConfig().setIgnoreCorruptedStatistics(false),
                new OrcFileWriterConfig(),
                new ParquetFileWriterConfig(),
                new CacheConfig())
                .getSessionProperties());
        assertThatThrownBy(() -> statisticsProvider.getTableStatistics(
                session,
                TABLE,
                ImmutableMap.of(),
                ImmutableMap.of(),
                ImmutableList.of(partition(partitionName))))
                .isInstanceOf(PrestoException.class)
                .hasFieldOrPropertyWithValue("errorCode", HIVE_CORRUPTED_COLUMN_STATISTICS.toErrorCode());
        TestingConnectorSession ignoreSession = new TestingConnectorSession(new HiveSessionProperties(
                new HiveClientConfig().setIgnoreCorruptedStatistics(true),
                new OrcFileWriterConfig(),
                new ParquetFileWriterConfig(),
                new CacheConfig())
                .getSessionProperties());
        assertEquals(
                statisticsProvider.getTableStatistics(
                        ignoreSession,
                        TABLE,
                        ImmutableMap.of(),
                        ImmutableMap.of(),
                        ImmutableList.of(partition(partitionName))),
                TableStatistics.empty());
    }