feat(ingestion/jdbc-connector): Adding a connector for JDBC connections with customizable platform and sqlglot dialect.

metadata-ingestion/docs/sources/jdbc/jdbc.md

@@ -0,0 +1,57 @@
+### Starter Recipe
+
+```yaml
+source:
+  type: jdbc
+  config:
+    # JDBC Driver Configuration
+    driver:
+      driver_class: org.postgresql.Driver  # Replace with your database's driver class
+      # Either specify maven_coordinates or driver_path
+      maven_coordinates: org.postgresql:postgresql:42.7.1
+      # driver_path: "/path/to/driver.jar"
+
+    # Connection Configuration
+    connection:
+      uri: "jdbc:postgresql://localhost:5432//mydb"  # Replace with your database URI
+      username: "user"
+      password: "pass"
+
+      # Optional SSL Configuration
+      ssl_config:
+        cert_path: "/path/to/cert"
+        # cert_type: "pem"    # pem, jks, or p12
+        # cert_password: ""
+
+      # Additional JDBC Properties
+      properties:
+        applicationName: "datahub_jdbc_ingestion"
+
+    # Additional JVM Arguments
+    jvm_args:
+      - "-Xmx1g"
+      - "-Djavax.net.ssl.trustStore=/etc/ssl/certs/java/cacerts"
+
+    # Optional: SQL dialect for query parsing
+    sqlglot_dialect: "postgres"  # Replace with your database's dialect
+
+    # Optional Filters
+    schema_pattern:
+      allow:
+        - "schema1"
+        - "schema2"
+
+    table_pattern:
+      allow:
+        - "schema1.table1"
+      deny:
+        - "schema1.temp_.*"
+
+    # Feature flags
+    include_tables: true
+    include_views: true
+    include_stored_procedures: false
+
+sink:
+  # sink configs
+```

metadata-ingestion/docs/sources/jdbc/jdbc_pre.md

@@ -0,0 +1,25 @@
+### Setup
+
+This integration pulls metadata from databases via JDBC connections. It supports various database systems through their respective JDBC drivers.
+
+You'll need:
+1. A running database instance that you want to connect to
+2. The appropriate JDBC driver for your database
+3. Valid credentials with permissions to read metadata
+
+#### Steps to Get Started
+
+1. **JDBC Driver Setup**:
+   - Option 1: Download the JDBC driver JAR file for your database
+   - Option 2: Use Maven coordinates to automatically download the driver
+
+2. **Permissions Required**:
+   - READ access to system catalogs/metadata views
+   - Ability to execute metadata queries
+   - Access to relevant schemas and tables
+
+3. **Connection Information**:
+   - JDBC connection URL
+   - Username and password (if using basic authentication)
+   - SSL configuration (if required)
+   - Any additional JDBC properties needed for your specific database

metadata-ingestion/docs/sources/jdbc/jdbc_recipe.yml

@@ -0,0 +1,53 @@
+source:
+  type: jdbc
+  config:
+    # JDBC Driver Configuration
+    driver:
+      driver_class: org.postgresql.Driver  # Replace with your database's driver class
+      # Either specify maven_coordinates or driver_path
+      maven_coordinates: org.postgresql:postgresql:42.7.1
+      # driver_path: "/path/to/driver.jar"
+
+    # Connection Configuration
+    connection:
+      uri: "jdbc:postgresql://localhost:5432//mydb"  # Replace with your database URI
+      username: "user"
+      password: "pass"
+
+      # Optional SSL Configuration
+      ssl_config:
+        cert_path: "/path/to/cert"
+        # cert_type: "pem"    # pem, jks, or p12
+        # cert_password: ""
+
+      # Additional JDBC Properties
+      properties:
+        applicationName: "datahub_jdbc_ingestion"
+
+    # Additional JVM Arguments
+    jvm_args:
+      - "-Xmx1g"
+      - "-Djavax.net.ssl.trustStore=/etc/ssl/certs/java/cacerts"
+
+    # Optional: SQL dialect for query parsing
+    sqlglot_dialect: "postgres"  # Replace with your database's dialect
+
+    # Optional Filters
+    schema_pattern:
+      allow:
+        - "schema1"
+        - "schema2"
+
+    table_pattern:
+      allow:
+        - "schema1.table1"
+      deny:
+        - "schema1.temp_.*"
+
+    # Feature flags
+    include_tables: true
+    include_views: true
+    include_stored_procedures: false
+
+sink:
+  # sink configs

metadata-ingestion/setup.py

@@ -450,6 +450,11 @@
     | pyhive_common
     | {"psycopg2-binary", "pymysql>=1.0.2"},
     "iceberg": iceberg_common,
+
+    "jdbc": usage_common
+    | classification_lib
+    | sqlglot_lib
+    | {"beautifulsoup4", "JayDeBeApi", "JPype1", "requests"},
     "iceberg-catalog": aws_common,
     "json-schema": set(),
     "kafka": kafka_common | kafka_protobuf,
@@ -636,6 +641,7 @@
             "iceberg",
             "iceberg-catalog",
             "mlflow",
+            "jdbc",
             "json-schema",
             "ldap",
             "looker",
@@ -743,6 +749,7 @@
         "hana = datahub.ingestion.source.sql.hana:HanaSource",
         "hive = datahub.ingestion.source.sql.hive:HiveSource",
         "hive-metastore = datahub.ingestion.source.sql.hive_metastore:HiveMetastoreSource",
+        "jdbc = datahub.ingestion.source.jdbc.source:JDBCSource",
         "json-schema = datahub.ingestion.source.schema.json_schema:JsonSchemaSource",
         "kafka = datahub.ingestion.source.kafka.kafka:KafkaSource",
         "kafka-connect = datahub.ingestion.source.kafka_connect.kafka_connect:KafkaConnectSource",

