(flink) branch release-1.20 updated: [FLINK-37098] Fix UnresolvedSchema of QueryOperationCatalogView

[dwysakowicz]

This is an automated email from the ASF dual-hosted git repository.

dwysakowicz pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.20 by this push:
     new 77c96aadab7 [FLINK-37098] Fix UnresolvedSchema of 
QueryOperationCatalogView
77c96aadab7 is described below

commit 77c96aadab763e31cc304c62931be7a961c6e640
Author: Dawid Wysakowicz <dwysakow...@apache.org>
AuthorDate: Mon Jan 20 15:56:28 2025 +0100

    [FLINK-37098] Fix UnresolvedSchema of QueryOperationCatalogView
---
 .../table/catalog/QueryOperationCatalogView.java     |  8 +++++++-
 .../table/planner/catalog/JavaCatalogTableTest.java  | 20 ++++++++++++++++++++
 2 files changed, 27 insertions(+), 1 deletion(-)

diff --git 
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/QueryOperationCatalogView.java
 
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/QueryOperationCatalogView.java
index 0a820bb3ca9..02e31d5112b 100644
--- 
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/QueryOperationCatalogView.java
+++ 
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/QueryOperationCatalogView.java
@@ -52,7 +52,13 @@ public final class QueryOperationCatalogView implements 
CatalogView {
 
     @Override
     public Schema getUnresolvedSchema() {
-        return 
Schema.newBuilder().fromResolvedSchema(queryOperation.getResolvedSchema()).build();
+        return Optional.ofNullable(originalView)
+                .map(CatalogView::getUnresolvedSchema)
+                .orElseGet(
+                        () ->
+                                Schema.newBuilder()
+                                        
.fromResolvedSchema(queryOperation.getResolvedSchema())
+                                        .build());
     }
 
     @Override
diff --git 
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/JavaCatalogTableTest.java
 
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/JavaCatalogTableTest.java
index 3b17fb8d464..61b2b4b8ec6 100644
--- 
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/JavaCatalogTableTest.java
+++ 
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/JavaCatalogTableTest.java
@@ -53,6 +53,7 @@ import java.util.stream.Collectors;
 
 import static org.apache.flink.table.api.Expressions.$;
 import static org.apache.flink.table.api.Expressions.lit;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * Tests for resolving types of computed columns (including time attributes) 
of tables from catalog.
@@ -202,6 +203,25 @@ class JavaCatalogTableTest extends TableTestBase {
         testUtil.verifyExecPlan("SELECT * FROM `cat`.`default`.v");
     }
 
+    @TestTemplate
+    void testShowCreateViewUsesCorrectColumnNames() {
+        TableTestUtil testUtil = getTestUtil();
+        TableEnvironment tableEnvironment = testUtil.getTableEnv();
+        tableEnvironment.registerCatalog("cat", new CustomCatalog("cat"));
+        tableEnvironment.executeSql(
+                "CREATE VIEW `cat`.`default`.v (`customer_id`, `product_id`) 
AS " + "SELECT 1, 1");
+        String result =
+                tableEnvironment
+                        .executeSql("SHOW CREATE VIEW `cat`.`default`.v")
+                        .collect()
+                        .next()
+                        .getFieldAs(0);
+        assertThat(result)
+                .isEqualTo(
+                        "CREATE VIEW `cat`.`default`.`v`(`customer_id`, 
`product_id`) as\n"
+                                + "SELECT 1, 1");
+    }
+
     private static class CustomCatalog extends GenericInMemoryCatalog {
         public CustomCatalog(String name) {
             super(name);