You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/06/27 12:09:37 UTC
[GitHub] [flink] swuferhong opened a new pull request, #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
swuferhong opened a new pull request, #20084:
URL: https://github.com/apache/flink/pull/20084
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
First: this pr is let HiveTableSource extend from SupportsStatisticReport to make hive source have statistics report ability;
Second: this pr try to abstract test methods and test class relationships for statistics report tests.
## Brief change log
*(for example:)*
- *The TaskInfo is stored in the blob store on job creation time as a persistent artifact*
- *Deployments RPC transmits only the blob storage reference*
- *TaskManagers retrieve the TaskInfo from the blob cache*
## Verifying this change
Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (100MB)*
- *Extended integration test for recovery after master (JobManager) failure*
- *Added test that validates that TaskInfo is transferred only once across recoveries*
- *Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): yes
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on PR #20084:
URL: https://github.com/apache/flink/pull/20084#issuecomment-1188522296
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r928354682
##########
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>");
+ dataMap.put("timestamp", timestampDate);
+ dataMap.put("binary", binaryDate);
+
+ return dataMap;
+ }
+
+ private static void assertHiveTableOrcFormatTableStatsEquals(
Review Comment:
> Is it possible to extact the common code between `assertHiveTableOrcFormatTableStatsEquals` and `assertHiveTableParquetFormatTableStatsEquals`. Also, could it be moved to `StatisticsReportTestBase`?
Now, the types of statistical values of a specific type returned by Orc and Parquet format are inconsistent. Eg. For type Integer, Orc `getMax()` return type Long, while Parquet return type Integer. So, I don't think it's necessary to abstract 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on PR #20084:
URL: https://github.com/apache/flink/pull/20084#issuecomment-1194882181
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r908013925
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveInputFormat.java:
##########
@@ -132,10 +132,10 @@ private RowType tableRowType() {
return RowType.of(types, fieldNames);
}
- private BulkFormat<RowData, ? super HiveSourceSplit> createBulkFormatForSplit(
+ public BulkFormat<RowData, ? super HiveSourceSplit> createBulkFormatForSplit(
Review Comment:
why the access level should be changed ?
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceBuilder.java:
##########
@@ -315,7 +315,7 @@ private RowType getProducedRowType() {
return (RowType) producedSchema.toRowDataType().bridgedTo(RowData.class).getLogicalType();
}
- private BulkFormat<RowData, HiveSourceSplit> createDefaultBulkFormat() {
+ public BulkFormat<RowData, HiveSourceSplit> createDefaultBulkFormat() {
Review Comment:
ditto
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveInputFormat.java:
##########
@@ -302,7 +302,8 @@ public TypeInformation<RowData> getProducedType() {
}
}
- private class HiveReader implements BulkFormat.Reader<RowData> {
+ /** HiveReader. */
+ public class HiveReader implements BulkFormat.Reader<RowData> {
Review Comment:
useless change ?
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +279,93 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(HiveOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ return ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ return getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ if (inputSplits.isEmpty()) {
Review Comment:
the branch is checked before
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveOptions.java:
##########
@@ -130,6 +131,16 @@ public class HiveOptions {
"This is a synonym for the deprecated 'streaming-source.consume-order' option.")
.build());
+ public static final ConfigOption<FileSystemConnectorOptions.FileStatisticsType>
Review Comment:
use directly the defination in FileSystemConnectorOptions
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +279,93 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(HiveOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ return ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ return getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ if (inputSplits.isEmpty()) {
+ return TableStats.UNKNOWN;
+ }
+ // TODO now we assume that one hive external table has only one storage file format
+ HiveTablePartition hiveTablePartition = inputSplits.get(0).getHiveTablePartition();
+ StorageDescriptor sd = hiveTablePartition.getStorageDescriptor();
+ List<Path> files =
+ inputSplits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
+ InputFormat mapredInputFormat;
+ try {
+ mapredInputFormat =
+ (InputFormat)
+ Class.forName(
+ sd.getInputFormat(),
Review Comment:
do we have any better solution instead of reflection. cc @luoyuxia could you provide some suggestions ?
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveInputFormat.java:
##########
@@ -327,6 +328,10 @@ private HiveReader(HiveSourceSplit split) throws IOException {
serializer = new RowDataSerializer(producedRowType);
}
+ public HiveMapredSplitReader getHiveMapredSplitReader() {
Review Comment:
useless change ?
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +279,93 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(HiveOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
Review Comment:
we should consider how to handle the case after limit push and filter push down
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r934307986
##########
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
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r928345419
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Hive using Flink's
+ // InputFormat to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ // TODO now we assume that one hive external table has only one storage file format
Review Comment:
>
Yes, for hive table source, now we only need to report stats for parquet and orc format.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on PR #20084:
URL: https://github.com/apache/flink/pull/20084#issuecomment-1185395900
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
luoyuxia commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r908283449
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +279,93 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(HiveOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ return ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ return getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ if (inputSplits.isEmpty()) {
+ return TableStats.UNKNOWN;
+ }
+ // TODO now we assume that one hive external table has only one storage file format
+ HiveTablePartition hiveTablePartition = inputSplits.get(0).getHiveTablePartition();
+ StorageDescriptor sd = hiveTablePartition.getStorageDescriptor();
+ List<Path> files =
+ inputSplits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
+ InputFormat mapredInputFormat;
+ try {
+ mapredInputFormat =
+ (InputFormat)
+ Class.forName(
+ sd.getInputFormat(),
Review Comment:
We can only create the InputFormat by reflection. No other better way.
[Hive](https://github.com/apache/hive/blob/master/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java#L387) also use reflection to get the instance.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r915868826
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +279,93 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(HiveOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ return ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ return getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ if (inputSplits.isEmpty()) {
+ return TableStats.UNKNOWN;
+ }
+ // TODO now we assume that one hive external table has only one storage file format
+ HiveTablePartition hiveTablePartition = inputSplits.get(0).getHiveTablePartition();
+ StorageDescriptor sd = hiveTablePartition.getStorageDescriptor();
+ List<Path> files =
+ inputSplits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
+ InputFormat mapredInputFormat;
+ try {
+ mapredInputFormat =
+ (InputFormat)
+ Class.forName(
+ sd.getInputFormat(),
+ true,
+ Thread.currentThread().getContextClassLoader())
+ .newInstance();
+ if (mapredInputFormat instanceof TextInputFormat) {
+ // only support csv format.
+ return CsvFormatStatisticsReportUtil.getTableStatistics(files);
+ } else if (mapredInputFormat instanceof OrcInputFormat) {
+ return OrcFormatStatisticsReportUtil.getTableStatistics(
+ files, producedDataType, jobConf);
+ } else if (mapredInputFormat instanceof MapredParquetInputFormat) {
Review Comment:
> What if `DeprecatedParquetInputFormat`? I think it'll be better to check by `sd.getSerdeInfo.getSerializationLib`just like `HiveInputFormat#useParquetVectorizedRead`
Ok, Thanks, I try to use `sd.getSerdeInfo.getSerializationLib`. By doing this, we can also avoid to use reflection.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r915870874
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +279,93 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(HiveOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ return ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ return getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ if (inputSplits.isEmpty()) {
+ return TableStats.UNKNOWN;
+ }
+ // TODO now we assume that one hive external table has only one storage file format
+ HiveTablePartition hiveTablePartition = inputSplits.get(0).getHiveTablePartition();
+ StorageDescriptor sd = hiveTablePartition.getStorageDescriptor();
+ List<Path> files =
+ inputSplits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
+ InputFormat mapredInputFormat;
+ try {
+ mapredInputFormat =
+ (InputFormat)
+ Class.forName(
+ sd.getInputFormat(),
Review Comment:
> do we have any better solution instead of reflection. cc @luoyuxia could you provide some suggestions ?
In line with @luoyuxia 's suggestion. I try to use `sd.getSerdeInfo.getSerializationLib` instead of this. By doing that, we can avoid to use reflection.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on PR #20084:
URL: https://github.com/apache/flink/pull/20084#issuecomment-1200100903
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r928355032
##########
flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFilesystemStatisticsReportTest.java:
##########
@@ -18,90 +18,36 @@
package org.apache.flink.formats.csv;
-import org.apache.flink.table.api.TableConfig;
-import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.plan.stats.TableStats;
-import org.apache.flink.table.planner.plan.schema.TableSourceTable;
import org.apache.flink.table.planner.plan.stats.FlinkStatistic;
-import org.apache.flink.table.planner.utils.BatchTableTestUtil;
-import org.apache.flink.table.planner.utils.TableTestBase;
-import org.apache.flink.table.planner.utils.TableTestUtil;
-import org.apache.flink.util.Preconditions;
+import org.apache.flink.table.types.DataType;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelVisitor;
-import org.apache.calcite.rel.core.TableScan;
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.io.IOException;
-import java.nio.file.Files;
-import java.util.Arrays;
-import java.util.List;
+import java.util.concurrent.ExecutionException;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Test for statistics functionality in {@link CsvFormatFactory} in the case of file system source.
*/
-public class CsvFormatFilesystemStatisticsReportTest extends TableTestBase {
- private BatchTableTestUtil util;
- private TableEnvironment tEnv;
- @TempDir private static File path;
+public class CsvFormatFilesystemStatisticsReportTest extends CsvFormatStatisticsReportTest {
Review Comment:
> Since `CsvFormatFilesystemStatisticsReportTest` extends `CsvFormatStatisticsReportTest`, it wll also run the tests in `CsvFormatStatisticsReportTest`. Is it expected?
Yes, the result is in line with expectations.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r928375302
##########
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 "
Review Comment:
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r915845313
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveInputFormat.java:
##########
@@ -302,7 +302,8 @@ public TypeInformation<RowData> getProducedType() {
}
}
- private class HiveReader implements BulkFormat.Reader<RowData> {
+ /** HiveReader. */
+ public class HiveReader implements BulkFormat.Reader<RowData> {
Review Comment:
> useless change ?
Mistake!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r934363561
##########
flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/util/CsvFormatStatisticsReportUtil.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.formats.csv.util;
+
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.plan.stats.TableStats;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.List;
+
+/** Utils for Csv format statistics report. */
+public class CsvFormatStatisticsReportUtil {
+ public static TableStats getTableStatistics(List<Path> files) {
+ // For Csv format, it's a heavy operation to obtain accurate statistics by scanning all
+ // files. So, We obtain the estimated statistics by sampling, the specific way is to
+ // sample the first 100 lines and calculate their row size, then compare row size with
+ // total file size to get the estimated row count.
+ final int totalSampleLineCnt = 100;
+ try {
+ long totalFileSize = 0;
+ int sampledRowCnt = 0;
+ long sampledRowSize = 0;
+ for (Path file : files) {
+ FileSystem fs = FileSystem.get(file.toUri());
+ FileStatus status = fs.getFileStatus(file);
+ totalFileSize += status.getLen();
+
+ // sample the line size
+ if (sampledRowCnt < totalSampleLineCnt) {
+ try (InputStreamReader isr =
+ new InputStreamReader(
+ Files.newInputStream(new File(file.toUri()).toPath()));
+ BufferedReader br = new BufferedReader(isr)) {
+ String line;
+ while (sampledRowCnt < totalSampleLineCnt
+ && (line = br.readLine()) != null) {
+ sampledRowCnt += 1;
+ sampledRowSize += (line.getBytes(StandardCharsets.UTF_8).length + 1);
+ }
+ }
+ }
+ }
+
+ // If line break is "\r\n", br.readLine() will ignore '\n' which make sampledRowSize
+ // smaller than totalFileSize. This will influence test result.
+ if (sampledRowCnt < totalSampleLineCnt) {
+ sampledRowSize = totalFileSize;
+ }
+ if (sampledRowSize == 0) {
+ return TableStats.UNKNOWN;
+ }
+
+ int realSampledLineCnt = Math.min(totalSampleLineCnt, sampledRowCnt);
+ long estimatedRowCount = totalFileSize * realSampledLineCnt / sampledRowSize;
+ return new TableStats(estimatedRowCount);
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
Review Comment:
> add some log for exception case
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r930653031
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Hive using Flink's
+ // InputFormat to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
Review Comment:
> I think it will be better to log the exception
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r928346759
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Hive using Flink's
+ // InputFormat to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ // TODO now we assume that one hive external table has only one storage file format
+ String serializationLib =
+ inputSplits
+ .get(0)
+ .getHiveTablePartition()
+ .getStorageDescriptor()
+ .getSerdeInfo()
+ .getSerializationLib()
+ .toLowerCase();
+ List<Path> files =
+ inputSplits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
+ // Now we only support Csv, Parquet, Orc formats.
+ if (serializationLib.contains("parquet")) {
+ return ParquetFormatStatisticsReportUtil.getTableStatistics(
+ files, producedDataType, jobConf, hiveVersion.startsWith("3"));
Review Comment:
> Why check `hiveVersion.startsWith("3")`?
The purpose of checking `hiveVersion.startsWith("3")` is to judge whether utcTimestamp is used.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
luoyuxia commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r928066455
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceBuilder.java:
##########
@@ -315,7 +315,7 @@ private RowType getProducedRowType() {
return (RowType) producedSchema.toRowDataType().bridgedTo(RowData.class).getLogicalType();
}
- private BulkFormat<RowData, HiveSourceSplit> createDefaultBulkFormat() {
+ public BulkFormat<RowData, HiveSourceSplit> createDefaultBulkFormat() {
Review Comment:
Can be `protected`.
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Hive using Flink's
+ // InputFormat to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
Review Comment:
```suggestion
// If HiveInputFormat's variable useMapRedReader is true, Flink will use hadoop mapred record reader to read data.
```
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Hive using Flink's
+ // InputFormat to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ // TODO now we assume that one hive external table has only one storage file format
+ String serializationLib =
+ inputSplits
+ .get(0)
+ .getHiveTablePartition()
+ .getStorageDescriptor()
+ .getSerdeInfo()
+ .getSerializationLib()
+ .toLowerCase();
+ List<Path> files =
+ inputSplits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
+ // Now we only support Csv, Parquet, Orc formats.
+ if (serializationLib.contains("parquet")) {
+ return ParquetFormatStatisticsReportUtil.getTableStatistics(
+ files, producedDataType, jobConf, hiveVersion.startsWith("3"));
Review Comment:
Why check `hiveVersion.startsWith("3")`?
##########
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>");
+ dataMap.put("timestamp", timestampDate);
+ dataMap.put("binary", binaryDate);
+
+ return dataMap;
+ }
+
+ private static void assertHiveTableOrcFormatTableStatsEquals(
Review Comment:
Is it possible to extact the common code between `assertHiveTableOrcFormatTableStatsEquals`
and `assertHiveTableParquetFormatTableStatsEquals`.
Also, could it be moved to `StatisticsReportTestBase`?
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Hive using Flink's
Review Comment:
```suggestion
// If HiveInputFormat's variable useMapRedReader is false, it will use Flink's
```
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Hive using Flink's
+ // InputFormat to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
Review Comment:
I think it will be better to log the exception
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Hive using Flink's
+ // InputFormat to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return TableStats.UNKNOWN;
Review Comment:
If `inputSplits.size() == 0`, why not return `TableStats(0)`?
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Hive using Flink's
+ // InputFormat to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ // TODO now we assume that one hive external table has only one storage file format
Review Comment:
What if Hive's partitions have different formats? Seems it will return `UNKONW`?
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Hive using Flink's
+ // InputFormat to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ // TODO now we assume that one hive external table has only one storage file format
+ String serializationLib =
+ inputSplits
+ .get(0)
+ .getHiveTablePartition()
+ .getStorageDescriptor()
+ .getSerdeInfo()
+ .getSerializationLib()
+ .toLowerCase();
+ List<Path> files =
+ inputSplits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
+ // Now we only support Csv, Parquet, Orc formats.
Review Comment:
Csv is also supported? Seems there's no code cover it.
##########
flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFilesystemStatisticsReportTest.java:
##########
@@ -18,90 +18,36 @@
package org.apache.flink.formats.csv;
-import org.apache.flink.table.api.TableConfig;
-import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.plan.stats.TableStats;
-import org.apache.flink.table.planner.plan.schema.TableSourceTable;
import org.apache.flink.table.planner.plan.stats.FlinkStatistic;
-import org.apache.flink.table.planner.utils.BatchTableTestUtil;
-import org.apache.flink.table.planner.utils.TableTestBase;
-import org.apache.flink.table.planner.utils.TableTestUtil;
-import org.apache.flink.util.Preconditions;
+import org.apache.flink.table.types.DataType;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelVisitor;
-import org.apache.calcite.rel.core.TableScan;
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.io.IOException;
-import java.nio.file.Files;
-import java.util.Arrays;
-import java.util.List;
+import java.util.concurrent.ExecutionException;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Test for statistics functionality in {@link CsvFormatFactory} in the case of file system source.
*/
-public class CsvFormatFilesystemStatisticsReportTest extends TableTestBase {
- private BatchTableTestUtil util;
- private TableEnvironment tEnv;
- @TempDir private static File path;
+public class CsvFormatFilesystemStatisticsReportTest extends CsvFormatStatisticsReportTest {
Review Comment:
Since `CsvFormatFilesystemStatisticsReportTest` extends `CsvFormatStatisticsReportTest`, it wll also run the tests in `CsvFormatStatisticsReportTest`. Is it expected?
##########
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?
##########
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 "
Review Comment:
nit: Using String.formart(xx)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #20084:
URL: https://github.com/apache/flink/pull/20084#issuecomment-1167279288
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "70ec374b2ce5c5da6a146540079834dc9f20a276",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "70ec374b2ce5c5da6a146540079834dc9f20a276",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 70ec374b2ce5c5da6a146540079834dc9f20a276 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r922935948
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +279,93 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(HiveOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
Review Comment:
> we should consider how to handle the case after limit push and filter push down
Now, hive source don't support filter push down. For limit push down, `PushLimitIntoTableSourceScanRule` happened after `FlinkRecomputeStatisticsProgram`, and `PushLimitIntoTableSourceScanRule` can re-compute the new row count. So, I think there is no need to add re-compute stats logic in HiveTableSource.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r915829129
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceBuilder.java:
##########
@@ -315,7 +315,7 @@ private RowType getProducedRowType() {
return (RowType) producedSchema.toRowDataType().bridgedTo(RowData.class).getLogicalType();
}
- private BulkFormat<RowData, HiveSourceSplit> createDefaultBulkFormat() {
+ public BulkFormat<RowData, HiveSourceSplit> createDefaultBulkFormat() {
Review Comment:
> ditto
I think this modify is necessary. Because $createDefaultBulkFormat() can only used by $buildWithDefaultBulkFormat(), however $buildWithDefaultBulkFormat() a HiveSource not a BulkFormat. So, it should be a public method which can be used to create default bulk format.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe closed pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
godfreyhe closed pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
URL: https://github.com/apache/flink/pull/20084
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r934370330
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java:
##########
@@ -359,11 +359,19 @@ public TableStats reportStatistics() {
splits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
if (bulkReaderFormat instanceof FileBasedStatisticsReportableInputFormat) {
- return ((FileBasedStatisticsReportableInputFormat) bulkReaderFormat)
- .reportStatistics(files, producedDataType);
+ TableStats tableStats =
+ ((FileBasedStatisticsReportableInputFormat) bulkReaderFormat)
+ .reportStatistics(files, producedDataType);
+ return limit == null
Review Comment:
> add some log for exception case
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on PR #20084:
URL: https://github.com/apache/flink/pull/20084#issuecomment-1186725864
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on PR #20084:
URL: https://github.com/apache/flink/pull/20084#issuecomment-1202234686
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r934294704
##########
flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/util/CsvFormatStatisticsReportUtil.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.formats.csv.util;
+
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.plan.stats.TableStats;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.List;
+
+/** Utils for Csv format statistics report. */
+public class CsvFormatStatisticsReportUtil {
+ public static TableStats getTableStatistics(List<Path> files) {
+ // For Csv format, it's a heavy operation to obtain accurate statistics by scanning all
+ // files. So, We obtain the estimated statistics by sampling, the specific way is to
+ // sample the first 100 lines and calculate their row size, then compare row size with
+ // total file size to get the estimated row count.
+ final int totalSampleLineCnt = 100;
+ try {
+ long totalFileSize = 0;
+ int sampledRowCnt = 0;
+ long sampledRowSize = 0;
+ for (Path file : files) {
+ FileSystem fs = FileSystem.get(file.toUri());
+ FileStatus status = fs.getFileStatus(file);
+ totalFileSize += status.getLen();
+
+ // sample the line size
+ if (sampledRowCnt < totalSampleLineCnt) {
+ try (InputStreamReader isr =
+ new InputStreamReader(
+ Files.newInputStream(new File(file.toUri()).toPath()));
+ BufferedReader br = new BufferedReader(isr)) {
+ String line;
+ while (sampledRowCnt < totalSampleLineCnt
+ && (line = br.readLine()) != null) {
+ sampledRowCnt += 1;
+ sampledRowSize += (line.getBytes(StandardCharsets.UTF_8).length + 1);
+ }
+ }
+ }
+ }
+
+ // If line break is "\r\n", br.readLine() will ignore '\n' which make sampledRowSize
+ // smaller than totalFileSize. This will influence test result.
+ if (sampledRowCnt < totalSampleLineCnt) {
+ sampledRowSize = totalFileSize;
+ }
+ if (sampledRowSize == 0) {
+ return TableStats.UNKNOWN;
+ }
+
+ int realSampledLineCnt = Math.min(totalSampleLineCnt, sampledRowCnt);
+ long estimatedRowCount = totalFileSize * realSampledLineCnt / sampledRowSize;
+ return new TableStats(estimatedRowCount);
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
Review Comment:
add some log for exception case
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Flink will use hadoop
+ // mapRed record to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return new TableStats(0);
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ // TODO now we assume that one hive external table has only one storage file format
+ String serializationLib =
+ inputSplits
+ .get(0)
+ .getHiveTablePartition()
+ .getStorageDescriptor()
+ .getSerdeInfo()
+ .getSerializationLib()
+ .toLowerCase();
+ List<Path> files =
+ inputSplits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
+ // Now we only support Parquet, Orc formats.
+ if (serializationLib.contains("parquet")) {
+ return ParquetFormatStatisticsReportUtil.getTableStatistics(
+ files, producedDataType, jobConf, hiveVersion.startsWith("3"));
+ } else if (serializationLib.contains("orc")) {
+ return OrcFormatStatisticsReportUtil.getTableStatistics(
+ files, producedDataType, jobConf);
+ } else {
+ // Now, only support Orc and Parquet Formats.
+ return TableStats.UNKNOWN;
Review Comment:
add some log here
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Hive using Flink's
+ // InputFormat to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
Review Comment:
+1
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java:
##########
@@ -359,11 +359,19 @@ public TableStats reportStatistics() {
splits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
if (bulkReaderFormat instanceof FileBasedStatisticsReportableInputFormat) {
- return ((FileBasedStatisticsReportableInputFormat) bulkReaderFormat)
- .reportStatistics(files, producedDataType);
+ TableStats tableStats =
+ ((FileBasedStatisticsReportableInputFormat) bulkReaderFormat)
+ .reportStatistics(files, producedDataType);
+ return limit == null
Review Comment:
add some log for exception case
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Flink will use hadoop
+ // mapRed record to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
Review Comment:
Since we has built the HiveSourceBuilder with limit, is `limit` also needed here ?
##########
flink-formats/flink-orc/src/main/java/org/apache/flink/orc/util/OrcFormatStatisticsReportUtil.java:
##########
@@ -0,0 +1,214 @@
+/*
+ * 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.orc.util;
+
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.plan.stats.ColumnStats;
+import org.apache.flink.table.plan.stats.TableStats;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.orc.ColumnStatistics;
+import org.apache.orc.DateColumnStatistics;
+import org.apache.orc.DecimalColumnStatistics;
+import org.apache.orc.DoubleColumnStatistics;
+import org.apache.orc.IntegerColumnStatistics;
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.StringColumnStatistics;
+import org.apache.orc.TimestampColumnStatistics;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.impl.ColumnStatisticsImpl;
+
+import java.io.IOException;
+import java.sql.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Utils for Orc format statistics report. */
+public class OrcFormatStatisticsReportUtil {
+ public static TableStats getTableStatistics(
+ List<Path> files, DataType producedDataType, Configuration hadoopConfig) {
+ try {
+ long rowCount = 0;
+ Map<String, ColumnStatistics> columnStatisticsMap = new HashMap<>();
+ RowType producedRowType = (RowType) producedDataType.getLogicalType();
+ for (Path file : files) {
+ rowCount +=
+ updateStatistics(hadoopConfig, file, columnStatisticsMap, producedRowType);
+ }
+
+ Map<String, ColumnStats> columnStatsMap =
+ convertToColumnStats(rowCount, columnStatisticsMap, producedRowType);
+
+ return new TableStats(rowCount, columnStatsMap);
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
Review Comment:
add some log for exception
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Flink will use hadoop
+ // mapRed record to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return new TableStats(0);
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ // TODO now we assume that one hive external table has only one storage file format
+ String serializationLib =
+ inputSplits
+ .get(0)
+ .getHiveTablePartition()
+ .getStorageDescriptor()
+ .getSerdeInfo()
+ .getSerializationLib()
+ .toLowerCase();
+ List<Path> files =
+ inputSplits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
+ // Now we only support Parquet, Orc formats.
+ if (serializationLib.contains("parquet")) {
Review Comment:
case sensitive?
##########
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
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on PR #20084:
URL: https://github.com/apache/flink/pull/20084#issuecomment-1200113237
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on PR #20084:
URL: https://github.com/apache/flink/pull/20084#issuecomment-1200334904
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
luoyuxia commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r908295003
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +279,93 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(HiveOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ return ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ return getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ } else {
+ return TableStats.UNKNOWN;
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ if (inputSplits.isEmpty()) {
+ return TableStats.UNKNOWN;
+ }
+ // TODO now we assume that one hive external table has only one storage file format
+ HiveTablePartition hiveTablePartition = inputSplits.get(0).getHiveTablePartition();
+ StorageDescriptor sd = hiveTablePartition.getStorageDescriptor();
+ List<Path> files =
+ inputSplits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
+ InputFormat mapredInputFormat;
+ try {
+ mapredInputFormat =
+ (InputFormat)
+ Class.forName(
+ sd.getInputFormat(),
+ true,
+ Thread.currentThread().getContextClassLoader())
+ .newInstance();
+ if (mapredInputFormat instanceof TextInputFormat) {
+ // only support csv format.
+ return CsvFormatStatisticsReportUtil.getTableStatistics(files);
+ } else if (mapredInputFormat instanceof OrcInputFormat) {
+ return OrcFormatStatisticsReportUtil.getTableStatistics(
+ files, producedDataType, jobConf);
+ } else if (mapredInputFormat instanceof MapredParquetInputFormat) {
Review Comment:
What if `DeprecatedParquetInputFormat`?
I think it'll be better to check by `sd.getSerdeInfo.getSerializationLib`just like `HiveInputFormat#useParquetVectorizedRead`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on PR #20084:
URL: https://github.com/apache/flink/pull/20084#issuecomment-1186408458
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r922985005
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +279,93 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(HiveOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
Review Comment:
>
Adding limit push down logic in HiveTableSource for reporting stats. But, this not work now because of optimizing order.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r928358278
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Hive using Flink's
+ // InputFormat to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return TableStats.UNKNOWN;
Review Comment:
> If `inputSplits.size() == 0`, why not return `TableStats(0)`?
Personally, I think it's better to return `TableStats.UNKNOWN` instead of `TableStats(0)`. Because `TableStats(0)` is useless, but `TableStats.UNKNOWN` more like a label, which means we have not obtained useful stats, stands a globally unified value.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r928344559
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceBuilder.java:
##########
@@ -315,7 +315,7 @@ private RowType getProducedRowType() {
return (RowType) producedSchema.toRowDataType().bridgedTo(RowData.class).getLogicalType();
}
- private BulkFormat<RowData, HiveSourceSplit> createDefaultBulkFormat() {
+ public BulkFormat<RowData, HiveSourceSplit> createDefaultBulkFormat() {
Review Comment:
> Can be `protected`.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on PR #20084:
URL: https://github.com/apache/flink/pull/20084#issuecomment-1197600539
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r915845064
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveInputFormat.java:
##########
@@ -132,10 +132,10 @@ private RowType tableRowType() {
return RowType.of(types, fieldNames);
}
- private BulkFormat<RowData, ? super HiveSourceSplit> createBulkFormatForSplit(
+ public BulkFormat<RowData, ? super HiveSourceSplit> createBulkFormatForSplit(
Review Comment:
Mistake!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r934377036
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Flink will use hadoop
+ // mapRed record to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return new TableStats(0);
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ // TODO now we assume that one hive external table has only one storage file format
+ String serializationLib =
+ inputSplits
+ .get(0)
+ .getHiveTablePartition()
+ .getStorageDescriptor()
+ .getSerdeInfo()
+ .getSerializationLib()
+ .toLowerCase();
+ List<Path> files =
+ inputSplits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
+ // Now we only support Parquet, Orc formats.
+ if (serializationLib.contains("parquet")) {
+ return ParquetFormatStatisticsReportUtil.getTableStatistics(
+ files, producedDataType, jobConf, hiveVersion.startsWith("3"));
+ } else if (serializationLib.contains("orc")) {
+ return OrcFormatStatisticsReportUtil.getTableStatistics(
+ files, producedDataType, jobConf);
+ } else {
+ // Now, only support Orc and Parquet Formats.
+ return TableStats.UNKNOWN;
Review Comment:
> add some log here
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r934427212
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Flink will use hadoop
+ // mapRed record to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
+ // If no limit push down, return recompute table stats.
+ return tableStats;
+ } else {
+ // If table have limit push down, return new table stats without table column
+ // stats.
+ long newRowCount = Math.min(limit, tableStats.getRowCount());
+ return new TableStats(newRowCount);
+ }
+ } else {
+ return new TableStats(0);
+ }
+
+ } catch (Exception e) {
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private TableStats getMapRedInputFormatStatistics(
+ List<HiveSourceSplit> inputSplits, DataType producedDataType) {
+ // TODO now we assume that one hive external table has only one storage file format
+ String serializationLib =
+ inputSplits
+ .get(0)
+ .getHiveTablePartition()
+ .getStorageDescriptor()
+ .getSerdeInfo()
+ .getSerializationLib()
+ .toLowerCase();
+ List<Path> files =
+ inputSplits.stream().map(FileSourceSplit::path).collect(Collectors.toList());
+ // Now we only support Parquet, Orc formats.
+ if (serializationLib.contains("parquet")) {
Review Comment:
> case sensitive?
Yes, `serLib.contains("xxx")` is widely used to judge file formats in hive source.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r934433419
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -261,6 +273,99 @@ public DynamicTableSource copy() {
return source;
}
+ @Override
+ public TableStats reportStatistics() {
+ try {
+ // only support BOUNDED source
+ if (isStreamingSource()) {
+ return TableStats.UNKNOWN;
+ }
+ if (flinkConf.get(FileSystemConnectorOptions.SOURCE_REPORT_STATISTICS)
+ != FileSystemConnectorOptions.FileStatisticsType.ALL) {
+ return TableStats.UNKNOWN;
+ }
+
+ HiveSourceBuilder sourceBuilder =
+ new HiveSourceBuilder(jobConf, flinkConf, tablePath, hiveVersion, catalogTable)
+ .setProjectedFields(projectedFields)
+ .setLimit(limit);
+ int threadNum =
+ flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM);
+ List<HiveTablePartition> hivePartitionsToRead =
+ getAllPartitions(
+ jobConf,
+ hiveVersion,
+ tablePath,
+ catalogTable.getPartitionKeys(),
+ remainingPartitions);
+ BulkFormat<RowData, HiveSourceSplit> defaultBulkFormat =
+ sourceBuilder.createDefaultBulkFormat();
+ List<HiveSourceSplit> inputSplits =
+ HiveSourceFileEnumerator.createInputSplits(
+ 0, hivePartitionsToRead, threadNum, jobConf);
+ if (inputSplits.size() != 0) {
+ TableStats tableStats;
+ if (defaultBulkFormat instanceof FileBasedStatisticsReportableInputFormat) {
+ // If HiveInputFormat's variable useMapRedReader is false, Flink will use hadoop
+ // mapRed record to read data.
+ tableStats =
+ ((FileBasedStatisticsReportableInputFormat) defaultBulkFormat)
+ .reportStatistics(
+ inputSplits.stream()
+ .map(FileSourceSplit::path)
+ .collect(Collectors.toList()),
+ catalogTable.getSchema().toRowDataType());
+ } else {
+ // If HiveInputFormat's variable useMapRedReader is true, Hive using MapRed
+ // InputFormat to read data.
+ tableStats =
+ getMapRedInputFormatStatistics(
+ inputSplits, catalogTable.getSchema().toRowDataType());
+ }
+ if (limit == null) {
Review Comment:
> Since we has built the HiveSourceBuilder with limit, is `limit` also needed here ?
Yes, `HiveSourceBuilder.setLimit()` is only used for choosing `LimitableBulkFormat` or `LimitableBulkFormat`. We will read all files even if `limit != null`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a diff in pull request #20084: [FLINK-27988][table-planner] Let HiveTableSource extend from SupportStatisticsReport
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on code in PR #20084:
URL: https://github.com/apache/flink/pull/20084#discussion_r935538496
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java:
##########
@@ -362,20 +367,29 @@ public TableStats reportStatistics() {
TableStats tableStats =
((FileBasedStatisticsReportableInputFormat) bulkReaderFormat)
.reportStatistics(files, producedDataType);
+ if (tableStats.equals(TableStats.UNKNOWN)) {
+ return tableStats;
+ }
return limit == null
? tableStats
: new TableStats(Math.min(limit, tableStats.getRowCount()));
} else if (deserializationFormat instanceof FileBasedStatisticsReportableInputFormat) {
TableStats tableStats =
((FileBasedStatisticsReportableInputFormat) deserializationFormat)
.reportStatistics(files, producedDataType);
+ if (tableStats.equals(TableStats.UNKNOWN)) {
+ return tableStats;
+ }
return limit == null
? tableStats
: new TableStats(Math.min(limit, tableStats.getRowCount()));
} else {
return TableStats.UNKNOWN;
}
} catch (Exception e) {
+ LOG.info(
+ String.format(
+ "Reporting statistics failed for file system table source: %s", e));
Review Comment:
LOG.warn(Reporting statistics failed for file system table source: {}", e.getMessage());
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/file/table/FileSystemStatisticsReportTest.java:
##########
@@ -287,4 +300,17 @@ public void testFilterPartitionPushDownAndReportStatisticsDisabled() {
getStatisticsFromOptimizedPlan("select * from PartTable where a > 10 and b = 1");
assertThat(statistic.getTableStats()).isEqualTo(TableStats.UNKNOWN);
}
+
+ @Test
+ public void testFileSystemSourceWithoutData() {
+ FlinkStatistic statistic = getStatisticsFromOptimizedPlan("select * from emptyTable");
+ assertThat(statistic.getTableStats()).isEqualTo(TableStats.UNKNOWN);
+ }
+
+ @Test
+ public void testFileSystemSourceWithoutDataWithLimitPushDown() {
+ FlinkStatistic statistic =
Review Comment:
add TODO the update status in limit push rule when statistics is unknown
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java:
##########
@@ -336,6 +338,7 @@ public TableStats reportStatistics() {
}
} catch (Exception e) {
+ LOG.info(String.format("Reporting statistics failed for hive table source: %s", e));
Review Comment:
ditto
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org