metadata-ingestion/src/datahub/ingestion/source/jdbc/config.py

@@ -0,0 +1,172 @@
+import re
+from typing import Dict, List, Optional
+
+from pydantic import Field, validator
+from sqlglot import dialects
+
+from datahub.configuration.common import AllowDenyPattern, ConfigModel
+from datahub.configuration.source_common import (
+    EnvConfigMixin,
+    PlatformInstanceConfigMixin,
+)
+from datahub.ingestion.source.state.stale_entity_removal_handler import (
+    StatefulStaleMetadataRemovalConfig,
+)
+from datahub.ingestion.source.state.stateful_ingestion_base import (
+    StatefulIngestionConfigBase,
+)
+from datahub.ingestion.source.usage.usage_common import BaseUsageConfig
+
+
+class SSLConfig(ConfigModel):
+    """SSL Certificate configuration"""
+
+    cert_path: Optional[str] = Field(
+        default=None,
+        description="Path to SSL certificate file",
+    )
+    cert_content: Optional[str] = Field(
+        default=None,
+        description="Base64 encoded certificate content",
+    )
+    cert_type: str = Field(
+        default="pem",
+        description="Certificate type (pem, jks, p12)",
+    )
+    cert_password: Optional[str] = Field(
+        default=None,
+        description="Certificate password if required",
+    )
+
+    @validator("cert_type")
+    def validate_cert_type(cls, v: str) -> str:
+        valid_types = ["pem", "jks", "p12"]
+        if v.lower() not in valid_types:
+            raise ValueError(f"cert_type must be one of: {', '.join(valid_types)}")
+        return v.lower()
+
+
+class JDBCConnectionConfig(ConfigModel):
+    """JDBC Connection configuration"""
+
+    uri: str = Field(
+        description="JDBC URI (jdbc:protocol://host:port/database)",
+    )
+    username: Optional[str] = Field(
+        default=None,
+        description="Database username",
+    )
+    password: Optional[str] = Field(
+        default=None,
+        description="Database password",
+    )
+    properties: Dict[str, str] = Field(
+        default_factory=dict,
+        description="Additional JDBC properties",
+    )
+    ssl_config: Optional[SSLConfig] = Field(
+        default=None,
+        description="SSL configuration",
+    )
+
+    @validator("uri")
+    def validate_uri(cls, v: str) -> str:
+        if not v.startswith("jdbc:"):
+            raise ValueError("URI must start with 'jdbc:'")
+        return v
+
+
+class JDBCDriverConfig(ConfigModel):
+    """JDBC Driver configuration"""
+
+    driver_class: str = Field(
+        description="Fully qualified JDBC driver class name",
+    )
+    driver_path: Optional[str] = Field(
+        default=None,
+        description="Path to JDBC driver JAR",
+    )
+    maven_coordinates: Optional[str] = Field(
+        default=None,
+        description="Maven coordinates (groupId:artifactId:version)",
+    )
+
+    @validator("driver_class")
+    def validate_driver_class(cls, v: str) -> str:
+        if not v:
+            raise ValueError("driver_class must be specified")
+        return v
+
+    @validator("maven_coordinates")
+    def validate_maven_coordinates(cls, v: Optional[str]) -> Optional[str]:
+        if v and not re.match(r"^[^:]+:[^:]+:[^:]+$", v):
+            raise ValueError(
+                "maven_coordinates must be in format 'groupId:artifactId:version'"
+            )
+        return v
+
+
+class JDBCSourceConfig(
+    StatefulIngestionConfigBase,
+    EnvConfigMixin,
+    PlatformInstanceConfigMixin,
+):
+    """Configuration for JDBC metadata extraction"""
+
+    driver: JDBCDriverConfig = Field(
+        description="JDBC driver configuration",
+    )
+    connection: JDBCConnectionConfig = Field(
+        description="Database connection configuration",
+    )
+    platform: str = Field(
+        description="Name of platform being ingested.",
+    )
+    include_tables: bool = Field(
+        default=True,
+        description="Include tables in extraction",
+    )
+    include_views: bool = Field(
+        default=True,
+        description="Include views in extraction",
+    )
+    include_stored_procedures: bool = Field(
+        default=False,
+        description="Include stored procedures in extraction",
+    )
+    sqlglot_dialect: Optional[str] = Field(
+        default=None,
+        description="sqlglot dialect to use for SQL transpiling",
+    )
+    jvm_args: List[str] = Field(
+        default=[],
+        description="JVM arguments for JDBC driver",
+    )
+    schema_pattern: AllowDenyPattern = Field(
+        default=AllowDenyPattern.allow_all(),
+        description="Regex patterns for schemas",
+    )
+    table_pattern: AllowDenyPattern = Field(
+        default=AllowDenyPattern.allow_all(),
+        description="Regex patterns for tables",
+    )
+    view_pattern: AllowDenyPattern = Field(
+        default=AllowDenyPattern.allow_all(),
+        description="Regex patterns for views",
+    )
+    usage: BaseUsageConfig = Field(
+        description="Usage statistics configuration",
+        default=BaseUsageConfig(),
+    )
+    stateful_ingestion: Optional[StatefulStaleMetadataRemovalConfig] = None
+
+    @validator("sqlglot_dialect")
+    def validate_dialect(cls, v):
+        if v is None:
+            return v
+        valid_dialects = [d for d in dir(dialects) if not d.startswith("_")]
+        if v not in valid_dialects:
+            raise ValueError(
+                f"Invalid dialect '{v}'. Must be one of: {', '.join(sorted(valid_dialects))}"
+            )
+        return v