Fix 'schemas' table not being properly created when Delta Lake is use…

…d for bookkeeping.
AbsaOSS · Aug 29, 2024 · 63c330a · 63c330a
1 parent 065f214
commit 63c330a
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 2 deletions.
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperDeltaTable.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperDeltaTable.scala
@@ -100,8 +100,8 @@ class BookkeeperDeltaTable(database: Option[String],
   }
 
   private def initSchemasDirectory(): Unit = {
-    if (!spark.catalog.tableExists(schemasTable)) {
-      writeEmptyDataset[TableSchemaJson](schemasTable)
+    if (!spark.catalog.tableExists(schemasFullTableName)) {
+      writeEmptyDataset[TableSchemaJson](schemasFullTableName)
     }
   }
 }
diff --git a/...rc/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperDeltaTableLongSuite.scala b/...rc/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperDeltaTableLongSuite.scala
@@ -43,6 +43,13 @@ class BookkeeperDeltaTableLongSuite extends BookkeeperCommonSuite with SparkTest
       val rndInt = Math.abs(Random.nextInt())
       getBookkeeper(s"tbl${rndInt}_")
     }
+
+    "test tables are created properly" in {
+      getBookkeeper(s"tbl0000_")
+
+      assert(spark.catalog.tableExists("tbl0000_bookkeeping"))
+      assert(spark.catalog.tableExists("tbl0000_schemas"))
+    }
   }
 
   private def cleanUpWarehouse(): Unit = {