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/12/01 07:03:13 UTC
(incubator-paimon) 39/46: [flink] Fix Table Doesn't Exist Exception when Reading System Tables with Flink Generic Catalog. (#2428)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.6
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit d7d123260a7d6d3a23986ab09a99dea601db841a
Author: Kerwin <37...@users.noreply.github.com>
AuthorDate: Thu Nov 30 21:17:41 2023 +0800
[flink] Fix Table Doesn't Exist Exception when Reading System Tables with Flink Generic Catalog. (#2428)
---
.../org/apache/paimon/flink/FlinkGenericCatalog.java | 6 ++++++
.../paimon/hive/FlinkGenericCatalogITCase.java | 20 ++++++++++++++++++++
2 files changed, 26 insertions(+)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalog.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalog.java
index dadc80b7b..0db0376ef 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalog.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalog.java
@@ -324,12 +324,18 @@ public class FlinkGenericCatalog extends AbstractCatalog {
@Override
public CatalogTableStatistics getTableStatistics(ObjectPath tablePath)
throws TableNotExistException, CatalogException {
+ if (isPaimonTable(tablePath)) {
+ return paimon.getTableStatistics(tablePath);
+ }
return flink.getTableStatistics(tablePath);
}
@Override
public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath)
throws TableNotExistException, CatalogException {
+ if (isPaimonTable(tablePath)) {
+ return paimon.getTableColumnStatistics(tablePath);
+ }
return flink.getTableColumnStatistics(tablePath);
}
diff --git a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/FlinkGenericCatalogITCase.java b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/FlinkGenericCatalogITCase.java
index 9f319eae2..10fd291fc 100644
--- a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/FlinkGenericCatalogITCase.java
+++ b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/FlinkGenericCatalogITCase.java
@@ -117,4 +117,24 @@ public class FlinkGenericCatalogITCase extends AbstractTestBase {
sql("CREATE TABLE bh (f0 INT, f1 INT) WITH ('connector'='blackhole')");
sql("INSERT INTO bh SELECT * FROM paimon_t");
}
+
+ @Test
+ public void testReadPaimonSystemTable() {
+ sql(
+ "CREATE TABLE paimon_t (\n"
+ + " user_id BIGINT,\n"
+ + " item_id BIGINT,\n"
+ + " behavior STRING,\n"
+ + " dt STRING,\n"
+ + " PRIMARY KEY (dt, user_id) NOT ENFORCED\n"
+ + ") PARTITIONED BY (dt) "
+ + " WITH ('connector'='paimon', 'file.format' = 'avro' )");
+ sql("INSERT INTO paimon_t VALUES (1, 2, 'click', '2023-11-01')");
+ sql("INSERT INTO paimon_t VALUES (2, 3, 'click', '2023-11-02')");
+
+ List<Row> result =
+ sql("SELECT snapshot_id, schema_id, commit_kind FROM paimon_t$snapshots");
+
+ assertThat(result).containsExactly(Row.of(1L, 0L, "APPEND"), Row.of(2L, 0L, "APPEND"));
+ }
}