You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2023/05/16 06:05:13 UTC
[incubator-paimon] branch master updated: [flink] Support table statistics in table store (#426)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new f60c616b9 [flink] Support table statistics in table store (#426)
f60c616b9 is described below
commit f60c616b9d3d34bc9c33a6c5ac1c2de93cedb669
Author: Shammon FY <zj...@gmail.com>
AuthorDate: Tue May 16 14:05:09 2023 +0800
[flink] Support table statistics in table store (#426)
---
.../paimon/flink/source/DataTableSource.java | 64 ++++++
.../paimon/flink/source/DataTableSource.java | 64 ++++++
...bleSource.java => AbstractDataTableSource.java} | 25 ++-
.../paimon/flink/source/DataTableSource.java | 228 ++-------------------
.../statistics/AppendOnlyTableStatisticsTest.java | 44 ++++
.../ChangelogValueCountTableStatisticsTest.java | 44 ++++
.../ChangelogWithKeyTableStatisticsTest.java | 71 +++++++
.../FileStoreTableStatisticsTestBase.java | 181 ++++++++++++++++
8 files changed, 502 insertions(+), 219 deletions(-)
diff --git a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/source/DataTableSource.java b/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
new file mode 100644
index 000000000..ce7ccd23d
--- /dev/null
+++ b/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
@@ -0,0 +1,64 @@
+/*
+ * 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.paimon.flink.source;
+
+import org.apache.paimon.flink.log.LogStoreTableFactory;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.table.Table;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
+
+import javax.annotation.Nullable;
+
+/** Data table source. */
+public class DataTableSource extends AbstractDataTableSource {
+ public DataTableSource(
+ ObjectIdentifier tableIdentifier,
+ Table table,
+ boolean streaming,
+ DynamicTableFactory.Context context,
+ @Nullable LogStoreTableFactory logStoreTableFactory) {
+ super(tableIdentifier, table, streaming, context, logStoreTableFactory);
+ }
+
+ protected DataTableSource(
+ ObjectIdentifier tableIdentifier,
+ Table table,
+ boolean streaming,
+ DynamicTableFactory.Context context,
+ @Nullable LogStoreTableFactory logStoreTableFactory,
+ @Nullable Predicate predicate,
+ @Nullable int[][] projectFields,
+ @Nullable Long limit,
+ @Nullable WatermarkStrategy<RowData> watermarkStrategy) {
+ super(
+ tableIdentifier,
+ table,
+ streaming,
+ context,
+ logStoreTableFactory,
+ predicate,
+ projectFields,
+ limit,
+ watermarkStrategy);
+ }
+}
diff --git a/paimon-flink/paimon-flink-1.15/src/main/java/org/apache/paimon/flink/source/DataTableSource.java b/paimon-flink/paimon-flink-1.15/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
new file mode 100644
index 000000000..ce7ccd23d
--- /dev/null
+++ b/paimon-flink/paimon-flink-1.15/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
@@ -0,0 +1,64 @@
+/*
+ * 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.paimon.flink.source;
+
+import org.apache.paimon.flink.log.LogStoreTableFactory;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.table.Table;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
+
+import javax.annotation.Nullable;
+
+/** Data table source. */
+public class DataTableSource extends AbstractDataTableSource {
+ public DataTableSource(
+ ObjectIdentifier tableIdentifier,
+ Table table,
+ boolean streaming,
+ DynamicTableFactory.Context context,
+ @Nullable LogStoreTableFactory logStoreTableFactory) {
+ super(tableIdentifier, table, streaming, context, logStoreTableFactory);
+ }
+
+ protected DataTableSource(
+ ObjectIdentifier tableIdentifier,
+ Table table,
+ boolean streaming,
+ DynamicTableFactory.Context context,
+ @Nullable LogStoreTableFactory logStoreTableFactory,
+ @Nullable Predicate predicate,
+ @Nullable int[][] projectFields,
+ @Nullable Long limit,
+ @Nullable WatermarkStrategy<RowData> watermarkStrategy) {
+ super(
+ tableIdentifier,
+ table,
+ streaming,
+ context,
+ logStoreTableFactory,
+ predicate,
+ projectFields,
+ limit,
+ watermarkStrategy);
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AbstractDataTableSource.java
similarity index 94%
copy from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
copy to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AbstractDataTableSource.java
index 0002cb91b..fc5fa21fd 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AbstractDataTableSource.java
@@ -73,17 +73,20 @@ import static org.apache.paimon.flink.FlinkConnectorOptions.SCAN_WATERMARK_IDLE_
* LogHybridSourceFactory.FlinkHybridFirstSource} and kafka log source created by {@link
* LogSourceProvider}.
*/
-public class DataTableSource extends FlinkTableSource
+public abstract class AbstractDataTableSource extends FlinkTableSource
implements LookupTableSource, SupportsWatermarkPushDown {
private final ObjectIdentifier tableIdentifier;
- private final boolean streaming;
+ protected final boolean streaming;
private final DynamicTableFactory.Context context;
@Nullable private final LogStoreTableFactory logStoreTableFactory;
@Nullable private WatermarkStrategy<RowData> watermarkStrategy;
- public DataTableSource(
+ protected long rowCount = -1;
+ protected List<Split> splits = null;
+
+ public AbstractDataTableSource(
ObjectIdentifier tableIdentifier,
Table table,
boolean streaming,
@@ -101,7 +104,7 @@ public class DataTableSource extends FlinkTableSource
null);
}
- private DataTableSource(
+ protected AbstractDataTableSource(
ObjectIdentifier tableIdentifier,
Table table,
boolean streaming,
@@ -212,13 +215,12 @@ public class DataTableSource extends FlinkTableSource
FlinkSourceBuilder sourceBuilder, StreamExecutionEnvironment env) {
Options options = Options.fromMap(this.table.options());
Integer parallelism = options.get(FlinkConnectorOptions.SCAN_PARALLELISM);
- List<Split> splits = null;
if (options.get(FlinkConnectorOptions.INFER_SCAN_PARALLELISM)) {
// for streaming mode, set the default parallelism to the bucket number.
if (streaming) {
parallelism = options.get(CoreOptions.BUCKET);
} else {
- splits = table.newReadBuilder().withFilter(predicate).newScan().plan().splits();
+ splitsForScan();
if (null != splits) {
parallelism = splits.size();
}
@@ -235,6 +237,17 @@ public class DataTableSource extends FlinkTableSource
return sourceBuilder.withParallelism(parallelism).withSplits(splits).withEnv(env).build();
}
+ protected void splitsForScan() {
+ if (splits == null && rowCount == -1) {
+ splits = table.newReadBuilder().withFilter(predicate).newScan().plan().splits();
+ if (splits == null) {
+ rowCount = 0;
+ } else {
+ rowCount = splits.stream().mapToLong(Split::rowCount).sum();
+ }
+ }
+ }
+
@Override
public DynamicTableSource copy() {
return new DataTableSource(
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
index 0002cb91b..2a7050e64 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
@@ -18,90 +18,33 @@
package org.apache.paimon.flink.source;
-import org.apache.paimon.CoreOptions;
-import org.apache.paimon.CoreOptions.ChangelogProducer;
-import org.apache.paimon.CoreOptions.LogChangelogMode;
-import org.apache.paimon.CoreOptions.LogConsistency;
-import org.apache.paimon.flink.FlinkConnectorOptions;
-import org.apache.paimon.flink.FlinkConnectorOptions.WatermarkEmitStrategy;
-import org.apache.paimon.flink.PaimonDataStreamScanProvider;
-import org.apache.paimon.flink.log.LogSourceProvider;
+import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.flink.log.LogStoreTableFactory;
-import org.apache.paimon.flink.lookup.FileStoreLookupFunction;
-import org.apache.paimon.flink.lookup.LookupRuntimeProviderFactory;
-import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
-import org.apache.paimon.table.ChangelogValueCountFileStoreTable;
-import org.apache.paimon.table.ChangelogWithKeyFileStoreTable;
import org.apache.paimon.table.Table;
-import org.apache.paimon.table.source.Split;
-import org.apache.paimon.utils.Projection;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.catalog.ObjectIdentifier;
-import org.apache.flink.table.connector.ChangelogMode;
-import org.apache.flink.table.connector.source.DynamicTableSource;
-import org.apache.flink.table.connector.source.LookupTableSource;
-import org.apache.flink.table.connector.source.abilities.SupportsWatermarkPushDown;
+import org.apache.flink.table.connector.source.abilities.SupportsStatisticReport;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.plan.stats.TableStats;
import javax.annotation.Nullable;
-import java.time.Duration;
-import java.util.List;
-import java.util.stream.IntStream;
-
-import static org.apache.paimon.CoreOptions.CHANGELOG_PRODUCER;
-import static org.apache.paimon.CoreOptions.LOG_CHANGELOG_MODE;
-import static org.apache.paimon.CoreOptions.LOG_CONSISTENCY;
-import static org.apache.paimon.CoreOptions.LOG_SCAN_REMOVE_NORMALIZE;
-import static org.apache.paimon.flink.FlinkConnectorOptions.SCAN_WATERMARK_ALIGNMENT_GROUP;
-import static org.apache.paimon.flink.FlinkConnectorOptions.SCAN_WATERMARK_ALIGNMENT_MAX_DRIFT;
-import static org.apache.paimon.flink.FlinkConnectorOptions.SCAN_WATERMARK_ALIGNMENT_UPDATE_INTERVAL;
-import static org.apache.paimon.flink.FlinkConnectorOptions.SCAN_WATERMARK_EMIT_STRATEGY;
-import static org.apache.paimon.flink.FlinkConnectorOptions.SCAN_WATERMARK_IDLE_TIMEOUT;
-
-/**
- * Table source to create {@link StaticFileStoreSource} or {@link ContinuousFileStoreSource} under
- * batch mode or change-tracking is disabled. For streaming mode with change-tracking enabled and
- * FULL scan mode, it will create a {@link
- * org.apache.flink.connector.base.source.hybrid.HybridSource} of {@code
- * LogHybridSourceFactory.FlinkHybridFirstSource} and kafka log source created by {@link
- * LogSourceProvider}.
- */
-public class DataTableSource extends FlinkTableSource
- implements LookupTableSource, SupportsWatermarkPushDown {
-
- private final ObjectIdentifier tableIdentifier;
- private final boolean streaming;
- private final DynamicTableFactory.Context context;
- @Nullable private final LogStoreTableFactory logStoreTableFactory;
-
- @Nullable private WatermarkStrategy<RowData> watermarkStrategy;
-
+/** Data table source with table statistics. */
+public class DataTableSource extends AbstractDataTableSource implements SupportsStatisticReport {
public DataTableSource(
ObjectIdentifier tableIdentifier,
Table table,
boolean streaming,
DynamicTableFactory.Context context,
@Nullable LogStoreTableFactory logStoreTableFactory) {
- this(
- tableIdentifier,
- table,
- streaming,
- context,
- logStoreTableFactory,
- null,
- null,
- null,
- null);
+ super(tableIdentifier, table, streaming, context, logStoreTableFactory);
}
- private DataTableSource(
+ @VisibleForTesting
+ public DataTableSource(
ObjectIdentifier tableIdentifier,
Table table,
boolean streaming,
@@ -111,133 +54,7 @@ public class DataTableSource extends FlinkTableSource
@Nullable int[][] projectFields,
@Nullable Long limit,
@Nullable WatermarkStrategy<RowData> watermarkStrategy) {
- super(table, predicate, projectFields, limit);
- this.tableIdentifier = tableIdentifier;
- this.streaming = streaming;
- this.context = context;
- this.logStoreTableFactory = logStoreTableFactory;
- this.predicate = predicate;
- this.projectFields = projectFields;
- this.limit = limit;
- this.watermarkStrategy = watermarkStrategy;
- }
-
- @Override
- public ChangelogMode getChangelogMode() {
- if (!streaming) {
- // batch merge all, return insert only
- return ChangelogMode.insertOnly();
- }
-
- if (table instanceof AppendOnlyFileStoreTable) {
- return ChangelogMode.insertOnly();
- } else if (table instanceof ChangelogValueCountFileStoreTable) {
- return ChangelogMode.all();
- } else if (table instanceof ChangelogWithKeyFileStoreTable) {
- Options options = Options.fromMap(table.options());
-
- if (options.get(LOG_SCAN_REMOVE_NORMALIZE)) {
- return ChangelogMode.all();
- }
-
- if (logStoreTableFactory == null
- && options.get(CHANGELOG_PRODUCER) != ChangelogProducer.NONE) {
- return ChangelogMode.all();
- }
-
- // optimization: transaction consistency and all changelog mode avoid the generation of
- // normalized nodes. See FlinkTableSink.getChangelogMode validation.
- return options.get(LOG_CONSISTENCY) == LogConsistency.TRANSACTIONAL
- && options.get(LOG_CHANGELOG_MODE) == LogChangelogMode.ALL
- ? ChangelogMode.all()
- : ChangelogMode.upsert();
- } else {
- throw new UnsupportedOperationException(
- "Unsupported Table subclass "
- + table.getClass().getName()
- + " for streaming mode.");
- }
- }
-
- @Override
- public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
- LogSourceProvider logSourceProvider = null;
- if (logStoreTableFactory != null) {
- logSourceProvider =
- logStoreTableFactory.createSourceProvider(context, scanContext, projectFields);
- }
-
- WatermarkStrategy<RowData> watermarkStrategy = this.watermarkStrategy;
- Options options = Options.fromMap(table.options());
- if (watermarkStrategy != null) {
- WatermarkEmitStrategy emitStrategy = options.get(SCAN_WATERMARK_EMIT_STRATEGY);
- if (emitStrategy == WatermarkEmitStrategy.ON_EVENT) {
- watermarkStrategy = new OnEventWatermarkStrategy(watermarkStrategy);
- }
- Duration idleTimeout = options.get(SCAN_WATERMARK_IDLE_TIMEOUT);
- if (idleTimeout != null) {
- watermarkStrategy = watermarkStrategy.withIdleness(idleTimeout);
- }
- String watermarkAlignGroup = options.get(SCAN_WATERMARK_ALIGNMENT_GROUP);
- if (watermarkAlignGroup != null) {
- try {
- watermarkStrategy =
- WatermarkAlignUtils.withWatermarkAlignment(
- watermarkStrategy,
- watermarkAlignGroup,
- options.get(SCAN_WATERMARK_ALIGNMENT_MAX_DRIFT),
- options.get(SCAN_WATERMARK_ALIGNMENT_UPDATE_INTERVAL));
- } catch (NoSuchMethodError error) {
- throw new RuntimeException(
- "Flink 1.14 does not support watermark alignment, please check your Flink version.",
- error);
- }
- }
- }
-
- FlinkSourceBuilder sourceBuilder =
- new FlinkSourceBuilder(tableIdentifier, table)
- .withContinuousMode(streaming)
- .withLogSourceProvider(logSourceProvider)
- .withProjection(projectFields)
- .withPredicate(predicate)
- .withLimit(limit)
- .withWatermarkStrategy(watermarkStrategy);
-
- return new PaimonDataStreamScanProvider(
- !streaming, env -> configureSource(sourceBuilder, env));
- }
-
- private DataStream<RowData> configureSource(
- FlinkSourceBuilder sourceBuilder, StreamExecutionEnvironment env) {
- Options options = Options.fromMap(this.table.options());
- Integer parallelism = options.get(FlinkConnectorOptions.SCAN_PARALLELISM);
- List<Split> splits = null;
- if (options.get(FlinkConnectorOptions.INFER_SCAN_PARALLELISM)) {
- // for streaming mode, set the default parallelism to the bucket number.
- if (streaming) {
- parallelism = options.get(CoreOptions.BUCKET);
- } else {
- splits = table.newReadBuilder().withFilter(predicate).newScan().plan().splits();
- if (null != splits) {
- parallelism = splits.size();
- }
- if (null != limit && limit > 0) {
- int limitCount =
- limit >= Integer.MAX_VALUE ? Integer.MAX_VALUE : limit.intValue();
- parallelism = Math.min(parallelism, limitCount);
- }
-
- parallelism = null == parallelism ? 1 : Math.max(1, parallelism);
- }
- }
-
- return sourceBuilder.withParallelism(parallelism).withSplits(splits).withEnv(env).build();
- }
-
- @Override
- public DynamicTableSource copy() {
- return new DataTableSource(
+ super(
tableIdentifier,
table,
streaming,
@@ -250,27 +67,12 @@ public class DataTableSource extends FlinkTableSource
}
@Override
- public String asSummaryString() {
- return "Paimon-DataSource";
- }
-
- @Override
- public void applyWatermark(WatermarkStrategy<RowData> watermarkStrategy) {
- this.watermarkStrategy = watermarkStrategy;
- }
-
- @Override
- public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
- if (limit != null) {
- throw new RuntimeException(
- "Limit push down should not happen in Lookup source, but it is " + limit);
+ public TableStats reportStatistics() {
+ if (streaming) {
+ return TableStats.UNKNOWN;
}
- int[] projection =
- projectFields == null
- ? IntStream.range(0, table.rowType().getFieldCount()).toArray()
- : Projection.of(projectFields).toTopLevelIndexes();
- int[] joinKey = Projection.of(context.getKeys()).toTopLevelIndexes();
- return LookupRuntimeProviderFactory.create(
- new FileStoreLookupFunction(table, projection, joinKey, predicate));
+
+ splitsForScan();
+ return new TableStats(rowCount);
}
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/AppendOnlyTableStatisticsTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/AppendOnlyTableStatisticsTest.java
new file mode 100644
index 000000000..7f1ebeb84
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/AppendOnlyTableStatisticsTest.java
@@ -0,0 +1,44 @@
+/*
+ * 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.paimon.flink.source.statistics;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.WriteMode;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+
+/** Statistics tests for {@link AppendOnlyFileStoreTable}. */
+public class AppendOnlyTableStatisticsTest extends FileStoreTableStatisticsTestBase {
+ @Override
+ protected FileStoreTable createStoreTable() throws Exception {
+ Options conf = new Options();
+ conf.set(CoreOptions.PATH, tablePath.toString());
+ conf.set(CoreOptions.WRITE_MODE, WriteMode.APPEND_ONLY);
+ conf.set(CoreOptions.BUCKET, 1);
+ TableSchema tableSchema =
+ new SchemaManager(LocalFileIO.create(), tablePath)
+ .createTable(schemaBuilder().partitionKeys("pt").build());
+ return FileStoreTableFactory.create(LocalFileIO.create(), tablePath, tableSchema);
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/ChangelogValueCountTableStatisticsTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/ChangelogValueCountTableStatisticsTest.java
new file mode 100644
index 000000000..b9fa521bf
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/ChangelogValueCountTableStatisticsTest.java
@@ -0,0 +1,44 @@
+/*
+ * 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.paimon.flink.source.statistics;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.WriteMode;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.ChangelogValueCountFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+
+/** Statistics tests for {@link ChangelogValueCountFileStoreTable}. */
+public class ChangelogValueCountTableStatisticsTest extends FileStoreTableStatisticsTestBase {
+ @Override
+ FileStoreTable createStoreTable() throws Exception {
+ Options conf = new Options();
+ conf.set(CoreOptions.PATH, tablePath.toString());
+ conf.set(CoreOptions.WRITE_MODE, WriteMode.CHANGE_LOG);
+ conf.set(CoreOptions.BUCKET, 1);
+ TableSchema tableSchema =
+ new SchemaManager(LocalFileIO.create(), tablePath)
+ .createTable(schemaBuilder().partitionKeys("pt").build());
+ return FileStoreTableFactory.create(LocalFileIO.create(), tablePath, tableSchema);
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/ChangelogWithKeyTableStatisticsTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/ChangelogWithKeyTableStatisticsTest.java
new file mode 100644
index 000000000..38d1a47af
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/ChangelogWithKeyTableStatisticsTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.paimon.flink.source.statistics;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.WriteMode;
+import org.apache.paimon.flink.source.DataTableSource;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.ChangelogWithKeyFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Statistics tests for {@link ChangelogWithKeyFileStoreTable}. */
+public class ChangelogWithKeyTableStatisticsTest extends FileStoreTableStatisticsTestBase {
+
+ /** {@link ChangelogWithKeyFileStoreTable} does not support filter value. */
+ @Test
+ public void testTableFilterValueStatistics() throws Exception {
+ FileStoreTable table = writeData();
+ PredicateBuilder builder = new PredicateBuilder(table.schema().logicalRowType());
+ DataTableSource keyFilterSource =
+ new DataTableSource(
+ identifier,
+ table,
+ false,
+ null,
+ null,
+ builder.greaterThan(2, 500L),
+ null,
+ null,
+ null);
+ Assertions.assertThat(keyFilterSource.reportStatistics().getRowCount()).isEqualTo(9L);
+ // TODO validate column statistics
+ }
+
+ @Override
+ FileStoreTable createStoreTable() throws Exception {
+ Options conf = new Options();
+ conf.set(CoreOptions.PATH, tablePath.toString());
+ conf.set(CoreOptions.WRITE_MODE, WriteMode.CHANGE_LOG);
+ conf.set(CoreOptions.BUCKET, 1);
+ TableSchema tableSchema =
+ new SchemaManager(LocalFileIO.create(), tablePath)
+ .createTable(
+ schemaBuilder().partitionKeys("pt").primaryKey("pt", "a").build());
+ return FileStoreTableFactory.create(LocalFileIO.create(), tablePath, tableSchema);
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/FileStoreTableStatisticsTestBase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/FileStoreTableStatisticsTestBase.java
new file mode 100644
index 000000000..56bfb53bd
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/FileStoreTableStatisticsTestBase.java
@@ -0,0 +1,181 @@
+/*
+ * 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.paimon.flink.source.statistics;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.flink.source.DataTableSource;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.VarCharType;
+
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.plan.stats.TableStats;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.Arrays;
+import java.util.UUID;
+
+/** Statistics tests for {@link FileStoreTable}. */
+public abstract class FileStoreTableStatisticsTestBase {
+ protected final ObjectIdentifier identifier = ObjectIdentifier.of("c", "d", "t");
+
+ @TempDir java.nio.file.Path tempDir;
+
+ protected Path tablePath;
+ protected String commitUser;
+
+ @BeforeEach
+ public void before() {
+ tablePath = new Path(tempDir.toString() + "/" + UUID.randomUUID());
+ commitUser = UUID.randomUUID().toString();
+ }
+
+ @Test
+ public void testTableScanStatistics() throws Exception {
+ FileStoreTable table = writeData();
+ DataTableSource scanSource = new DataTableSource(identifier, table, false, null, null);
+ Assertions.assertThat(scanSource.reportStatistics().getRowCount()).isEqualTo(9L);
+ // TODO validate column statistics
+ }
+
+ @Test
+ public void testTableStreamingStatistics() throws Exception {
+ FileStoreTable table = writeData();
+ DataTableSource streamSource = new DataTableSource(identifier, table, true, null, null);
+ Assertions.assertThat(streamSource.reportStatistics()).isEqualTo(TableStats.UNKNOWN);
+ }
+
+ @Test
+ public void testTableFilterPartitionStatistics() throws Exception {
+ FileStoreTable table = writeData();
+ PredicateBuilder builder = new PredicateBuilder(table.schema().logicalRowType());
+ DataTableSource partitionFilterSource =
+ new DataTableSource(
+ identifier,
+ table,
+ false,
+ null,
+ null,
+ builder.equal(0, 1),
+ null,
+ null,
+ null);
+ Assertions.assertThat(partitionFilterSource.reportStatistics().getRowCount()).isEqualTo(5L);
+ // TODO validate column statistics
+ }
+
+ @Test
+ public void testTableFilterKeyStatistics() throws Exception {
+ FileStoreTable table = writeData();
+ PredicateBuilder builder = new PredicateBuilder(table.schema().logicalRowType());
+ DataTableSource keyFilterSource =
+ new DataTableSource(
+ identifier,
+ table,
+ false,
+ null,
+ null,
+ builder.equal(1, 50),
+ null,
+ null,
+ null);
+ Assertions.assertThat(keyFilterSource.reportStatistics().getRowCount()).isEqualTo(2L);
+ // TODO validate column statistics
+ }
+
+ @Test
+ public void testTableFilterValueStatistics() throws Exception {
+ FileStoreTable table = writeData();
+ PredicateBuilder builder = new PredicateBuilder(table.schema().logicalRowType());
+ DataTableSource keyFilterSource =
+ new DataTableSource(
+ identifier,
+ table,
+ false,
+ null,
+ null,
+ builder.greaterThan(2, 500L),
+ null,
+ null,
+ null);
+ Assertions.assertThat(keyFilterSource.reportStatistics().getRowCount()).isEqualTo(4L);
+ // TODO validate column statistics
+ }
+
+ protected FileStoreTable writeData() throws Exception {
+ FileStoreTable table = createStoreTable();
+
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ write.write(rowData(1, 10, 100L, "S1"));
+ write.write(rowData(1, 20, 200L, null));
+ write.write(rowData(2, 30, 300L, "S3"));
+ commit.commit(0, write.prepareCommit(true, 0));
+
+ write.write(rowData(1, 40, 400L, "S4"));
+ write.write(rowData(2, 50, null, "S5"));
+ write.write(rowData(2, 60, 600L, "S6"));
+ commit.commit(1, write.prepareCommit(true, 1));
+
+ write.write(rowData(1, 70, 700L, "S7"));
+ write.write(rowData(2, 80, null, "S8"));
+ write.write(rowData(1, 90, 900L, null));
+ commit.commit(2, write.prepareCommit(true, 2));
+
+ write.close();
+
+ return table;
+ }
+
+ abstract FileStoreTable createStoreTable() throws Exception;
+
+ protected InternalRow rowData(Object... values) {
+ return GenericRow.of(
+ Arrays.stream(values)
+ .map(
+ v -> {
+ if (v instanceof String) {
+ return BinaryString.fromString((String) v);
+ } else {
+ return v;
+ }
+ })
+ .toArray());
+ }
+
+ protected Schema.Builder schemaBuilder() {
+ return Schema.newBuilder()
+ .column("pt", new IntType())
+ .column("a", new IntType())
+ .column("b", new BigIntType(true))
+ .column("c", new VarCharType(100));
+ }
+}