swuferhong commented on code in PR #20084: URL: https://github.com/apache/flink/pull/20084#discussion_r928351394
########## flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceStatisticsReportTest.java: ########## @@ -0,0 +1,412 @@ +/* + * 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("create database " + dbName); + + tEnv.executeSql( + "create table " + + catalogName + + "." + + dbName + + "." + + sourceTable + + "(" + + String.join(", ", ddlTypesMapToStringList(ddlTypesMap())) + + ")"); + + DataType dataType = + tEnv.from(catalogName + "." + dbName + "." + sourceTable) + .getResolvedSchema() + .toPhysicalRowDataType(); + tEnv.fromValues(dataType, getData()) + .executeInsert(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( + "select * from " + catalogName + "." + dbName + "." + sourceTable); + assertThat(statistic.getTableStats()).isEqualTo(TableStats.UNKNOWN); + } + + @Test + public void testFlinkOrcFormatHiveTableSourceStatisticsReport() throws Exception { + tEnv.executeSql( + "create table hive.db1.orcTable " + + " (" + + String.join(", ", ddlTypesMapToStringList(ddlTypesMap())) + + ") stored as orc"); + tEnv.executeSql( + "insert into hive.db1.orcTable select * from " + + 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( + "create table hive.db1.parquetTable" + + " (" + + String.join(", ", ddlTypesMapToStringList(ddlTypesMap())) + + ") stored as parquet"); + tEnv.executeSql( + "insert into hive.db1.parquetTable select * from " + + 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( + "create table hive.db1.orcTable" + + " (" + + String.join(", ", ddlTypesMapToStringList(ddlTypesMap())) + + ") stored as orc"); + tEnv.executeSql( + "insert into hive.db1.orcTable select * from " + + 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( + "create table hive.db1.parquetTable" + + " (" + + String.join(", ", ddlTypesMapToStringList(ddlTypesMap())) + + ") stored as parquet"); + tEnv.executeSql( + "insert into hive.db1.parquetTable select * from " + + 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() { + FlinkStatistic statistic = + getStatisticsFromOptimizedPlan( + "select * from " + + catalogName + + "." + + dbName + + "." + + sourceTable + + " limit 1"); + assertThat(statistic.getTableStats()).isEqualTo(new TableStats(1)); + } + + @Override + protected Map<String, String> ddlTypesMap() { + // hive table ddl now don't support type: TIMESTAMP(3), TIMESTAMP(9), TIMESTAMP WITHOUT TIME + // ZONE, TIMESTAMP WITH LOCAL TIME ZONE AND ROW. So we remove these types. + Map<String, String> ddlTypesMap = super.ddlTypesMap(); + String timestampTypeName = ddlTypesMap.remove("timestamp(3)"); + ddlTypesMap.remove("timestamp(9)"); + ddlTypesMap.remove("timestamp without time zone"); + ddlTypesMap.remove("timestamp with local time zone"); + String binaryTypeName = ddlTypesMap.remove("binary(1)"); + ddlTypesMap.remove("varbinary(1)"); + ddlTypesMap.remove("time"); + ddlTypesMap.remove("row<col1 string, col2 int>"); + ddlTypesMap.put("timestamp", timestampTypeName); + ddlTypesMap.put("binary", binaryTypeName); + + return ddlTypesMap; + } + + @Override + protected Map<String, List<Object>> getDataMap() { + // hive table ddl now don't support type: TIMESTAMP(3), TIMESTAMP(9), TIMESTAMP WITHOUT TIME + // ZONE, TIMESTAMP WITH LOCAL TIME ZONE AND ROW. So we remove these types related data. + Map<String, List<Object>> dataMap = super.getDataMap(); + List<Object> timestampDate = dataMap.remove("timestamp(3)"); + dataMap.remove("timestamp(9)"); + dataMap.remove("timestamp without time zone"); + dataMap.remove("timestamp with local time zone"); + List<Object> binaryDate = dataMap.remove("binary(1)"); + dataMap.remove("varbinary(1)"); + dataMap.remove("time"); + dataMap.remove("row<col1 string, col2 int>"); Review Comment: > Why remove this? In `StatisticsReportTestBase`, we try to support full Flink types, but hive source does not support some of these types, so it needs to be removed in this method. -- 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]
