[FLINK-37098] Fix UnresolvedSchema of QueryOperationCatalogView

twalthr · Jan 21, 2025 · e000678 · e000678
1 parent 632f3fe
commit e000678
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 1 deletion.
diff --git a/...able-api-java/src/main/java/org/apache/flink/table/catalog/QueryOperationCatalogView.java b/...able-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/...le-planner/src/test/java/org/apache/flink/table/planner/catalog/JavaCatalogTableTest.java b/...le-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);