You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/08/02 11:43:06 UTC
[GitHub] [flink] JingGe opened a new pull request, #20422: [FLINK-28778][SQL/API] Bulk fetch of table and column statistics for given partitions
JingGe opened a new pull request, #20422:
URL: https://github.com/apache/flink/pull/20422
## What is the purpose of the change
Currently the statistics information about tables can only be fetched from the catalog by each given partition iteratively. Since getting statistics information from catalogs is a very heavy operation, in order to improve the query performance, we’d better provide functionality to fetch the statistics information of a table for all given partitions in one shot.
Based on the manual performance test, for 2000 partitions, the cost will be improved from 10s to 2s. The improvement result is 500%.
## Brief change log
- Implementations for new added methods will be done in all classes that implement the Catalog interface.
- For the concrete HiveCatalog, __HIVE_DEFAULT_PARTITION__ will be taken care of while fetching the column statistics. - - All currently supported types for partition column will be taken into consideration.
- The logic of getting TableStates in FlinkRecomputeStatisticsProgram will be optimized from calling the catalog iteratively to bulk fetch.
- Logic of TableStates conversion will be consolidated into the CatalogTableStatisticsConverter class to make the domain design a little bit cleaner.
## Verifying this change
This change is already covered by existing tests, such as HiveCatalogHiveMetadataTest.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (**yes** / no)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (**yes** / no)
- If yes, how is the feature documented? (not applicable / **docs** / JavaDocs / not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a diff in pull request #20422: [FLINK-28778][SQL/API] Bulk fetch of table and column statistics for given partitions
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on code in PR #20422:
URL: https://github.com/apache/flink/pull/20422#discussion_r939681964
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/GenericInMemoryCatalog.java:
##########
@@ -720,6 +720,25 @@ public CatalogTableStatistics getPartitionStatistics(
return result != null ? result.copy() : CatalogTableStatistics.UNKNOWN;
}
+ /**
+ * For in memory catalog, just iteratively call {@link #getPartitionStatistics(ObjectPath,
+ * CatalogPartitionSpec)} for each partition.
+ */
+ @Override
+ public List<CatalogTableStatistics> bulkGetPartitionStatistics(
Review Comment:
the implementation is totally same the the default one, I think we can remove it
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkRecomputeStatisticsProgram.java:
##########
@@ -183,36 +181,29 @@ private TableStats getPartitionsTableStats(
} else {
partitionList = partitionPushDownSpec.getPartitions();
}
- for (Map<String, String> partition : partitionList) {
- Optional<TableStats> partitionStats =
- getPartitionStats(catalog, tablePath, partition);
- if (!partitionStats.isPresent()) {
- // clear all information before
- newTableStat = null;
- break;
- } else {
- newTableStat =
- newTableStat == null
- ? partitionStats.get()
- : newTableStat.merge(partitionStats.get());
- }
- }
+ return getPartitionStats(
+ table.contextResolvedTable().getCatalog().get(),
+ table.contextResolvedTable().getIdentifier().toObjectPath(),
+ partitionList)
+ .get();
}
- return newTableStat;
+ return TableStats.UNKNOWN;
}
private Optional<TableStats> getPartitionStats(
- Catalog catalog, ObjectPath tablePath, Map<String, String> partition) {
+ Catalog catalog, ObjectPath tablePath, List<Map<String, String>> partition) {
try {
- CatalogPartitionSpec spec = new CatalogPartitionSpec(partition);
- CatalogTableStatistics partitionStat = catalog.getPartitionStatistics(tablePath, spec);
- CatalogColumnStatistics partitionColStat =
- catalog.getPartitionColumnStatistics(tablePath, spec);
- TableStats stats =
- CatalogTableStatisticsConverter.convertToTableStats(
- partitionStat, partitionColStat);
- return Optional.of(stats);
+
+ final List<CatalogPartitionSpec> partitionSpecs =
+ partition.stream()
+ .map(p -> new CatalogPartitionSpec(p))
Review Comment:
nit: .map(CatalogPartitionSpec::new)
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/CatalogTableStatisticsConverter.java:
##########
@@ -61,6 +63,27 @@ public static TableStats convertToTableStats(
return new TableStats(rowCount, columnStatsMap);
}
+ public static TableStats convertToAccumulatedTableStates(
+ List<CatalogTableStatistics> tableStatisticsList,
+ List<CatalogColumnStatistics> catalogColumnStatisticsList) {
+ final Optional<TableStats> rowCountMergedTableStats =
+ tableStatisticsList.stream()
+ .map(p -> CatalogTableStatisticsConverter.convertToTableStats(p, null))
+ .reduce((s1, s2) -> s1.merge(s2));
+
+ final Optional<TableStats> columnStatsMergedTableStats =
+ catalogColumnStatisticsList.stream()
+ .map(
+ p ->
+ CatalogTableStatisticsConverter.convertToTableStats(
+ CatalogTableStatistics.EMPTY, p))
+ .reduce((s1, s2) -> s1.merge(s2));
Review Comment:
We can just convert a CatalogTableStatistics and its corresponding CatalogColumnStatistics into a TableStats, and then call TableStats#merge. If that, CatalogTableStatistics.EMPTY, StatisticDataUtils can be removed. There's no calculation here, and even CatalogColumnStatisticDataBuilder is no needed. This pr will become very simple
##########
flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogHiveMetadataTest.java:
##########
@@ -240,6 +246,84 @@ void testAlterPartitionColumnStatistics() throws Exception {
catalog.getPartitionColumnStatistics(path1, partitionSpec));
}
+ private ResolvedSchema createSchemaForStatistics() {
+ return new ResolvedSchema(
+ Arrays.asList(
+ Column.physical("first", DataTypes.STRING()),
+ Column.physical("four", DataTypes.BOOLEAN()),
+ Column.physical("five", DataTypes.DOUBLE()),
+ Column.physical("second", DataTypes.DATE()),
+ Column.physical("third", DataTypes.INT())),
+ Collections.emptyList(),
+ null);
+ }
+
+ @Test
+ public void testGetTableAndColumnStatistics() throws Exception {
+
+ // create a test table
+ catalog.createDatabase(db1, createDb(), false);
+ final ResolvedSchema resolvedSchema = createSchemaForStatistics();
+ final CatalogTable origin =
+ CatalogTable.of(
+ Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
+ TEST_COMMENT,
+ createPartitionKeys(),
+ getBatchTableProperties());
+ CatalogTable catalogTable = new ResolvedCatalogTable(origin, resolvedSchema);
+
+ catalog.createTable(path1, catalogTable, false);
+
+ int second = 29;
+ final int partitionFileSize = 20;
+ final String partitionName = "third";
+
+ List<Map<String, String>> partitionSpecs = new ArrayList<>();
+ Map<String, String> partitionSpec = null;
+
+ while (second-- > 10) {
+ int third = 20;
+ while (third-- > 0) {
+ partitionSpec = new HashMap<String, String>();
+ partitionSpec.put("second", "2010-04-" + second + " 00:00:00");
+ partitionSpec.put(partitionName, Integer.toString(third));
+ createPartition(partitionSpec);
+ partitionSpecs.add(partitionSpec);
+ }
+ }
+
+ final List<CatalogPartitionSpec> catalogPartitionSpecList =
+ partitionSpecs.stream()
+ .map(p -> new CatalogPartitionSpec(p))
+ .collect(Collectors.toList());
+
+ final List<CatalogTableStatistics> partitionStatistics =
+ catalog.bulkGetPartitionStatistics(path1, catalogPartitionSpecList);
+ assertThat(partitionStatistics.size()).isEqualTo(partitionSpecs.size());
+
+ final List<CatalogColumnStatistics> tableColumnStatistics =
+ catalog.bulkGetPartitionColumnStatistics(path1, catalogPartitionSpecList);
+
+ assertThat(tableColumnStatistics.size()).isEqualTo(catalogPartitionSpecList.size() + 1);
+
+ final CatalogColumnStatistics catalogColumnStatistics =
+ tableColumnStatistics.get(tableColumnStatistics.size() - 1);
+ CatalogColumnStatisticsDataLong columnStatisticsDataLong =
+ (CatalogColumnStatisticsDataLong)
+ catalogColumnStatistics.getColumnStatisticsData().get(partitionName);
+
+ assertThat(columnStatisticsDataLong.getNdv()).isEqualTo(catalogPartitionSpecList.size());
+ assertThat(columnStatisticsDataLong.getMin()).isEqualTo(0);
+ assertThat(columnStatisticsDataLong.getMax()).isEqualTo(partitionFileSize - 1);
Review Comment:
other columns' statistics does not been verified
##########
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java:
##########
@@ -438,6 +440,38 @@ public CatalogTableStatistics getPartitionStatistics(
return CatalogTableStatistics.UNKNOWN;
}
+ @Override
+ public List<CatalogTableStatistics> bulkGetPartitionStatistics(
Review Comment:
I think we can just use the default implementation, and remove this method.
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java:
##########
@@ -1699,6 +1716,69 @@ public CatalogTableStatistics getPartitionStatistics(
}
}
+ @Override
+ public List<CatalogTableStatistics> bulkGetPartitionStatistics(
+ ObjectPath tablePath, List<CatalogPartitionSpec> partitionSpecs)
+ throws PartitionNotExistException, CatalogException {
+
+ return getTableAndPartitions(tablePath, partitionSpecs).f1.stream()
+ .map(p -> createCatalogTableStatistics(p.getParameters()))
+ .collect(Collectors.toList());
+ }
+
+ /** @return Partitions and names of partitions. */
+ private Tuple3<Table, List<Partition>, List<String>> getTableAndPartitions(
Review Comment:
We can just split this method into two methods, one is to get the HiveTable, and another is to get partitions. That could make the result type more simple and readable
##########
flink-table/flink-table-common/src/test/java/org/apache/flink/table/factories/TestCatalogFactory.java:
##########
@@ -316,13 +316,27 @@ public CatalogTableStatistics getPartitionStatistics(
throw new UnsupportedOperationException();
}
+ @Override
+ public List<CatalogTableStatistics> bulkGetPartitionStatistics(
+ ObjectPath tablePath, List<CatalogPartitionSpec> partitionSpecs)
+ throws PartitionNotExistException, CatalogException {
+ throw new UnsupportedClassVersionError();
Review Comment:
should throw `UnsupportedOperationException`
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java:
##########
@@ -1734,6 +1814,152 @@ public CatalogColumnStatistics getPartitionColumnStatistics(
}
}
+ @Override
+ public List<CatalogColumnStatistics> bulkGetPartitionColumnStatistics(
+ ObjectPath tablePath, List<CatalogPartitionSpec> partitionSpecs)
+ throws PartitionNotExistException, CatalogException {
+
+ checkNotNull(partitionSpecs);
+
+ List<CatalogColumnStatistics> result = new ArrayList<>(partitionSpecs.size());
+
+ final Tuple3<Table, List<Partition>, List<String>> partitionsTuple3 =
+ getTableAndPartitions(tablePath, partitionSpecs);
+
+ try {
+
+ final List<FieldSchema> distinctColumns =
+ partitionsTuple3.f1.stream()
+ .map(p -> p.getSd().getCols())
+ .flatMap(Collection::stream)
+ .distinct()
+ .collect(Collectors.toList());
+
+ Map<String, List<ColumnStatisticsObj>> partitionColumnStatistics =
+ client.getPartitionColumnStatistics(
+ partitionsTuple3.f1.get(0).getDbName(),
+ partitionsTuple3.f1.get(0).getTableName(),
+ partitionsTuple3.f2,
+ getFieldNames(distinctColumns));
+
+ for (String partitionName : partitionsTuple3.f2) {
+ List<ColumnStatisticsObj> columnStatisticsObjs =
+ partitionColumnStatistics.get(partitionName);
+ if (columnStatisticsObjs != null && !columnStatisticsObjs.isEmpty()) {
+ result.add(
+ new CatalogColumnStatistics(
+ HiveStatsUtil.createCatalogColumnStats(
+ columnStatisticsObjs, hiveVersion)));
+ } else {
+ result.add(CatalogColumnStatistics.UNKNOWN);
+ }
+ }
+
+ result.add(
+ buildPartitionedByColumnSatistics(
+ tablePath,
+ partitionsTuple3
+ .f0
+ .getPartitionKeys()
+ .get(partitionsTuple3.f0.getPartitionKeys().size() - 1),
+ partitionsTuple3.f2));
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format(
+ "Failed to get table stats of table %s 's partitions %s",
+ tablePath.getFullName(), String.valueOf(partitionSpecs)),
+ e);
+ }
+
+ return result;
+ }
+
+ private CatalogColumnStatistics buildPartitionedByColumnSatistics(
+ ObjectPath tablePath, FieldSchema partitionedByColumn, List<String> partitionNames) {
+
+ Long nullCount = null;
+ Long ndv = HiveStatsUtil.max(0L, Long.valueOf(partitionNames.size()));
+
+ CatalogColumnStatisticDataBuilder<? extends CatalogColumnStatisticsDataBase> builder = null;
+
+ final CatalogColumnStatisticsDataLong.Builder longBuilder =
+ CatalogColumnStatisticsDataLong.builder();
+ longBuilder.ndv(ndv);
+ final CatalogColumnStatisticsDataDate.Builder dateBuilder =
+ CatalogColumnStatisticsDataDate.builder();
+ dateBuilder.ndv(ndv);
+ final CatalogColumnStatisticsDataDouble.Builder doubleBuilder =
+ CatalogColumnStatisticsDataDouble.builder();
+ doubleBuilder.ndv(ndv);
+ final CatalogColumnStatisticsDataString.Builder stringBuilder =
+ CatalogColumnStatisticsDataString.builder();
+ stringBuilder.ndv(ndv);
+ final CatalogColumnStatisticsDataBoolean.Builder booleanBuilder =
+ CatalogColumnStatisticsDataBoolean.builder();
+ final CatalogColumnStatisticsDataBinary.Builder binaryBuilder =
+ CatalogColumnStatisticsDataBinary.builder();
+
+ for (String partitionName : partitionNames) {
+ String[] partName = partitionName.split(Path.SEPARATOR);
+ String[] partVal = partName[partName.length - 1].split("=");
+ String value = partVal[1];
+ if (value.equals(HIVE_DEFAULT_PARTITION)) {
+ try {
+ nullCount =
Review Comment:
Not set nullCount
##########
flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogHiveMetadataTest.java:
##########
@@ -240,6 +246,84 @@ void testAlterPartitionColumnStatistics() throws Exception {
catalog.getPartitionColumnStatistics(path1, partitionSpec));
}
+ private ResolvedSchema createSchemaForStatistics() {
+ return new ResolvedSchema(
+ Arrays.asList(
+ Column.physical("first", DataTypes.STRING()),
+ Column.physical("four", DataTypes.BOOLEAN()),
+ Column.physical("five", DataTypes.DOUBLE()),
+ Column.physical("second", DataTypes.DATE()),
+ Column.physical("third", DataTypes.INT())),
+ Collections.emptyList(),
+ null);
+ }
+
+ @Test
+ public void testGetTableAndColumnStatistics() throws Exception {
+
+ // create a test table
+ catalog.createDatabase(db1, createDb(), false);
+ final ResolvedSchema resolvedSchema = createSchemaForStatistics();
+ final CatalogTable origin =
+ CatalogTable.of(
+ Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
+ TEST_COMMENT,
+ createPartitionKeys(),
+ getBatchTableProperties());
+ CatalogTable catalogTable = new ResolvedCatalogTable(origin, resolvedSchema);
+
+ catalog.createTable(path1, catalogTable, false);
+
+ int second = 29;
+ final int partitionFileSize = 20;
+ final String partitionName = "third";
+
+ List<Map<String, String>> partitionSpecs = new ArrayList<>();
+ Map<String, String> partitionSpec = null;
+
+ while (second-- > 10) {
+ int third = 20;
+ while (third-- > 0) {
+ partitionSpec = new HashMap<String, String>();
+ partitionSpec.put("second", "2010-04-" + second + " 00:00:00");
+ partitionSpec.put(partitionName, Integer.toString(third));
+ createPartition(partitionSpec);
+ partitionSpecs.add(partitionSpec);
+ }
+ }
+
+ final List<CatalogPartitionSpec> catalogPartitionSpecList =
+ partitionSpecs.stream()
+ .map(p -> new CatalogPartitionSpec(p))
+ .collect(Collectors.toList());
+
+ final List<CatalogTableStatistics> partitionStatistics =
+ catalog.bulkGetPartitionStatistics(path1, catalogPartitionSpecList);
+ assertThat(partitionStatistics.size()).isEqualTo(partitionSpecs.size());
+
+ final List<CatalogColumnStatistics> tableColumnStatistics =
+ catalog.bulkGetPartitionColumnStatistics(path1, catalogPartitionSpecList);
+
+ assertThat(tableColumnStatistics.size()).isEqualTo(catalogPartitionSpecList.size() + 1);
+
+ final CatalogColumnStatistics catalogColumnStatistics =
+ tableColumnStatistics.get(tableColumnStatistics.size() - 1);
+ CatalogColumnStatisticsDataLong columnStatisticsDataLong =
+ (CatalogColumnStatisticsDataLong)
+ catalogColumnStatistics.getColumnStatisticsData().get(partitionName);
+
+ assertThat(columnStatisticsDataLong.getNdv()).isEqualTo(catalogPartitionSpecList.size());
+ assertThat(columnStatisticsDataLong.getMin()).isEqualTo(0);
+ assertThat(columnStatisticsDataLong.getMax()).isEqualTo(partitionFileSize - 1);
+
+ final List<CatalogPartitionSpec> catalogPartitionSpecs = catalog.listPartitions(path1);
Review Comment:
This line can be removed ?
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/Catalog.java:
##########
@@ -566,6 +565,31 @@ CatalogTableStatistics getPartitionStatistics(
ObjectPath tablePath, CatalogPartitionSpec partitionSpec)
throws PartitionNotExistException, CatalogException;
+ /**
+ * Get a list of statistics of given partitions.
+ *
+ * @param tablePath path of the table
+ * @param partitionSpecs partition specs of partitions that will be used to filter out all other
+ * unrelated statistics, i.e. the statistics fetch will be limited within the given
+ * partitions
+ * @return list of statistics of given partitions
+ * @throws PartitionNotExistException if one partition does not exist
+ * @throws CatalogException in case of any runtime exception
+ */
+ default List<CatalogTableStatistics> bulkGetPartitionStatistics(
Review Comment:
This method should be tested via GenericInMemoryCatalog
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/stats/StatisticDataUtils.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.table.catalog.stats;
+
+/** Utils class for catalog column statistics. */
+public class StatisticDataUtils {
+
+ public static Double acculmulateAverage(
+ Double oldAvgLength, Long oldCount, Long newLength, Long newCount) {
+ if (oldAvgLength == null || oldCount == null) {
+ return Double.valueOf(newLength);
+ }
+ if (newLength == null || newCount == null) {
+ return oldAvgLength;
+ }
+ return oldCount + newCount == 0
+ ? null
+ : (oldAvgLength * oldCount + newLength * newCount) / (oldCount + newCount);
+ }
+
+ public static Long min(Long oldLength, Long newLength) {
Review Comment:
The parameter name should be more generic
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/Catalog.java:
##########
@@ -311,11 +314,7 @@ void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfE
void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
throws TableNotExistException, CatalogException;
- /**
- * If true, tables which do not specify a connector will be translated to managed tables.
- *
- * @see CatalogBaseTable.TableKind#MANAGED
- */
Review Comment:
Unnecessary change ?
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/plan/stats/ColumnStats.java:
##########
@@ -211,6 +211,11 @@ public ColumnStats copy() {
* @return The merged column stats.
*/
public ColumnStats merge(ColumnStats other) {
+
+ if (other == null) {
Review Comment:
if other is null or unknown, the ColumnStats should return UNKNOWN
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong commented on pull request #20422: [FLINK-28778][SQL/API] Bulk fetch of table and column statistics for given partitions
Posted by GitBox <gi...@apache.org>.
wuchong commented on PR #20422:
URL: https://github.com/apache/flink/pull/20422#issuecomment-1208993062
Continue with #20501
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong closed pull request #20422: [FLINK-28778][SQL/API] Bulk fetch of table and column statistics for given partitions
Posted by GitBox <gi...@apache.org>.
wuchong closed pull request #20422: [FLINK-28778][SQL/API] Bulk fetch of table and column statistics for given partitions
URL: https://github.com/apache/flink/pull/20422
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #20422: [FLINK-28778][SQL/API] Bulk fetch of table and column statistics for given partitions
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #20422:
URL: https://github.com/apache/flink/pull/20422#issuecomment-1202380753
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f1da34372ebcb83aafda286cd8fb595c8bb148ea",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f1da34372ebcb83aafda286cd8fb595c8bb148ea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f1da34372ebcb83aafda286cd8fb595c8bb148ea UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingGe commented on pull request #20422: [FLINK-28778][SQL/API] Bulk fetch of table and column statistics for given partitions
Posted by GitBox <gi...@apache.org>.
JingGe commented on PR #20422:
URL: https://github.com/apache/flink/pull/20422#issuecomment-1202719566
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #20422: [FLINK-28778][SQL/API] Bulk fetch of table and column statistics for given partitions
Posted by GitBox <gi...@apache.org>.
luoyuxia commented on code in PR #20422:
URL: https://github.com/apache/flink/pull/20422#discussion_r939763843
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java:
##########
@@ -146,10 +159,14 @@
/** A catalog implementation for Hive. */
public class HiveCatalog extends AbstractCatalog {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveCatalog.class);
+
+ public static final String HIVE_DEFAULT_PARTITION = "__HIVE_DEFAULT_PARTITION__";
Review Comment:
The name for Hive default partition is configurable in Hive itself. You can use
`getHiveConf().getVar(HiveConf.ConfVars.DEFAULTPARTITIONNAME)`
to get it.
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java:
##########
@@ -1734,6 +1814,152 @@ public CatalogColumnStatistics getPartitionColumnStatistics(
}
}
+ @Override
+ public List<CatalogColumnStatistics> bulkGetPartitionColumnStatistics(
+ ObjectPath tablePath, List<CatalogPartitionSpec> partitionSpecs)
+ throws PartitionNotExistException, CatalogException {
+
+ checkNotNull(partitionSpecs);
+
+ List<CatalogColumnStatistics> result = new ArrayList<>(partitionSpecs.size());
+
+ final Tuple3<Table, List<Partition>, List<String>> partitionsTuple3 =
Review Comment:
Split this method may improve readability.
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java:
##########
@@ -1699,6 +1716,69 @@ public CatalogTableStatistics getPartitionStatistics(
}
}
+ @Override
+ public List<CatalogTableStatistics> bulkGetPartitionStatistics(
+ ObjectPath tablePath, List<CatalogPartitionSpec> partitionSpecs)
+ throws PartitionNotExistException, CatalogException {
+
+ return getTableAndPartitions(tablePath, partitionSpecs).f1.stream()
+ .map(p -> createCatalogTableStatistics(p.getParameters()))
+ .collect(Collectors.toList());
+ }
+
+ /** @return Partitions and names of partitions. */
+ private Tuple3<Table, List<Partition>, List<String>> getTableAndPartitions(
+ ObjectPath tablePath, List<CatalogPartitionSpec> partitionSpecs)
+ throws PartitionNotExistException, CatalogException {
+
+ List<Partition> partitions = new ArrayList<>(partitionSpecs.size());
+ Tuple2<Table, List<String>> partitionsNamesTuple2 = null;
+
+ try {
+ partitionsNamesTuple2 = getTableAndPartitionsNames(tablePath, partitionSpecs);
+
+ partitions.addAll(
+ client.getPartitionsByNames(
+ partitionsNamesTuple2.f0.getDbName(),
+ partitionsNamesTuple2.f0.getTableName(),
+ partitionsNamesTuple2.f1));
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format(
+ "Failed to get partition stats of table %s 's partitions %s",
+ tablePath.getFullName(), String.valueOf(partitionSpecs)),
Review Comment:
nit:
`String.valueOf` is unnecessary.
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java:
##########
@@ -1734,6 +1814,152 @@ public CatalogColumnStatistics getPartitionColumnStatistics(
}
}
+ @Override
+ public List<CatalogColumnStatistics> bulkGetPartitionColumnStatistics(
+ ObjectPath tablePath, List<CatalogPartitionSpec> partitionSpecs)
+ throws PartitionNotExistException, CatalogException {
+
+ checkNotNull(partitionSpecs);
+
+ List<CatalogColumnStatistics> result = new ArrayList<>(partitionSpecs.size());
+
+ final Tuple3<Table, List<Partition>, List<String>> partitionsTuple3 =
+ getTableAndPartitions(tablePath, partitionSpecs);
+
+ try {
+
+ final List<FieldSchema> distinctColumns =
Review Comment:
Is it for get all columns of the table? If it's, I think we can use `partitionsTuple3.f0.getSd().getCols()`
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java:
##########
@@ -1734,6 +1814,152 @@ public CatalogColumnStatistics getPartitionColumnStatistics(
}
}
+ @Override
+ public List<CatalogColumnStatistics> bulkGetPartitionColumnStatistics(
+ ObjectPath tablePath, List<CatalogPartitionSpec> partitionSpecs)
+ throws PartitionNotExistException, CatalogException {
+
+ checkNotNull(partitionSpecs);
+
+ List<CatalogColumnStatistics> result = new ArrayList<>(partitionSpecs.size());
+
+ final Tuple3<Table, List<Partition>, List<String>> partitionsTuple3 =
+ getTableAndPartitions(tablePath, partitionSpecs);
+
+ try {
+
+ final List<FieldSchema> distinctColumns =
+ partitionsTuple3.f1.stream()
+ .map(p -> p.getSd().getCols())
+ .flatMap(Collection::stream)
+ .distinct()
+ .collect(Collectors.toList());
+
+ Map<String, List<ColumnStatisticsObj>> partitionColumnStatistics =
+ client.getPartitionColumnStatistics(
+ partitionsTuple3.f1.get(0).getDbName(),
+ partitionsTuple3.f1.get(0).getTableName(),
+ partitionsTuple3.f2,
+ getFieldNames(distinctColumns));
+
+ for (String partitionName : partitionsTuple3.f2) {
+ List<ColumnStatisticsObj> columnStatisticsObjs =
+ partitionColumnStatistics.get(partitionName);
+ if (columnStatisticsObjs != null && !columnStatisticsObjs.isEmpty()) {
+ result.add(
+ new CatalogColumnStatistics(
+ HiveStatsUtil.createCatalogColumnStats(
+ columnStatisticsObjs, hiveVersion)));
+ } else {
+ result.add(CatalogColumnStatistics.UNKNOWN);
+ }
+ }
+
+ result.add(
Review Comment:
A little confused about this line. Is it for add column statistic for the partition columns? If so, why only add one element?
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java:
##########
@@ -1734,6 +1814,152 @@ public CatalogColumnStatistics getPartitionColumnStatistics(
}
}
+ @Override
+ public List<CatalogColumnStatistics> bulkGetPartitionColumnStatistics(
+ ObjectPath tablePath, List<CatalogPartitionSpec> partitionSpecs)
+ throws PartitionNotExistException, CatalogException {
+
+ checkNotNull(partitionSpecs);
+
+ List<CatalogColumnStatistics> result = new ArrayList<>(partitionSpecs.size());
+
+ final Tuple3<Table, List<Partition>, List<String>> partitionsTuple3 =
+ getTableAndPartitions(tablePath, partitionSpecs);
+
+ try {
+
+ final List<FieldSchema> distinctColumns =
+ partitionsTuple3.f1.stream()
+ .map(p -> p.getSd().getCols())
+ .flatMap(Collection::stream)
+ .distinct()
+ .collect(Collectors.toList());
+
+ Map<String, List<ColumnStatisticsObj>> partitionColumnStatistics =
+ client.getPartitionColumnStatistics(
+ partitionsTuple3.f1.get(0).getDbName(),
+ partitionsTuple3.f1.get(0).getTableName(),
+ partitionsTuple3.f2,
+ getFieldNames(distinctColumns));
+
+ for (String partitionName : partitionsTuple3.f2) {
+ List<ColumnStatisticsObj> columnStatisticsObjs =
+ partitionColumnStatistics.get(partitionName);
+ if (columnStatisticsObjs != null && !columnStatisticsObjs.isEmpty()) {
+ result.add(
+ new CatalogColumnStatistics(
+ HiveStatsUtil.createCatalogColumnStats(
+ columnStatisticsObjs, hiveVersion)));
+ } else {
+ result.add(CatalogColumnStatistics.UNKNOWN);
+ }
+ }
+
+ result.add(
+ buildPartitionedByColumnSatistics(
+ tablePath,
+ partitionsTuple3
+ .f0
+ .getPartitionKeys()
+ .get(partitionsTuple3.f0.getPartitionKeys().size() - 1),
+ partitionsTuple3.f2));
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format(
+ "Failed to get table stats of table %s 's partitions %s",
+ tablePath.getFullName(), String.valueOf(partitionSpecs)),
+ e);
+ }
+
+ return result;
+ }
+
+ private CatalogColumnStatistics buildPartitionedByColumnSatistics(
+ ObjectPath tablePath, FieldSchema partitionedByColumn, List<String> partitionNames) {
+
+ Long nullCount = null;
+ Long ndv = HiveStatsUtil.max(0L, Long.valueOf(partitionNames.size()));
+
+ CatalogColumnStatisticDataBuilder<? extends CatalogColumnStatisticsDataBase> builder = null;
+
+ final CatalogColumnStatisticsDataLong.Builder longBuilder =
+ CatalogColumnStatisticsDataLong.builder();
+ longBuilder.ndv(ndv);
+ final CatalogColumnStatisticsDataDate.Builder dateBuilder =
+ CatalogColumnStatisticsDataDate.builder();
+ dateBuilder.ndv(ndv);
+ final CatalogColumnStatisticsDataDouble.Builder doubleBuilder =
+ CatalogColumnStatisticsDataDouble.builder();
+ doubleBuilder.ndv(ndv);
+ final CatalogColumnStatisticsDataString.Builder stringBuilder =
+ CatalogColumnStatisticsDataString.builder();
+ stringBuilder.ndv(ndv);
+ final CatalogColumnStatisticsDataBoolean.Builder booleanBuilder =
+ CatalogColumnStatisticsDataBoolean.builder();
+ final CatalogColumnStatisticsDataBinary.Builder binaryBuilder =
+ CatalogColumnStatisticsDataBinary.builder();
+
+ for (String partitionName : partitionNames) {
+ String[] partName = partitionName.split(Path.SEPARATOR);
+ String[] partVal = partName[partName.length - 1].split("=");
Review Comment:
why always choose the last partName?
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java:
##########
@@ -1734,6 +1814,152 @@ public CatalogColumnStatistics getPartitionColumnStatistics(
}
}
+ @Override
+ public List<CatalogColumnStatistics> bulkGetPartitionColumnStatistics(
+ ObjectPath tablePath, List<CatalogPartitionSpec> partitionSpecs)
+ throws PartitionNotExistException, CatalogException {
+
+ checkNotNull(partitionSpecs);
+
+ List<CatalogColumnStatistics> result = new ArrayList<>(partitionSpecs.size());
+
+ final Tuple3<Table, List<Partition>, List<String>> partitionsTuple3 =
+ getTableAndPartitions(tablePath, partitionSpecs);
+
+ try {
+
+ final List<FieldSchema> distinctColumns =
+ partitionsTuple3.f1.stream()
+ .map(p -> p.getSd().getCols())
+ .flatMap(Collection::stream)
+ .distinct()
+ .collect(Collectors.toList());
+
+ Map<String, List<ColumnStatisticsObj>> partitionColumnStatistics =
+ client.getPartitionColumnStatistics(
+ partitionsTuple3.f1.get(0).getDbName(),
+ partitionsTuple3.f1.get(0).getTableName(),
+ partitionsTuple3.f2,
+ getFieldNames(distinctColumns));
+
+ for (String partitionName : partitionsTuple3.f2) {
+ List<ColumnStatisticsObj> columnStatisticsObjs =
+ partitionColumnStatistics.get(partitionName);
+ if (columnStatisticsObjs != null && !columnStatisticsObjs.isEmpty()) {
+ result.add(
+ new CatalogColumnStatistics(
+ HiveStatsUtil.createCatalogColumnStats(
+ columnStatisticsObjs, hiveVersion)));
+ } else {
+ result.add(CatalogColumnStatistics.UNKNOWN);
+ }
+ }
+
+ result.add(
+ buildPartitionedByColumnSatistics(
+ tablePath,
+ partitionsTuple3
+ .f0
+ .getPartitionKeys()
+ .get(partitionsTuple3.f0.getPartitionKeys().size() - 1),
+ partitionsTuple3.f2));
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format(
+ "Failed to get table stats of table %s 's partitions %s",
+ tablePath.getFullName(), String.valueOf(partitionSpecs)),
+ e);
+ }
+
+ return result;
+ }
+
+ private CatalogColumnStatistics buildPartitionedByColumnSatistics(
+ ObjectPath tablePath, FieldSchema partitionedByColumn, List<String> partitionNames) {
+
+ Long nullCount = null;
+ Long ndv = HiveStatsUtil.max(0L, Long.valueOf(partitionNames.size()));
Review Comment:
`Long.valueOf(partitionNames.size())` can be reduced to `(long) partitionNames.size()`.
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java:
##########
@@ -1734,6 +1814,152 @@ public CatalogColumnStatistics getPartitionColumnStatistics(
}
}
+ @Override
+ public List<CatalogColumnStatistics> bulkGetPartitionColumnStatistics(
+ ObjectPath tablePath, List<CatalogPartitionSpec> partitionSpecs)
+ throws PartitionNotExistException, CatalogException {
+
+ checkNotNull(partitionSpecs);
+
+ List<CatalogColumnStatistics> result = new ArrayList<>(partitionSpecs.size());
+
+ final Tuple3<Table, List<Partition>, List<String>> partitionsTuple3 =
+ getTableAndPartitions(tablePath, partitionSpecs);
+
+ try {
+
+ final List<FieldSchema> distinctColumns =
+ partitionsTuple3.f1.stream()
+ .map(p -> p.getSd().getCols())
+ .flatMap(Collection::stream)
+ .distinct()
+ .collect(Collectors.toList());
+
+ Map<String, List<ColumnStatisticsObj>> partitionColumnStatistics =
+ client.getPartitionColumnStatistics(
+ partitionsTuple3.f1.get(0).getDbName(),
+ partitionsTuple3.f1.get(0).getTableName(),
+ partitionsTuple3.f2,
+ getFieldNames(distinctColumns));
+
+ for (String partitionName : partitionsTuple3.f2) {
+ List<ColumnStatisticsObj> columnStatisticsObjs =
+ partitionColumnStatistics.get(partitionName);
+ if (columnStatisticsObjs != null && !columnStatisticsObjs.isEmpty()) {
+ result.add(
+ new CatalogColumnStatistics(
+ HiveStatsUtil.createCatalogColumnStats(
+ columnStatisticsObjs, hiveVersion)));
+ } else {
+ result.add(CatalogColumnStatistics.UNKNOWN);
+ }
+ }
+
+ result.add(
+ buildPartitionedByColumnSatistics(
+ tablePath,
+ partitionsTuple3
+ .f0
+ .getPartitionKeys()
+ .get(partitionsTuple3.f0.getPartitionKeys().size() - 1),
+ partitionsTuple3.f2));
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format(
+ "Failed to get table stats of table %s 's partitions %s",
+ tablePath.getFullName(), String.valueOf(partitionSpecs)),
+ e);
+ }
+
+ return result;
+ }
+
+ private CatalogColumnStatistics buildPartitionedByColumnSatistics(
+ ObjectPath tablePath, FieldSchema partitionedByColumn, List<String> partitionNames) {
+
+ Long nullCount = null;
+ Long ndv = HiveStatsUtil.max(0L, Long.valueOf(partitionNames.size()));
+
+ CatalogColumnStatisticDataBuilder<? extends CatalogColumnStatisticsDataBase> builder = null;
+
+ final CatalogColumnStatisticsDataLong.Builder longBuilder =
+ CatalogColumnStatisticsDataLong.builder();
+ longBuilder.ndv(ndv);
+ final CatalogColumnStatisticsDataDate.Builder dateBuilder =
+ CatalogColumnStatisticsDataDate.builder();
+ dateBuilder.ndv(ndv);
+ final CatalogColumnStatisticsDataDouble.Builder doubleBuilder =
+ CatalogColumnStatisticsDataDouble.builder();
+ doubleBuilder.ndv(ndv);
+ final CatalogColumnStatisticsDataString.Builder stringBuilder =
+ CatalogColumnStatisticsDataString.builder();
+ stringBuilder.ndv(ndv);
+ final CatalogColumnStatisticsDataBoolean.Builder booleanBuilder =
+ CatalogColumnStatisticsDataBoolean.builder();
+ final CatalogColumnStatisticsDataBinary.Builder binaryBuilder =
+ CatalogColumnStatisticsDataBinary.builder();
+
+ for (String partitionName : partitionNames) {
+ String[] partName = partitionName.split(Path.SEPARATOR);
+ String[] partVal = partName[partName.length - 1].split("=");
+ String value = partVal[1];
+ if (value.equals(HIVE_DEFAULT_PARTITION)) {
+ try {
+ nullCount =
+ getPartitionStatistics(
+ tablePath,
+ new CatalogPartitionSpec(
+ Collections.singletonMap(
+ partitionedByColumn.getName(),
+ HIVE_DEFAULT_PARTITION)))
+ .getRowCount();
+ } catch (Exception e) {
+ LOG.info(
+ "Fail to get getPartitionColumnStatistics for partition {}={}",
+ partitionedByColumn.getName(),
+ HIVE_DEFAULT_PARTITION);
+ }
+ } else {
+ final String type = partitionedByColumn.getType();
+ if ("int".equalsIgnoreCase(type)) {
Review Comment:
Avoid to use "int" directly. May be we can do the check using LogicalType like `HiveStatsUtil#getColumnStatisticsData`
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/stats/CatalogColumnStatisticsDataBoolean.java:
##########
@@ -54,4 +54,45 @@ public CatalogColumnStatisticsDataBoolean copy() {
return new CatalogColumnStatisticsDataBoolean(
trueCount, falseCount, getNullCount(), new HashMap<>(getProperties()));
}
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /** {@link CatalogColumnStatisticsDataBoolean} builder static inner class. */
+ public static final class Builder
+ implements CatalogColumnStatisticDataBuilder<CatalogColumnStatisticsDataBoolean> {
+ private Long nullCount;
+ private Map<String, String> properties;
+ private Long trueCount = 0L;
+ private Long falseCount = 0L;
+
+ private Builder() {}
+
+ public Builder nullCount(Long nullCount) {
+ this.nullCount = nullCount;
+ return this;
+ }
+
+ public Builder properties(Map<String, String> properties) {
+ this.properties = properties;
+ return this;
+ }
+
+ public Builder increasTrueCount(Long trueCount) {
Review Comment:
nit:
```suggestion
public Builder increaseTrueCount(Long trueCount) {
```
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java:
##########
@@ -1734,6 +1814,152 @@ public CatalogColumnStatistics getPartitionColumnStatistics(
}
}
+ @Override
+ public List<CatalogColumnStatistics> bulkGetPartitionColumnStatistics(
+ ObjectPath tablePath, List<CatalogPartitionSpec> partitionSpecs)
+ throws PartitionNotExistException, CatalogException {
+
+ checkNotNull(partitionSpecs);
+
+ List<CatalogColumnStatistics> result = new ArrayList<>(partitionSpecs.size());
+
+ final Tuple3<Table, List<Partition>, List<String>> partitionsTuple3 =
+ getTableAndPartitions(tablePath, partitionSpecs);
+
+ try {
+
+ final List<FieldSchema> distinctColumns =
Review Comment:
If it's for get partition columns, we can use `partitionsTuple3.f0.getPartitionKeys();`
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/stats/StatisticDataUtils.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.table.catalog.stats;
+
+/** Utils class for catalog column statistics. */
+public class StatisticDataUtils {
+
+ public static Double acculmulateAverage(
Review Comment:
nit:
```suggestion
public static Double accumulateAverage(
```
##########
flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogHiveMetadataTest.java:
##########
@@ -240,6 +246,84 @@ void testAlterPartitionColumnStatistics() throws Exception {
catalog.getPartitionColumnStatistics(path1, partitionSpec));
}
+ private ResolvedSchema createSchemaForStatistics() {
+ return new ResolvedSchema(
+ Arrays.asList(
+ Column.physical("first", DataTypes.STRING()),
+ Column.physical("four", DataTypes.BOOLEAN()),
+ Column.physical("five", DataTypes.DOUBLE()),
+ Column.physical("second", DataTypes.DATE()),
+ Column.physical("third", DataTypes.INT())),
+ Collections.emptyList(),
+ null);
+ }
+
+ @Test
+ public void testGetTableAndColumnStatistics() throws Exception {
+
+ // create a test table
+ catalog.createDatabase(db1, createDb(), false);
+ final ResolvedSchema resolvedSchema = createSchemaForStatistics();
+ final CatalogTable origin =
+ CatalogTable.of(
+ Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
+ TEST_COMMENT,
+ createPartitionKeys(),
+ getBatchTableProperties());
+ CatalogTable catalogTable = new ResolvedCatalogTable(origin, resolvedSchema);
+
+ catalog.createTable(path1, catalogTable, false);
+
+ int second = 29;
+ final int partitionFileSize = 20;
+ final String partitionName = "third";
+
+ List<Map<String, String>> partitionSpecs = new ArrayList<>();
+ Map<String, String> partitionSpec = null;
+
+ while (second-- > 10) {
+ int third = 20;
+ while (third-- > 0) {
+ partitionSpec = new HashMap<String, String>();
+ partitionSpec.put("second", "2010-04-" + second + " 00:00:00");
+ partitionSpec.put(partitionName, Integer.toString(third));
+ createPartition(partitionSpec);
+ partitionSpecs.add(partitionSpec);
+ }
+ }
+
+ final List<CatalogPartitionSpec> catalogPartitionSpecList =
+ partitionSpecs.stream()
+ .map(p -> new CatalogPartitionSpec(p))
+ .collect(Collectors.toList());
+
+ final List<CatalogTableStatistics> partitionStatistics =
+ catalog.bulkGetPartitionStatistics(path1, catalogPartitionSpecList);
+ assertThat(partitionStatistics.size()).isEqualTo(partitionSpecs.size());
+
+ final List<CatalogColumnStatistics> tableColumnStatistics =
+ catalog.bulkGetPartitionColumnStatistics(path1, catalogPartitionSpecList);
+
+ assertThat(tableColumnStatistics.size()).isEqualTo(catalogPartitionSpecList.size() + 1);
+
+ final CatalogColumnStatistics catalogColumnStatistics =
+ tableColumnStatistics.get(tableColumnStatistics.size() - 1);
+ CatalogColumnStatisticsDataLong columnStatisticsDataLong =
+ (CatalogColumnStatisticsDataLong)
+ catalogColumnStatistics.getColumnStatisticsData().get(partitionName);
+
+ assertThat(columnStatisticsDataLong.getNdv()).isEqualTo(catalogPartitionSpecList.size());
+ assertThat(columnStatisticsDataLong.getMin()).isEqualTo(0);
+ assertThat(columnStatisticsDataLong.getMax()).isEqualTo(partitionFileSize - 1);
Review Comment:
+1
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java:
##########
@@ -1699,6 +1716,69 @@ public CatalogTableStatistics getPartitionStatistics(
}
}
+ @Override
+ public List<CatalogTableStatistics> bulkGetPartitionStatistics(
+ ObjectPath tablePath, List<CatalogPartitionSpec> partitionSpecs)
+ throws PartitionNotExistException, CatalogException {
+
+ return getTableAndPartitions(tablePath, partitionSpecs).f1.stream()
+ .map(p -> createCatalogTableStatistics(p.getParameters()))
+ .collect(Collectors.toList());
+ }
+
+ /** @return Partitions and names of partitions. */
+ private Tuple3<Table, List<Partition>, List<String>> getTableAndPartitions(
Review Comment:
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org