[FLINK-14552][table] Enable partition statistics in blink planner

flink-table/flink-table-common/src/main/java/org/apache/flink/table/plan/stats/ColumnStats.java

@@ -22,6 +22,8 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
+import java.util.function.BinaryOperator;
 
 /**
  * Column statistics.
@@ -217,6 +219,92 @@ public ColumnStats copy() {
 		}
 	}
 
+	/**
+	 * Merges two column stats.
+	 * When the stats are unknown, whatever the other are, we need return unknown stats.
+	 * The unknown definition for column stats is null.
+	 *
+	 * @param other The other column stats to merge.
+	 * @return The merged column stats.
+	 */
+	public ColumnStats merge(ColumnStats other) {
+		Long ndv = combineIfNonNull(Long::sum, this.ndv, other.ndv);
+		Long nullCount = combineIfNonNull(Long::sum, this.nullCount, other.nullCount);
+		Double avgLen = combineIfNonNull((a1, a2) -> (a1 + a2) / 2, this.avgLen, other.avgLen);
+		Integer maxLen = combineIfNonNull(Math::max, this.maxLen, other.maxLen);
+
+		Number maxValue = combineIfNonNull(
+				(n1, n2) -> n1.doubleValue() > n2.doubleValue() ? n1 : n2,
+				this.maxValue,
+				other.maxValue);
+		Number minValue = combineIfNonNull(
+				(n1, n2) -> n1.doubleValue() < n2.doubleValue() ? n1 : n2,
+				this.minValue,
+				other.minValue);
+
+		@SuppressWarnings("unchecked")
+		Comparable max = combineIfNonNull(
+				(c1, c2) -> ((Comparable) c1).compareTo(c2) > 0 ? c1 : c2,
+				this.max,
+				other.max);
+		@SuppressWarnings("unchecked")
+		Comparable min = combineIfNonNull(
+				(c1, c2) -> ((Comparable) c1).compareTo(c2) < 0 ? c1 : c2,
+				this.min,
+				other.min);
+
+		if (max != null || min != null) {
+			return new ColumnStats(
+					ndv,
+					nullCount,
+					avgLen,
+					maxLen,
+					max,
+					min
+			);
+		} else {
+			return new ColumnStats(
+					ndv,
+					nullCount,
+					avgLen,
+					maxLen,
+					maxValue,
+					minValue
+			);
+		}
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o) {
+			return true;
+		}
+		if (o == null || getClass() != o.getClass()) {
+			return false;
+		}
+		ColumnStats that = (ColumnStats) o;
+		return Objects.equals(ndv, that.ndv) &&
+				Objects.equals(nullCount, that.nullCount) &&
+				Objects.equals(avgLen, that.avgLen) &&
+				Objects.equals(maxLen, that.maxLen) &&
+				Objects.equals(maxValue, that.maxValue) &&
+				Objects.equals(max, that.max) &&
+				Objects.equals(minValue, that.minValue) &&
+				Objects.equals(min, that.min);
+	}
+
+	@Override
+	public int hashCode() {
+		return Objects.hash(ndv, nullCount, avgLen, maxLen, maxValue, max, minValue, min);
+	}
+
+	private static <T> T combineIfNonNull(BinaryOperator<T> op, T t1, T t2) {
+		if (t1 == null || t2 == null) {
+			return null;
+		}
+		return op.apply(t1, t2);
+	}
+
 	/**
 	 * ColumnStats builder.
 	 */

flink-table/flink-table-common/src/main/java/org/apache/flink/table/plan/stats/TableStats.java

@@ -20,14 +20,23 @@
 
 import org.apache.flink.annotation.PublicEvolving;
 
+import javax.annotation.Nonnull;
+
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * Table statistics.
  */
 @PublicEvolving
 public final class TableStats {
+
+	/**
+	 * Unknown definition for table stats:
+	 * Unknown {@link #rowCount} is -1.
+	 * Unknown {@link #colStats} is not exist in map.
+	 */
 	public static final TableStats UNKNOWN = new TableStats(-1, new HashMap<>());
 
 	/**
@@ -69,4 +78,54 @@ public TableStats copy() {
 		return copy;
 	}
 
+	/**
+	 * Merges two table stats.
+	 * When the stats are unknown, whatever the other are, we need return unknown stats.
+	 * See {@link #UNKNOWN}.
+	 *
+	 * @param other The other table stats to merge.
+	 * @return The merged table stats.
+	 */
+	@Nonnull
+	public TableStats merge(TableStats other) {
+		Map<String, ColumnStats> colStats = new HashMap<>();
+		for (Map.Entry<String, ColumnStats> entry : this.colStats.entrySet()) {
+			String col = entry.getKey();
+			ColumnStats stats = entry.getValue();
+			ColumnStats otherStats = other.colStats.get(col);
+			if (otherStats != null) {
+				colStats.put(col, stats.merge(otherStats));
+			}
+		}
+		return new TableStats(
+				this.rowCount >= 0 && other.rowCount >= 0 ?
+						this.rowCount + other.rowCount : UNKNOWN.rowCount,
+				colStats);
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o) {
+			return true;
+		}
+		if (o == null || getClass() != o.getClass()) {
+			return false;
+		}
+		TableStats that = (TableStats) o;
+		return rowCount == that.rowCount &&
+				Objects.equals(colStats, that.colStats);
+	}
+
+	@Override
+	public int hashCode() {
+		return Objects.hash(rowCount, colStats);
+	}
+
+	@Override
+	public String toString() {
+		return "TableStats{" +
+				"rowCount=" + rowCount +
+				", colStats=" + colStats +
+				'}';
+	}
 }

