Flink refine jdbc table name

Signed-off-by: Zhenqiu Huang <huangzhenqiu0825@gmail.com>
OpenLineage · Mar 13, 2024 · f0e1c99 · f0e1c99
1 parent 4ba9374
commit f0e1c99
Showing 1 changed file with 4 additions and 2 deletions.
diff --git a/...on/flink/shared/src/main/java/io/openlineage/flink/visitor/wrapper/JdbcSourceWrapper.java b/...on/flink/shared/src/main/java/io/openlineage/flink/visitor/wrapper/JdbcSourceWrapper.java
@@ -39,8 +39,8 @@ public String getConnectionUrl() {
 
   public Optional<String> getTableName() {
     Optional<String> queryOpt = Optional.empty();
+    Optional<JdbcConnectionOptions> connectionOptionsOpt = getConnectionOptions();
     if (source instanceof JdbcRowDataLookupFunction) {
-      Optional<JdbcConnectionOptions> connectionOptionsOpt = getConnectionOptions();
       return connectionOptionsOpt
           .map(
               connectionOptions ->
@@ -56,7 +56,9 @@ public Optional<String> getTableName() {
 
     return queryOpt
         .flatMap(query -> OpenLineageSql.parse(List.of(query)))
-        .map(sqlMeta -> sqlMeta.inTables().isEmpty() ? "" : sqlMeta.inTables().get(0).name());
+        .map(
+            sqlMeta ->
+                sqlMeta.inTables().isEmpty() ? "" : sqlMeta.inTables().get(0).qualifiedName());
   }
 
   private Optional<JdbcConnectionOptions> getConnectionOptions() {