You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2020/02/24 11:27:59 UTC
[flink] branch release-1.10 updated: [FLINK-16161][hive] Statistics
zero should be unknown in HiveCatalog
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.10 by this push:
new f7dcb02 [FLINK-16161][hive] Statistics zero should be unknown in HiveCatalog
f7dcb02 is described below
commit f7dcb02f933b94d773b00ea844ba66099c45b8f1
Author: JingsongLi <lz...@aliyun.com>
AuthorDate: Mon Feb 24 16:36:19 2020 +0800
[FLINK-16161][hive] Statistics zero should be unknown in HiveCatalog
---
.../flink/table/catalog/hive/HiveCatalog.java | 24 +++++++--------
.../table/catalog/hive/util/HiveStatsUtil.java | 22 +++++++++++++-
.../catalog/hive/HiveCatalogHiveMetadataTest.java | 34 ++++++++++++++++++++++
3 files changed, 66 insertions(+), 14 deletions(-)
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java
index 37bb276..83a5299 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java
@@ -108,6 +108,8 @@ import java.util.Set;
import java.util.stream.Collectors;
import static org.apache.flink.table.catalog.config.CatalogConfig.FLINK_PROPERTY_PREFIX;
+import static org.apache.flink.table.catalog.hive.util.HiveStatsUtil.parsePositiveIntStat;
+import static org.apache.flink.table.catalog.hive.util.HiveStatsUtil.parsePositiveLongStat;
import static org.apache.flink.table.filesystem.PartitionPathUtils.unescapePathName;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -1241,14 +1243,10 @@ public class HiveCatalog extends AbstractCatalog {
* @return whether need to update stats.
*/
private boolean statsChanged(CatalogTableStatistics newTableStats, Map<String, String> parameters) {
- String oldRowCount = parameters.getOrDefault(StatsSetupConst.ROW_COUNT, HiveStatsUtil.DEFAULT_UNKNOWN_STATS_CONST);
- String oldTotalSize = parameters.getOrDefault(StatsSetupConst.TOTAL_SIZE, HiveStatsUtil.DEFAULT_UNKNOWN_STATS_CONST);
- String oldNumFiles = parameters.getOrDefault(StatsSetupConst.NUM_FILES, HiveStatsUtil.DEFAULT_UNKNOWN_STATS_CONST);
- String oldRawDataSize = parameters.getOrDefault(StatsSetupConst.RAW_DATA_SIZE, HiveStatsUtil.DEFAULT_UNKNOWN_STATS_CONST);
- return newTableStats.getRowCount() != Long.parseLong(oldRowCount)
- || newTableStats.getTotalSize() != Long.parseLong(oldTotalSize)
- || newTableStats.getFileCount() != Integer.parseInt(oldNumFiles)
- || newTableStats.getRawDataSize() != Long.parseLong(oldRawDataSize);
+ return newTableStats.getRowCount() != parsePositiveLongStat(parameters, StatsSetupConst.ROW_COUNT)
+ || newTableStats.getTotalSize() != parsePositiveLongStat(parameters, StatsSetupConst.TOTAL_SIZE)
+ || newTableStats.getFileCount() != parsePositiveIntStat(parameters, StatsSetupConst.NUM_FILES)
+ || newTableStats.getRawDataSize() != parsePositiveLongStat(parameters, StatsSetupConst.NUM_FILES);
}
/**
@@ -1264,11 +1262,11 @@ public class HiveCatalog extends AbstractCatalog {
}
private static CatalogTableStatistics createCatalogTableStatistics(Map<String, String> parameters) {
- long rowCount = Long.parseLong(parameters.getOrDefault(StatsSetupConst.ROW_COUNT, HiveStatsUtil.DEFAULT_UNKNOWN_STATS_CONST));
- long totalSize = Long.parseLong(parameters.getOrDefault(StatsSetupConst.TOTAL_SIZE, HiveStatsUtil.DEFAULT_UNKNOWN_STATS_CONST));
- int numFiles = Integer.parseInt(parameters.getOrDefault(StatsSetupConst.NUM_FILES, HiveStatsUtil.DEFAULT_UNKNOWN_STATS_CONST));
- long rawDataSize = Long.parseLong(parameters.getOrDefault(StatsSetupConst.RAW_DATA_SIZE, HiveStatsUtil.DEFAULT_UNKNOWN_STATS_CONST));
- return new CatalogTableStatistics(rowCount, numFiles, totalSize, rawDataSize);
+ return new CatalogTableStatistics(
+ parsePositiveLongStat(parameters, StatsSetupConst.ROW_COUNT),
+ parsePositiveIntStat(parameters, StatsSetupConst.NUM_FILES),
+ parsePositiveLongStat(parameters, StatsSetupConst.TOTAL_SIZE),
+ parsePositiveLongStat(parameters, StatsSetupConst.RAW_DATA_SIZE));
}
@Override
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveStatsUtil.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveStatsUtil.java
index d8107b7..9fd4394 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveStatsUtil.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveStatsUtil.java
@@ -63,7 +63,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
public class HiveStatsUtil {
private static final Logger LOG = LoggerFactory.getLogger(HiveStatsUtil.class);
- public static final String DEFAULT_UNKNOWN_STATS_CONST = "-1";
+ private static final int DEFAULT_UNKNOWN_STATS_VALUE = -1;
private HiveStatsUtil() {}
@@ -292,4 +292,24 @@ public class HiveStatsUtil {
throw new CatalogException(String.format("Flink does not support converting ColumnStats '%s' for Hive column " +
"type '%s' yet", colStat, colType));
}
+
+ public static int parsePositiveIntStat(Map<String, String> parameters, String key) {
+ String value = parameters.get(key);
+ if (value == null) {
+ return DEFAULT_UNKNOWN_STATS_VALUE;
+ } else {
+ int v = Integer.parseInt(value);
+ return v > 0 ? v : DEFAULT_UNKNOWN_STATS_VALUE;
+ }
+ }
+
+ public static long parsePositiveLongStat(Map<String, String> parameters, String key) {
+ String value = parameters.get(key);
+ if (value == null) {
+ return DEFAULT_UNKNOWN_STATS_VALUE;
+ } else {
+ long v = Long.parseLong(value);
+ return v > 0 ? v : DEFAULT_UNKNOWN_STATS_VALUE;
+ }
+ }
}
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogHiveMetadataTest.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogHiveMetadataTest.java
index 21faf29..a576eb1 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogHiveMetadataTest.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogHiveMetadataTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.catalog.CatalogPartitionSpec;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.CatalogTableImpl;
+import org.apache.flink.table.catalog.config.CatalogConfig;
import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataBase;
@@ -32,9 +33,11 @@ import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataDate;
import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataDouble;
import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataLong;
import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataString;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
import org.apache.flink.table.catalog.stats.Date;
import org.apache.flink.util.StringUtils;
+import org.apache.hadoop.hive.common.StatsSetupConst;
import org.apache.hadoop.hive.metastore.api.Table;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -42,6 +45,7 @@ import org.junit.Test;
import java.util.HashMap;
import java.util.Map;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
/**
@@ -128,6 +132,36 @@ public class HiveCatalogHiveMetadataTest extends HiveCatalogMetadataTestBase {
checkEquals(catalogColumnStatistics, catalog.getPartitionColumnStatistics(path1, partitionSpec));
}
+ @Test
+ public void testHiveStatistics() throws Exception {
+ catalog.createDatabase(db1, createDb(), false);
+ checkStatistics(0, -1);
+ checkStatistics(1, 1);
+ checkStatistics(1000, 1000);
+ }
+
+ private void checkStatistics(int inputStat, int expectStat) throws Exception {
+ catalog.dropTable(path1, true);
+
+ Map<String, String> properties = new HashMap<>();
+ properties.put(CatalogConfig.IS_GENERIC, "false");
+ properties.put(StatsSetupConst.ROW_COUNT, String.valueOf(inputStat));
+ properties.put(StatsSetupConst.NUM_FILES, String.valueOf(inputStat));
+ properties.put(StatsSetupConst.TOTAL_SIZE, String.valueOf(inputStat));
+ properties.put(StatsSetupConst.RAW_DATA_SIZE, String.valueOf(inputStat));
+ CatalogTable catalogTable = new CatalogTableImpl(
+ TableSchema.builder().field("f0", DataTypes.INT()).build(),
+ properties,
+ "");
+ catalog.createTable(path1, catalogTable, false);
+
+ CatalogTableStatistics statistics = catalog.getTableStatistics(path1);
+ assertEquals(expectStat, statistics.getRowCount());
+ assertEquals(expectStat, statistics.getFileCount());
+ assertEquals(expectStat, statistics.getRawDataSize());
+ assertEquals(expectStat, statistics.getTotalSize());
+ }
+
// ------ utils ------
@Override