flink-table/flink-table-common/src/test/java/org/apache/flink/table/plan/stats/TableStatsTest.java

@@ -0,0 +1,96 @@
+/*
+ * 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.plan.stats;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Test for {@link TableStats}.
+ */
+public class TableStatsTest {
+
+	@Test
+	public void testMerge() {
+		Map<String, ColumnStats> colStats1 = new HashMap<>();
+		colStats1.put("a", new ColumnStats(4L, 5L, 2D, 3, 15, 2));
+		TableStats stats1 = new TableStats(30, colStats1);
+
+		Map<String, ColumnStats> colStats2 = new HashMap<>();
+		colStats2.put("a", new ColumnStats(3L, 15L, 12D, 23, 35, 6));
+		TableStats stats2 = new TableStats(32, colStats2);
+
+		Map<String, ColumnStats> colStatsMerge = new HashMap<>();
+		colStatsMerge.put("a", new ColumnStats(7L, 20L, 7D, 23, 35, 2));
+		Assert.assertEquals(new TableStats(62, colStatsMerge), stats1.merge(stats2));
+	}
+
+	@Test
+	public void testMergeLackColumnStats() {
+		Map<String, ColumnStats> colStats1 = new HashMap<>();
+		colStats1.put("a", new ColumnStats(4L, 5L, 2D, 3, 15, 2));
+		colStats1.put("b", new ColumnStats(4L, 5L, 2D, 3, 15, 2));
+		TableStats stats1 = new TableStats(30, colStats1);
+
+		Map<String, ColumnStats> colStats2 = new HashMap<>();
+		colStats2.put("a", new ColumnStats(3L, 15L, 12D, 23, 35, 6));
+		TableStats stats2 = new TableStats(32, colStats2);
+
+		Map<String, ColumnStats> colStatsMerge = new HashMap<>();
+		colStatsMerge.put("a", new ColumnStats(7L, 20L, 7D, 23, 35, 2));
+		Assert.assertEquals(new TableStats(62, colStatsMerge), stats1.merge(stats2));
+	}
+
+	@Test
+	public void testMergeUnknownRowCount() {
+		TableStats stats1 = new TableStats(-1, new HashMap<>());
+		TableStats stats2 = new TableStats(32, new HashMap<>());
+		Assert.assertEquals(new TableStats(-1, new HashMap<>()), stats1.merge(stats2));
+
+		stats1 = new TableStats(-1, new HashMap<>());
+		stats2 = new TableStats(-1, new HashMap<>());
+		Assert.assertEquals(new TableStats(-1, new HashMap<>()), stats1.merge(stats2));
+
+		stats1 = new TableStats(-3, new HashMap<>());
+		stats2 = new TableStats(-2, new HashMap<>());
+		Assert.assertEquals(new TableStats(-1, new HashMap<>()), stats1.merge(stats2));
+	}
+
+	@Test
+	public void testMergeColumnStatsUnknown() {
+		ColumnStats columnStats0 = new ColumnStats(4L, 5L, 2D, 3, 15, 2);
+		ColumnStats columnStats1 = new ColumnStats(4L, null, 2D, 3, 15, 2);
+		ColumnStats columnStats2 = new ColumnStats(4L, 5L, 2D, null, 15, 2);
+		ColumnStats columnStats3 = new ColumnStats(null, 5L, 2D, 3, 15, 2);
+		ColumnStats columnStats4 = new ColumnStats(4L, 5L, 2D, 3, null, 2);
+		ColumnStats columnStats5 = new ColumnStats(4L, 5L, 2D, 3, 15, null);
+		ColumnStats columnStats6 = new ColumnStats(4L, 5L, null, 3, 15, 2);
+
+		Assert.assertEquals(new ColumnStats(8L, null, 2D, 3, 15, 2), columnStats0.merge(columnStats1));
+		Assert.assertEquals(new ColumnStats(8L, 10L, 2D, null, 15, 2), columnStats0.merge(columnStats2));
+		Assert.assertEquals(new ColumnStats(null, 10L, 2D, 3, 15, 2), columnStats0.merge(columnStats3));
+		Assert.assertEquals(new ColumnStats(8L, 10L, 2D, 3, null, 2), columnStats0.merge(columnStats4));
+		Assert.assertEquals(new ColumnStats(8L, 10L, 2D, 3, 15, null), columnStats0.merge(columnStats5));
+		Assert.assertEquals(new ColumnStats(8L, 10L, null, 3, 15, 2), columnStats0.merge(columnStats6));
+		Assert.assertEquals(new ColumnStats(8L, 10L, null, 3, 15, 2), columnStats6.merge(columnStats6));
+	}
+}

flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/FlinkCalciteCatalogReader.java

@@ -23,6 +23,7 @@
 import org.apache.flink.table.catalog.CatalogTable;
 import org.apache.flink.table.catalog.CatalogView;
 import org.apache.flink.table.catalog.ConnectorCatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
 import org.apache.flink.table.catalog.QueryOperationCatalogView;
 import org.apache.flink.table.planner.catalog.CatalogSchemaTable;
 import org.apache.flink.table.planner.catalog.QueryOperationCatalogViewTable;
@@ -109,8 +110,8 @@ private static FlinkPreparingTableBase toPreparingTable(
 			ConnectorCatalogTable<?, ?> connectorTable = (ConnectorCatalogTable<?, ?>) baseTable;
 			if ((connectorTable).getTableSource().isPresent()) {
 				return convertSourceTable(relOptSchema,
-					names,
 					rowType,
+					schemaTable.getTableIdentifier(),
 					connectorTable,
 					schemaTable.getStatistic(),
 					schemaTable.isStreamingMode());
@@ -155,8 +156,8 @@ private static FlinkPreparingTableBase convertCatalogView(
 
 	private static FlinkPreparingTableBase convertSourceTable(
 			RelOptSchema relOptSchema,
-			List<String> names,
 			RelDataType rowType,
+			ObjectIdentifier tableIdentifier,
 			ConnectorCatalogTable<?, ?> table,
 			FlinkStatistic statistic,
 			boolean isStreamingMode) {
@@ -173,7 +174,7 @@ private static FlinkPreparingTableBase convertSourceTable(
 
 		return new TableSourceTable<>(
 			relOptSchema,
-			names,
+			tableIdentifier,
 			rowType,
 			statistic,
 			tableSource,

flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java

@@ -22,9 +22,11 @@
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.table.api.TableException;
 import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogManager;
 import org.apache.flink.table.catalog.ConnectorCatalogTable;
 import org.apache.flink.table.catalog.FunctionLookup;
 import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
 import org.apache.flink.table.expressions.CallExpression;
 import org.apache.flink.table.expressions.Expression;
 import org.apache.flink.table.expressions.ExpressionDefaultVisitor;
@@ -54,6 +56,7 @@
 import org.apache.flink.table.operations.WindowAggregateQueryOperation;
 import org.apache.flink.table.operations.WindowAggregateQueryOperation.ResolvedGroupWindow;
 import org.apache.flink.table.operations.utils.QueryOperationDefaultVisitor;
+import org.apache.flink.table.planner.calcite.FlinkContext;
 import org.apache.flink.table.planner.calcite.FlinkRelBuilder;
 import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
 import org.apache.flink.table.planner.expressions.PlannerProctimeAttribute;
@@ -349,21 +352,19 @@ public <U> RelNode visit(TableSourceQueryOperation<U> tableSourceOperation) {
 			}
 
 			FlinkStatistic statistic;
-			List<String> names;
+			ObjectIdentifier tableIdentifier;
 			if (tableSourceOperation instanceof RichTableSourceQueryOperation &&
 				((RichTableSourceQueryOperation<U>) tableSourceOperation).getIdentifier() != null) {
-				ObjectIdentifier identifier = ((RichTableSourceQueryOperation<U>) tableSourceOperation).getIdentifier();
+				tableIdentifier = ((RichTableSourceQueryOperation<U>) tableSourceOperation).getIdentifier();
 				statistic = ((RichTableSourceQueryOperation<U>) tableSourceOperation).getStatistic();
-				names = Arrays.asList(
-					identifier.getCatalogName(),
-					identifier.getDatabaseName(),
-					identifier.getObjectName());
 			} else {
 				statistic = FlinkStatistic.UNKNOWN();
 				// TableSourceScan requires a unique name of a Table for computing a digest.
 				// We are using the identity hash of the TableSource object.
 				String refId = "Unregistered_TableSource_" + System.identityHashCode(tableSource);
-				names = Collections.singletonList(refId);
+				CatalogManager catalogManager = relBuilder.getCluster().getPlanner().getContext()
+						.unwrap(FlinkContext.class).getCatalogManager();
+				tableIdentifier = catalogManager.qualifyIdentifier(UnresolvedIdentifier.of(refId));
 			}
 
 			RelDataType rowType = TableSourceUtil.getSourceRowType(relBuilder.getTypeFactory(),
@@ -372,7 +373,7 @@ public <U> RelNode visit(TableSourceQueryOperation<U> tableSourceOperation) {
 				!isBatch);
 			TableSourceTable<?> tableSourceTable = new TableSourceTable<>(
 				relBuilder.getRelOptSchema(),
-				names,
+				tableIdentifier,
 				rowType,
 				statistic,
 				tableSource,