Qualify table name in relation when fetching extra option

Signed-off-by: Chen Dai <[email protected]>
opensearch-project · Oct 19, 2023 · 101ef1b · 101ef1b
1 parent 0cc2f25
commit 101ef1b
Showing 1 changed file with 6 additions and 3 deletions.
diff --git a/...integration/src/main/scala/org/opensearch/flint/spark/mv/FlintSparkMaterializedView.scala b/...integration/src/main/scala/org/opensearch/flint/spark/mv/FlintSparkMaterializedView.scala
@@ -84,7 +84,7 @@ case class FlintSparkMaterializedView(
         aggregate.copy(child = watermark(timeCol, aggregate.child))
 
       case relation: UnresolvedRelation if !relation.isStreaming =>
-        relation.copy(isStreaming = true, options = optionsWithExtra(relation))
+        relation.copy(isStreaming = true, options = optionsWithExtra(spark, relation))
     }
     logicalPlanToDataFrame(spark, streamingPlan)
   }
@@ -98,9 +98,12 @@ case class FlintSparkMaterializedView(
     EventTimeWatermark(timeCol, IntervalUtils.fromIntervalString(delay), child)
   }
 
-  private def optionsWithExtra(relation: UnresolvedRelation): CaseInsensitiveStringMap = {
+  private def optionsWithExtra(
+      spark: SparkSession,
+      relation: UnresolvedRelation): CaseInsensitiveStringMap = {
     val originalOptions = relation.options.asCaseSensitiveMap
-    val extraOptions = options.extraSourceOptions(relation.tableName).asJava
+    val tableName = qualifyTableName(spark, relation.tableName)
+    val extraOptions = options.extraSourceOptions(tableName).asJava
     new CaseInsensitiveStringMap((originalOptions ++ extraOptions).asJava)
   }