This is an automated email from the ASF dual-hosted git repository.
godfrey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 5f2d088a271 [FLINK-27988][table-planner] Let HiveTableSource extend
from SupportStatisticsReport
5f2d088a271 is described below
commit 5f2d088a2713ced5c6ce072db92f4378f73bc739
Author: zhengyunhong.zyh <[email protected]>
AuthorDate: Mon Jun 27 19:58:01 2022 +0800
[FLINK-27988][table-planner] Let HiveTableSource extend from
SupportStatisticsReport
This closes #20084
---
.../file/table/FileSystemTableSource.java | 29 +-
.../flink/connectors/hive/HiveSourceBuilder.java | 2 +-
.../flink/connectors/hive/HiveTableSource.java | 112 +++++-
.../connectors/hive/read/HiveInputFormat.java | 6 +-
.../hive/HiveTableSourceStatisticsReportTest.java | 418 +++++++++++++++++++++
.../flink/formats/csv/CsvFileFormatFactory.java | 56 +--
.../csv/util/CsvFormatStatisticsReportUtil.java | 89 +++++
.../CsvFormatFilesystemStatisticsReportTest.java | 74 +---
.../formats/csv/CsvFormatStatisticsReportTest.java | 92 +++--
.../flink/orc/OrcColumnarRowInputFormat.java | 14 +-
.../org/apache/flink/orc/OrcFileFormatFactory.java | 203 +---------
.../orc/util/OrcFormatStatisticsReportUtil.java | 220 +++++++++++
.../parquet/ParquetColumnarRowInputFormat.java | 13 +-
.../formats/parquet/ParquetFileFormatFactory.java | 268 +------------
.../parquet/ParquetVectorizedInputFormat.java | 4 +-
.../utils/ParquetFormatStatisticsReportUtil.java | 307 +++++++++++++++
.../PushPartitionIntoTableSourceScanRule.java | 2 +-
.../file/table/FileSystemStatisticsReportTest.java | 148 ++++----
.../formats/testcsv/TestCsvFormatFactory.java | 6 +
.../planner/utils/StatisticsReportTestBase.java | 2 +-
20 files changed, 1372 insertions(+), 693 deletions(-)
diff --git
a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java
b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java
index 3532372e2e6..7a59323e955 100644
---
a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java
+++
b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java
@@ -55,6 +55,9 @@ import org.apache.flink.table.plan.stats.TableStats;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.utils.PartitionPathUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nullable;
import java.io.Serializable;
@@ -86,6 +89,8 @@ public class FileSystemTableSource extends
AbstractFileSystemTable
SupportsReadingMetadata,
SupportsStatisticReport {
+ private static final Logger LOG =
LoggerFactory.getLogger(FileSystemTableSource.class);
+
@Nullable private final DecodingFormat<BulkFormat<RowData,
FileSourceSplit>> bulkReaderFormat;
@Nullable private final DecodingFormat<DeserializationSchema<RowData>>
deserializationFormat;
@@ -359,15 +364,31 @@ public class FileSystemTableSource extends
AbstractFileSystemTable
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);
+ if (tableStats.equals(TableStats.UNKNOWN)) {
+ return tableStats;
+ }
+ return limit == null
+ ? tableStats
+ : new TableStats(Math.min(limit,
tableStats.getRowCount()));
} else if (deserializationFormat instanceof
FileBasedStatisticsReportableInputFormat) {
- return ((FileBasedStatisticsReportableInputFormat)
deserializationFormat)
- .reportStatistics(files, producedDataType);
+ 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.warn(
+ "Reporting statistics failed for file system table source:
{}", e.getMessage());
return TableStats.UNKNOWN;
}
}
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceBuilder.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceBuilder.java
index 75b6e741533..4afff3b181e 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceBuilder.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceBuilder.java
@@ -315,7 +315,7 @@ public class HiveSourceBuilder {
return (RowType)
producedSchema.toRowDataType().bridgedTo(RowData.class).getLogicalType();
}
- private BulkFormat<RowData, HiveSourceSplit> createDefaultBulkFormat() {
+ protected BulkFormat<RowData, HiveSourceSplit> createDefaultBulkFormat() {
return LimitableBulkFormat.create(
new HiveInputFormat(
new JobConfWrapper(jobConf),
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
index cc5d63962c8..cc407efcec0 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
@@ -25,11 +25,18 @@ import
org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.file.src.FileSourceSplit;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.table.ContinuousPartitionFetcher;
import org.apache.flink.connector.file.table.DefaultPartTimeExtractor;
+import org.apache.flink.connector.file.table.FileSystemConnectorOptions;
import org.apache.flink.connectors.hive.read.HiveContinuousPartitionContext;
import org.apache.flink.connectors.hive.read.HivePartitionFetcherContextBase;
+import org.apache.flink.connectors.hive.read.HiveSourceSplit;
import org.apache.flink.connectors.hive.util.HivePartitionUtils;
+import org.apache.flink.core.fs.Path;
+import
org.apache.flink.formats.parquet.utils.ParquetFormatStatisticsReportUtil;
+import org.apache.flink.orc.util.OrcFormatStatisticsReportUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -41,14 +48,17 @@ import
org.apache.flink.table.catalog.hive.client.HiveShimLoader;
import org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.ProviderContext;
+import
org.apache.flink.table.connector.format.FileBasedStatisticsReportableInputFormat;
import org.apache.flink.table.connector.source.DataStreamScanProvider;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import org.apache.flink.table.connector.source.abilities.SupportsLimitPushDown;
import
org.apache.flink.table.connector.source.abilities.SupportsPartitionPushDown;
import
org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown;
+import
org.apache.flink.table.connector.source.abilities.SupportsStatisticReport;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.plan.stats.TableStats;
import org.apache.flink.table.types.DataType;
import org.apache.flink.util.Preconditions;
@@ -56,6 +66,8 @@ import
org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.mapred.JobConf;
import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
@@ -64,6 +76,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.stream.Collectors;
import static
org.apache.flink.connector.file.table.FileSystemConnectorOptions.PARTITION_TIME_EXTRACTOR_TIMESTAMP_FORMATTER;
import static
org.apache.flink.connectors.hive.HiveOptions.STREAMING_SOURCE_CONSUME_START_OFFSET;
@@ -75,8 +88,10 @@ public class HiveTableSource
implements ScanTableSource,
SupportsPartitionPushDown,
SupportsProjectionPushDown,
- SupportsLimitPushDown {
+ SupportsLimitPushDown,
+ SupportsStatisticReport {
+ private static final Logger LOG =
LoggerFactory.getLogger(HiveTableSource.class);
private static final String HIVE_TRANSFORMATION = "hive";
protected final JobConf jobConf;
@@ -261,6 +276,101 @@ public class HiveTableSource
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);
+ 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) {
+ LOG.warn("Reporting statistics failed for hive table source: {}",
e.getMessage());
+ 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.
+ LOG.info(
+ "Now for hive table source, reporting statistics only
support Orc and Parquet formats.");
+ return TableStats.UNKNOWN;
+ }
+ }
+
/** PartitionFetcher.Context for {@link ContinuousPartitionFetcher}. */
@SuppressWarnings("unchecked")
public static class HiveContinuousPartitionFetcherContext<T extends
Comparable<T>>
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveInputFormat.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveInputFormat.java
index 3b93bdf413e..d88ba944c67 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveInputFormat.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveInputFormat.java
@@ -135,7 +135,7 @@ public class HiveInputFormat implements BulkFormat<RowData,
HiveSourceSplit> {
private BulkFormat<RowData, ? super HiveSourceSplit>
createBulkFormatForSplit(
HiveSourceSplit split) {
if (!useMapRedReader &&
useParquetVectorizedRead(split.getHiveTablePartition())) {
- LOG.debug(String.format("Use native parquet reader for %s.",
split.toString()));
+ LOG.debug(String.format("Use native parquet reader for %s.",
split));
return ParquetColumnarRowInputFormat.createPartitionedFormat(
jobConfWrapper.conf(),
producedRowType,
@@ -146,7 +146,7 @@ public class HiveInputFormat implements BulkFormat<RowData,
HiveSourceSplit> {
hiveVersion.startsWith("3"),
false);
} else if (!useMapRedReader &&
useOrcVectorizedRead(split.getHiveTablePartition())) {
- LOG.debug(String.format("Use native orc reader for %s.",
split.toString()));
+ LOG.debug(String.format("Use native orc reader for %s.", split));
return createOrcFormat();
} else {
if (useMapRedReader) {
@@ -157,7 +157,7 @@ public class HiveInputFormat implements BulkFormat<RowData,
HiveSourceSplit> {
LOG.debug(
String.format(
"Use MapReduce RecordReader reader because the
conditions of vectorized read are not met for %s.",
- split.toString()));
+ split));
}
return new HiveMapRedBulkFormat();
}
diff --git
a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceStatisticsReportTest.java
b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceStatisticsReportTest.java
new file mode 100644
index 00000000000..4917d50bda3
--- /dev/null
+++
b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceStatisticsReportTest.java
@@ -0,0 +1,418 @@
+/*
+ * 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() {
+ FlinkStatistic statistic =
+ getStatisticsFromOptimizedPlan(
+ String.format(
+ "select * from %s.%s.%s limit 1",
+ catalogName, dbName, sourceTable));
+ assertThat(statistic.getTableStats()).isEqualTo(new TableStats(1));
+ }
+
+ @Test
+ public void testHiveTableSourceWithoutData() {
+ tEnv.executeSql(
+ String.format(
+ "create table hive.db1.orcTable ( %s ) stored as orc",
+ String.join(", ",
ddlTypesMapToStringList(ddlTypesMap()))));
+ FlinkStatistic statistic =
+ getStatisticsFromOptimizedPlan(
+ String.format("select * from %s.%s.%s", catalogName,
dbName, "orcTable"));
+ assertThat(statistic.getTableStats()).isEqualTo(new TableStats(0));
+ }
+
+ @Test
+ public void testHiveTableSourceWithoutDataWithLimitPushDown() {
+ tEnv.executeSql(
+ String.format(
+ "create table hive.db1.orcTable ( %s ) stored as orc",
+ String.join(", ",
ddlTypesMapToStringList(ddlTypesMap()))));
+ FlinkStatistic statistic =
+ getStatisticsFromOptimizedPlan(
+ String.format(
+ "select * from %s.%s.%s limit 1", catalogName,
dbName, "orcTable"));
+ assertThat(statistic.getTableStats()).isEqualTo(new TableStats(0));
+ }
+
+ @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");
+ String rowName = ddlTypesMap.remove("row<col1 string, col2 int>");
+ ddlTypesMap.put("timestamp", timestampTypeName);
+ ddlTypesMap.put("binary", binaryTypeName);
+ ddlTypesMap.put("STRUCT<col1 : string, col2 : int>", rowName);
+
+ 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> timestampData = dataMap.remove("timestamp(3)");
+ dataMap.remove("timestamp(9)");
+ dataMap.remove("timestamp without time zone");
+ dataMap.remove("timestamp with local time zone");
+ List<Object> binaryData = dataMap.remove("binary(1)");
+ dataMap.remove("varbinary(1)");
+ dataMap.remove("time");
+ List<Object> rowData = dataMap.remove("row<col1 string, col2 int>");
+ dataMap.put("timestamp", timestampData);
+ dataMap.put("binary", binaryData);
+ dataMap.put("row<col1 string, col2 int>", rowData);
+
+ return dataMap;
+ }
+
+ private static void assertHiveTableOrcFormatTableStatsEquals(
+ TableStats tableStats, int expectedRowCount, long nullCount) {
+ Map<String, ColumnStats> expectedColumnStatsMap = new HashMap<>();
+ expectedColumnStatsMap.put(
+ "f_boolean", new
ColumnStats.Builder().setNullCount(nullCount).build());
+ expectedColumnStatsMap.put(
+ "f_tinyint",
+ new
ColumnStats.Builder().setMax(3L).setMin(1L).setNullCount(0L).build());
+ expectedColumnStatsMap.put(
+ "f_smallint",
+ new
ColumnStats.Builder().setMax(128L).setMin(100L).setNullCount(0L).build());
+ expectedColumnStatsMap.put(
+ "f_int",
+ new ColumnStats.Builder()
+ .setMax(45536L)
+ .setMin(31000L)
+ .setNullCount(nullCount)
+ .build());
+ expectedColumnStatsMap.put(
+ "f_bigint",
+ new ColumnStats.Builder()
+ .setMax(1238123899121L)
+ .setMin(1238123899000L)
+ .setNullCount(0L)
+ .build());
+ expectedColumnStatsMap.put(
+ "f_float",
+ new ColumnStats.Builder()
+ .setMax(33.33300018310547D)
+ .setMin(33.31100082397461D)
+ .setNullCount(nullCount)
+ .build());
+ expectedColumnStatsMap.put(
+ "f_double",
+ new
ColumnStats.Builder().setMax(10.1D).setMin(1.1D).setNullCount(0L).build());
+ expectedColumnStatsMap.put(
+ "f_string",
+ new
ColumnStats.Builder().setMax("def").setMin("abcd").setNullCount(0L).build());
+ expectedColumnStatsMap.put(
+ "f_decimal5",
+ new ColumnStats.Builder()
+ .setMax(new BigDecimal("223.45"))
+ .setMin(new BigDecimal("123.45"))
+ .setNullCount(0L)
+ .build());
+ expectedColumnStatsMap.put(
+ "f_decimal14",
+ new ColumnStats.Builder()
+ .setMax(new BigDecimal("123333333355.33"))
+ .setMin(new BigDecimal("123333333333.33"))
+ .setNullCount(0L)
+ .build());
+ expectedColumnStatsMap.put(
+ "f_decimal38",
+ new ColumnStats.Builder()
+ .setMax(new
BigDecimal("123433343334333433343334333433343334.34"))
+ .setMin(new
BigDecimal("123433343334333433343334333433343334.33"))
+ .setNullCount(nullCount)
+ .build());
+ expectedColumnStatsMap.put(
+ "f_date",
+ new ColumnStats.Builder()
+ .setMax(Date.valueOf("1990-10-16"))
+ .setMin(Date.valueOf("1990-10-14"))
+ .setNullCount(0L)
+ .build());
+ expectedColumnStatsMap.put(
+ "f_timestamp3",
+ new ColumnStats.Builder()
+ .setMax(
+ DateTimeUtils.parseTimestampData("1990-10-16
12:12:43.123", 3)
+ .toTimestamp())
+ .setMin(
+ DateTimeUtils.parseTimestampData("1990-10-14
12:12:43.123", 3)
+ .toTimestamp())
+ .setNullCount(0L)
+ .build());
+ expectedColumnStatsMap.put("f_binary", null);
+ expectedColumnStatsMap.put("f_array", null);
+ expectedColumnStatsMap.put("f_map", null);
+ expectedColumnStatsMap.put("f_row", null);
+
+ assertThat(tableStats).isEqualTo(new TableStats(expectedRowCount,
expectedColumnStatsMap));
+ }
+
+ private static void assertHiveTableParquetFormatTableStatsEquals(
+ TableStats tableStats, int expectedRowCount, long nullCount) {
+ Map<String, ColumnStats> expectedColumnStatsMap = new HashMap<>();
+ expectedColumnStatsMap.put(
+ "f_boolean", new
ColumnStats.Builder().setNullCount(nullCount).build());
+ expectedColumnStatsMap.put(
+ "f_tinyint",
+ new
ColumnStats.Builder().setMax(3).setMin(1).setNullCount(0L).build());
+ expectedColumnStatsMap.put(
+ "f_smallint",
+ new
ColumnStats.Builder().setMax(128).setMin(100).setNullCount(0L).build());
+ expectedColumnStatsMap.put(
+ "f_int",
+ new ColumnStats.Builder()
+ .setMax(45536)
+ .setMin(31000)
+ .setNullCount(nullCount)
+ .build());
+ expectedColumnStatsMap.put(
+ "f_bigint",
+ new ColumnStats.Builder()
+ .setMax(1238123899121L)
+ .setMin(1238123899000L)
+ .setNullCount(0L)
+ .build());
+ expectedColumnStatsMap.put(
+ "f_float",
+ new ColumnStats.Builder()
+ .setMax(33.333F)
+ .setMin(33.311F)
+ .setNullCount(nullCount)
+ .build());
+ expectedColumnStatsMap.put(
+ "f_double",
+ new
ColumnStats.Builder().setMax(10.1D).setMin(1.1D).setNullCount(0L).build());
+ expectedColumnStatsMap.put(
+ "f_string",
+ new
ColumnStats.Builder().setMax("def").setMin("abcd").setNullCount(0L).build());
+ expectedColumnStatsMap.put(
+ "f_decimal5",
+ new ColumnStats.Builder()
+ .setMax(new BigDecimal("223.45"))
+ .setMin(new BigDecimal("123.45"))
+ .setNullCount(0L)
+ .build());
+ expectedColumnStatsMap.put(
+ "f_decimal14",
+ new ColumnStats.Builder()
+ .setMax(new BigDecimal("123333333355.33"))
+ .setMin(new BigDecimal("123333333333.33"))
+ .setNullCount(0L)
+ .build());
+ expectedColumnStatsMap.put(
+ "f_decimal38",
+ new ColumnStats.Builder()
+ .setMax(new
BigDecimal("123433343334333433343334333433343334.34"))
+ .setMin(new
BigDecimal("123433343334333433343334333433343334.33"))
+ .setNullCount(nullCount)
+ .build());
+ expectedColumnStatsMap.put(
+ "f_date",
+ new ColumnStats.Builder()
+ .setMax(Date.valueOf("1990-10-16"))
+ .setMin(Date.valueOf("1990-10-14"))
+ .setNullCount(0L)
+ .build());
+ // Now parquet store timestamp as type int96, and int96 now not
support statistics, so
+ // timestamp not support statistics now.
+ expectedColumnStatsMap.put("f_timestamp3", null);
+
+ expectedColumnStatsMap.put("f_binary", new
ColumnStats.Builder().setNullCount(0L).build());
+ expectedColumnStatsMap.put("f_array", null);
+ expectedColumnStatsMap.put("f_map", null);
+ expectedColumnStatsMap.put("f_row", null);
+ assertThat(tableStats).isEqualTo(new TableStats(expectedRowCount,
expectedColumnStatsMap));
+ }
+}
diff --git
a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java
b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java
index 2054c182a01..391e1c49f20 100644
---
a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java
+++
b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java
@@ -30,11 +30,10 @@ import
org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.table.factories.BulkReaderFormatFactory;
import org.apache.flink.connector.file.table.factories.BulkWriterFormatFactory;
import org.apache.flink.connector.file.table.format.BulkDecodingFormat;
-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.formats.common.Converter;
import
org.apache.flink.formats.csv.RowDataToCsvConverters.RowDataToCsvConverter;
+import org.apache.flink.formats.csv.util.CsvFormatStatisticsReportUtil;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.Projection;
import org.apache.flink.table.connector.format.EncodingFormat;
@@ -55,11 +54,6 @@ import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.Csv
import org.apache.commons.lang3.StringEscapeUtils;
-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.Collections;
import java.util.List;
import java.util.Set;
@@ -154,53 +148,7 @@ public class CsvFileFormatFactory implements
BulkReaderFormatFactory, BulkWriter
@Override
public TableStats reportStatistics(List<Path> files, DataType
producedDataType) {
- // 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;
- }
+ return CsvFormatStatisticsReportUtil.getTableStatistics(files);
}
}
diff --git
a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/util/CsvFormatStatisticsReportUtil.java
b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/util/CsvFormatStatisticsReportUtil.java
new file mode 100644
index 00000000000..bb0fa317acc
--- /dev/null
+++
b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/util/CsvFormatStatisticsReportUtil.java
@@ -0,0 +1,89 @@
+/*
+ * 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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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 {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(CsvFormatStatisticsReportUtil.class);
+
+ 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) {
+ LOG.warn("Reporting statistics failed for Csv format: {}",
e.getMessage());
+ return TableStats.UNKNOWN;
+ }
+ }
+}
diff --git
a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFilesystemStatisticsReportTest.java
b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFilesystemStatisticsReportTest.java
index 476319d6ce4..72ecbeeabbc 100644
---
a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFilesystemStatisticsReportTest.java
+++
b/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 {
@BeforeEach
- public void setup() throws IOException {
- util = batchTestUtil(TableConfig.getDefault());
- tEnv = util.getTableEnv();
- String pathName = writeData(path, Arrays.asList("1,1,hi", "2,1,hello",
"3,2,hello world"));
-
- String ddl =
- String.format(
- "CREATE TABLE sourceTable (\n"
- + " a bigint,\n"
- + " b int,\n"
- + " c varchar\n"
- + ") with (\n"
- + " 'connector' = 'filesystem',"
- + " 'format' = 'csv',"
- + " 'path' = '%s')",
- pathName);
- tEnv.executeSql(ddl);
+ public void setup(@TempDir File file) throws Exception {
+ super.setup(file);
}
@Test
- public void testCsvFileSystemStatisticsReport() {
+ public void testCsvFileSystemStatisticsReport()
+ throws ExecutionException, InterruptedException {
+ // insert data and get statistics by get plan.
+ DataType dataType =
tEnv.from("sourceTable").getResolvedSchema().toPhysicalRowDataType();
+ tEnv.fromValues(dataType,
getData()).executeInsert("sourceTable").await();
FlinkStatistic statistic = getStatisticsFromOptimizedPlan("select *
from sourceTable");
assertThat(statistic.getTableStats()).isEqualTo(new TableStats(3));
}
-
- private String writeData(File path, List<String> data) throws IOException {
- String file = path.getAbsolutePath() + "/00-00.tmp";
- Files.write(new File(file).toPath(), String.join("\n",
data).getBytes());
- return file;
- }
-
- private FlinkStatistic getStatisticsFromOptimizedPlan(String sql) {
- RelNode relNode = TableTestUtil.toRelNode(tEnv.sqlQuery(sql));
- RelNode optimized = util.getPlanner().optimize(relNode);
- FlinkStatisticVisitor visitor = new FlinkStatisticVisitor();
- visitor.go(optimized);
- return visitor.result;
- }
-
- private static class FlinkStatisticVisitor extends RelVisitor {
- private FlinkStatistic result = null;
-
- @Override
- public void visit(RelNode node, int ordinal, RelNode parent) {
- if (node instanceof TableScan) {
- Preconditions.checkArgument(result == null);
- TableSourceTable table = (TableSourceTable) node.getTable();
- result = table.getStatistic();
- }
- super.visit(node, ordinal, parent);
- }
- }
}
diff --git
a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatStatisticsReportTest.java
b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatStatisticsReportTest.java
index 6a166e2d0b2..a96f938dc7f 100644
---
a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatStatisticsReportTest.java
+++
b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatStatisticsReportTest.java
@@ -21,9 +21,12 @@ package org.apache.flink.formats.csv;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.table.plan.stats.TableStats;
+import org.apache.flink.table.planner.utils.StatisticsReportTestBase;
+import org.apache.flink.table.types.DataType;
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;
@@ -33,61 +36,58 @@ import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
/** Test for statistics functionality in {@link CsvFormatFactory}. */
-public class CsvFormatStatisticsReportTest {
+public class CsvFormatStatisticsReportTest extends StatisticsReportTestBase {
private static CsvFileFormatFactory.CsvBulkDecodingFormat
csvBulkDecodingFormat;
@BeforeEach
- public void setup() {
+ public void setup(@TempDir File file) throws Exception {
+ super.setup(file);
+ createFileSystemSource();
Configuration configuration = new Configuration();
csvBulkDecodingFormat = new
CsvFileFormatFactory.CsvBulkDecodingFormat(configuration);
}
- @Test
- public void testCsvFormatStatsReportWithSingleFile() throws IOException {
- String fileContent =
- "#description of the data\n"
- + "header1|header2|header3|\n"
- + "this is|1|2.0|\n"
- + "//a comment\n"
- + "a test|3|4.0|\n"
- + "#next|5|6.0|\n";
-
- Path tempFile = createTempFile(fileContent);
+ @Override
+ protected String[] properties() {
+ List<String> ret = new ArrayList<>();
+ ret.add("'format' = 'csv'");
+ return ret.toArray(new String[0]);
+ }
+ @Test
+ public void testCsvFormatStatsReportWithSingleFile() throws Exception {
+ // insert data and get statistics.
+ DataType dataType =
tEnv.from("sourceTable").getResolvedSchema().toPhysicalRowDataType();
+ tEnv.fromValues(dataType,
getData()).executeInsert("sourceTable").await();
+ assertThat(folder.listFiles()).isNotNull().hasSize(1);
+ File[] files = folder.listFiles();
+ assert files != null;
TableStats tableStats =
-
csvBulkDecodingFormat.reportStatistics(Collections.singletonList(tempFile),
null);
- assertThat(tableStats).isEqualTo(new TableStats(6));
+ csvBulkDecodingFormat.reportStatistics(
+ Collections.singletonList(new
Path(files[0].toURI().toString())), null);
+ assertThat(tableStats).isEqualTo(new TableStats(3));
}
@Test
- public void testCsvFormatStatsReportWithMultiFile() throws IOException {
- String fileContent1 =
- "#description of the data\r\n"
- + "header1|header2|header3|\r\n"
- + "this is|1|2.0|\r\n"
- + "//a comment\r\n"
- + "a test|3|4.0|\r\n"
- + "#next|5|6.0|\r\n";
- String fileContent2 =
- "#description of the data\r\n"
- + "header1|header2|header3|\r\n"
- + "this is|1|2.0|\r\n"
- + "//a comment\r\n"
- + "a test|3|4.0|\r\n"
- + "#next|5|6.0|\r\n";
- Path tempFile1 = createTempFile(fileContent1);
- Path tempFile2 = createTempFile(fileContent2);
- List<Path> files = new ArrayList<>();
- files.add(tempFile1);
- files.add(tempFile2);
-
- TableStats tableStats = csvBulkDecodingFormat.reportStatistics(files,
null);
- assertThat(tableStats).isEqualTo(new TableStats(12));
+ public void testCsvFormatStatsReportWithMultiFile() throws Exception {
+ // insert data and get statistics.
+ DataType dataType =
tEnv.from("sourceTable").getResolvedSchema().toPhysicalRowDataType();
+ tEnv.fromValues(dataType,
getData()).executeInsert("sourceTable").await();
+ tEnv.fromValues(dataType,
getData()).executeInsert("sourceTable").await();
+ assertThat(folder.listFiles()).isNotNull().hasSize(2);
+ File[] files = folder.listFiles();
+ List<Path> paths = new ArrayList<>();
+ assert files != null;
+ paths.add(new Path(files[0].toURI().toString()));
+ paths.add(new Path(files[1].toURI().toString()));
+ TableStats tableStats = csvBulkDecodingFormat.reportStatistics(paths,
null);
+ assertThat(tableStats).isEqualTo(new TableStats(6));
}
@Test
@@ -109,6 +109,22 @@ public class CsvFormatStatisticsReportTest {
assertThat(tableStats).isEqualTo(TableStats.UNKNOWN);
}
+ @Override
+ protected Map<String, String> ddlTypesMap() {
+ // now Csv format don't support type MAP, so we remove this type.
+ Map<String, String> ddlTypes = super.ddlTypesMap();
+ ddlTypes.remove("map<string, int>");
+ return ddlTypes;
+ }
+
+ @Override
+ protected Map<String, List<Object>> getDataMap() {
+ // now Csv format don't support type MAP, so we remove data belong to
this type.
+ Map<String, List<Object>> dataMap = super.getDataMap();
+ dataMap.remove("map<string, int>");
+ return dataMap;
+ }
+
private static Path createTempFile(String content) throws IOException {
File tempFile = File.createTempFile("test_contents", "tmp");
tempFile.deleteOnExit();
diff --git
a/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/OrcColumnarRowInputFormat.java
b/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/OrcColumnarRowInputFormat.java
index d719a1c6daa..d7963db1102 100644
---
a/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/OrcColumnarRowInputFormat.java
+++
b/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/OrcColumnarRowInputFormat.java
@@ -23,13 +23,18 @@ import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.util.Pool;
import org.apache.flink.connector.file.table.ColumnarRowIterator;
import org.apache.flink.connector.file.table.PartitionFieldExtractor;
+import org.apache.flink.core.fs.Path;
import org.apache.flink.orc.shim.OrcShim;
+import org.apache.flink.orc.util.OrcFormatStatisticsReportUtil;
import org.apache.flink.orc.vector.ColumnBatchFactory;
import org.apache.flink.orc.vector.OrcVectorizedBatchWrapper;
+import
org.apache.flink.table.connector.format.FileBasedStatisticsReportableInputFormat;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.columnar.ColumnarRowData;
import org.apache.flink.table.data.columnar.vector.ColumnVector;
import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
+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;
@@ -56,7 +61,8 @@ import static
org.apache.flink.orc.vector.AbstractOrcColumnVector.createFlinkVec
* different and types of extra fields need to be added.
*/
public class OrcColumnarRowInputFormat<BatchT, SplitT extends FileSourceSplit>
- extends AbstractOrcFileInputFormat<RowData, BatchT, SplitT> {
+ extends AbstractOrcFileInputFormat<RowData, BatchT, SplitT>
+ implements FileBasedStatisticsReportableInputFormat {
private static final long serialVersionUID = 1L;
@@ -94,6 +100,12 @@ public class OrcColumnarRowInputFormat<BatchT, SplitT
extends FileSourceSplit>
return this.producedTypeInfo;
}
+ @Override
+ public TableStats reportStatistics(List<Path> files, DataType
producedDataType) {
+ return OrcFormatStatisticsReportUtil.getTableStatistics(
+ files, producedDataType,
hadoopConfigWrapper.getHadoopConfig());
+ }
+
// ------------------------------------------------------------------------
/** One batch of ORC columnar vectors and Flink column vectors. */
diff --git
a/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/OrcFileFormatFactory.java
b/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/OrcFileFormatFactory.java
index 02057fd66be..5eb37873265 100644
---
a/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/OrcFileFormatFactory.java
+++
b/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/OrcFileFormatFactory.java
@@ -29,6 +29,7 @@ import
org.apache.flink.connector.file.table.factories.BulkWriterFormatFactory;
import org.apache.flink.connector.file.table.format.BulkDecodingFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.orc.shim.OrcShim;
+import org.apache.flink.orc.util.OrcFormatStatisticsReportUtil;
import org.apache.flink.orc.vector.RowDataVectorizer;
import org.apache.flink.orc.writer.OrcBulkWriterFactory;
import org.apache.flink.table.connector.ChangelogMode;
@@ -43,34 +44,18 @@ import
org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.factories.DynamicTableFactory;
-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.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
-import java.util.Map;
import java.util.Properties;
import java.util.Set;
@@ -103,6 +88,13 @@ public class OrcFileFormatFactory implements
BulkReaderFormatFactory, BulkWriter
return orcProperties;
}
+ private static Configuration getOrcConfiguration(ReadableConfig
formatOptions) {
+ Properties properties = getOrcProperties(formatOptions);
+ Configuration hadoopConf = new Configuration();
+ properties.forEach((k, v) -> hadoopConf.set(k.toString(),
v.toString()));
+ return hadoopConf;
+ }
+
@Override
public BulkDecodingFormat<RowData> createDecodingFormat(
DynamicTableFactory.Context context, ReadableConfig formatOptions)
{
@@ -165,13 +157,9 @@ public class OrcFileFormatFactory implements
BulkReaderFormatFactory, BulkWriter
}
}
- Properties properties = getOrcProperties(formatOptions);
- Configuration conf = new Configuration();
- properties.forEach((k, v) -> conf.set(k.toString(), v.toString()));
-
return OrcColumnarRowInputFormat.createPartitionedFormat(
OrcShim.defaultShim(),
- conf,
+ getOrcConfiguration(formatOptions),
(RowType) producedDataType.getLogicalType(),
Collections.emptyList(),
null,
@@ -193,177 +181,8 @@ public class OrcFileFormatFactory implements
BulkReaderFormatFactory, BulkWriter
@Override
public TableStats reportStatistics(List<Path> files, DataType
producedDataType) {
- try {
- Properties properties = getOrcProperties(formatOptions);
- Configuration hadoopConfig = new Configuration();
- properties.forEach((k, v) -> hadoopConfig.set(k.toString(),
v.toString()));
-
- 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;
- }
- }
-
- private long updateStatistics(
- Configuration hadoopConf,
- Path file,
- Map<String, ColumnStatistics> columnStatisticsMap,
- RowType producedRowType)
- throws IOException {
- org.apache.hadoop.fs.Path path = new
org.apache.hadoop.fs.Path(file.toUri());
- Reader reader =
- OrcFile.createReader(
- path,
- OrcFile.readerOptions(hadoopConf)
-
.maxLength(OrcConf.MAX_FILE_LENGTH.getLong(hadoopConf)));
- ColumnStatistics[] statistics = reader.getStatistics();
- TypeDescription schema = reader.getSchema();
- List<String> fieldNames = schema.getFieldNames();
- List<TypeDescription> columnTypes = schema.getChildren();
- for (String column : producedRowType.getFieldNames()) {
- int fieldIdx = fieldNames.indexOf(column);
- if (fieldIdx >= 0) {
- int colId = columnTypes.get(fieldIdx).getId();
- ColumnStatistics statistic = statistics[colId];
- updateStatistics(statistic, column, columnStatisticsMap);
- }
- }
-
- return reader.getNumberOfRows();
- }
-
- private void updateStatistics(
- ColumnStatistics statistic,
- String column,
- Map<String, ColumnStatistics> columnStatisticsMap) {
- ColumnStatistics previousStatistics =
columnStatisticsMap.get(column);
- if (previousStatistics == null) {
- columnStatisticsMap.put(column, statistic);
- } else {
- if (previousStatistics instanceof ColumnStatisticsImpl) {
- ((ColumnStatisticsImpl) previousStatistics)
- .merge((ColumnStatisticsImpl) statistic);
- }
- }
- }
-
- private Map<String, ColumnStats> convertToColumnStats(
- long totalRowCount,
- Map<String, ColumnStatistics> columnStatisticsMap,
- RowType logicalType) {
- Map<String, ColumnStats> columnStatsMap = new HashMap<>();
- for (String column : logicalType.getFieldNames()) {
- ColumnStatistics columnStatistics =
columnStatisticsMap.get(column);
- if (columnStatistics == null) {
- continue;
- }
- ColumnStats columnStats =
- convertToColumnStats(
- totalRowCount,
-
logicalType.getTypeAt(logicalType.getFieldIndex(column)),
- columnStatistics);
- columnStatsMap.put(column, columnStats);
- }
-
- return columnStatsMap;
- }
-
- private ColumnStats convertToColumnStats(
- long totalRowCount, LogicalType logicalType, ColumnStatistics
columnStatistics) {
- ColumnStats.Builder builder =
- new
ColumnStats.Builder().setNdv(null).setAvgLen(null).setMaxLen(null);
- if (!columnStatistics.hasNull()) {
- builder.setNullCount(0L);
- } else {
- builder.setNullCount(totalRowCount -
columnStatistics.getNumberOfValues());
- }
-
- // For complex types: ROW, ARRAY, MAP. The returned statistics
have wrong null count
- // value, so now complex types stats return null.
- switch (logicalType.getTypeRoot()) {
- case BOOLEAN:
- break;
- case TINYINT:
- case SMALLINT:
- case INTEGER:
- case BIGINT:
- if (columnStatistics instanceof IntegerColumnStatistics) {
- builder.setMax(((IntegerColumnStatistics)
columnStatistics).getMaximum())
- .setMin(((IntegerColumnStatistics)
columnStatistics).getMinimum());
- break;
- } else {
- return null;
- }
- case FLOAT:
- case DOUBLE:
- if (columnStatistics instanceof DoubleColumnStatistics) {
- builder.setMax(((DoubleColumnStatistics)
columnStatistics).getMaximum())
- .setMin(((DoubleColumnStatistics)
columnStatistics).getMinimum());
- break;
- } else {
- return null;
- }
- case CHAR:
- case VARCHAR:
- if (columnStatistics instanceof StringColumnStatistics) {
- builder.setMax(((StringColumnStatistics)
columnStatistics).getMaximum())
- .setMin(((StringColumnStatistics)
columnStatistics).getMinimum());
- break;
- } else {
- return null;
- }
- case DATE:
- if (columnStatistics instanceof DateColumnStatistics) {
- Date maximum =
- (Date) ((DateColumnStatistics)
columnStatistics).getMaximum();
- Date minimum =
- (Date) ((DateColumnStatistics)
columnStatistics).getMinimum();
- builder.setMax(maximum).setMin(minimum);
- break;
- } else {
- return null;
- }
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- case TIMESTAMP_WITH_TIME_ZONE:
- if (columnStatistics instanceof TimestampColumnStatistics)
{
- builder.setMax(((TimestampColumnStatistics)
columnStatistics).getMaximum())
- .setMin(
- ((TimestampColumnStatistics)
columnStatistics)
- .getMinimum());
- break;
- } else {
- return null;
- }
- case DECIMAL:
- if (columnStatistics instanceof DecimalColumnStatistics) {
- builder.setMax(
- ((DecimalColumnStatistics)
columnStatistics)
- .getMaximum()
- .bigDecimalValue())
- .setMin(
- ((DecimalColumnStatistics)
columnStatistics)
- .getMinimum()
- .bigDecimalValue());
- break;
- } else {
- return null;
- }
- default:
- return null;
- }
- return builder.build();
+ return OrcFormatStatisticsReportUtil.getTableStatistics(
+ files, producedDataType,
getOrcConfiguration(formatOptions));
}
}
}
diff --git
a/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/util/OrcFormatStatisticsReportUtil.java
b/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/util/OrcFormatStatisticsReportUtil.java
new file mode 100644
index 00000000000..a4abfd63ad2
--- /dev/null
+++
b/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/util/OrcFormatStatisticsReportUtil.java
@@ -0,0 +1,220 @@
+/*
+ * 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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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 {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(OrcFormatStatisticsReportUtil.class);
+
+ 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) {
+ LOG.warn("Reporting statistics failed for Orc format: {}",
e.getMessage());
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private static long updateStatistics(
+ Configuration hadoopConf,
+ Path file,
+ Map<String, ColumnStatistics> columnStatisticsMap,
+ RowType producedRowType)
+ throws IOException {
+ org.apache.hadoop.fs.Path path = new
org.apache.hadoop.fs.Path(file.toUri());
+ Reader reader =
+ OrcFile.createReader(
+ path,
+ OrcFile.readerOptions(hadoopConf)
+
.maxLength(OrcConf.MAX_FILE_LENGTH.getLong(hadoopConf)));
+ ColumnStatistics[] statistics = reader.getStatistics();
+ TypeDescription schema = reader.getSchema();
+ List<String> fieldNames = schema.getFieldNames();
+ List<TypeDescription> columnTypes = schema.getChildren();
+ for (String column : producedRowType.getFieldNames()) {
+ int fieldIdx = fieldNames.indexOf(column);
+ if (fieldIdx >= 0) {
+ int colId = columnTypes.get(fieldIdx).getId();
+ ColumnStatistics statistic = statistics[colId];
+ updateStatistics(statistic, column, columnStatisticsMap);
+ }
+ }
+
+ return reader.getNumberOfRows();
+ }
+
+ private static void updateStatistics(
+ ColumnStatistics statistic,
+ String column,
+ Map<String, ColumnStatistics> columnStatisticsMap) {
+ ColumnStatistics previousStatistics = columnStatisticsMap.get(column);
+ if (previousStatistics == null) {
+ columnStatisticsMap.put(column, statistic);
+ } else {
+ if (previousStatistics instanceof ColumnStatisticsImpl) {
+ ((ColumnStatisticsImpl)
previousStatistics).merge((ColumnStatisticsImpl) statistic);
+ }
+ }
+ }
+
+ private static Map<String, ColumnStats> convertToColumnStats(
+ long totalRowCount,
+ Map<String, ColumnStatistics> columnStatisticsMap,
+ RowType logicalType) {
+ Map<String, ColumnStats> columnStatsMap = new HashMap<>();
+ for (String column : logicalType.getFieldNames()) {
+ ColumnStatistics columnStatistics =
columnStatisticsMap.get(column);
+ if (columnStatistics == null) {
+ continue;
+ }
+ ColumnStats columnStats =
+ convertToColumnStats(
+ totalRowCount,
+
logicalType.getTypeAt(logicalType.getFieldIndex(column)),
+ columnStatistics);
+ columnStatsMap.put(column, columnStats);
+ }
+
+ return columnStatsMap;
+ }
+
+ private static ColumnStats convertToColumnStats(
+ long totalRowCount, LogicalType logicalType, ColumnStatistics
columnStatistics) {
+ ColumnStats.Builder builder =
+ new
ColumnStats.Builder().setNdv(null).setAvgLen(null).setMaxLen(null);
+ if (!columnStatistics.hasNull()) {
+ builder.setNullCount(0L);
+ } else {
+ builder.setNullCount(totalRowCount -
columnStatistics.getNumberOfValues());
+ }
+
+ // For complex types: ROW, ARRAY, MAP. The returned statistics have
wrong null count
+ // value, so now complex types stats return null.
+ switch (logicalType.getTypeRoot()) {
+ case BOOLEAN:
+ break;
+ case TINYINT:
+ case SMALLINT:
+ case INTEGER:
+ case BIGINT:
+ if (columnStatistics instanceof IntegerColumnStatistics) {
+ builder.setMax(((IntegerColumnStatistics)
columnStatistics).getMaximum())
+ .setMin(((IntegerColumnStatistics)
columnStatistics).getMinimum());
+ break;
+ } else {
+ return null;
+ }
+ case FLOAT:
+ case DOUBLE:
+ if (columnStatistics instanceof DoubleColumnStatistics) {
+ builder.setMax(((DoubleColumnStatistics)
columnStatistics).getMaximum())
+ .setMin(((DoubleColumnStatistics)
columnStatistics).getMinimum());
+ break;
+ } else {
+ return null;
+ }
+ case CHAR:
+ case VARCHAR:
+ if (columnStatistics instanceof StringColumnStatistics) {
+ builder.setMax(((StringColumnStatistics)
columnStatistics).getMaximum())
+ .setMin(((StringColumnStatistics)
columnStatistics).getMinimum());
+ break;
+ } else {
+ return null;
+ }
+ case DATE:
+ if (columnStatistics instanceof DateColumnStatistics) {
+ Date maximum = (Date) ((DateColumnStatistics)
columnStatistics).getMaximum();
+ Date minimum = (Date) ((DateColumnStatistics)
columnStatistics).getMinimum();
+ builder.setMax(maximum).setMin(minimum);
+ break;
+ } else {
+ return null;
+ }
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_TIME_ZONE:
+ if (columnStatistics instanceof TimestampColumnStatistics) {
+ builder.setMax(((TimestampColumnStatistics)
columnStatistics).getMaximum())
+ .setMin(((TimestampColumnStatistics)
columnStatistics).getMinimum());
+ break;
+ } else {
+ return null;
+ }
+ case DECIMAL:
+ if (columnStatistics instanceof DecimalColumnStatistics) {
+ builder.setMax(
+ ((DecimalColumnStatistics)
columnStatistics)
+ .getMaximum()
+ .bigDecimalValue())
+ .setMin(
+ ((DecimalColumnStatistics)
columnStatistics)
+ .getMinimum()
+ .bigDecimalValue());
+ break;
+ } else {
+ return null;
+ }
+ default:
+ return null;
+ }
+ return builder.build();
+ }
+}
diff --git
a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormat.java
b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormat.java
index 5caa2cb773c..85550adac35 100644
---
a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormat.java
+++
b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormat.java
@@ -24,13 +24,17 @@ import org.apache.flink.connector.file.src.util.Pool;
import org.apache.flink.connector.file.table.ColumnarRowIterator;
import org.apache.flink.connector.file.table.PartitionFieldExtractor;
import org.apache.flink.core.fs.Path;
+import
org.apache.flink.formats.parquet.utils.ParquetFormatStatisticsReportUtil;
import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
import org.apache.flink.formats.parquet.vector.ColumnBatchFactory;
+import
org.apache.flink.table.connector.format.FileBasedStatisticsReportableInputFormat;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.columnar.ColumnarRowData;
import org.apache.flink.table.data.columnar.vector.ColumnVector;
import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
import
org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
+import org.apache.flink.table.plan.stats.TableStats;
+import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.hadoop.conf.Configuration;
@@ -45,7 +49,8 @@ import static
org.apache.flink.formats.parquet.vector.ParquetSplitReaderUtil.cre
* ColumnarRowData} to provide a row view of column batch.
*/
public class ParquetColumnarRowInputFormat<SplitT extends FileSourceSplit>
- extends ParquetVectorizedInputFormat<RowData, SplitT> {
+ extends ParquetVectorizedInputFormat<RowData, SplitT>
+ implements FileBasedStatisticsReportableInputFormat {
private static final long serialVersionUID = 1L;
@@ -117,6 +122,12 @@ public class ParquetColumnarRowInputFormat<SplitT extends
FileSourceSplit>
return producedTypeInfo;
}
+ @Override
+ public TableStats reportStatistics(List<Path> files, DataType
producedDataType) {
+ return ParquetFormatStatisticsReportUtil.getTableStatistics(
+ files, producedDataType, hadoopConfig.conf(), isUtcTimestamp);
+ }
+
private static class ColumnarRowReaderBatch extends
ParquetReaderBatch<RowData> {
private final ColumnarRowIterator result;
diff --git
a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetFileFormatFactory.java
b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetFileFormatFactory.java
index d401da5a8f9..14f257899c1 100644
---
a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetFileFormatFactory.java
+++
b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetFileFormatFactory.java
@@ -29,7 +29,7 @@ import
org.apache.flink.connector.file.table.factories.BulkWriterFormatFactory;
import org.apache.flink.connector.file.table.format.BulkDecodingFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.parquet.row.ParquetRowDataBuilder;
-import org.apache.flink.formats.parquet.vector.reader.TimestampColumnReader;
+import
org.apache.flink.formats.parquet.utils.ParquetFormatStatisticsReportUtil;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.Projection;
import org.apache.flink.table.connector.format.EncodingFormat;
@@ -38,48 +38,19 @@ import
org.apache.flink.table.connector.format.ProjectableDecodingFormat;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
import org.apache.flink.table.factories.DynamicTableFactory;
-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.DecimalType;
-import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
-import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.hadoop.conf.Configuration;
-import org.apache.parquet.Preconditions;
-import org.apache.parquet.column.statistics.BinaryStatistics;
-import org.apache.parquet.column.statistics.DoubleStatistics;
-import org.apache.parquet.column.statistics.FloatStatistics;
-import org.apache.parquet.column.statistics.IntStatistics;
-import org.apache.parquet.column.statistics.LongStatistics;
-import org.apache.parquet.column.statistics.Statistics;
-import org.apache.parquet.hadoop.ParquetFileReader;
-import org.apache.parquet.hadoop.metadata.BlockMetaData;
-import org.apache.parquet.hadoop.metadata.ParquetMetadata;
-import org.apache.parquet.io.api.Binary;
-import org.apache.parquet.schema.MessageType;
-import org.apache.parquet.schema.Type;
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.math.BigInteger;
-import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
-import java.sql.Date;
-import java.sql.Time;
-import java.sql.Timestamp;
import java.util.Collections;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
-import java.util.Map;
import java.util.Properties;
import java.util.Set;
-import java.util.stream.Collectors;
import static org.apache.flink.configuration.ConfigOptions.key;
@@ -185,238 +156,11 @@ public class ParquetFileFormatFactory implements
BulkReaderFormatFactory, BulkWr
@Override
public TableStats reportStatistics(List<Path> files, DataType
producedDataType) {
- try {
- Configuration hadoopConfig =
getParquetConfiguration(formatOptions);
- Map<String, Statistics<?>> columnStatisticsMap = new
HashMap<>();
- RowType producedRowType = (RowType)
producedDataType.getLogicalType();
- long rowCount = 0;
- for (Path file : files) {
- rowCount += updateStatistics(hadoopConfig, file,
columnStatisticsMap);
- }
- Map<String, ColumnStats> columnStatsMap =
- convertToColumnStats(columnStatisticsMap,
producedRowType);
- return new TableStats(rowCount, columnStatsMap);
- } catch (Exception e) {
- return TableStats.UNKNOWN;
- }
- }
-
- private Map<String, ColumnStats> convertToColumnStats(
- Map<String, Statistics<?>> columnStatisticsMap, RowType
producedRowType) {
- Map<String, ColumnStats> columnStatMap = new HashMap<>();
- for (String column : producedRowType.getFieldNames()) {
- Statistics<?> statistics = columnStatisticsMap.get(column);
- if (statistics == null) {
- continue;
- }
- ColumnStats columnStats =
- convertToColumnStats(
-
producedRowType.getTypeAt(producedRowType.getFieldIndex(column)),
- statistics);
- columnStatMap.put(column, columnStats);
- }
- return columnStatMap;
- }
-
- private ColumnStats convertToColumnStats(
- LogicalType logicalType, Statistics<?> statistics) {
- ColumnStats.Builder builder =
- new
ColumnStats.Builder().setNullCount(statistics.getNumNulls());
-
- // For complex types: ROW, ARRAY, MAP. The returned statistics
have wrong null count
- // value, so now complex types stats return null.
- switch (logicalType.getTypeRoot()) {
- case BOOLEAN:
- case BINARY:
- case VARBINARY:
- break;
- case TINYINT:
- case SMALLINT:
- case INTEGER:
- case BIGINT:
- if (statistics instanceof IntStatistics) {
- builder.setMin(((IntStatistics) statistics).getMin())
- .setMax(((IntStatistics) statistics).getMax());
- } else if (statistics instanceof LongStatistics) {
- builder.setMin(((LongStatistics) statistics).getMin())
- .setMax(((LongStatistics)
statistics).getMax());
- } else {
- return null;
- }
- break;
- case DOUBLE:
- if (statistics instanceof DoubleStatistics) {
- builder.setMin(((DoubleStatistics)
statistics).getMin())
- .setMax(((DoubleStatistics)
statistics).getMax());
- break;
- } else {
- return null;
- }
- case FLOAT:
- if (statistics instanceof FloatStatistics) {
- builder.setMin(((FloatStatistics) statistics).getMin())
- .setMax(((FloatStatistics)
statistics).getMax());
- break;
- } else {
- return null;
- }
- case DATE:
- if (statistics instanceof IntStatistics) {
- Date min =
- Date.valueOf(
- DateTimeUtils.formatDate(
- ((IntStatistics)
statistics).getMin()));
- Date max =
- Date.valueOf(
- DateTimeUtils.formatDate(
- ((IntStatistics)
statistics).getMax()));
- builder.setMin(min).setMax(max);
- break;
- } else {
- return null;
- }
- case TIME_WITHOUT_TIME_ZONE:
- if (statistics instanceof IntStatistics) {
- Time min =
- Time.valueOf(
- DateTimeUtils.toLocalTime(
- ((IntStatistics)
statistics).getMin()));
- Time max =
- Time.valueOf(
- DateTimeUtils.toLocalTime(
- ((IntStatistics)
statistics).getMax()));
- builder.setMin(min).setMax(max);
- break;
- } else {
- return null;
- }
- case CHAR:
- case VARCHAR:
- if (statistics instanceof BinaryStatistics) {
- Binary min = ((BinaryStatistics)
statistics).genericGetMin();
- Binary max = ((BinaryStatistics)
statistics).genericGetMax();
- if (min != null) {
- builder.setMin(min.toStringUsingUTF8());
- } else {
- builder.setMin(null);
- }
- if (max != null) {
- builder.setMax(max.toStringUsingUTF8());
- } else {
- builder.setMax(null);
- }
- break;
- } else {
- return null;
- }
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- case TIMESTAMP_WITH_TIME_ZONE:
- if (statistics instanceof LongStatistics) {
- builder.setMin(new Timestamp(((LongStatistics)
statistics).getMin()))
- .setMax(new Timestamp(((LongStatistics)
statistics).getMax()));
- } else if (statistics instanceof BinaryStatistics) {
- Binary min = ((BinaryStatistics)
statistics).genericGetMin();
- Binary max = ((BinaryStatistics)
statistics).genericGetMax();
- if (min != null) {
- builder.setMin(binaryToTimestamp(min,
formatOptions.get(UTC_TIMEZONE)));
- } else {
- builder.setMin(null);
- }
- if (max != null) {
- builder.setMax(binaryToTimestamp(max,
formatOptions.get(UTC_TIMEZONE)));
- } else {
- builder.setMax(null);
- }
- } else {
- return null;
- }
- break;
- case DECIMAL:
- if (statistics instanceof IntStatistics) {
- builder.setMin(BigDecimal.valueOf(((IntStatistics)
statistics).getMin()))
- .setMax(BigDecimal.valueOf(((IntStatistics)
statistics).getMax()));
- } else if (statistics instanceof LongStatistics) {
- builder.setMin(BigDecimal.valueOf(((LongStatistics)
statistics).getMin()))
- .setMax(BigDecimal.valueOf(((LongStatistics)
statistics).getMax()));
- } else if (statistics instanceof BinaryStatistics) {
- Binary min = ((BinaryStatistics)
statistics).genericGetMin();
- Binary max = ((BinaryStatistics)
statistics).genericGetMax();
- if (min != null) {
- builder.setMin(
- binaryToDecimal(min, ((DecimalType)
logicalType).getScale()));
- } else {
- builder.setMin(null);
- }
- if (max != null) {
- builder.setMax(
- binaryToDecimal(max, ((DecimalType)
logicalType).getScale()));
- } else {
- builder.setMax(null);
- }
- } else {
- return null;
- }
- break;
- default:
- return null;
- }
- return builder.build();
- }
-
- private long updateStatistics(
- Configuration hadoopConfig,
- Path file,
- Map<String, Statistics<?>> columnStatisticsMap)
- throws IOException {
- org.apache.hadoop.fs.Path hadoopPath = new
org.apache.hadoop.fs.Path(file.toUri());
- ParquetMetadata metadata =
ParquetFileReader.readFooter(hadoopConfig, hadoopPath);
- MessageType schema = metadata.getFileMetaData().getSchema();
- List<String> columns =
- schema.asGroupType().getFields().stream()
- .map(Type::getName)
- .collect(Collectors.toList());
- List<BlockMetaData> blocks = metadata.getBlocks();
- long rowCount = 0;
- for (BlockMetaData block : blocks) {
- rowCount += block.getRowCount();
- for (int i = 0; i < columns.size(); ++i) {
- updateStatistics(
- block.getColumns().get(i).getStatistics(),
- columns.get(i),
- columnStatisticsMap);
- }
- }
- return rowCount;
- }
-
- private void updateStatistics(
- Statistics<?> statistics,
- String column,
- Map<String, Statistics<?>> columnStatisticsMap) {
- Statistics<?> previousStatistics = columnStatisticsMap.get(column);
- if (previousStatistics == null) {
- columnStatisticsMap.put(column, statistics);
- } else {
- previousStatistics.mergeStatistics(statistics);
- }
- }
-
- private static BigDecimal binaryToDecimal(Binary decimal, int scale) {
- BigInteger bigInteger = new BigInteger(decimal.getBytesUnsafe());
- return new BigDecimal(bigInteger, scale);
- }
-
- private static Timestamp binaryToTimestamp(Binary timestamp, boolean
utcTimestamp) {
- Preconditions.checkArgument(timestamp.length() == 12, "Must be 12
bytes");
- ByteBuffer buf = timestamp.toByteBuffer();
- buf.order(ByteOrder.LITTLE_ENDIAN);
- long timeOfDayNanos = buf.getLong();
- int julianDay = buf.getInt();
-
- TimestampData timestampData =
- TimestampColumnReader.int96ToTimestamp(utcTimestamp,
timeOfDayNanos, julianDay);
- return timestampData.toTimestamp();
+ return ParquetFormatStatisticsReportUtil.getTableStatistics(
+ files,
+ producedDataType,
+ getParquetConfiguration(formatOptions),
+ formatOptions.get(UTC_TIMEZONE));
}
}
}
diff --git
a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
index a6394589fac..0d8654ddb11 100644
---
a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
+++
b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
@@ -81,12 +81,12 @@ public abstract class ParquetVectorizedInputFormat<T,
SplitT extends FileSourceS
private static final Logger LOG =
LoggerFactory.getLogger(ParquetVectorizedInputFormat.class);
private static final long serialVersionUID = 1L;
- private final SerializableConfiguration hadoopConfig;
+ protected final SerializableConfiguration hadoopConfig;
private final String[] projectedFields;
private final LogicalType[] projectedTypes;
private final ColumnBatchFactory<SplitT> batchFactory;
private final int batchSize;
- private final boolean isUtcTimestamp;
+ protected final boolean isUtcTimestamp;
private final boolean isCaseSensitive;
private final Set<Integer> unknownFieldsIndices = new HashSet<>();
diff --git
a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/ParquetFormatStatisticsReportUtil.java
b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/ParquetFormatStatisticsReportUtil.java
new file mode 100644
index 00000000000..d426e2a29b3
--- /dev/null
+++
b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/ParquetFormatStatisticsReportUtil.java
@@ -0,0 +1,307 @@
+/*
+ * 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.parquet.utils;
+
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.formats.parquet.vector.reader.TimestampColumnReader;
+import org.apache.flink.table.data.TimestampData;
+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.DecimalType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.utils.DateTimeUtils;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.Preconditions;
+import org.apache.parquet.column.statistics.BinaryStatistics;
+import org.apache.parquet.column.statistics.DoubleStatistics;
+import org.apache.parquet.column.statistics.FloatStatistics;
+import org.apache.parquet.column.statistics.IntStatistics;
+import org.apache.parquet.column.statistics.LongStatistics;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.Type;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Utils for Parquet format statistics report. */
+public class ParquetFormatStatisticsReportUtil {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ParquetFormatStatisticsReportUtil.class);
+
+ public static TableStats getTableStatistics(
+ List<Path> files,
+ DataType producedDataType,
+ Configuration hadoopConfig,
+ boolean isUtcTimestamp) {
+ try {
+ Map<String, Statistics<?>> columnStatisticsMap = new HashMap<>();
+ RowType producedRowType = (RowType)
producedDataType.getLogicalType();
+ long rowCount = 0;
+ for (Path file : files) {
+ rowCount += updateStatistics(hadoopConfig, file,
columnStatisticsMap);
+ }
+ Map<String, ColumnStats> columnStatsMap =
+ convertToColumnStats(columnStatisticsMap, producedRowType,
isUtcTimestamp);
+ return new TableStats(rowCount, columnStatsMap);
+ } catch (Exception e) {
+ LOG.warn("Reporting statistics failed for Parquet format: {}",
e.getMessage());
+ return TableStats.UNKNOWN;
+ }
+ }
+
+ private static Map<String, ColumnStats> convertToColumnStats(
+ Map<String, Statistics<?>> columnStatisticsMap,
+ RowType producedRowType,
+ boolean isUtcTimestamp) {
+ Map<String, ColumnStats> columnStatMap = new HashMap<>();
+ for (String column : producedRowType.getFieldNames()) {
+ Statistics<?> statistics = columnStatisticsMap.get(column);
+ if (statistics == null) {
+ continue;
+ }
+ ColumnStats columnStats =
+ convertToColumnStats(
+
producedRowType.getTypeAt(producedRowType.getFieldIndex(column)),
+ statistics,
+ isUtcTimestamp);
+ columnStatMap.put(column, columnStats);
+ }
+ return columnStatMap;
+ }
+
+ private static ColumnStats convertToColumnStats(
+ LogicalType logicalType, Statistics<?> statistics, boolean
isUtcTimestamp) {
+ ColumnStats.Builder builder =
+ new
ColumnStats.Builder().setNullCount(statistics.getNumNulls());
+
+ // For complex types: ROW, ARRAY, MAP. The returned statistics have
wrong null count
+ // value, so now complex types stats return null.
+ switch (logicalType.getTypeRoot()) {
+ case BOOLEAN:
+ case BINARY:
+ case VARBINARY:
+ break;
+ case TINYINT:
+ case SMALLINT:
+ case INTEGER:
+ case BIGINT:
+ if (statistics instanceof IntStatistics) {
+ builder.setMin(((IntStatistics) statistics).getMin())
+ .setMax(((IntStatistics) statistics).getMax());
+ } else if (statistics instanceof LongStatistics) {
+ builder.setMin(((LongStatistics) statistics).getMin())
+ .setMax(((LongStatistics) statistics).getMax());
+ } else {
+ return null;
+ }
+ break;
+ case DOUBLE:
+ if (statistics instanceof DoubleStatistics) {
+ builder.setMin(((DoubleStatistics) statistics).getMin())
+ .setMax(((DoubleStatistics) statistics).getMax());
+ break;
+ } else {
+ return null;
+ }
+ case FLOAT:
+ if (statistics instanceof FloatStatistics) {
+ builder.setMin(((FloatStatistics) statistics).getMin())
+ .setMax(((FloatStatistics) statistics).getMax());
+ break;
+ } else {
+ return null;
+ }
+ case DATE:
+ if (statistics instanceof IntStatistics) {
+ Date min =
+ Date.valueOf(
+ DateTimeUtils.formatDate(
+ ((IntStatistics)
statistics).getMin()));
+ Date max =
+ Date.valueOf(
+ DateTimeUtils.formatDate(
+ ((IntStatistics)
statistics).getMax()));
+ builder.setMin(min).setMax(max);
+ break;
+ } else {
+ return null;
+ }
+ case TIME_WITHOUT_TIME_ZONE:
+ if (statistics instanceof IntStatistics) {
+ Time min =
+ Time.valueOf(
+ DateTimeUtils.toLocalTime(
+ ((IntStatistics)
statistics).getMin()));
+ Time max =
+ Time.valueOf(
+ DateTimeUtils.toLocalTime(
+ ((IntStatistics)
statistics).getMax()));
+ builder.setMin(min).setMax(max);
+ break;
+ } else {
+ return null;
+ }
+ case CHAR:
+ case VARCHAR:
+ if (statistics instanceof BinaryStatistics) {
+ Binary min = ((BinaryStatistics)
statistics).genericGetMin();
+ Binary max = ((BinaryStatistics)
statistics).genericGetMax();
+ if (min != null) {
+ builder.setMin(min.toStringUsingUTF8());
+ } else {
+ builder.setMin(null);
+ }
+ if (max != null) {
+ builder.setMax(max.toStringUsingUTF8());
+ } else {
+ builder.setMax(null);
+ }
+ break;
+ } else {
+ return null;
+ }
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ case TIMESTAMP_WITH_TIME_ZONE:
+ if (statistics instanceof LongStatistics) {
+ builder.setMin(new Timestamp(((LongStatistics)
statistics).getMin()))
+ .setMax(new Timestamp(((LongStatistics)
statistics).getMax()));
+ } else if (statistics instanceof BinaryStatistics) {
+ Binary min = ((BinaryStatistics)
statistics).genericGetMin();
+ Binary max = ((BinaryStatistics)
statistics).genericGetMax();
+ if (min != null) {
+ builder.setMin(binaryToTimestamp(min, isUtcTimestamp));
+ } else {
+ builder.setMin(null);
+ }
+ if (max != null) {
+ builder.setMax(binaryToTimestamp(max, isUtcTimestamp));
+ } else {
+ builder.setMax(null);
+ }
+ } else {
+ return null;
+ }
+ break;
+ case DECIMAL:
+ if (statistics instanceof IntStatistics) {
+ builder.setMin(BigDecimal.valueOf(((IntStatistics)
statistics).getMin()))
+ .setMax(BigDecimal.valueOf(((IntStatistics)
statistics).getMax()));
+ } else if (statistics instanceof LongStatistics) {
+ builder.setMin(BigDecimal.valueOf(((LongStatistics)
statistics).getMin()))
+ .setMax(BigDecimal.valueOf(((LongStatistics)
statistics).getMax()));
+ } else if (statistics instanceof BinaryStatistics) {
+ Binary min = ((BinaryStatistics)
statistics).genericGetMin();
+ Binary max = ((BinaryStatistics)
statistics).genericGetMax();
+ if (min != null) {
+ builder.setMin(
+ binaryToDecimal(min, ((DecimalType)
logicalType).getScale()));
+ } else {
+ builder.setMin(null);
+ }
+ if (max != null) {
+ builder.setMax(
+ binaryToDecimal(max, ((DecimalType)
logicalType).getScale()));
+ } else {
+ builder.setMax(null);
+ }
+ } else {
+ return null;
+ }
+ break;
+ default:
+ return null;
+ }
+ return builder.build();
+ }
+
+ private static long updateStatistics(
+ Configuration hadoopConfig, Path file, Map<String, Statistics<?>>
columnStatisticsMap)
+ throws IOException {
+ org.apache.hadoop.fs.Path hadoopPath = new
org.apache.hadoop.fs.Path(file.toUri());
+ ParquetMetadata metadata = ParquetFileReader.readFooter(hadoopConfig,
hadoopPath);
+ MessageType schema = metadata.getFileMetaData().getSchema();
+ List<String> columns =
+ schema.asGroupType().getFields().stream()
+ .map(Type::getName)
+ .collect(Collectors.toList());
+ List<BlockMetaData> blocks = metadata.getBlocks();
+ long rowCount = 0;
+ for (BlockMetaData block : blocks) {
+ rowCount += block.getRowCount();
+ for (int i = 0; i < columns.size(); ++i) {
+ updateStatistics(
+ block.getColumns().get(i).getStatistics(),
+ columns.get(i),
+ columnStatisticsMap);
+ }
+ }
+ return rowCount;
+ }
+
+ private static void updateStatistics(
+ Statistics<?> statistics,
+ String column,
+ Map<String, Statistics<?>> columnStatisticsMap) {
+ Statistics<?> previousStatistics = columnStatisticsMap.get(column);
+ if (previousStatistics == null) {
+ columnStatisticsMap.put(column, statistics);
+ } else {
+ previousStatistics.mergeStatistics(statistics);
+ }
+ }
+
+ private static BigDecimal binaryToDecimal(Binary decimal, int scale) {
+ BigInteger bigInteger = new BigInteger(decimal.getBytesUnsafe());
+ return new BigDecimal(bigInteger, scale);
+ }
+
+ private static Timestamp binaryToTimestamp(Binary timestamp, boolean
utcTimestamp) {
+ Preconditions.checkArgument(timestamp.length() == 12, "Must be 12
bytes");
+ ByteBuffer buf = timestamp.toByteBuffer();
+ buf.order(ByteOrder.LITTLE_ENDIAN);
+ long timeOfDayNanos = buf.getLong();
+ int julianDay = buf.getInt();
+
+ TimestampData timestampData =
+ TimestampColumnReader.int96ToTimestamp(utcTimestamp,
timeOfDayNanos, julianDay);
+ return timestampData.toTimestamp();
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoTableSourceScanRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoTableSourceScanRule.java
index 2bcb97ddd66..7c0fe1e302e 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoTableSourceScanRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoTableSourceScanRule.java
@@ -185,7 +185,7 @@ public class PushPartitionIntoTableSourceScanRule extends
RelOptRule {
TableSourceTable newTableSourceTable =
tableSourceTable.copy(
dynamicTableSource,
- // the statistics will be updated in
FlinkCollectStatisticsProgram
+ // The statistics will be updated in
FlinkRecomputeStatisticsProgram.
tableSourceTable.getStatistic(),
new SourceAbilitySpec[] {partitionPushDownSpec});
LogicalTableScan newScan =
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/file/table/FileSystemStatisticsReportTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/file/table/FileSystemStatisticsReportTest.java
index f53e361fb26..9bb5c207207 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/file/table/FileSystemStatisticsReportTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/file/table/FileSystemStatisticsReportTest.java
@@ -18,29 +18,18 @@
package org.apache.flink.connector.file.table;
-import org.apache.flink.table.api.TableConfig;
-import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.catalog.CatalogPartitionImpl;
import org.apache.flink.table.catalog.CatalogPartitionSpec;
import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
-import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
import org.apache.flink.table.plan.stats.TableStats;
-import org.apache.flink.table.planner.plan.optimize.program.FlinkBatchProgram;
-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.planner.utils.StatisticsReportTestBase;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelVisitor;
-import org.apache.calcite.rel.core.TableScan;
-import org.junit.Before;
-import org.junit.Test;
+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;
@@ -53,19 +42,14 @@ import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
/** Test for statistics functionality in {@link FileSystemTableSource}. */
-public class FileSystemStatisticsReportTest extends TableTestBase {
-
- private BatchTableTestUtil util;
- private TableEnvironment tEnv;
-
- @Before
- public void setup() throws Exception {
- util = batchTestUtil(TableConfig.getDefault());
- util.buildBatchProgram(FlinkBatchProgram.JOIN_REORDER());
- tEnv = util.getTableEnv();
- String path1 = tempFolder().newFile().getAbsolutePath();
- writeData(new File(path1), Arrays.asList("1,1,hi", "2,1,hello",
"3,2,hello world"));
-
+public class FileSystemStatisticsReportTest extends StatisticsReportTestBase {
+
+ @BeforeEach
+ public void setup(@TempDir File file) throws Exception {
+ super.setup(file);
+ String filePath1 =
+ createFileAndWriteData(
+ file, "00-00.tmp", Arrays.asList("1,1,hi",
"2,1,hello", "3,2,hello world"));
String ddl1 =
String.format(
"CREATE TABLE NonPartTable (\n"
@@ -76,12 +60,13 @@ public class FileSystemStatisticsReportTest extends
TableTestBase {
+ " 'connector' = 'filesystem',"
+ " 'format' = 'testcsv',"
+ " 'path' = '%s')",
- path1);
+ filePath1);
tEnv.executeSql(ddl1);
- String path2 = tempFolder().newFolder().getAbsolutePath();
- writeData(new File(path2, "b=1"), Arrays.asList("1,1,hi",
"2,1,hello"));
- writeData(new File(path2, "b=2"), Arrays.asList("3,2,hello world"));
+ File partitionDataPath = new File(file, "partitionData");
+ partitionDataPath.mkdirs();
+ writeData(new File(partitionDataPath, "b=1"), Arrays.asList("1,1,hi",
"2,1,hello"));
+ writeData(new File(partitionDataPath, "b=2"),
Collections.singletonList("3,2,hello world"));
String ddl2 =
String.format(
"CREATE TABLE PartTable (\n"
@@ -92,25 +77,26 @@ public class FileSystemStatisticsReportTest extends
TableTestBase {
+ " 'connector' = 'filesystem',"
+ " 'format' = 'testcsv',"
+ " 'path' = '%s')",
- path2);
+ partitionDataPath.toURI());
tEnv.executeSql(ddl2);
tEnv.getCatalog(tEnv.getCurrentCatalog())
- .get()
+ .orElseThrow(Exception::new)
.createPartition(
new ObjectPath(tEnv.getCurrentDatabase(), "PartTable"),
new CatalogPartitionSpec(Collections.singletonMap("b",
"1")),
new CatalogPartitionImpl(new HashMap<>(), ""),
false);
tEnv.getCatalog(tEnv.getCurrentCatalog())
- .get()
+ .orElseThrow(Exception::new)
.createPartition(
new ObjectPath(tEnv.getCurrentDatabase(), "PartTable"),
new CatalogPartitionSpec(Collections.singletonMap("b",
"2")),
new CatalogPartitionImpl(new HashMap<>(), ""),
false);
- String path3 = tempFolder().newFile().getAbsolutePath();
- writeData(new File(path1), Arrays.asList("1,1,hi", "2,1,hello",
"3,2,hello world"));
+ String filePath2 =
+ createFileAndWriteData(
+ file, "00-01.tmp", Arrays.asList("1,1,hi",
"2,1,hello", "3,2,hello world"));
String ddl3 =
String.format(
@@ -123,8 +109,34 @@ public class FileSystemStatisticsReportTest extends
TableTestBase {
+ " 'format' = 'testcsv',"
+ " 'source.report-statistics' = 'NONE',"
+ " 'path' = '%s')",
- path3);
+ filePath2);
tEnv.executeSql(ddl3);
+
+ String emptyPath = createFileAndWriteData(file, "00-02.tmp",
Collections.emptyList());
+ String ddl4 =
+ String.format(
+ "CREATE TABLE emptyTable (\n"
+ + " a bigint,\n"
+ + " b int,\n"
+ + " c varchar\n"
+ + ") with (\n"
+ + " 'connector' = 'filesystem',"
+ + " 'format' = 'testcsv',"
+ + " 'path' = '%s')",
+ emptyPath);
+ tEnv.executeSql(ddl4);
+ }
+
+ @Override
+ protected String[] properties() {
+ return new String[0];
+ }
+
+ private String createFileAndWriteData(File path, String fileName,
List<String> data)
+ throws IOException {
+ String file = path.getAbsolutePath() + "/" + fileName;
+ Files.write(new File(file).toPath(), String.join("\n",
data).getBytes());
+ return file;
}
private void writeData(File file, List<String> data) throws IOException {
@@ -134,7 +146,7 @@ public class FileSystemStatisticsReportTest extends
TableTestBase {
@Test
public void testCatalogStatisticsExist() throws Exception {
tEnv.getCatalog(tEnv.getCurrentCatalog())
- .get()
+ .orElseThrow(Exception::new)
.alterTableStatistics(
new ObjectPath(tEnv.getCurrentDatabase(),
"NonPartTable"),
new CatalogTableStatistics(10L, 1, 100L, 100L),
@@ -158,9 +170,9 @@ public class FileSystemStatisticsReportTest extends
TableTestBase {
}
@Test
- public void testFilterPushDownAndCatalogStatisticsExist() throws
TableNotExistException {
+ public void testFilterPushDownAndCatalogStatisticsExist() throws Exception
{
tEnv.getCatalog(tEnv.getCurrentCatalog())
- .get()
+ .orElseThrow(Exception::new)
.alterTableStatistics(
new ObjectPath(tEnv.getCurrentDatabase(),
"NonPartTable"),
new CatalogTableStatistics(10L, 1, 100L, 100L),
@@ -190,17 +202,23 @@ public class FileSystemStatisticsReportTest extends
TableTestBase {
}
@Test
- public void testNoPartitionPushDownAndCatalogStatisticsExist()
- throws PartitionNotExistException {
+ public void testLimitPushDownAndCatalogStatisticsDoNotExist() {
+ FlinkStatistic statistic =
+ getStatisticsFromOptimizedPlan("select * from NonPartTable
limit 1");
+ assertThat(statistic.getTableStats()).isEqualTo(new TableStats(1));
+ }
+
+ @Test
+ public void testNoPartitionPushDownAndCatalogStatisticsExist() throws
Exception {
tEnv.getCatalog(tEnv.getCurrentCatalog())
- .get()
+ .orElseThrow(Exception::new)
.alterPartitionStatistics(
new ObjectPath(tEnv.getCurrentDatabase(), "PartTable"),
new CatalogPartitionSpec(Collections.singletonMap("b",
"1")),
new CatalogTableStatistics(6L, 1, 100L, 100L),
false);
tEnv.getCatalog(tEnv.getCurrentCatalog())
- .get()
+ .orElseThrow(Exception::new)
.alterPartitionStatistics(
new ObjectPath(tEnv.getCurrentDatabase(), "PartTable"),
new CatalogPartitionSpec(Collections.singletonMap("b",
"2")),
@@ -222,16 +240,16 @@ public class FileSystemStatisticsReportTest extends
TableTestBase {
}
@Test
- public void testPartitionPushDownAndCatalogStatisticsExist() throws
PartitionNotExistException {
+ public void testPartitionPushDownAndCatalogStatisticsExist() throws
Exception {
tEnv.getCatalog(tEnv.getCurrentCatalog())
- .get()
+ .orElseThrow(Exception::new)
.alterPartitionStatistics(
new ObjectPath(tEnv.getCurrentDatabase(), "PartTable"),
new CatalogPartitionSpec(Collections.singletonMap("b",
"1")),
new CatalogTableStatistics(6L, 1, 100L, 100L),
false);
tEnv.getCatalog(tEnv.getCurrentCatalog())
- .get()
+ .orElseThrow(Exception::new)
.alterPartitionStatistics(
new ObjectPath(tEnv.getCurrentDatabase(), "PartTable"),
new CatalogPartitionSpec(Collections.singletonMap("b",
"2")),
@@ -246,14 +264,14 @@ public class FileSystemStatisticsReportTest extends
TableTestBase {
@Test
public void testFilterPartitionPushDownPushDownAndCatalogStatisticsExist()
throws Exception {
tEnv.getCatalog(tEnv.getCurrentCatalog())
- .get()
+ .orElseThrow(Exception::new)
.alterPartitionStatistics(
new ObjectPath(tEnv.getCurrentDatabase(), "PartTable"),
new CatalogPartitionSpec(Collections.singletonMap("b",
"1")),
new CatalogTableStatistics(6L, 1, 100L, 100L),
false);
tEnv.getCatalog(tEnv.getCurrentCatalog())
- .get()
+ .orElseThrow(Exception::new)
.alterPartitionStatistics(
new ObjectPath(tEnv.getCurrentDatabase(), "PartTable"),
new CatalogPartitionSpec(Collections.singletonMap("b",
"2")),
@@ -283,25 +301,19 @@ public class FileSystemStatisticsReportTest extends
TableTestBase {
assertThat(statistic.getTableStats()).isEqualTo(TableStats.UNKNOWN);
}
- private FlinkStatistic getStatisticsFromOptimizedPlan(String sql) {
- RelNode relNode = TableTestUtil.toRelNode(tEnv.sqlQuery(sql));
- RelNode optimized = util.getPlanner().optimize(relNode);
- FlinkStatisticVisitor visitor = new FlinkStatisticVisitor();
- visitor.go(optimized);
- return visitor.result;
+ @Test
+ public void testFileSystemSourceWithoutData() {
+ FlinkStatistic statistic = getStatisticsFromOptimizedPlan("select *
from emptyTable");
+ assertThat(statistic.getTableStats()).isEqualTo(TableStats.UNKNOWN);
}
- private static class FlinkStatisticVisitor extends RelVisitor {
- private FlinkStatistic result = null;
-
- @Override
- public void visit(RelNode node, int ordinal, RelNode parent) {
- if (node instanceof TableScan) {
- Preconditions.checkArgument(result == null);
- TableSourceTable table = (TableSourceTable) node.getTable();
- result = table.getStatistic();
- }
- super.visit(node, ordinal, parent);
- }
+ @Test
+ public void testFileSystemSourceWithoutDataWithLimitPushDown() {
+ // TODO for source support limit push down and query have limit
condition, In
+ // PushLimitIntoTableSourceScanRule will give stats a new rowCount
value even if this table
+ // have no data.
+ FlinkStatistic statistic =
+ getStatisticsFromOptimizedPlan("select * from emptyTable limit
1");
+ assertThat(statistic.getTableStats()).isEqualTo(new TableStats(1));
}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/formats/testcsv/TestCsvFormatFactory.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/formats/testcsv/TestCsvFormatFactory.java
index 45c64b9fced..2755760b293 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/formats/testcsv/TestCsvFormatFactory.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/formats/testcsv/TestCsvFormatFactory.java
@@ -41,6 +41,9 @@ import org.apache.flink.table.plan.stats.TableStats;
import
org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext;
import org.apache.flink.table.types.DataType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.BufferedReader;
import java.io.File;
import java.io.InputStreamReader;
@@ -59,6 +62,8 @@ import java.util.Set;
public class TestCsvFormatFactory
implements DeserializationFormatFactory, SerializationFormatFactory {
+ private static final Logger LOG =
LoggerFactory.getLogger(TestCsvFormatFactory.class);
+
public static final String IDENTIFIER = "testcsv";
@Override
@@ -173,6 +178,7 @@ public class TestCsvFormatFactory
long estimatedRowCount = totalFileSize * realSampledLineCnt /
sampledRowSize;
return new TableStats(estimatedRowCount);
} catch (Exception e) {
+ LOG.warn("Reporting statistics failed for Csv format: {}",
e.getMessage());
return TableStats.UNKNOWN;
}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/StatisticsReportTestBase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/StatisticsReportTestBase.java
index 764d9689242..d27c739ddd5 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/StatisticsReportTestBase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/StatisticsReportTestBase.java
@@ -294,7 +294,7 @@ public abstract class StatisticsReportTestBase extends
TestLogger {
}
}
- private static String[] ddlTypesMapToStringList(Map<String, String>
ddlTypesMap) {
+ protected static String[] ddlTypesMapToStringList(Map<String, String>
ddlTypesMap) {
String[] types = new String[ddlTypesMap.size()];
int i = 0;
for (Map.Entry<String, String> entry : ddlTypesMap.entrySet()) {
