This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 90e879928e [flink] Support TIMESTAMP columns in ANALYZE TABLE column
statistics (#10613)
90e879928e is described below
commit 90e879928e1f3f77573f9913a0f50219efd57958
Author: YangJie <[email protected]>
AuthorDate: Sun Oct 18 04:51:57 2026 -0400
[flink] Support TIMESTAMP columns in ANALYZE TABLE column statistics
(#10613)
---
.../apache/paimon/flink/utils/TableStatsUtil.java | 16 +++++++++++++
.../paimon/flink/FlinkAnalyzeTableITCase.java | 27 ++++++++++++++++++++++
2 files changed, 43 insertions(+)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableStatsUtil.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableStatsUtil.java
index aadcd0e30f..c37e1456d3 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableStatsUtil.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableStatsUtil.java
@@ -158,6 +158,22 @@ public class TableStatsUtil {
null != longColStat.getNullCount() ?
longColStat.getNullCount() : null,
null,
null);
+ } else if
(typeRoot.equals(DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE)) {
+ // Flink reports a TIMESTAMP column as the epoch seconds of
its wall-clock time
+ return ColStats.newColStats(
+ field.id(),
+ null != longColStat.getNdv() ? longColStat.getNdv() :
null,
+ null != longColStat.getMin()
+ ?
org.apache.paimon.data.Timestamp.fromEpochMillis(
+ longColStat.getMin() * 1000L)
+ : null,
+ null != longColStat.getMax()
+ ?
org.apache.paimon.data.Timestamp.fromEpochMillis(
+ longColStat.getMax() * 1000L)
+ : null,
+ null != longColStat.getNullCount() ?
longColStat.getNullCount() : null,
+ null,
+ null);
} else if
(typeRoot.equals(DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE)) {
return ColStats.newColStats(
field.id(),
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkAnalyzeTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkAnalyzeTableITCase.java
index 5a99925de5..65d40eca2a 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkAnalyzeTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkAnalyzeTableITCase.java
@@ -20,6 +20,7 @@ package org.apache.paimon.flink;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.Timestamp;
import org.apache.paimon.stats.ColStats;
import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.FileStoreTable;
@@ -28,6 +29,7 @@ import org.apache.paimon.utils.DateTimeUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.time.LocalDateTime;
import java.util.Collections;
import java.util.Map;
import java.util.Optional;
@@ -86,6 +88,31 @@ public class FlinkAnalyzeTableITCase extends
CatalogITCaseBase {
assertThat(decimalStats).isEqualTo(ColStats.newColStats(1, 0L, null,
null, 1L, null, null));
}
+ @Test
+ public void testAnalyzeTimestampColumn() throws
Catalog.TableNotExistException {
+ sql("CREATE TABLE T (id INT, ts TIMESTAMP)");
+ sql(
+ "INSERT INTO T VALUES (1, TIMESTAMP '2023-01-12
20:00:01.123456'), "
+ + "(2, TIMESTAMP '2023-01-13 08:30:00.999999')");
+
+ sql("ANALYZE TABLE T COMPUTE STATISTICS FOR ALL COLUMNS");
+
+ // Flink collects whole seconds
+ ColStats<?> tsStats =
paimonTable("T").statistics().get().colStats().get("ts");
+ assertThat(tsStats)
+ .isEqualTo(
+ ColStats.newColStats(
+ 1,
+ 2L,
+ Timestamp.fromLocalDateTime(
+
LocalDateTime.parse("2023-01-12T20:00:01")),
+ Timestamp.fromLocalDateTime(
+
LocalDateTime.parse("2023-01-13T08:30:00")),
+ 0L,
+ null,
+ null));
+ }
+
@Test
public void testAnalyzeTableColumn() throws Catalog.TableNotExistException
{
sql(
