From e000678091bc7d95ef4305dc89385ce45c50ef56 Mon Sep 17 00:00:00 2001
From: Dawid Wysakowicz <dwysakowicz@apache.org>
Date: Mon, 20 Jan 2025 15:56:28 +0100
Subject: [PATCH] [FLINK-37098] Fix UnresolvedSchema of
 QueryOperationCatalogView

---
 .../catalog/QueryOperationCatalogView.java    |  8 ++++++-
 .../planner/catalog/JavaCatalogTableTest.java | 23 +++++++++++++++++++
 2 files changed, 30 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 0a820bb3ca9e0..02e31d5112b7f 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 QueryOperation getQueryOperation() {
 
     @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 b535bbc0500af..0ddbde44dfb4a 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 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,28 @@ void testTimeAttributeOfViewSelect() {
         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` (\n"
+                                + "  `customer_id`,\n"
+                                + "  `product_id`\n"
+                                + ")\n"
+                                + "AS SELECT 1, 1\n");
+    }
+
     private static class CustomCatalog extends GenericInMemoryCatalog {
         public CustomCatalog(String name) {
             super(name);