diff --git a/ingestion/examples/workflows/oracle.json b/ingestion/examples/workflows/oracle.json
new file mode 100644
index 00000000000..68493a207a0
--- /dev/null
+++ b/ingestion/examples/workflows/oracle.json
@@ -0,0 +1,25 @@
+{
+    "source": {
+      "type": "oracle",
+      "config": {
+        "host_port":"host:1521",
+        "username": "pdbadmin",
+        "password": "password",
+        "service_name": "local_oracle",
+        "service_type": "Oracle",
+        "oracle_service_name": "ORCLPDB1"
+      }
+    },
+    "sink": {
+      "type": "metadata-rest",
+      "config": {
+      }
+    },
+    "metadata_server": {
+      "type": "metadata-server",
+      "config": {
+        "api_endpoint": "http://localhost:8585/api",
+          "auth_provider_type": "no-auth"
+      }
+    }
+  }
diff --git a/ingestion/src/metadata/ingestion/source/oracle.py b/ingestion/src/metadata/ingestion/source/oracle.py
index 7b2ae7af86a..b65df7f4cd3 100644
--- a/ingestion/src/metadata/ingestion/source/oracle.py
+++ b/ingestion/src/metadata/ingestion/source/oracle.py
@@ -10,7 +10,10 @@
 #  limitations under the License.
 
 # This import verifies that the dependencies are available.
+from typing import Optional
+
 import cx_Oracle  # noqa: F401
+import pydantic
 
 from metadata.ingestion.ometa.openmetadata_rest import MetadataServerConfig
 from metadata.ingestion.source.sql_source import SQLConnectionConfig, SQLSource
@@ -19,9 +22,23 @@ from metadata.ingestion.source.sql_source import SQLConnectionConfig, SQLSource
 class OracleConfig(SQLConnectionConfig):
     # defaults
     scheme = "oracle+cx_oracle"
+    oracle_service_name: Optional[str] = None
+    query: Optional[str] = "select * from {}.{} where ROWNUM <= 50"
+
+    @pydantic.validator("oracle_service_name")
+    def check_oracle_service_name(cls, v, values):
+        if values.get("database") and v:
+            raise ValueError(
+                "Please provide database or oracle_service_name but not both"
+            )
+        return v
 
     def get_connection_url(self):
-        return super().get_connection_url()
+        url = super().get_connection_url()
+        if self.oracle_service_name:
+            assert not self.database
+            url = f"{url}/?service_name={self.oracle_service_name}"
+        return url
 
 
 class OracleSource(SQLSource):