swuferhong commented on code in PR #20084: URL: https://github.com/apache/flink/pull/20084#discussion_r934479463
########## flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceStatisticsReportTest.java: ########## @@ -0,0 +1,393 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connectors.hive; + +import org.apache.flink.table.api.SqlDialect; +import org.apache.flink.table.catalog.hive.HiveCatalog; +import org.apache.flink.table.catalog.hive.HiveTestUtils; +import org.apache.flink.table.plan.stats.ColumnStats; +import org.apache.flink.table.plan.stats.TableStats; +import org.apache.flink.table.planner.plan.stats.FlinkStatistic; +import org.apache.flink.table.planner.utils.StatisticsReportTestBase; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.utils.DateTimeUtils; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.math.BigDecimal; +import java.sql.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Test for statistics functionality in {@link HiveTableSource}. */ +public class HiveTableSourceStatisticsReportTest extends StatisticsReportTestBase { + + private static HiveCatalog hiveCatalog; + private static final String catalogName = "hive"; + private static final String dbName = "db1"; + private static final String sourceTable = "sourceTable"; + + @BeforeEach + public void setup(@TempDir File file) throws Exception { + super.setup(file); + hiveCatalog = HiveTestUtils.createHiveCatalog(); + hiveCatalog.open(); + + tEnv.getConfig().setSqlDialect(SqlDialect.HIVE); + tEnv.registerCatalog(catalogName, hiveCatalog); + tEnv.useCatalog(catalogName); + tEnv.executeSql(String.format("create database %s", dbName)); + + tEnv.executeSql( + String.format( + "create table %s.%s.%s ( %s )", + catalogName, + dbName, + sourceTable, + String.join(", ", ddlTypesMapToStringList(ddlTypesMap())))); + + DataType dataType = + tEnv.from(String.format("%s.%s.%s", catalogName, dbName, sourceTable)) + .getResolvedSchema() + .toPhysicalRowDataType(); + tEnv.fromValues(dataType, getData()) + .executeInsert(String.format("%s.%s.%s", catalogName, dbName, sourceTable)) + .await(); + } + + @AfterEach + public void after() { + super.after(); + if (null != hiveCatalog) { + hiveCatalog.close(); + } + } + + @Override + protected String[] properties() { + return new String[0]; + } + + @Test + public void testMapRedCsvFormatHiveTableSourceStatisticsReport() { + FlinkStatistic statistic = + getStatisticsFromOptimizedPlan( + String.format("select * from %s.%s.%s", catalogName, dbName, sourceTable)); + assertThat(statistic.getTableStats()).isEqualTo(TableStats.UNKNOWN); + } + + @Test + public void testFlinkOrcFormatHiveTableSourceStatisticsReport() throws Exception { + tEnv.executeSql( + String.format( + "create table hive.db1.orcTable ( %s ) stored as orc", + String.join(", ", ddlTypesMapToStringList(ddlTypesMap())))); + tEnv.executeSql( + String.format( + "insert into hive.db1.orcTable select * from %s.%s.%s", + catalogName, dbName, sourceTable)) + .await(); + + // Hive to read Orc file + FlinkStatistic statistic = + getStatisticsFromOptimizedPlan( + "select * from hive.db1.orcTable where f_smallint > 100"); + assertHiveTableOrcFormatTableStatsEquals(statistic.getTableStats(), 3, 1L); + } + + @Test + public void testFlinkParquetFormatHiveTableSourceStatisticsReport() throws Exception { + tEnv.executeSql( + String.format( + "create table hive.db1.parquetTable ( %s ) stored as parquet", + String.join(", ", ddlTypesMapToStringList(ddlTypesMap())))); + tEnv.executeSql( + String.format( + "insert into hive.db1.parquetTable select * from %s.%s.%s", + catalogName, dbName, sourceTable)) + .await(); + + // Hive to read Parquet file + FlinkStatistic statistic = + getStatisticsFromOptimizedPlan( + "select * from hive.db1.parquetTable where f_smallint > 100"); + assertHiveTableParquetFormatTableStatsEquals(statistic.getTableStats(), 3, 1L); + } + + @Test + public void testMapRedOrcFormatHiveTableSourceStatisticsReport() throws Exception { + // Use mapRed parquet format. + tEnv.getConfig().set(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER, true); + tEnv.executeSql( + String.format( + "create table hive.db1.orcTable ( %s ) stored as orc", + String.join(", ", ddlTypesMapToStringList(ddlTypesMap())))); + tEnv.executeSql( + String.format( + "insert into hive.db1.orcTable select * from %s.%s.%s", + catalogName, dbName, sourceTable)) + .await(); + + // Hive to read Orc file + FlinkStatistic statistic = + getStatisticsFromOptimizedPlan( + "select * from hive.db1.orcTable where f_smallint > 100"); + assertHiveTableOrcFormatTableStatsEquals(statistic.getTableStats(), 3, 1L); + } + + @Test + public void testMapRedParquetFormatHiveTableSourceStatisticsReport() throws Exception { + // Use mapRed parquet format. + tEnv.getConfig().set(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER, true); + tEnv.executeSql( + String.format( + "create table hive.db1.parquetTable ( %s ) stored as parquet", + String.join(", ", ddlTypesMapToStringList(ddlTypesMap())))); + tEnv.executeSql( + String.format( + "insert into hive.db1.parquetTable select * from %s.%s.%s", + catalogName, dbName, sourceTable)) + .await(); + + // Hive to read Parquet file. + FlinkStatistic statistic = + getStatisticsFromOptimizedPlan( + "select * from hive.db1.parquetTable where f_smallint > 100"); + assertHiveTableParquetFormatTableStatsEquals(statistic.getTableStats(), 3, 1L); + } + + @Test + public void testHiveTableSourceWithLimitPushDown() { Review Comment: > add a case with limit push down and statistics is unknown done! ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/file/table/FileSystemStatisticsReportTest.java: ########## @@ -190,17 +189,23 @@ public void testFilterPushDownAndReportStatisticsDisabled() { } @Test - public void testNoPartitionPushDownAndCatalogStatisticsExist() - throws PartitionNotExistException { + public void testLimitPushDownAndCatalogStatisticsDoNotExist() { Review Comment: > add a test about limit push down and statistics is unknown done! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
