fix oracle jdbc url

Author: m-abulazm

src/databricks/labs/lakebridge/reconcile/connectors/jdbc_reader.py

@@ -9,7 +9,7 @@ class JDBCReaderMixin:
     _spark: SparkSession
 
     # TODO update the url
-    def _get_jdbc_reader(self, query, jdbc_url, driver, prepare_query=None):
+    def _get_jdbc_reader(self, query, jdbc_url, driver, additional_options: dict | None = None):
         driver_class = {
             "oracle": "oracle.jdbc.driver.OracleDriver",
             "snowflake": "net.snowflake.client.jdbc.SnowflakeDriver",
@@ -21,8 +21,9 @@ def _get_jdbc_reader(self, query, jdbc_url, driver, prepare_query=None):
             .option("driver", driver_class.get(driver, driver))
             .option("dbtable", f"({query}) tmp")
         )
-        if prepare_query is not None:
-            reader = reader.option('prepareQuery', prepare_query)
+        if isinstance(additional_options, dict):
+            for key, value in additional_options.items():
+                reader = reader.option(key, value)
         return reader
 
     @staticmethod

src/databricks/labs/lakebridge/reconcile/connectors/oracle.py

@@ -49,8 +49,7 @@ def __init__(
     @property
     def get_jdbc_url(self) -> str:
         return (
-            f"jdbc:{OracleDataSource._DRIVER}:thin:{self._get_secret('user')}"
-            f"/{self._get_secret('password')}@//{self._get_secret('host')}"
+            f"jdbc:{OracleDataSource._DRIVER}:thin@//{self._get_secret('host')}"
             f":{self._get_secret('port')}/{self._get_secret('database')}"
         )
 
@@ -86,7 +85,7 @@ def get_schema(
         schema_query = re.sub(
             r'\s+',
             ' ',
-            OracleDataSource._SCHEMA_QUERY.format(table=table, owner=schema),
+            OracleDataSource._SCHEMA_QUERY.format(table=table.lower(), owner=schema.lower()),
         )
         try:
             logger.debug(f"Fetching schema using query: \n`{schema_query}`")
@@ -109,7 +108,12 @@ def _get_timestamp_options() -> dict[str, str]:
         }
 
     def reader(self, query: str) -> DataFrameReader:
-        return self._get_jdbc_reader(query, self.get_jdbc_url, OracleDataSource._DRIVER)
+        user = self._get_secret('user')
+        password = self._get_secret('password')
+        logger.debug(f"Using user: {user} to connect to Oracle")
+        return self._get_jdbc_reader(
+            query, self.get_jdbc_url, OracleDataSource._DRIVER, {"user": user, "password": password}
+        )
 
     def normalize_identifier(self, identifier: str) -> NormalizedIdentifier:
         normalized = DialectUtils.normalize_identifier(

src/databricks/labs/lakebridge/reconcile/connectors/tsql.py

@@ -134,7 +134,7 @@ def get_schema(
             return self.log_and_throw_exception(e, "schema", schema_query)
 
     def reader(self, query: str, prepare_query_str="") -> DataFrameReader:
-        return self._get_jdbc_reader(query, self.get_jdbc_url, self._DRIVER, prepare_query_str)
+        return self._get_jdbc_reader(query, self.get_jdbc_url, self._DRIVER, {"prepareQuery": prepare_query_str})
 
     def normalize_identifier(self, identifier: str) -> NormalizedIdentifier:
         return DialectUtils.normalize_identifier(

tests/unit/reconcile/connectors/test_oracle.py

@@ -72,12 +72,14 @@ def test_read_data_with_options():
     spark.read.format.assert_called_with("jdbc")
     spark.read.format().option.assert_called_with(
         "url",
-        "jdbc:oracle:thin:my_user/my_password@//my_host:777/my_database",
+        "jdbc:oracle:thin@//my_host:777/my_database",
     )
     spark.read.format().option().option.assert_called_with("driver", "oracle.jdbc.driver.OracleDriver")
     spark.read.format().option().option().option.assert_called_with("dbtable", "(select 1 from data.employee) tmp")
-    actual_args = spark.read.format().option().option().option().options.call_args.kwargs
-    expected_args = {
+    spark.read.format().option().option().option().option.assert_called_with("user", "my_user")
+    spark.read.format().option().option().option().option().option.assert_called_with("password", "my_password")
+    jdbc_actual_args = spark.read.format().option().option().option().option().option().options.call_args.kwargs
+    jdbc_expected_args = {
         "numPartitions": 50,
         "partitionColumn": "s_nationkey",
         "lowerBound": '0',
@@ -88,8 +90,8 @@ def test_read_data_with_options():
         r"'''YYYY-MM-DD''');dbms_session.set_nls('nls_timestamp_format', '''YYYY-MM-DD "
         r"HH24:MI:SS''');END;",
     }
-    assert actual_args == expected_args
-    spark.read.format().option().option().option().options().load.assert_called_once()
+    assert jdbc_actual_args == jdbc_expected_args
+    spark.read.format().option().option().option().option().option().options().load.assert_called_once()
 
 
 def test_get_schema():
@@ -141,7 +143,7 @@ def test_read_data_exception_handling():
         filters=None,
     )
 
-    spark.read.format().option().option().option().options().load.side_effect = RuntimeError("Test Exception")
+    spark.read.format().option().option().option().option().option().options().load.side_effect = RuntimeError("Test Exception")
 
     # Call the read_data method with the Tables configuration and assert that a PySparkException is raised
     with pytest.raises(
@@ -156,7 +158,7 @@ def test_get_schema_exception_handling():
     engine, spark, ws, scope = initial_setup()
     ords = OracleDataSource(engine, spark, ws, scope)
 
-    spark.read.format().option().option().option().load.side_effect = RuntimeError("Test Exception")
+    spark.read.format().option().option().option().option().option().load.side_effect = RuntimeError("Test Exception")
 
     # Call the get_schema method with predefined table, schema, and catalog names and assert that a PySparkException
     # is raised