feat(ingest/dremio): Dremio Source Ingestion

datahub-web-react/src/app/ingest/source/builder/sources.json

@@ -309,6 +309,6 @@
         "displayName": "Dremio",
         "description": "Import Spaces, Sources, Tables and statistics from Dremio.",
         "docsUrl": "https://datahubproject.io/docs/metadata-ingestion/",
-        "recipe": "source:\n    type: dremio\n    config:\n        # Coordinates\n        hostname: null\n        port: null\n        #Credentials with personal access token\n        authentication_method: PAT\n        password: pass\n        #Or Credentials with basic auth\n        #authentication_method: password\n        #username: null\n        #password: null\n        stateful_ingestion:\n            enabled: true"
+        "recipe": "source:\n    type: dremio\n    config:\n        # Coordinates\n        hostname: null\n        port: null\n\n        #Credentials with personal access token\n        authentication_method: PAT\n        password: pass\n        #Or Credentials with basic auth\n        #authentication_method: password\n        #username: null\n        #password: null\n        #tls: true-->https/false-->http\n\n        stateful_ingestion:\n            enabled: true"
     }
 ]

metadata-ingestion/src/datahub/ingestion/source/dremio/dremio_api.py

@@ -223,6 +223,9 @@ def authenticate(self, connection_args: "DremioSourceConfig") -> None:
                 self.report.failure("Failed to authenticate", str(e))
                 sleep(1)  # Optional: exponential backoff
 
+        self.report.failure(
+            "Credentials cannot be refreshed. Please check your username and password."
+        )
         raise DremioAPIException(
             "Credentials cannot be refreshed. Please check your username and password."
         )
@@ -263,9 +266,15 @@ def execute_query(self, query: str, timeout: int = 300) -> List[Dict[str, Any]]:
                     return future.result(timeout=timeout)
                 except concurrent.futures.TimeoutError:
                     self.cancel_query(job_id)
+                    self.report.failure(
+                        f"Query execution timed out after {timeout} seconds"
+                    )
                     raise TimeoutError(
                         f"Query execution timed out after {timeout} seconds"
                     )
+                except RuntimeError as e:
+                    self.report.failure("Query Execution failed", exc=e)
+                    raise DremioAPIException("Query Execution failed: {str(e)}")
 
         except requests.RequestException as e:
             raise DremioAPIException(f"Error executing query: {str(e)}")

metadata-ingestion/src/datahub/ingestion/source/dremio/dremio_config.py

@@ -10,6 +10,9 @@
     PlatformInstanceConfigMixin,
 )
 from datahub.ingestion.source.ge_profiling_config import GEProfilingConfig
+from datahub.ingestion.source.state.stale_entity_removal_handler import (
+    StatefulStaleMetadataRemovalConfig,
+)
 from datahub.ingestion.source.state.stateful_ingestion_base import (
     StatefulIngestionConfigBase,
 )
@@ -152,6 +155,8 @@ class DremioSourceConfig(
         default=BaseUsageConfig(),
     )
 
+    stateful_ingestion: Optional[StatefulStaleMetadataRemovalConfig] = None
+
     # Profiling
     profile_pattern: AllowDenyPattern = Field(
         default=AllowDenyPattern.allow_all(),
@@ -172,13 +177,3 @@ class DremioSourceConfig(
         default=False,
         description="Whether to include query-based lineage information.",
     )
-
-    include_table_rename_lineage: bool = Field(
-        default=True,
-        description="Whether to include table rename lineage",
-    )
-
-    include_copy_lineage: bool = Field(
-        default=True,
-        description="Whether to include copy lineage",
-    )

metadata-ingestion/src/datahub/ingestion/source/dremio/dremio_source.py

@@ -3,13 +3,14 @@
 import re
 from collections import defaultdict
 from concurrent.futures import ThreadPoolExecutor, as_completed
-from typing import Any, Dict, Iterable, List, Optional
+from typing import Dict, Iterable, List, Optional
 
 from datahub.emitter.mce_builder import (
     make_data_platform_urn,
     make_dataset_urn_with_platform_instance,
 )
 from datahub.emitter.mcp import MetadataChangeProposalWrapper
+from datahub.ingestion.api.common import PipelineContext
 from datahub.ingestion.api.decorators import (
     SupportStatus,
     capability,
@@ -105,7 +106,7 @@ class DremioSource(StatefulIngestionSourceBase):
     config: DremioSourceConfig
     report: DremioSourceReport
 
-    def __init__(self, config: DremioSourceConfig, ctx: Any):
+    def __init__(self, config: DremioSourceConfig, ctx: PipelineContext):
         super().__init__(config, ctx)
         self.default_db = "dremio"
         self.config = config
@@ -145,7 +146,7 @@ def __init__(self, config: DremioSourceConfig, ctx: Any):
         )
 
     @classmethod
-    def create(cls, config_dict: Dict[str, Any], ctx: Any) -> "DremioSource":
+    def create(cls, config_dict: Dict, ctx: PipelineContext) -> "DremioSource":
         config = DremioSourceConfig.parse_obj(config_dict)
         return cls(config, ctx)
 
@@ -599,9 +600,3 @@ def get_report(self) -> SourceReport:
         Get the source report.
         """
         return self.report
-
-    def close(self) -> None:
-        """
-        Close any resources held by the source.
-        """
-        pass

metadata-ingestion/tests/integration/dremio/docker-compose.yml

@@ -17,19 +17,40 @@ services:
       - 9001:9001
       - 9000:9000
     command: ["server", "/data", "--console-address", ":9001"]
+
+  #MySql
+  mysqldb:
+    image: mysql:8.0
+    environment:
+      MYSQL_ROOT_PASSWORD: rootpwd123
+      MYSQL_USER: foo
+      MYSQL_PASSWORD: datahub123
+    container_name: test_mysql
+    hostname: test_mysql
+    ports:
+      - "23306:3306"
+    volumes:
+      - ./setup/mysqld.cnf:/etc/mysql/mysql.conf.d/mysqld.cnf
+      - ./setup/setup.sql:/docker-entrypoint-initdb.d/setup.sql
+    networks:
+      dremio-network:
+
   # Dremio
   dremio:
     platform: linux/x86_64
     image: dremio/dremio-oss:latest
+    container_name: dremio
     ports:
       - 9047:9047
       - 31010:31010
       - 32010:32010
-    container_name: dremio
+    environment:
+      - DREMIO_JAVA_SERVER_EXTRA_OPTS=-Dpaths.dist=file:///opt/dremio/data/dist
     networks:
       dremio-network:
     depends_on:
       - minio
+      - mysqldb
 
 networks:
   dremio-network: